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