| """Stress-test the risk score for overfitting and leakage (same fit as train_scorer.ts, reimplemented with numpy). |
| |
| Usage (repo root): python training/validate_scorer.py |
| """ |
| import json |
| import numpy as np |
|
|
| rows = [json.loads(l) for l in open("training/data/features.jsonl")] |
| X = np.array([[e["sales"], e["pressure"], e["agent"], e["selfAssert"], e["catScam"] + e["catMkt"], e["catLegit"], e["specific"], e["official"], e["protected"]] |
| for e in (r["evidence"] for r in rows)]) |
| y = np.array([1 if r["label"] == "junk" else 0 for r in rows]) |
| src = np.array([r["source"] for r in rows]) |
| cat = np.array([r["category"] for r in rows]) |
| verdict = X[:, 4] |
|
|
|
|
| def fit(Xt, yt, iters=4000, lr=0.5, l2=0.01, legit_w=3.0): |
| s = np.where(yt == 1, 1.0, legit_w) |
| w, b = np.zeros(Xt.shape[1]), 0.0 |
| for _ in range(iters): |
| p = 1 / (1 + np.exp(-(Xt @ w + b))) |
| g = s * (p - yt) |
| b -= lr * g.sum() / s.sum() |
| w -= lr * (Xt.T @ g / s.sum() + l2 * w) |
| return w, b |
|
|
|
|
| def predict(m, Xs): |
| return 1 / (1 + np.exp(-(Xs @ m[0] + m[1]))) |
|
|
|
|
| def auc(score, labels): |
| pos, neg = score[labels == 1], score[labels == 0] |
| return float(((pos[:, None] > neg[None, :]).sum() + 0.5 * (pos[:, None] == neg[None, :]).sum()) / (len(pos) * len(neg))) |
|
|
|
|
| def oof(idx, seed, k=5): |
| rng = np.random.default_rng(seed) |
| folds = rng.permutation(len(idx)) % k |
| out = np.zeros(len(idx)) |
| for f in range(k): |
| tr, te = idx[folds != f], idx[folds == f] |
| out[folds == f] = predict(fit(X[tr], y[tr]), X[te]) |
| return out |
|
|
|
|
| def summary(score, labels): |
| maxl = score[labels == 0].max() |
| return f"AUC {auc(score, labels):.4f} | max legit {maxl:.2f} | junk above max legit {(score[labels == 1] > maxl).sum()}/{(labels == 1).sum()}" |
|
|
|
|
| allidx = np.arange(len(rows)) |
| print("A. Memorization check: score on training data vs. on unseen data") |
| print(" train (fit on all, score all):", summary(predict(fit(X, y), X), y)) |
| print(" held-out (5-fold): ", summary(oof(allidx, 0), y)) |
|
|
| print("B. Stability: 10 different random 5-fold splits") |
| aucs = [auc(oof(allidx, s), y) for s in range(10)] |
| print(f" AUC mean {np.mean(aucs):.4f}, min {min(aucs):.4f}, max {max(aucs):.4f}") |
|
|
| print("C. Leakage check: shuffle the labels; a sound pipeline must fall to ~0.50") |
| rng = np.random.default_rng(1) |
| perm = [] |
| for i in range(20): |
| ys = rng.permutation(y) |
| folds = rng.permutation(len(y)) % 5 |
| sc = np.zeros(len(y)) |
| for f in range(5): |
| tr, te = folds != f, folds == f |
| sc[te] = predict(fit(X[tr], ys[tr], iters=1500), X[te]) |
| perm.append(auc(sc, ys)) |
| print(f" shuffled-label AUC mean {np.mean(perm):.3f} (range {min(perm):.3f}–{max(perm):.3f}) over 20 shuffles") |
|
|
| print("D. Same-author test: synthetic only (real-style and junk both written by the same LLM)") |
| syn = np.where(src == "synthetic-deepseek")[0] |
| print(" learned (5-fold): ", summary(oof(syn, 0), y[syn])) |
| print(" Jev verdict alone:", summary(verdict[syn], y[syn])) |
|
|
| print("E. Transfer test: train WITHOUT any real FTC robocalls, then score the 300 FTC robocalls") |
| noftc = np.where(src != "ncsu-ftc")[0] |
| m = fit(X[noftc], y[noftc]) |
| ftc = np.where(src == "ncsu-ftc")[0] |
| legit_oof = oof(noftc, 0)[y[noftc] == 0] |
| ftc_scores = predict(m, X[ftc]) |
| print(f" FTC robocalls scored above every real-style alert: {(ftc_scores > legit_oof.max()).sum()}/300 (max legit {legit_oof.max():.2f}); median FTC score {np.median(ftc_scores):.2f}") |
| print(f" Jev verdict alone on the same: {(verdict[ftc] > verdict[y == 0].max()).sum()}/300 above max legit {verdict[y == 0].max():.2f}") |
|
|
| print("F. New-tactic test: hold out one entire junk category, train on the rest, score it") |
| legit_idx = np.where(y == 0)[0] |
| for c in ["bank_impersonation", "gov_impersonation", "retail_impersonation", "injected_scam", "sales_pitch", "ftc_robocall"]: |
| held = np.where(cat == c)[0] |
| tr = np.setdiff1d(allidx, held) |
| sc_all = oof(tr, 0) |
| maxl = sc_all[y[tr] == 0].max() |
| s = predict(fit(X[tr], y[tr]), X[held]) |
| print(f" {c:22} n={len(held):3} above max legit {(s > maxl).sum():3}/{len(held)} median {np.median(s):.2f}") |
|
|
| print("G. Uncertainty: bootstrap 95% interval for the AUC gain over Jev's verdict") |
| sc = oof(allidx, 0) |
| rng = np.random.default_rng(2) |
| gains = [] |
| for _ in range(1000): |
| i = rng.integers(0, len(y), len(y)) |
| if y[i].min() == y[i].max(): |
| continue |
| gains.append(auc(sc[i], y[i]) - auc(verdict[i], y[i])) |
| print(f" AUC gain {np.mean(gains):+.4f}, 95% interval [{np.percentile(gains, 2.5):+.4f}, {np.percentile(gains, 97.5):+.4f}]") |
|
|
