Blog

#!/usr/bin/env python3
"""1편의 (크기, 엔트로피, gzip 크기)에 바이트 통계 특성 네 개를 더해 다시 수집한다."""
import gzip, math, random
from collections import Counter
from pathlib import Path

DIRS = ["/usr/bin", "/usr/lib/x86_64-linux-gnu", "/usr/share/doc",
        "/usr/share/man", "/usr/share/icons", "/usr/include"]

def features(data: bytes) -> dict:
    n = len(data)
    counts = Counter(data)
    ent = -sum((c / n) * math.log2(c / n) for c in counts.values())

    # 2-gram 엔트로피: 바이트 하나가 아니라 이웃한 쌍의 다양성 (비용 때문에 앞 64KB만)
    head = data[:65536]
    bigrams = Counter(zip(head, head[1:]))
    m = sum(bigrams.values())
    bient = -sum((c / m) * math.log2(c / m) for c in bigrams.values()) if m else 0.0

    # 같은 바이트가 연달아 나오는 비율 (0으로 채운 영역, 들여쓰기 등)
    runs = sum(1 for a, b in zip(head, head[1:]) if a == b)

    return {
        "size":     n,
        "entropy":  ent,
        "bigram":   bient,
        "ascii":    sum(counts[c] for c in range(32, 127)) / n,
        "uniq":     len(counts) / 256.0,
        "run":      runs / max(1, len(head) - 1),
        "gz_size":  len(gzip.compress(data, 9)),
    }

def collect(limit_per_dir=60, min_size=4096, max_size=2_000_000):
    random.seed(42)                      # 1편과 같은 시드 → 같은 파일 집합
    rows = []
    for d in DIRS:
        files = [p for p in Path(d).rglob("*") if p.is_file() and not p.is_symlink()]
        random.shuffle(files)
        taken = 0
        for p in files:
            if taken >= limit_per_dir:
                break
            try:
                data = p.read_bytes()
            except OSError:
                continue
            if not (min_size <= len(data) <= max_size):
                continue
            row = {"ext": p.suffix.lower() or "(none)"}
            row.update(features(data))
            rows.append(row)
            taken += 1
    return rows

if __name__ == "__main__":
    rows = collect()
    cols = ["ext", "size", "entropy", "bigram", "ascii", "uniq", "run", "gz_size"]
    with open("files2.csv", "w") as fp:
        fp.write(",".join(cols) + "\n")
        for r in rows:
            fp.write(",".join(f"{r[c]:.4f}" if isinstance(r[c], float) else str(r[c])
                              for c in cols) + "\n")
    print(f"수집한 파일: {len(rows)}개 -> files2.csv")

"""files2.csv(특성 6개)를 읽어 학습/평가로 나눈다."""
import csv
import numpy as np

COLS = ["entropy", "bigram", "ascii", "uniq", "run"]

def load(path="files2.csv", seed=0, train_ratio=0.7):
    rows = list(csv.DictReader(open(path)))
    ext  = np.array([r["ext"] for r in rows])
    size = np.array([float(r["size"]) for r in rows])
    F    = {c: np.array([float(r[c]) for r in rows]) for c in COLS}
    F["logsize"] = np.log10(size)
    ratio = np.array([float(r["gz_size"]) for r in rows]) / size

    idx = np.random.default_rng(seed).permutation(len(rows))
    n = int(len(rows) * train_ratio)
    return ext, F, ratio, idx[:n], idx[n:]

#!/usr/bin/env python3
"""분류에 MSE 대신 로그손실을 쓰는 이유를 기울기 크기로 확인한다."""
import numpy as np

def sigmoid(z):
    return 1 / (1 + np.exp(-z))

# 정답이 1인 표본 하나. z가 음수일수록 '확신에 차서 틀린' 상태다.
print(f"{'z':>6}{'예측 p':>10}{'MSE 손실':>11}{'MSE 기울기':>13}"
      f"{'로그손실':>11}{'로그손실 기울기':>16}")
