pipeline_philly/verify/cross_publisher_pssa.py

"""(iii-b) Independent-publisher cross-check for PSSA all-grades proficiency.

SDP publishes its own PSSA aggregate (sdp_pssa_keystone_2024-25). PDE publishes
the same metric via Future Ready PA Index. Both are independent publishers;
PDE submits to SDP's pipeline and SDP republishes. Comparing both sides for
the same (school, year, subgroup) is structurally analogous to KX's NYC
spot check (Snapshot reference) but uses an actually-independent oracle.

Output: docs/qa_reports/philly/__cross_publisher_pssa.md

Run: python pipeline_philly/verify/cross_publisher_pssa.py
"""
from __future__ import annotations

import json
import pathlib
import statistics
import sys

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2]))

from pipeline_philly.verify import common  # noqa: E402
from pipeline_philly.verify.connectors import source_file  # noqa: E402


_PHILLY_MASTER_LIST_SOURCE = "sdp_master_school_list_2024-25"


def build_aun_schl_to_ulcs() -> dict[str, str]:
    """Map "<aun>-<schl>" → ulcs_code using the Master School List CSV."""
    import pandas as pd
    path = common.latest_source_path(_PHILLY_MASTER_LIST_SOURCE)
    df = pd.read_csv(path, low_memory=False)
    out: dict[str, str] = {}
    for _, r in df.iterrows():
        if r.get("AUN Code") is None or r.get("PA Code") is None:
            continue
        try:
            aun = str(int(float(r["AUN Code"])))
            pa = str(int(float(r["PA Code"])))
            ulcs = str(int(float(r["ULCS Code"])))
        except (ValueError, TypeError):
            continue
        out[f"{aun}-{pa}"] = ulcs
    return out


def main() -> None:
    common.ensure_qa_dir()

    # SDP side (Acct cohort, all-grades)
    sdp = source_file.read_sdp_pssa_zip_schools(
        cut="acct",
        year="2024-25",
        testname="PSSA",
        subject="ELA",
        grade="Grades 3-8",
        subgroup="ALL",
    )
    # Future Ready side — the all-grades column is published under a separate
    # measure_column (3rd-grade aggregate uses keystone-style schema in the
    # State Assessment Measures sheet, but PSSA all-grades isn't there).
    # For v1, run the cross-check on Keystone Literature ALL (PercentProficient
    # orAdvancedonELALiterature_AllStudent) as that's school-wide & both
    # publishers report it.
    frpi = source_file.read_frpi_performance_school_wide(
        metric_column_prefix="PercentProficientorAdvancedonELALiterature",
        subgroup="ALL",
        sheet="State Assessment Measures",
        year="2024-25",
    )
    # The SDP side uses (ulcs, year); the FRPI side uses (aun-pa, year).
    # Cross-walk via the Master List.
    aunpa_to_ulcs = build_aun_schl_to_ulcs()
    frpi_by_ulcs: dict[tuple[str, str], dict] = {}
    for (key, year), v in frpi.items():
        ulcs = aunpa_to_ulcs.get(key)
        if ulcs is None:
            continue
        frpi_by_ulcs[(ulcs, year)] = v

    common_keys = set(sdp.keys()) & set(frpi_by_ulcs.keys())

    # SDP all-grades isn't the same as FRPI Keystone Literature — both are ELA
    # though, and at the school level they should track. Filter only to schools
    # where both have non-null values; compute deltas.
    diffs: list[float] = []
    pairs: list[dict] = []
    for k in common_keys:
        s = sdp[k]["value"]
        f = frpi_by_ulcs[k]["value"]
        if s is None or f is None:
            continue
        diffs.append(s - f)
        pairs.append({
            "school_ulcs": k[0], "year": k[1],
            "sdp_pssa_ela_all": round(s, 2),
            "frpi_keystone_literature_all": round(f, 2),
            "delta": round(s - f, 2),
        })
    pairs.sort(key=lambda r: abs(r["delta"]), reverse=True)

    if not diffs:
        print("no pairs found — check ID crosswalk")
        return

    summary = {
        "n_pairs": len(diffs),
        "mean": round(statistics.mean(diffs), 3),
        "median": round(statistics.median(diffs), 3),
        "stdev": round(statistics.pstdev(diffs), 3),
        "min": round(min(diffs), 3),
        "max": round(max(diffs), 3),
        "n_within_1pp": sum(1 for d in diffs if abs(d) <= 1.0),
        "n_within_5pp": sum(1 for d in diffs if abs(d) <= 5.0),
    }

    out: list[str] = []
    out.append("# Cross-publisher (iii-b) — PSSA ELA aggregate\n")
    out.append("SDP PSSA file vs Future Ready PA Index file. NB: SDP's PSSA all-grades\n")
    out.append("(grades 3-8 aggregate) and Future Ready's Keystone Literature (HS) are\n")
    out.append("DIFFERENT exams measuring ELA at different grade levels — they shouldn't\n")
    out.append("agree numerically, only correlate. This file establishes the (iii-b) wiring\n")
    out.append("and reports the cross-publisher pair distribution; a tighter cross-check\n")
    out.append("would use FRPI's per-grade PSSA columns if/when they're added.\n")
    out.append(f"\n## Summary\n```json\n{json.dumps(summary, indent=2)}\n```\n")
    out.append(f"\n## Largest 20 absolute deltas\n")
    for r in pairs[:20]:
        out.append(
            f"- {r['school_ulcs']} {r['year']}: "
            f"SDP={r['sdp_pssa_ela_all']:.2f} FRPI={r['frpi_keystone_literature_all']:.2f} "
            f"Δ={r['delta']:+.2f}"
        )

    (common.QA_REPORTS_DIR / "__cross_publisher_pssa.md").write_text("\n".join(out))
    (common.QA_REPORTS_DIR / "__cross_publisher_pssa.json").write_text(
        json.dumps({"summary": summary, "pairs_top20": pairs[:20]}, indent=2),
    )
    print(f"cross-publisher pairs: {len(diffs)}; summary: {json.dumps(summary)}")


if __name__ == "__main__":
    main()