pipeline_philly/verify/reconcile_acct_actual.py

"""O.8 Acct vs Actual population-cut diff report.

Per O.8 (locked 2026-06-03), SDP PSSA is ingested in two population cuts:
  - Acct (accountability cohort, Oct-31 register attribution) — canonical
  - Actual (all tested students) — co-loaded for transparency

This script writes a per-school diff report at every (school, year, metric,
subgroup) where both cuts have a non-null served value. It's an additional
reconciliation gate distinct from (i)-(iv) (see 05_validation_spec.md
"Acct vs Actual"), structurally analogous to a population-cut variant of
the (ii) base case.

Output: docs/qa_reports/philly/__acct_vs_actual.md (+ .json).

Run: python pipeline_philly/verify/reconcile_acct_actual.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

_SQL = """
    SELECT a.school_ulcs,
           a.year,
           a.metric_key,
           a.subgroup::text AS subgroup,
           a.value::float8 AS acct_value,
           a.denominator::float8 AS acct_n,
           u.value::float8 AS actual_value,
           u.denominator::float8 AS actual_n,
           s.name,
           s.school_type::text
      FROM philly_school_year_metrics a
      JOIN philly_school_year_metrics u
        ON u.school_ulcs = a.school_ulcs
       AND u.year = a.year
       AND u.metric_key = a.metric_key
       AND u.subgroup = a.subgroup
       AND u.population_cut = 'actual'
      JOIN philly_schools s ON s.ulcs_code = a.school_ulcs
     WHERE a.population_cut = 'acct'
       AND a.value IS NOT NULL
       AND u.value IS NOT NULL
"""


def main() -> None:
    common.ensure_qa_dir()
    dsn = common.served_db_dsn()
    try:
        import psycopg
    except ImportError as e:
        raise SystemExit("psycopg not installed") from e

    with psycopg.connect(dsn) as conn:
        conn.read_only = True
        with conn.cursor() as cur:
            cur.execute(_SQL)
            pairs = cur.fetchall()

    diffs: list[float] = []
    by_metric: dict[str, list[float]] = {}
    by_subgroup: dict[str, list[float]] = {}
    rows: list[dict] = []
    for ulcs, year, metric, subgroup, acct_v, acct_n, actual_v, actual_n, name, stype in pairs:
        d = float(acct_v) - float(actual_v)
        diffs.append(d)
        by_metric.setdefault(metric, []).append(d)
        by_subgroup.setdefault(subgroup, []).append(d)
        rows.append({
            "school_ulcs": ulcs,
            "name": name,
            "school_type": stype,
            "year": year,
            "metric": metric,
            "subgroup": subgroup,
            "acct": float(acct_v),
            "actual": float(actual_v),
            "delta": d,
            "acct_n": float(acct_n) if acct_n is not None else None,
            "actual_n": float(actual_n) if actual_n is not None else None,
        })

    rows.sort(key=lambda r: abs(r["delta"]), reverse=True)

    def summarize(name: str, values: list[float]) -> dict:
        if not values:
            return {"name": name, "n": 0}
        return {
            "name": name,
            "n": len(values),
            "mean": round(statistics.mean(values), 3),
            "median": round(statistics.median(values), 3),
            "stdev": round(statistics.pstdev(values), 3),
            "min": round(min(values), 3),
            "max": round(max(values), 3),
            "n_zero": sum(1 for v in values if abs(v) < 1e-6),
            "n_le_1pp": sum(1 for v in values if abs(v) <= 1.0),
            "n_gt_5pp": sum(1 for v in values if abs(v) > 5.0),
        }

    overall = summarize("ALL", diffs)
    per_metric = {m: summarize(m, vs) for m, vs in by_metric.items()}
    per_subgroup = {s: summarize(s, vs) for s, vs in by_subgroup.items()}

    md: list[str] = []
    md.append("# Acct vs Actual population-cut diff — Philly PSSA\n")
    md.append("Per O.8 (locked 2026-06-03): SDP publishes PSSA in two population cuts;\n")
    md.append("Acct is canonical, Actual is co-loaded. This report lists per-school deltas\n")
    md.append("at every (school, year, metric, subgroup) where both have a non-null served\n")
    md.append("value. NYC had to reverse-engineer this gap; Philly publishes both natively.\n")
    md.append("\n## Finding (2026-09-07): the two published cuts are the same file\n")
    md.append(
        "SDP's 2024-25 archive ships PSSA_Keystone_Schools_Acct_Suppressed.csv and\n"
        "PSSA_Keystone_Schools_Actual_Suppressed.csv with identical byte counts; sorted,\n"
        "the two files are line-for-line identical (28,525 rows; 0 differing denominators,\n"
        "0 differing proficiency values). The zero-delta result below is therefore NOT\n"
        "evidence that the accountability cohort and all-tested populations agree — it is\n"
        "evidence that one dataset was published under two names. Until SDP clarifies,\n"
        "the served 'acct' and 'actual' cuts carry the same numbers and this check has no\n"
        "discriminating power. Flagged on every PSSA/Keystone proof tree.\n"
    )
    md.append("\n## Overall\n")
    md.append(f"```json\n{json.dumps(overall, indent=2)}\n```\n")
    md.append("## By metric\n")
    for m, s in sorted(per_metric.items()):
        md.append(f"### {m}")
        md.append(f"```json\n{json.dumps(s, indent=2)}\n```")
    md.append("\n## By subgroup\n")
    for sg, s in sorted(per_subgroup.items()):
        md.append(f"### {sg}")
        md.append(f"```json\n{json.dumps(s, indent=2)}\n```")
    md.append(f"\n## Largest 30 absolute deltas\n")
    for r in rows[:30]:
        md.append(
            f"- {r['name'][:32]:32s} ({r['school_ulcs']}) {r['year']} "
            f"{r['metric']} {r['subgroup']}: "
            f"acct={r['acct']:.2f} actual={r['actual']:.2f} Δ={r['delta']:+.2f}",
        )

    (common.QA_REPORTS_DIR / "__acct_vs_actual.md").write_text("\n".join(md))
    overall["finding"] = "acct and actual source files are identical after sorting (28,525 rows) — publisher defect; check has no discriminating power"
    (common.QA_REPORTS_DIR / "__acct_vs_actual.json").write_text(
        json.dumps(
            {"overall": overall, "per_metric": per_metric, "per_subgroup": per_subgroup,
             "n_pairs": len(rows), "largest_30": rows[:30]},
            indent=2,
        ),
    )

    print(f"pairs: {len(rows)}")
    print(f"overall: {json.dumps(overall)}")


if __name__ == "__main__":
    main()