print("-" * 68)
for z in (-8, -4, -2, 0, 2, 4):
    p = sigmoid(z)
    # d/dz 이므로 시그모이드 미분 p(1-p)가 곱해진다
    mse_grad  = 2 * (p - 1) * p * (1 - p)
    ll_grad   = p - 1
    print(f"{z:>6}{p:>10.6f}{(p - 1) ** 2:>11.6f}{mse_grad:>13.2e}"
          f"{-np.log(p):>11.4f}{ll_grad:>16.4f}")

# 손실을 파라미터 하나짜리 문제로 두고 볼록한지 본다
x = np.array([-2.0, -1.0, 1.0, 2.0])
y = np.array([0.0, 0.0, 1.0, 1.0])
w = np.linspace(-6, 6, 2401)
mse = [(np.mean((sigmoid(wi * x) - y) ** 2)) for wi in w]
ll  = [float(-np.mean(y * np.log(sigmoid(wi * x)) +
                      (1 - y) * np.log(1 - sigmoid(wi * x)))) for wi in w]

def curvature_sign_changes(f):
    d2 = np.diff(np.array(f), 2)
    return int((np.sign(d2[:-1]) != np.sign(d2[1:])).sum())

print(f"\nw를 -6~6으로 훑으며 이계차분의 부호가 바뀐 횟수 (0이면 볼록)")
print(f"    MSE  손실 : {curvature_sign_changes(mse)}회")
print(f"    로그손실  : {curvature_sign_changes(ll)}회")

#!/usr/bin/env python3
"""'원본의 30% 미만으로 줄어드는가'를 로지스틱 회귀로 분류한다."""
import numpy as np
from data2 import load

NAMES = ["entropy", "bigram", "ascii", "uniq", "run"]

ext, F, ratio, tr, te = load()
y = (ratio < 0.30).astype(float)                 # 1 = 강하게 압축됨

X = np.column_stack([F[n] for n in NAMES])
mu, sd = X[tr].mean(0), X[tr].std(0)             # 표준화 통계는 학습 데이터에서만
Z = np.column_stack([(X - mu) / sd, np.ones(len(y))])

def sigmoid(z):
    return 1.0 / (1.0 + np.exp(-np.clip(z, -500, 500)))

def logloss(t, rows):
    p = np.clip(sigmoid(Z[rows] @ t), 1e-12, 1 - 1e-12)
    return float(-np.mean(y[rows] * np.log(p) + (1 - y[rows]) * np.log(1 - p)))

theta = np.zeros(Z.shape[1])
lr = 0.5
print(f"{'step':>6}{'학습 로그손실':>14}{'평가 로그손실':>14}{'평가 정확도':>12}")
for step in range(3001):
    if step in (0, 100, 500, 1000, 2000, 3000):
        acc = ((sigmoid(Z[te] @ theta) >= 0.5) == y[te]).mean()
        print(f"{step:>6}{logloss(theta, tr):>14.5f}{logloss(theta, te):>14.5f}{acc:>12.4f}")
    # 로그손실의 기울기는 (1/n) Z^T (sigmoid(Z theta) - y) — 선형회귀와 형태가 같다
    theta -= lr * (Z[tr].T @ (sigmoid(Z[tr] @ theta) - y[tr])) / len(tr)

print("\n" + f"{'특성':<10}{'계수':>10}")
for name, w in zip(NAMES + ["(절편)"], theta):
    print(f"{name:<10}{w:>10.4f}")
np.savez("logistic_out.npz", prob=sigmoid(Z[te] @ theta), y=y[te])

#!/usr/bin/env python3
"""특성을 하나씩 더해 가며 분류 성능이 어떻게 변하는지 본다."""
import numpy as np
from data2 import load

ext, F, ratio, tr, te = load()
y = (ratio < 0.30).astype(float)

