pipeline_philly/verify/reconcile_spree.py

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