"""Base case (ii) reconciliation: served DB vs exact ingested source file. For each component in components.yaml, reads the served value from philly_school_year_metrics, re-reads the same cell from the ingested source file, and compares within tolerance. Mirrors verify/reconcile.py (NYC). Writes a Markdown + JSON report per metric to docs/qa_reports/philly/__validation.md (+ .json). Run: python pipeline_philly/verify/reconcile.py --metric pssa_all_ela_proficiency python pipeline_philly/verify/reconcile.py --all """ from __future__ import annotations import argparse import json import pathlib import sys # Make pipeline_philly importable when this script is run as a path. sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2])) import yaml # noqa: E402 from pipeline_philly.verify import common # noqa: E402 from pipeline_philly.verify.connectors import served_db # noqa: E402 from pipeline_philly.verify.connectors import source_file # noqa: E402 def load_components() -> list[dict]: with open(common.COMPONENTS_YAML) as f: return yaml.safe_load(f) def read_source(component: dict) -> dict[tuple[str, str], dict]: """Dispatch a component to the right source-file reader.""" sid = component["served_source"] args = component.get("source_args", {}) subgroup = component["subgroup"] year = component.get("year") if sid == "sdp_pssa_keystone_2024-25": return source_file.read_sdp_pssa_zip_schools( cut=args["cut"], year=year, testname=args["testname"], subject=args["subject"], grade=args["grade"], subgroup=subgroup, ) elif sid == "sdp_attendance_90_school": return source_file.read_sdp_csv_long( source_id=sid, metric_value_col=args["metric_value_col"], subgroup=subgroup, ) elif sid == "sdp_graduation_school_2014-2025": return source_file.read_sdp_csv_long( source_id=sid, metric_value_col=args["metric_value_col"], school_col=args["school_col"], year_col=args["year_col"], group_col=args["group_col"], subgroup=subgroup, extra_filter=args.get("extra_filter"), ) else: raise SystemExit(f"no source-reader dispatch for source: {sid}") def reconcile_one(component: dict) -> dict: """Run base-case reconciliation for one metric. Returns summary dict.""" metric = component["metric_key"] subgroup = component["subgroup"] cut = component["population_cut"] tol = float(component["tolerance_abs"]) year_filter = component.get("year") served = served_db.fetch(metric, subgroup=subgroup, population_cut=cut) source = read_source(component) # If a single year was specified in the config, restrict served + source to it if year_filter: served = {k: v for k, v in served.items() if k[1] == year_filter} source = {k: v for k, v in source.items() if k[1] == year_filter} served_keys = set(served.keys()) source_keys = set(source.keys()) common_keys = served_keys & source_keys # Cell-level fidelity within common keys matches = 0 mismatches = [] suppressed = 0 for k in common_keys: s_val = served[k]["value"] f_val = source[k]["value"] if served[k]["suppressed"]: suppressed += 1 continue if s_val is None and f_val is None: matches += 1 continue if s_val is None or f_val is None: mismatches.append((k, s_val, f_val, None)) continue diff = s_val - f_val if abs(diff) <= tol: matches += 1 else: mismatches.append((k, s_val, f_val, diff)) return { "metric_key": metric, "subgroup": subgroup, "population_cut": cut, "tolerance_abs": tol, "n_served": len(served), "n_source": len(source), "n_common": len(common_keys), "matches": matches, "mismatches": mismatches, "suppressed": suppressed, "served_only_count": len(served_keys - source_keys), "source_only_count": len(source_keys - served_keys), "served_only_examples": sorted([f"{k[0]} {k[1]}" for k in (served_keys - source_keys)])[:20], "source_only_examples": sorted([f"{k[0]} {k[1]}" for k in (source_keys - served_keys)])[:20], "agreement_rate": matches / (matches + len(mismatches)) if (matches + len(mismatches)) else None, "status": "PASS" if matches and len(mismatches) == 0 else "MISMATCH", } def render_report(summary: dict) -> str: out: list[str] = [] out.append(f"# Validation (ii) base case — {summary['metric_key']}\n") out.append("served DB cell-by-cell vs an independent re-read of the exact ingested file.\n") out.append(f"## Summary\n") for k in [ "metric_key", "subgroup", "population_cut", "tolerance_abs", "n_served", "n_source", "n_common", "matches", "suppressed", "served_only_count", "source_only_count", "agreement_rate", "status", ]: out.append(f"- {k}: {summary.get(k)}") if summary.get("served_only_examples"): out.append("\n## Served-only examples\n") for e in summary["served_only_examples"]: out.append(f"- {e}") if summary.get("source_only_examples"): out.append("\n## Source-only examples\n") for e in summary["source_only_examples"]: out.append(f"- {e}") if summary["mismatches"]: out.append(f"\n## Mismatches ({len(summary['mismatches'])})\n") for (key, s, f, d) in summary["mismatches"][:50]: out.append(f"- {key[0]} {key[1]}: served={s} source={f} Δ={d}") if len(summary["mismatches"]) > 50: out.append(f"- ... and {len(summary['mismatches']) - 50} more") return "\n".join(out) def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--metric", default=None) ap.add_argument("--all", action="store_true") args = ap.parse_args() common.ensure_qa_dir() components = load_components() if args.metric: components = [c for c in components if c["metric_key"] == args.metric] if not components: print(f"no component matching --metric={args.metric}", file=sys.stderr) sys.exit(1) if not args.metric and not args.all: print("must pass --metric= or --all") sys.exit(1) for component in components: summary = reconcile_one(component) # Persist the report slug = component["metric_key"] md = render_report(summary) (common.QA_REPORTS_DIR / f"{slug}__validation.md").write_text(md) json_summary = {k: v for k, v in summary.items() if k != "mismatches"} json_summary["n_mismatches"] = len(summary["mismatches"]) (common.QA_REPORTS_DIR / f"{slug}__validation.json").write_text( json.dumps(json_summary, indent=2), ) print( f"{slug:38s} STATUS: {summary['status']:9s} " f"common={summary['n_common']:>5d} matches={summary['matches']:>5d} " f"mismatches={len(summary['mismatches']):>4d} " f"served_only={summary['served_only_count']:>3d} source_only={summary['source_only_count']:>3d}", ) if __name__ == "__main__": main()