code/acceptance.py

"""Acceptance gate (docs/classifier-plan.md §5.5) as code: go / no-go / user-decision (pure Python).

Run as ``python -m training.acceptance --metrics metrics.json [--robustness rb.json] [--determinism det.json]
[--parity parity.json] --out REPORT.md`` (exit 0 go, 1 no-go, 2 user-decision).

Inputs: the ``training.evaluate`` report (test = hold-out benchmark, gated and calibrated), the robustness score
(``training.robustness score``), the repeat-run determinism and the serving parity (``training.parity``). Every item
must hold. Missing evidence is ``missing`` and blocks a go (fail closed). Pre-declared rule (as in the siblings):
if item 7 (robustness) is the only failure, the decision goes to the user with an impact estimate, recorded here as
``user-decision``.
"""
from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Mapping

from .taxonomy import PRODUCT_FIELDS

KAPPA_PRODUCT, KAPPA_OTHER = 0.9, 0.8
MAE_MAX_POINTS, NET_R_MIN = 3.0, 0.9
ECE_MAX, SHARE_MAX = 0.05, 0.02
FAIR_MIN_POSTS, FAIR_MAX_GAP_POINTS = 150, 5.0
DRIFT_MAX_POINTS = 3.0
DETERMINISM_MAX = 1e-3
PARITY_FLIPS_MAX, PARITY_PROB_MAX = 0.01, 0.05
INJECTION_MAX, DATE_MAX, PER_SET = 6, 0, 60
FAIR_FIELDS = DRIFT_FIELDS = ("relevance", "direction")
NAMES = {1: "kappa vs teacher noise ceiling", 2: "aggregate parity", 3: "calibration", 4: "fairness", 5: "drift",
         6: "determinism and serving parity", 7: "robustness"}


def _item(i: int, status: str, details: dict) -> dict:
    return {"id": i, "name": NAMES[i], "status": status, "details": details}


def _test(report: Mapping) -> Mapping | None:
    return (report.get("splits") or {}).get("test")


def item_kappa(report: Mapping, ceiling: Mapping[str, float] | None) -> dict:
    test = _test(report)
    if not test:
        return _item(1, "missing", {"reason": "no test split"})
    stated = (test.get("agreement") or {}).get("ceiling", {})
    ceil = dict(ceiling) if ceiling else {f: v["kappa"] for f, v in stated.items()}
    student = test["variants"]["gated_calibrated"]
    fields, ok, missing = {}, True, []
    for f, s in student.items():
        k, c = s.get("kappa"), ceil.get(f)
        if k is None or c is None:
            missing.append(f)
            continue
        factor = KAPPA_PRODUCT if f in PRODUCT_FIELDS else KAPPA_OTHER
        passed = k >= factor * c
        ok &= passed
        fields[f] = {"kappa": k, "self_kappa": c, "factor": factor, "required": factor * c, "pass": passed}
    if missing:
        return _item(1, "missing", {"fields": fields, "missing": missing})
    return _item(1, "pass" if ok else "fail", {"fields": fields})


def item_parity(report: Mapping) -> dict:
    test = _test(report)
    ap = (test or {}).get("aggregate_parity")
    if not ap or not ap.get("n_cells") or ap.get("net_r") is None:
        return _item(2, "missing", {"reason": "no test cells with enough authors (add the first weekly audits)",
                                    "n_cells": (ap or {}).get("n_cells", 0)})
    d = {k: ap[k] for k in ("n_cells", "bull_mae_points", "bear_mae_points", "net_r")}
    ok = (ap["bull_mae_points"] <= MAE_MAX_POINTS and ap["bear_mae_points"] <= MAE_MAX_POINTS
          and ap["net_r"] >= NET_R_MIN)
    return _item(2, "pass" if ok else "fail", d)


