pipeline_miami/verify/cross_publisher_enrollment.py

"""(iii-b) Cross-publisher check: FLDOE October membership (enrollment_oct) vs
the federal copy in NCES CCD (enrollment_oct_ccd), school by school, for the
years both are loaded. Both derive from the same Survey 2 submission, so any
difference is a finding, not noise.

Run:  python -m pipeline_miami.verify.cross_publisher_enrollment
"""
from __future__ import annotations

import json
import statistics

from pipeline_miami.verify import common
from pipeline_miami.verify.connectors import served_db


def main() -> None:
    common.ensure_qa_dir()
    fl = served_db.fetch("enrollment_oct")
    ccd = served_db.fetch("enrollment_oct_ccd")
    years = sorted({k[1] for k in ccd})
    per_year = {}
    all_pairs = []
    for y in years:
        pairs = []
        for k, c in ccd.items():
            if k[1] != y:
                continue
            f = fl.get(k)
            if not f or f["value"] is None or c["value"] is None:
                continue
            pairs.append((k[0], f["value"], c["value"]))
        deltas = [f - c for _, f, c in pairs]
        exact = sum(1 for d in deltas if d == 0)
        within2pct = sum(1 for _, f, c in pairs if c and abs(f - c) / c <= 0.02)
        per_year[y] = {
            "n_pairs": len(pairs), "exact": exact, "exact_share": exact / len(pairs) if pairs else None,
            "within_2pct": within2pct, "median_abs_delta": statistics.median(abs(d) for d in deltas) if deltas else None,
            "sum_fldoe": sum(f for _, f, _ in pairs), "sum_ccd": sum(c for _, _, c in pairs),
            "largest_deltas": sorted(pairs, key=lambda p: -abs(p[1] - p[2]))[:10],
            "ccd_only": sorted(k[0] for k in ccd if k[1] == y and k not in fl)[:20],
            "fldoe_only_count": sum(1 for k in fl if k[1] == y and k not in ccd),
        }
        all_pairs += pairs
    n = len(all_pairs)
    exact = sum(1 for _, f, c in all_pairs if f == c)
    verdict = "PASS" if n and exact / n >= 0.95 else "REVIEW"
    summary = {"summary": {"n_pairs": n, "exact_share": exact / n if n else None, "verdict": verdict,
               "verdict_note": "FLDOE and CCD both publish the Survey 2 October count; exact agreement expected for ≥95% of schools. Remaining differences are late corrections to the state file or CCD's adult-education exclusion."},
               "by_year": per_year}
    (common.QA_REPORTS_DIR / "__cross_publisher_enrollment.json").write_text(json.dumps(summary, indent=2, default=str))
    out = ["# Cross-publisher (iii-b) — FLDOE membership vs NCES CCD\n", f"Verdict: **{verdict}** — {n} school-years compared, {exact} exact ({exact / n * 100 if n else 0:.1f}%).\n",
           "| year | pairs | exact | within 2% | median |Δ| | Σ FLDOE | Σ CCD |", "|---|---:|---:|---:|---:|---:|---:|"]
    for y, r in per_year.items():
        out.append(f"| {y} | {r['n_pairs']} | {r['exact']} | {r['within_2pct']} | {r['median_abs_delta']} | {int(r['sum_fldoe'])} | {int(r['sum_ccd'])} |")
    for y, r in per_year.items():
        out.append(f"\n## {y} — largest differences (school, FLDOE, CCD)")
        out += [f"- {p[0]}: {int(p[1])} vs {int(p[2])}" for p in r["largest_deltas"]]
    (common.QA_REPORTS_DIR / "__cross_publisher_enrollment.md").write_text("\n".join(out))
    print(json.dumps(summary["summary"], indent=1))
    for y, r in per_year.items():
        print(f"  {y}: pairs={r['n_pairs']} exact={r['exact']} within2%={r['within_2pct']} ccd_only={len(r['ccd_only'])} fldoe_only={r['fldoe_only_count']}")


if __name__ == "__main__":
    main()