scripts/eval_reality_anchor.py

"""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
``<!-- REALITY_ANCHOR_START -->`` / ``<!-- REALITY_ANCHOR_END -->`` 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 = "<!-- REALITY_ANCHOR_START -->"
_RA_END = "<!-- REALITY_ANCHOR_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())