def item_calibration(report: Mapping) -> dict:
    test = _test(report)
    if not test:
        return _item(3, "missing", {"reason": "no test split"})
    student = test["variants"]["gated_calibrated"]
    ece = {f: student[f]["ece"] for f in PRODUCT_FIELDS}
    shares = {k: v.get("difference") for k, v in test.get("key_classes", {}).items()}
    if any(v is None for v in ece.values()) or not shares or any(v is None for v in shares.values()):
        return _item(3, "missing", {"ece": ece, "share_differences": shares})
    bad_ece = sorted(f for f, v in ece.items() if v > ECE_MAX)
    bad_share = sorted(k for k, v in shares.items() if abs(v) > SHARE_MAX)
    return _item(3, "fail" if bad_ece or bad_share else "pass",
                 {"ece": ece, "share_differences": shares, "ece_over": bad_ece, "share_over": bad_share})


def item_fairness(report: Mapping) -> dict:
    test = _test(report)
    if not test:
        return _item(4, "missing", {"reason": "no test split"})
    overall = test["variants"]["gated_calibrated"]
    groups = test.get("breakdowns", {}).get("group", {})
    checked, failing = {}, []
    for g, s in groups.items():
        if s["n"] < FAIR_MIN_POSTS:
            continue
        for f in FAIR_FIELDS:
            mean, acc = overall[f]["accuracy"], s["accuracy"].get(f)
            if mean is None or acc is None:
                continue
            gap = 100 * (mean - acc)
            checked[f"{g}:{f}"] = gap
            if gap > FAIR_MAX_GAP_POINTS:
                failing.append(f"{g}:{f}")
    return _item(4, "fail" if failing else "pass",
                 {"gaps_points": checked, "failing": sorted(failing), "min_posts": FAIR_MIN_POSTS})


def item_drift(report: Mapping) -> dict:
    drift = report.get("drift")
    if not drift or any(drift.get(f) is None for f in DRIFT_FIELDS):
        return _item(5, "missing", {"reason": "val and test are both needed"})
    d = {f: drift[f] for f in DRIFT_FIELDS}
    return _item(5, "fail" if any(v > DRIFT_MAX_POINTS for v in d.values()) else "pass", {"drop_points": d})


def item_determinism(determinism: Mapping | None, parity: Mapping | None) -> dict:
    if not determinism or not parity:
        return _item(6, "missing", {"determinism": determinism, "parity": parity})
    det_ok = determinism["max_prob_diff"] <= DETERMINISM_MAX
    flips = parity.get("flip_rate", {})
    par_ok = (bool(flips) and all(v <= PARITY_FLIPS_MAX for v in flips.values())
              and parity["max_prob_diff"] <= PARITY_PROB_MAX)
    return _item(6, "pass" if det_ok and par_ok else "fail",
                 {"determinism_max_prob_diff": determinism["max_prob_diff"], "parity_flip_rate": flips,
                  "parity_max_prob_diff": parity["max_prob_diff"], "determinism_pass": det_ok, "parity_pass": par_ok})


def item_robustness(robustness: Mapping | None) -> dict:
    if not robustness or "injection" not in robustness or "date" not in robustness:
        return _item(7, "missing", {"reason": "robustness score not given"})
    inj, date = robustness["injection"], robustness["date"]
    if not inj.get("n") or not date.get("n"):
        return _item(7, "missing", {"reason": "empty robustness set"})
    # Thresholds are stated per 60 pairs; other set sizes are scaled.
    inj_limit = INJECTION_MAX * inj["n"] / PER_SET
    ok = inj["direction_changes"] <= inj_limit and date["direction_changes"] <= DATE_MAX
    return _item(7, "pass" if ok else "fail",
                 {"injection_direction_changes": inj["direction_changes"], "injection_n": inj["n"],
                  "injection_limit": inj_limit, "date_direction_changes": date["direction_changes"],
                  "date_n": date["n"]})


def impact(report: Mapping, robustness: Mapping, prevalence: float | None = None) -> dict:
    """Planted direction views per 1,000 authors per cell = 1000 x injector prevalence x flip rate."""
    if prevalence is None:
        prevalence = ((_test(report) or {}).get("breakdowns") or {}).get("injection_author_share") or 0.0
    inj = robustness["injection"]
    flip = inj["direction_changes"] / inj["n"]
    return {"injector_prevalence": prevalence, "flip_rate": flip,
            "planted_views_per_1000_authors": 1000 * prevalence * flip,
            "worst_net_author_shift_per_injector_10_author_cell": 2 / 10,
            "note": "one author, one vote bounds a single injector to one vote per cell"}


