"""Phase 5 plan 05-05: Reality Anchor evaluator + MODEL_CARD.md updater. Reads ``data/reality_anchor.parquet`` (produced by ``label_anchor.py``), computes per-class precision / recall / F1 / support, writes an idempotent ```` / ```` block into ``MODEL_CARD.md``, and enforces the D-ANCHOR-04 launch gate. Macro-F1 semantics (D-ANCHOR-04): macro over classes with ``n_real >= 3``, **weighted by n_real per class**. Classes with ``n_real < 3`` are reported separately with a small-sample caveat (and excluded from the weighted macro). Exit-code conventions: 0 — launch gate passed (weighted macro F1 ≥ --gate) 2 — precondition (parquet missing OR n < --min-n; default 20 per D-ANCHOR-01) 3 — every class has n < 3 (too sparse for launch; block written for visibility) 4 — launch gate failed (weighted macro F1 < --gate) """ from __future__ import annotations import argparse import sys from pathlib import Path import pyarrow.parquet as pq # Canonical class slug list — MUST match model.features.CLASSES insertion order. CLASSES = [ "auth_8021x_eap_fail", "ap_roam_rekey_fail", "radius_timeout", "captive_portal_expiry", "mac_randomization_reject", "dhcp_lease_churn", "dns_resolver_fail", "driver_power_save_wake", "rf_sticky_client", "isp_upstream_fail", ] _RA_START = "" _RA_END = "" def per_class_stats(preds: list[str], trues: list[str]) -> dict[str, dict]: """Per-class precision / recall / F1 / support over (preds, trues). A class with no true-support and no predicted-support has all metrics 0.0 and n_real=0 (rather than NaN) so the table renders cleanly. """ stats: dict[str, dict] = {c: {"tp": 0, "fp": 0, "fn": 0, "n": 0} for c in CLASSES} for p, t in zip(preds, trues): if t in stats: stats[t]["n"] += 1 if p == t: stats[t]["tp"] += 1 else: stats[t]["fn"] += 1 if p in stats and p != t: stats[p]["fp"] += 1 out: dict[str, dict] = {} for c, s in stats.items(): tp, fp, fn, n = s["tp"], s["fp"], s["fn"], s["n"] prec = tp / (tp + fp) if (tp + fp) > 0 else 0.0 rec = tp / (tp + fn) if (tp + fn) > 0 else 0.0 f1 = 2 * prec * rec / (prec + rec) if (prec + rec) > 0 else 0.0 out[c] = { "precision_real": round(prec, 3), "recall_real": round(rec, 3), "f1_real": round(f1, 3), "n_real": n, } return out def weighted_macro_f1_n3(per_class: dict[str, dict]) -> float | None: """D-ANCHOR-04: macro F1 over classes with n_real >= 3, weighted by n_real. Returns None if no class qualifies (every class has n < 3). """ qualified = {c: s for c, s in per_class.items() if s["n_real"] >= 3} if not qualified: return None total_n = sum(s["n_real"] for s in qualified.values()) return sum(s["f1_real"] * s["n_real"] for s in qualified.values()) / total_n def render_table_md(per_class: dict[str, dict]) -> str: lines = [ "| Class | precision_real | recall_real | f1_real | n_real |", "|-------|----------------|-------------|---------|--------|", ] for c in CLASSES: s = per_class[c] lines.append( f"| {c} | {s['precision_real']} | {s['recall_real']} | " f"{s['f1_real']} | {s['n_real']} |" ) return "\n".join(lines) def update_model_card( model_card: Path, per_class: dict[str, dict], macro: float | None, ) -> None: """Idempotently replace (or append) the REALITY_ANCHOR marker block.""" text = model_card.read_text(encoding="utf-8") table_md = render_table_md(per_class) macro_str = f"{macro:.3f}" if macro is not None else "n/a (every class n<3)" block = ( f"{_RA_START}\n\n" f"### Reality Anchor (real-world owner dogfood) — measured\n\n" f"**Weighted macro F1 (classes with n≥3):** {macro_str}\n\n" f"{table_md}\n\n" f"_Per D-ANCHOR-04 semantics: macro over classes with n_real≥3, " f"weighted by n_real per class. Classes with n_real<3 reported above " f"with small-sample caveat._\n\n" f"{_RA_END}" ) if _RA_START in text and _RA_END in text: head, _, rest = text.partition(_RA_START) _, _, tail = rest.partition(_RA_END) new = head + block + tail else: new = text.rstrip() + "\n\n" + block + "\n" model_card.write_text(new, encoding="utf-8") def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser( description="Evaluate Reality Anchor parquet, update MODEL_CARD.md, " "and enforce the D-ANCHOR-04 launch gate." ) ap.add_argument("--parquet", required=True, type=Path) ap.add_argument("--model-card", required=True, type=Path) ap.add_argument("--gate", type=float, default=0.60, help="D-ANCHOR-04 launch gate threshold (weighted macro F1)") ap.add_argument("--min-n", type=int, default=20, help="D-ANCHOR-01 minimum-viable threshold (n_total)") args = ap.parse_args(argv) if not args.parquet.exists(): print( f"Reality Anchor parquet not found at {args.parquet}; " "run export+label first.", file=sys.stderr, ) print( "Reality Anchor not yet seeded (n<20); ship-blocked per D-ANCHOR-04 " "until dogfood accumulates.", file=sys.stderr, ) return 2 table = pq.read_table(args.parquet) if table.num_rows < args.min_n: print( f"Reality Anchor too sparse: n={table.num_rows} < {args.min_n} " "(D-ANCHOR-01 floor).", file=sys.stderr, ) return 2 preds = table.column("predicted_class").to_pylist() trues = table.column("true_class").to_pylist() per_class = per_class_stats(preds, trues) macro = weighted_macro_f1_n3(per_class) if macro is None: print( "Reality Anchor too sparse for launch — n<3 for every class.", file=sys.stderr, ) # Still update the model card so the table is visible; exit 3. update_model_card(args.model_card, per_class, macro) return 3 update_model_card(args.model_card, per_class, macro) print(f"Reality Anchor weighted macro F1 (n≥3): {macro:.3f}") if macro < args.gate: print( f"D-ANCHOR-04 launch gate FAILED: {macro:.3f} < {args.gate}", file=sys.stderr, ) return 4 print(f"D-ANCHOR-04 launch gate OK: {macro:.3f} >= {args.gate}") return 0 if __name__ == "__main__": sys.exit(main())