def fit(names, steps=3000, lr=0.5):
    X = np.column_stack([F[n] for n in names])
    mu, sd = X[tr].mean(0), X[tr].std(0)
    Z = np.column_stack([(X - mu) / sd, np.ones(len(y))])
    t = np.zeros(Z.shape[1])
    sig = lambda z: 1 / (1 + np.exp(-np.clip(z, -500, 500)))
    for _ in range(steps):
        t -= lr * (Z[tr].T @ (sig(Z[tr] @ t) - y[tr])) / len(tr)
    p = sig(Z[te] @ t)
    ll = -np.mean(y[te] * np.log(np.clip(p, 1e-12, 1)) +
                  (1 - y[te]) * np.log(np.clip(1 - p, 1e-12, 1)))
    return ((p >= 0.5) == y[te]).mean(), float(ll)

major = max(y[te].mean(), 1 - y[te].mean())
print(f"평가 표본 {len(te)}개 중 양성 {int(y[te].sum())}개 → 다수 클래스 기준선 정확도 {major:.4f}\n")
print(f"{'특성 구성':<44}{'정확도':>9}{'로그손실':>11}")
print("-" * 64)
for names in (["entropy"], ["entropy", "logsize"], ["entropy", "bigram"],
              ["entropy", "bigram", "ascii"],
              ["entropy", "bigram", "ascii", "uniq", "run"],
              ["entropy", "bigram", "ascii", "uniq", "run", "logsize"]):
    acc, ll = fit(names)
    print(f"{' + '.join(names):<44}{acc:>9.4f}{ll:>11.5f}")

#!/usr/bin/env python3
"""정확도 하나로는 안 되는 이유: 혼동행렬과 임계값 조정."""
import numpy as np

d = np.load("logistic_out.npz")
prob, y = d["prob"], d["y"]

def counts(thr):
    pred = prob >= thr
    tp = int(((pred == 1) & (y == 1)).sum())
    fp = int(((pred == 1) & (y == 0)).sum())
    fn = int(((pred == 0) & (y == 1)).sum())
    tn = int(((pred == 0) & (y == 0)).sum())
    return tp, fp, fn, tn

tp, fp, fn, tn = counts(0.5)
print("임계값 0.5의 혼동행렬")
print(f"{'':>12}{'예측 양성':>10}{'예측 음성':>10}")
print(f"{'실제 양성':>10}{tp:>12}{fn:>10}")
print(f"{'실제 음성':>10}{fp:>12}{tn:>10}")

print(f"\n{'임계값':>8}{'정밀도':>9}{'재현율':>9}{'F1':>8}{'정확도':>9}  (TP/FP/FN/TN)")
print("-" * 62)
for thr in (0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8):
    tp, fp, fn, tn = counts(thr)
    prec = tp / (tp + fp) if tp + fp else 0.0
    rec  = tp / (tp + fn) if tp + fn else 0.0
    f1   = 2 * prec * rec / (prec + rec) if prec + rec else 0.0
    acc  = (tp + tn) / len(y)
    print(f"{thr:>8.1f}{prec:>9.3f}{rec:>9.3f}{f1:>8.3f}{acc:>9.3f}"
          f"  ({tp}/{fp}/{fn}/{tn})")

#!/usr/bin/env python3
"""클래스가 완전히 갈리는 라벨에서는 로지스틱 회귀 가중치가 멈추지 않는다."""
import numpy as np
from data2 import load

ext, F, ratio, tr, te = load()
e = F["entropy"]
Z = np.column_stack([(e - e[tr].mean()) / e[tr].std(), np.ones(len(e))])

for thr in (0.30, 0.50):
    y = (ratio < thr).astype(float)
    print(f"[라벨: ratio < {thr:.2f}]  양성 entropy 최대 {e[y == 1].max():.3f}"
          f" / 음성 entropy 최소 {e[y == 0].min():.3f}")
    theta = np.zeros(2)
    for step in range(1, 200_001):
        p = 1 / (1 + np.exp(-np.clip(Z[tr] @ theta, -500, 500)))
        theta -= 0.5 * Z[tr].T @ (p - y[tr]) / len(tr)
        if step in (1_000, 10_000, 100_000, 200_000):
            print(f"    {step:>7}스텝  w = {theta[0]:8.3f}")