"""(iii-a) Spot check — served DB vs SDP SPREE composite source. SPREE = SDP's School Progress Report on Education and Equity. It republishes the same PSSA/Keystone/grad/attendance numbers families see in the official report card, on a per-school per-metric grain (long-format CSV). Per O.3: SPREE is both a published-rank baseline (for the site's cross-referencing) AND the (iii-a) spot-check oracle. This script does the spot-check half. Same-publisher caveat applies (per 05_validation_spec.md): SPREE comes from SDP's own pipeline, so it's a presentation cross-reference, not an independent oracle. The independent (iii-b) check uses Future Ready PA Index (separate script). Output: docs/qa_reports/philly/__spree_spot_check.md (+ .json). Run: python pipeline_philly/verify/reconcile_spree.py """ from __future__ import annotations import json import pathlib import statistics import sys import pandas as pd sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2])) from pipeline_philly.verify import common # noqa: E402 SPREE_SOURCE = "sdp_spree_2023-24" TOLERANCE = 1.5 # ±pp, per KX's NYC convention # Map SPREE metric-string prefix → our metric_key. # Pattern: a SPREE metric string starts with a known prefix and may end with # an optional subgroup suffix like ", Black/African American". METRIC_PREFIX_MAP: dict[str, str] = { "PSSA ELA: % of Students Proficient or Advanced, Grades 3-8": "pssa_all_ela_proficiency", "PSSA Math: % of Students Proficient or Advanced, Grades 3-8": "pssa_all_math_proficiency", "PSSA ELA: % of Students Proficient or Advanced, Grade 3": "pssa_grade3_ela_proficiency", "PSSA Math: % of Students Proficient or Advanced, Grade 3": "pssa_grade3_math_proficiency", "PSSA ELA: % of Students Proficient or Advanced, Grade 4": "pssa_grade4_ela_proficiency", "PSSA Math: % of Students Proficient or Advanced, Grade 4": "pssa_grade4_math_proficiency", "PSSA ELA: % of Students Proficient or Advanced, Grade 5": "pssa_grade5_ela_proficiency", "PSSA Math: % of Students Proficient or Advanced, Grade 5": "pssa_grade5_math_proficiency", "PSSA ELA: % of Students Proficient or Advanced, Grade 6": "pssa_grade6_ela_proficiency", "PSSA Math: % of Students Proficient or Advanced, Grade 6": "pssa_grade6_math_proficiency", "PSSA ELA: % of Students Proficient or Advanced, Grade 7": "pssa_grade7_ela_proficiency", "PSSA Math: % of Students Proficient or Advanced, Grade 7": "pssa_grade7_math_proficiency", "PSSA ELA: % of Students Proficient or Advanced, Grade 8": "pssa_grade8_ela_proficiency", "PSSA Math: % of Students Proficient or Advanced, Grade 8": "pssa_grade8_math_proficiency", "Keystone Algebra I: % of Students Proficient or Advanced": "keystone_algebra_proficiency", "Keystone Literature: % of Students Proficient or Advanced": "keystone_literature_proficiency", "Keystone Biology: % of Students Proficient or Advanced": "keystone_biology_proficiency", "4-Year Cohort Graduation Rate": "graduation_rate_4yr", } # Suffix after ", " → our PhillySubgroup SUBGROUP_SUFFIX_MAP: dict[str, str] = { "": "ALL", # bare prefix = All Students "American Indian/Alaskan Native": "AMER_INDIAN_AK_NATIVE", "Asian": "ASIAN", "Black/African American": "BLACK", "Hispanic/Latino": "HISPANIC", "Multiracial": "TWO_OR_MORE_RACES", "Native Hawaiian/Pacific Islander": "HAWAIIAN_PAC_ISL", "White": "WHITE", "Economically Disadvantaged": "ECON_DISADV", "English Learners": "ELL", "Special Education": "IEP", } def parse_spree_metric(s: str) -> tuple[str, str] | None: """Map SPREE metric text → (metric_key, subgroup). None if unrecognized.""" s = s.strip() for prefix, metric_key in METRIC_PREFIX_MAP.items(): if s == prefix: return metric_key, "ALL" if s.startswith(prefix + ","): suffix = s[len(prefix) + 1:].strip() sg = SUBGROUP_SUFFIX_MAP.get(suffix) if sg: return metric_key, sg return None def main() -> None: common.ensure_qa_dir() spree_path = common.latest_source_path(SPREE_SOURCE) df = pd.read_csv(spree_path, low_memory=False) print(f"[spree] loaded {len(df):,} SPREE rows") # SPREE year is "2023-2024"; normalize to "2023-24" to match our DB df["year_canon"] = df["accountability_year"].str.replace(r"^(\d{4})-\d{2}(\d{2})$", r"\1-\2", regex=True) # Parse metrics df["parsed"] = df["metric"].map(lambda m: parse_spree_metric(str(m)) if m else None) df_mapped = df[df["parsed"].notna()].copy() print(f"[spree] mapped rows (recognized metric × subgroup): {len(df_mapped):,}") df_mapped[["metric_key", "subgroup"]] = pd.DataFrame(df_mapped["parsed"].tolist(), index=df_mapped.index) # Now query DB for the same (ulcs, year, metric_key, subgroup) tuples. # For PSSA + Keystone we have two population_cuts (acct, actual). The (iii-a) # check uses acct (canonical per O.8). dsn = common.served_db_dsn() try: import psycopg except ImportError as e: raise SystemExit("psycopg not installed") from e # SPREE's score values are stored as percent (0-100), so direct compare works. diffs: list[dict] = [] found = 0 not_in_db = 0 null_in_spree = 0 pairs_count = 0 matches = 0 mismatches = 0 with psycopg.connect(dsn) as conn: conn.read_only = True with conn.cursor() as cur: # Build a temp lookup of (ulcs, year, metric_key, subgroup, cut='acct') → value keys = [ (str(r["school_id"]).strip(), str(r["year_canon"]), r["metric_key"], r["subgroup"]) for _, r in df_mapped.iterrows() ] # Query the relevant slice cur.execute( """ SELECT school_ulcs, year, metric_key, subgroup::text AS subgroup, value FROM philly_school_year_metrics WHERE population_cut IN ('acct', 'n/a') AND value IS NOT NULL """, ) db_map: dict[tuple, float] = {} for ulcs, year, mk, sg, val in cur: db_map[(str(ulcs), str(year), mk, sg)] = float(val) for _, r in df_mapped.iterrows(): pairs_count += 1 ulcs = str(r["school_id"]).strip() year = str(r["year_canon"]) mk = r["metric_key"] sg = r["subgroup"] spree_v = pd.to_numeric(r["metric_score"], errors="coerce") if pd.isna(spree_v): null_in_spree += 1 continue db_v = db_map.get((ulcs, year, mk, sg)) if db_v is None: not_in_db += 1 continue found += 1 delta = float(db_v) - float(spree_v) if abs(delta) <= TOLERANCE: matches += 1 else: mismatches += 1 diffs.append({ "school_ulcs": ulcs, "year": year, "metric_key": mk, "subgroup": sg, "served": round(float(db_v), 2), "spree": round(float(spree_v), 2), "delta": round(delta, 2), }) deltas = [d["delta"] for d in diffs] summary = { "spree_source": SPREE_SOURCE, "tolerance_abs": TOLERANCE, "spree_rows_total": len(df), "spree_rows_mapped": len(df_mapped), "pairs_attempted": pairs_count, "null_in_spree": null_in_spree, "not_in_db": not_in_db, "pairs_compared": found, "matches_within_tol": matches, "mismatches": mismatches, "agreement_rate": round(matches / found, 4) if found > 0 else None, "mean_delta": round(statistics.mean(deltas), 3) if deltas else None, "median_delta": round(statistics.median(deltas), 3) if deltas else None, "max_abs_delta": round(max(abs(d) for d in deltas), 3) if deltas else None, } # Worst 30 by abs(delta) worst = sorted(diffs, key=lambda r: abs(r["delta"]), reverse=True)[:30] out: list[str] = [] out.append("# (iii-a) SPREE spot check — same-publisher presentation cross-reference\n") out.append("Compares served DB values against SDP SPREE composite. Same-publisher\n") out.append("(both feed from SDP's data pipeline) — this is a presentation check, NOT\n") out.append("an independent oracle. For independent cross-publisher check, see\n") out.append("__cross_publisher_pssa.md.\n") out.append(f"\n## Summary\n```json\n{json.dumps(summary, indent=2)}\n```\n") if worst: out.append("\n## Worst 30 mismatches (by absolute delta)\n") for r in worst: out.append( f"- ULCS {r['school_ulcs']} {r['year']} {r['metric_key']} {r['subgroup']}: " f"served={r['served']:.2f} spree={r['spree']:.2f} Δ={r['delta']:+.2f}", ) (common.QA_REPORTS_DIR / "__spree_spot_check.md").write_text("\n".join(out)) (common.QA_REPORTS_DIR / "__spree_spot_check.json").write_text( json.dumps({"summary": summary, "worst_30": worst}, indent=2), ) print(f"\n[spree] {json.dumps(summary)}") print(f"[spree] report -> docs/qa_reports/philly/__spree_spot_check.md") if __name__ == "__main__": main()