pipeline_miami/verify/cross_publisher_absenteeism.py

"""(iii-b) Cross-publisher check for the outcomes track: FLDOE "absent 10% or
more" (chronic_absenteeism_rate × its denominator = a count) vs the federal
copy of the same submission in EDFacts FS195/DG814 (EdDataExpress), school by
school, 2022-23.

Population difference fixed BEFORE the comparison was run (documented in
docs/cities/miami/findings.md, F.14): EDFacts DG814 covers K-12 (+ ungraded)
while the FLDOE file counts every student enrolled ≥10 days including PK.
So the exact comparison is made on schools with no October PK membership;
schools with PK are reported as a regression of the gap on PK membership
(a descriptive slope, not a prevalence estimate). A suppressed PK cell means 1–9 PK
students (not none): those schools are kept out of the exact comparison and
out of the regression (their PK count is unknown), and counted. EDFacts
perturbs cells ≤4 by ±1, so the pass rule is within ±1 for ≥90% of no-PK
schools.

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

import csv
import io
import json
import statistics
import zipfile

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

SOURCE_ID = "eddataexpress_chronic_absenteeism_2022-23"
YEAR = "2022-23"
DADE_LEAID = "1200390"


def edfacts_allsch() -> dict[str, float]:
    """NCES_SCH → chronically absent count (ALLSCH) for Miami-Dade."""
    path = common.latest_source_path(SOURCE_ID)
    out: dict[str, float] = {}
    with zipfile.ZipFile(path) as z:
        name = next(n for n in z.namelist() if "_SCH_" in n and n.endswith(".csv"))
        with z.open(name) as fh:
            for r in csv.DictReader(io.TextIOWrapper(fh, encoding="utf-8-sig")):
                if r["LEAID"] != DADE_LEAID or r["SUBGROUP"] != "ALLSCH":
                    continue
                try:
                    out[r["NCES_SCH"]] = float(r["NUMERIC_VALUE"])
                except (TypeError, ValueError):
                    continue
    return out


def nces_map() -> dict[str, str]:
    import psycopg

    out: dict[str, str] = {}
    with psycopg.connect(common.served_db_dsn()) as conn:
        conn.read_only = True
        with conn.cursor() as cur:
            cur.execute("SELECT nces_id, school_number FROM miami_schools WHERE nces_id IS NOT NULL")
            for nces, num in cur:
                out[str(nces)] = str(num)
    return out


def main() -> None:
    common.ensure_qa_dir()
    ede = edfacts_allsch()
    nces = nces_map()
    fl = served_db.fetch("chronic_absenteeism_rate")
    pk = served_db.fetch("enrollment_oct_gPK")
    rows = []
    unmatched = 0
    for nces_id, count in ede.items():
        num = nces.get(nces_id)
        if not num:
            unmatched += 1
            continue
        f = fl.get((num, YEAR))
        if not f or f["value"] is None or not f["n"]:
            continue
        fl_count = round(f["value"] / 100 * f["n"])
        p = pk.get((num, YEAR))
        pk_status = "suppressed" if (p and p["suppressed"]) else ("with_pk" if (p and p["value"]) else "no_pk")
        rows.append({"school": num, "edfacts": count, "fldoe": fl_count, "delta": count - fl_count,
                     "pk": (p["value"] or 0.0) if p and p["value"] is not None else 0.0, "pk_status": pk_status, "denominator": f["n"]})
    no_pk = [r for r in rows if r["pk_status"] == "no_pk"]
    with_pk = [r for r in rows if r["pk_status"] == "with_pk"]
    pk_suppressed = [r for r in rows if r["pk_status"] == "suppressed"]
    within1 = sum(1 for r in no_pk if abs(r["delta"]) <= 1)
    exact = sum(1 for r in no_pk if r["delta"] == 0)
    share = within1 / len(no_pk) if no_pk else 0.0
    verdict = "PASS" if no_pk and share >= 0.90 else "REVIEW"
    # Gap on PK membership, least squares (different annual and October denominators; no prevalence interpretation).
    slope = intercept = corr = None
    if len(with_pk) >= 10:
        xs = [r["pk"] for r in with_pk]
        ys = [-r["delta"] for r in with_pk]
        mx, my = statistics.fmean(xs), statistics.fmean(ys)
        sxx = sum((x - mx) ** 2 for x in xs)
        sxy = sum((x - mx) * (y - my) for x, y in zip(xs, ys))
        syy = sum((y - my) ** 2 for y in ys)
        slope = sxy / sxx if sxx else None
        intercept = my - slope * mx if slope is not None else None
        corr = sxy / (sxx * syy) ** 0.5 if sxx and syy else None
    summary = {
        "summary": {
            "year": YEAR, "verdict": verdict, "n_edfacts_dade": len(ede), "n_unmatched_nces": unmatched, "n_pairs": len(rows),
            "no_pk": {"n": len(no_pk), "exact": exact, "within_1": within1, "within_1_share": share,
                      "median_delta": statistics.median(r["delta"] for r in no_pk) if no_pk else None},
            "pk_suppressed": {"n": len(pk_suppressed), "median_delta": statistics.median(r["delta"] for r in pk_suppressed) if pk_suppressed else None},
            "with_pk": {"n": len(with_pk), "median_delta": statistics.median(r["delta"] for r in with_pk) if with_pk else None,
                        "gap_on_pk": {"slope": slope, "intercept": intercept, "r": corr}},
            "verdict_note": "EDFacts DG814 is K-12 (+ungraded); FLDOE's absent-10%+ count includes PK, so exact agreement is expected only where October PK membership is zero. Pass rule: within ±1 (EDFacts perturbation) for ≥90% of no-PK schools. For PK schools the gap is reported as a regression on PK membership; schools whose PK cell is suppressed (1–9 students) are in neither group.",
        },
        "largest_no_pk_deltas": sorted(no_pk, key=lambda r: -abs(r["delta"]))[:15],
    }
    (common.QA_REPORTS_DIR / "__cross_publisher_absenteeism.json").write_text(json.dumps(summary, indent=2, default=str))
    s = summary["summary"]
    out = [
        "# Cross-publisher (iii-b) — FLDOE absent-10%+ count vs EDFacts DG814 (2022-23)\n",
        f"Verdict: **{verdict}** — {s['n_pairs']} schools matched by NCES id ({s['n_unmatched_nces']} EDFacts rows without a directory match).\n",
        "| population | schools | exact | within ±1 | median Δ (EDFacts − FLDOE) |", "|---|---:|---:|---:|---:|",
        f"| no October PK membership | {s['no_pk']['n']} | {s['no_pk']['exact']} | {s['no_pk']['within_1']} ({share * 100:.1f}%) | {s['no_pk']['median_delta']} |",
        f"| with PK membership | {s['with_pk']['n']} | — | — | {s['with_pk']['median_delta']} |",
        f"| PK cell suppressed (1–9) | {s['pk_suppressed']['n']} | — | — | {s['pk_suppressed']['median_delta']} |",
        "",
        (f"Schools with PK: EDFacts undercount = {slope:.3f} × PK membership {intercept:+.2f} (r = {corr:.3f}); the annual-count gap and October membership use different denominators, so this slope is not a direct estimate of PK chronic-absence prevalence."
         if slope is not None else "Too few PK schools for the regression."),
        "", s["verdict_note"], "", "## Largest differences among no-PK schools (school, EDFacts, FLDOE)",
    ]
    out += [f"- {r['school']}: {int(r['edfacts'])} vs {int(r['fldoe'])}" for r in summary["largest_no_pk_deltas"]]
    (common.QA_REPORTS_DIR / "__cross_publisher_absenteeism.md").write_text("\n".join(out))
    print(json.dumps(s, indent=1, default=str))


if __name__ == "__main__":
    main()