| """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"}) |
| |
| 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()) |
|
|
