pipeline_miami/verify/run_all.py

"""Run the closure-track harness end to end and write the full-QA verdict
(docs/qa_reports/miami/__enrollment_full_qa.{json,md}) that gates the
enrollment analysis page.

Run:  python -m pipeline_miami.verify.run_all [--summary-only]
"""
from __future__ import annotations

import datetime as dt
import json
import subprocess
import sys

from pipeline_miami.verify import common

STEPS = [
    ("A_base_case", [sys.executable, "-m", "pipeline_miami.verify.reconcile", "--all"]),
    ("B_completeness", [sys.executable, "-m", "pipeline_miami.verify.completeness"]),
    ("C_cross_publisher", [sys.executable, "-m", "pipeline_miami.verify.cross_publisher_enrollment"]),
    ("D_computed", [sys.executable, "-m", "pipeline_miami.verify.reconcile_computed"]),
    ("E_invariants", [sys.executable, "-m", "pipeline_miami.verify.served_invariants"]),
    # Outcomes track (reported beside the closure verdict, does not gate it)
    ("O_cross_publisher_absenteeism", [sys.executable, "-m", "pipeline_miami.verify.cross_publisher_absenteeism"]),
    ("O_cross_publisher_race", [sys.executable, "-m", "pipeline_miami.verify.cross_publisher_race"]),
    ("O_residual_checks", [sys.executable, "-m", "pipeline_miami.verify.residual_checks"]),
]


