"""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] # legit scored out-of-fold, by models that also never saw FTC 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}]")