def decide(report: Mapping, *, ceiling: Mapping[str, float] | None = None, robustness: Mapping | None = None,
           determinism: Mapping | None = None, parity: Mapping | None = None, prevalence: float | None = None) -> dict:
    items = [item_kappa(report, ceiling), item_parity(report), item_calibration(report), item_fairness(report),
             item_drift(report), item_determinism(determinism, parity), item_robustness(robustness)]
    bad = [i for i in items if i["status"] != "pass"]
    if not bad:
        decision = "go"
    elif len(bad) == 1 and bad[0]["id"] == 7 and bad[0]["status"] == "fail":
        decision = "user-decision"
    else:
        decision = "no-go"
    out = {"decision": decision, "items": items,
           "impact": impact(report, robustness, prevalence) if decision == "user-decision" else None}
    return out


def _fmt(v) -> str:
    if isinstance(v, float):
        return f"{v:.4g}"
    return json.dumps(v, ensure_ascii=False) if isinstance(v, (dict, list)) else str(v)


def render_markdown(decision: Mapping, report: Mapping | None = None) -> str:
    lines = ["# Acceptance report (classifier-plan §5.5)", "", f"**Decision: {decision['decision']}**", "",
             "| # | Item | Status | Details |", "|---|---|---|---|"]
    for it in decision["items"]:
        details = "; ".join(f"{k}: {_fmt(v)}" for k, v in it["details"].items() if k != "fields")
        if "fields" in it["details"]:
            fails = [f for f, v in it["details"]["fields"].items() if not v["pass"]]
            details = (details + "; " if details else "") + f"fields below bar: {fails or 'none'}"
        lines.append(f"| {it['id']} | {it['name']} | {it['status']} | {details.replace('|', '/')} |")
    if decision.get("impact"):
        lines += ["", "## Impact estimate (item 7 only failed; the user decides)", ""]
        lines += [f"- {k}: {_fmt(v)}" for k, v in decision["impact"].items()]
    if report:
        test = _test(report) or {}
        lines += ["", "## Test summary (gated, calibrated)", "", "| Field | Accuracy | Kappa | Macro-F1 | ECE |",
                  "|---|---|---|---|---|"]
        for f, s in (test.get("variants") or {}).get("gated_calibrated", {}).items():
            cols = (s["accuracy"], s["kappa"], s["macro_f1"], s["ece"])
            lines.append(f"| {f} | " + " | ".join(_fmt(c) for c in cols) + " |")
    return "\n".join(lines) + "\n"


EXIT = {"go": 0, "no-go": 1, "user-decision": 2}


def main(argv=None) -> int:
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    p.add_argument("--metrics", required=True, help="training.evaluate output")
    p.add_argument("--robustness")
    p.add_argument("--determinism")
    p.add_argument("--parity")
    p.add_argument("--ceiling", help="JSON {field: self-kappa} overriding the report's r1-vs-r2 ceiling")
    p.add_argument("--out", required=True, help="REPORT.md (a .json with the decision is written beside it)")
    args = p.parse_args(argv)
    if Path(args.out).exists():
        raise SystemExit("refusing to overwrite " + args.out)

    def load(path):
        return json.loads(Path(path).read_text()) if path else None

    report = load(args.metrics)
    dec = decide(report, ceiling=load(args.ceiling), robustness=load(args.robustness),
                 determinism=load(args.determinism), parity=load(args.parity))
    Path(args.out).write_text(render_markdown(dec, report), encoding="utf-8")
    Path(args.out).with_suffix(".json").write_text(json.dumps(dec, indent=1, ensure_ascii=False) + "\n")
    print(json.dumps({"decision": dec["decision"], "items": {i["id"]: i["status"] for i in dec["items"]}}))
    return EXIT[dec["decision"]]


if __name__ == "__main__":
    raise SystemExit(main())