def main() -> None:
    common.ensure_qa_dir()
    # --summary-only: rebuild the verdict files from the per-component reports already on disk
    # (after re-running single components with `reconcile --metric KEY`).
    if "--summary-only" not in sys.argv:
        for name, cmd in STEPS:
            print(f"== {name} ==")
            subprocess.run(cmd, check=True, cwd=common.REPO_ROOT)
    q = common.QA_REPORTS_DIR
    import yaml

    comps_all = yaml.safe_load(open(common.COMPONENTS_YAML))
    comps = [c for c in comps_all if c.get("track", "closure") == "closure"]
    outcomes = [c for c in comps_all if c.get("track") == "outcomes"]
    base = {c["metric_key"]: json.load(open(q / f"{c['metric_key']}__validation.json")) for c in comps}
    # Outcome metrics are validated separately; their summary is written beside the closure-track verdict.
    ob = {c["metric_key"]: json.load(open(q / f"{c['metric_key']}__validation.json")) for c in outcomes}
    xabs = json.load(open(q / "__cross_publisher_absenteeism.json"))["summary"]
    xrace = json.load(open(q / "__cross_publisher_race.json"))
    rz = json.load(open(q / "__residual_checks.json"))
    inv = json.load(open(q / "__served_invariants.json"))
    (q / "__outcomes_validation.json").write_text(json.dumps({
        "base_case": {k: {"status": v["status"], "years": v["years"], "n_common": v["n_common"], "matches": v["matches"], "n_mismatches": v["n_mismatches"], "suppressed": v["suppressed"]} for k, v in ob.items()},
        "cross_publisher_absenteeism": xabs,
        "cross_publisher_race": {"verdict": xrace["summary"]["verdict"], "by_metric": {k: {kk: vv for kk, vv in v.items() if kk != "largest"} for k, v in xrace["by_metric"].items()}},
        "residual_checks": {"verdict": rz["verdict"], "R1_max_abs_dz": rz["R1_independent_refit"]["max_abs_dz"], "R1_max_abs_dn": rz["R1_independent_refit"].get("max_abs_dn"), "R1_fits": rz["R1_independent_refit"]["fits"], "R2": {k: v for k, v in rz["R2_served_equals_committed"].items()}, "R3": rz["R3_sanity"]},
        "served_invariants": {"verdict": inv["verdict"], "hard": inv["hard"], "shift_alarms": {"n": inv["shift_alarms"]["n"], "unexplained": inv["shift_alarms"]["unexplained"]}},
        "overall": "PASS" if all(v["status"] == "PASS" for v in ob.values()) and xabs["verdict"] == "PASS" and xrace["summary"]["verdict"] == "PASS" and rz["verdict"] == "PASS" and inv["verdict"] == "PASS" else "REVIEW",
    }, indent=2, default=str))
    (q / "__outcomes_validation.md").write_text(
        "# Validation — outcome metrics\n\n## (ii) Base case\n\n| metric | years | common | matches | mismatches | suppressed | status |\n|---|---|---:|---:|---:|---:|---|\n"
        + "\n".join(f"| `{k}` | {', '.join(v['years'])} | {v['n_common']} | {v['matches']} | {v['n_mismatches']} | {v['suppressed']} | {v['status']} |" for k, v in ob.items())
        + f"\n\n## (iii-b) Cross-publisher — FLDOE absent-10%+ vs EDFacts DG814, {xabs['year']}\n\nVerdict: **{xabs['verdict']}** — no-PK schools within ±1: {xabs['no_pk']['within_1']}/{xabs['no_pk']['n']} ({xabs['no_pk']['within_1_share'] * 100:.1f}%); PK schools' gap = {xabs['with_pk']['gap_on_pk']['slope']:.3f} × PK membership (r = {xabs['with_pk']['gap_on_pk']['r']:.2f}). Detail: `__cross_publisher_absenteeism.md`.\n"
        + f"\n## (iii-b) Cross-publisher — FLDOE race shares vs NCES CCD\n\nVerdict: **{xrace['summary']['verdict']}** — " + "; ".join(f"{k}: {v['within_2pp_share'] * 100:.1f}% within 2 pp (median Δ {v['median_delta']:+.2f})" for k, v in xrace['by_metric'].items()) + ". Detail: `__cross_publisher_race.md`.\n"
        + f"\n## Residual-z checks\n\nVerdict: **{rz['verdict']}** — independent N, eligibility and refit: max |Δz| {rz['R1_independent_refit']['max_abs_dz']:.1e} over {rz['R1_independent_refit']['fits']} fits; served = committed: {rz['R2_served_equals_committed']['matched']} cells, {rz['R2_served_equals_committed']['csv_only']} CSV-only, {rz['R2_served_equals_committed']['served_only']} served-only. Detail: `__residual_checks.md`.\n"
        + f"\n## (iv) Served-value invariants\n\nVerdict: **{inv['verdict']}** — range violations {inv['hard']['I1_range_violations']}, identity violations {inv['hard']['I2_identity_violations']}, stale rows {sum(r['rows'] for r in inv['hard']['I3_stale_rows'])}, shift alarms {inv['shift_alarms']['n']} ({inv['shift_alarms']['unexplained']} unexplained). Detail: `__served_invariants.md`.\n")
    comp = json.load(open(q / "__completeness.json"))
    cross = json.load(open(q / "__cross_publisher_enrollment.json"))
    computed = json.load(open(q / "__computed_value.json"))
    a_ok = all(b["status"] == "PASS" for b in base.values())
    b_ok = all(r["source_only_in_directory"] == 0 and r["served_only_cells"] == 0 for r in comp["rows"])
    c_ok = cross["summary"]["verdict"] == "PASS"
    d_ok = all(r["mismatches"] == 0 for r in computed)
    e_ok = inv["verdict"] != "FAIL"   # hard rules (range, identity, ownership) gate; shift alarms are listed for review
    verdict = {
        "generated_at": dt.datetime.now(dt.timezone.utc).isoformat(),
        "verdict": {"overall": "PASS" if (a_ok and b_ok and c_ok and d_ok and e_ok) else "FAIL", "A_base_case": a_ok, "B_completeness": b_ok, "C_cross_publisher": c_ok, "D_computed": d_ok, "E_invariants": e_ok},
        "sections": {
            "A_base_case": {k: {"status": v["status"], "years": v["years"], "n_common": v["n_common"], "matches": v["matches"], "n_mismatches": v["n_mismatches"], "suppressed": v["suppressed"], "served_only": v["served_only_count"], "source_only": v["source_only_count"]} for k, v in base.items()},
            "B_completeness": comp,
            "C_cross_publisher": cross["summary"],
            "D_computed": [{k: v for k, v in r.items() if k != "examples"} for r in computed],
            "E_invariants": {"verdict": inv["verdict"], "hard": inv["hard"], "shift_alarms": {"n": inv["shift_alarms"]["n"], "unexplained": inv["shift_alarms"]["unexplained"]}},
        },
    }
    (q / "__enrollment_full_qa.json").write_text(json.dumps(verdict, indent=2, default=str))
    out = [f"# QA — Miami-Dade enrollment and capacity: served cells vs sources\n", f"Generated: {verdict['generated_at']}\n", f"## Verdict: **{verdict['verdict']['overall']}**\n",
           "| Section | Result |", "|---|---|",
           f"| A. Base case (served DB vs re-read of each ingested file, {len(base)} components) | {'PASS' if a_ok else 'FAIL'} |",
           f"| B. Completeness (every source cell served; every served cell in source) | {'PASS' if b_ok else 'FAIL'} |",
           f"| C. Cross-publisher (FLDOE membership vs NCES CCD) | {'PASS' if c_ok else 'FAIL'} |",
           f"| D. Computed values (derived metrics recomputed) | {'PASS' if d_ok else 'FAIL'} |",
           f"| E. Served-value invariants (ranges, identities, ownership gate; shift alarms reported) | {inv['verdict']} ({inv['shift_alarms']['unexplained']} unexplained shift alarms) |", "", "## A. Base case\n", "| metric | years | common | matches | mismatches | suppressed |", "|---|---|---:|---:|---:|---:|"]
    out += [f"| `{k}` | {', '.join(v['years'])} | {v['n_common']} | {v['matches']} | {v['n_mismatches']} | {v['suppressed']} |" for k, v in base.items()]
    (q / "__enrollment_full_qa.md").write_text("\n".join(out))
    print(json.dumps(verdict["verdict"], indent=1))


if __name__ == "__main__":
    main()