pipeline_miami/verify/served_invariants.py
Served verbatim from the project repository. Internal working document conventions apply: documents may reference file paths, branch names, and findings-ledger anchors from the repo.
"""(iv) Served-value invariants: properties every served cell must satisfy
regardless of which file it came from, plus year-over-year shift alarms.
Added after the 2026-09-16 review, whose critical findings were all
invisible to the cell-level base case because they were consistent with
the file being re-read: a total that was exactly twice the sum of its
categories, survey composites above 100, a year whose small-group medians
jumped from 0 to 1, a year with 32 fewer schools. Each is a violation of
something that must hold whatever the file says.
Hard rules (any violation → FAIL):
I1 Range by unit: PERCENT in [0, 100], except signed percent changes
(keys containing "_change_": ≥ −100) and utilization ratios (keys
containing "utilization": ≥ 0, a full building can exceed 100%);
RATE_PER_100, COUNT, USD ≥ 0; school_grade_letter_index in [0, 4];
ratios > 0.
I2 Identities between served cells of the same school-year:
sesir_total_per_100 ≥ Σ of the served sesir_*_per_100 categories
(the categories are disjoint incident types, so their sum is a lower
bound on the total; a total that is a multiple of the file's total
is caught in the loader, which cross-checks the summed types against
the published total and refuses the file on disagreement);
race shares (7 × pct_*) sum to 100 ± 0.05 where all are served;
enrollment_oct = Σ enrollment_oct_g* ± 0.5 where all 14 grades are served;
ELL lower bound ≤ upper bound; an unsuppressed exact ELL share exists
if and only if its bounds coincide, and equals those bounds.
I3 Ownership: every metrics row belongs to its source's latest load
(a superseded load's rows are stale and must be deleted, not served).
Shift alarms (REVIEW unless explained in served_invariants_allowlist.yaml):
S1 For each metric with ≥ 3 served years, each year's served median is
compared with the PREVIOUS served year's (the first year with the
next): |Δ| > 3 × max(spread, floor). spread is the MAD of the series'
own year-to-year moves not touching the tested year (so a one-year
defect cannot inflate the spread it is judged against); with fewer
than three such moves it is the unit's pooled prior (the median MAD of
moves over the long series of that unit, measured at run time). floor
is the larger of a unit floor (0.25 pp for percents, 2.5% of the level
otherwise) and the pooled ratio of typical move to within-year IQR
(≈ 0.06 for percents) times the series' own IQR, so a test-score
series whose schools span 25 pp tolerates a 4–5 pp move while a share
that clusters at zero tolerates 0.75 pp. The count of served schools
is compared the same way with floor max(1% of the level, 2 schools). Reworked after the 2026-09-16b
review (N8), which injected the defects the step was built for and
showed the earlier rule (median of the other years, half an IQR, a
0.5 pp floor) missed every one: under the current rule an ELA median
moved by 8 pp, a small race share lifted by 0.96 pp and 25 schools
dropped from a membership year each alarm (spread of the moves 0.3–2 pp
and 2–6 schools). A single-year step alarms in the step year and again
when the series returns; a real break (a new test, a new definition) is
recorded in the allowlist with its kind and reason so the alarm is
explained rather than silenced. Allowlist entries are keyed by
(metric_key, year, kind) and entries that match no current alarm are
listed as stale.
Writes docs/qa_reports/miami/__served_invariants.{json,md}.
Run: python -m pipeline_miami.verify.served_invariants
"""
from __future__ import annotations
import fnmatch
import json
import statistics
import yaml
from pipeline_miami.verify import common
ALLOWLIST = common.REPO_ROOT / "pipeline_miami" / "verify" / "served_invariants_allowlist.yaml"
RACE_KEYS = ["pct_white", "pct_black", "pct_hispanic", "pct_asian", "pct_hawaiian_pac_isl", "pct_amer_indian", "pct_two_or_more"]
GRADES = ["PK", "KG", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12"]
SESIR_CATS = ["sesir_violence_per_100", "sesir_bullying_per_100", "sesir_weapons_per_100", "sesir_drugs_per_100"]
def fetch_all():
import psycopg
with psycopg.connect(common.served_db_dsn()) as conn:
conn.read_only = True
defs = {k: {"unit": u, "domain": d} for k, u, d in conn.execute("SELECT key, unit::text, domain::text FROM miami_metric_definitions")}
rows = conn.execute("SELECT school_number, year, metric_key, value::float8 FROM miami_school_year_metrics WHERE subgroup='ALL' AND NOT suppressed AND value IS NOT NULL").fetchall()
stale = conn.execute("""
WITH latest AS (SELECT DISTINCT ON (source_id) source_id, id FROM miami_data_loads ORDER BY source_id, loaded_at DESC)
SELECT l.source_id, count(*) FROM miami_school_year_metrics m JOIN miami_data_loads l ON l.id = m.source_load_id
WHERE l.id NOT IN (SELECT id FROM latest) GROUP BY 1 ORDER BY 2 DESC""").fetchall()
return defs, rows, stale
def mad(xs: list[float]) -> float:
m = statistics.median(xs)
return statistics.median(abs(x - m) for x in xs)
def ell_bounds_problem(cells: dict[str, float]) -> str | None:
keys = {"pct_ell", "pct_ell_lower_bound", "pct_ell_upper_bound"}
if not keys.intersection(cells):
return None
lower, upper, exact = (cells.get(k) for k in ("pct_ell_lower_bound", "pct_ell_upper_bound", "pct_ell"))
if lower is None or upper is None:
return "missing ELL interval endpoint"
if lower > upper + 1e-8:
return f"lower={lower} exceeds upper={upper}"
if abs(upper - lower) > 1e-8:
return f"uncertain interval [{lower}, {upper}] served as exact {exact}" if exact is not None else None
if exact is None or abs(exact - lower) > 1e-8:
return f"collapsed interval [{lower}, {upper}] disagrees with exact {exact}"
return None
def main() -> None:
common.ensure_qa_dir()
defs, rows, stale = fetch_all()
allow = yaml.safe_load(open(ALLOWLIST)) if ALLOWLIST.exists() else []
# Exact entries keyed by (metric_key, year, kind); pattern entries (a metric_key glob and/or a `years` list) explain
# a documented district-wide cause across several series. An entry without a kind is legacy and matches either.
allowed, allowed_patterns = {}, []
for a in (allow or []):
kind = a.get("kind", "*")
if "*" in a["metric_key"] or "years" in a or "year" not in a: # no year at all: every year of that metric
allowed_patterns.append(((a["metric_key"], tuple(str(y) for y in a["years"]) if a.get("years") else None, kind), a["reason"]))
else:
allowed[(a["metric_key"], str(a["year"]), kind)] = a["reason"]
matched: set = set()
by_cell: dict[tuple[str, str], dict[str, float]] = {}
for num, year, key, val in rows:
by_cell.setdefault((str(num), str(year)), {})[key] = float(val)
# I1 — range by unit
range_violations = []
for (num, year), cells in by_cell.items():
for key, v in cells.items():
unit = defs.get(key, {}).get("unit")
lo, hi = (-100.0, None) if "_change_" in key else (0.0, None) if "utilization" in key else (0.0, 100.0)
bad = (unit == "PERCENT" and (v < lo or (hi is not None and v > hi))) or (unit in ("RATE_PER_100", "COUNT", "USD") and v < 0) \
or (key == "school_grade_letter_index" and not (0.0 <= v <= 4.0)) or (key in ("k_to_g5_ratio", "students_per_teacher") and v <= 0)
if bad:
range_violations.append({"school": num, "year": year, "metric_key": key, "value": v, "unit": unit})
# I2 — identities
identity_violations = []
for (num, year), cells in by_cell.items():
ell_problem = ell_bounds_problem(cells)
if ell_problem:
identity_violations.append({"school": num, "year": year, "rule": "ELL exact share agrees with censoring interval", "detail": ell_problem})
if "sesir_total_per_100" in cells:
cats = {c: cells[c] for c in SESIR_CATS if c in cells}
if cats and sum(cats.values()) > cells["sesir_total_per_100"] + 1e-6:
identity_violations.append({"school": num, "year": year, "rule": "sesir_total >= sum of categories", "detail": f"sum={sum(cats.values()):.3f} > total={cells['sesir_total_per_100']:.3f}"})
if all(k in cells for k in RACE_KEYS):
s = sum(cells[k] for k in RACE_KEYS)
if abs(s - 100.0) > 0.05:
identity_violations.append({"school": num, "year": year, "rule": "race shares sum to 100", "detail": f"sum={s:.3f}"})
if "enrollment_oct" in cells and all(f"enrollment_oct_g{g}" in cells for g in GRADES):
s = sum(cells[f"enrollment_oct_g{g}"] for g in GRADES)
if abs(s - cells["enrollment_oct"]) > 0.5:
identity_violations.append({"school": num, "year": year, "rule": "enrollment_oct = sum of grades", "detail": f"sum={s:.0f} total={cells['enrollment_oct']:.0f}"})
# I3 — ownership
stale_rows = [{"source_id": s, "rows": int(n)} for s, n in stale]
# S1 — shift alarms per metric
per_metric: dict[str, dict[str, list[float]]] = {}
for (num, year), cells in by_cell.items():
for key, v in cells.items():
per_metric.setdefault(key, {}).setdefault(year, []).append(v)
alarms = []
def explain(key: str, y: str, kind: str):
for k in ((key, y, kind), (key, y, "*")):
r = allowed.get(k)
if r is not None:
matched.add(k)
return r
for (pat, ys, kd), r in allowed_patterns:
if fnmatch.fnmatchcase(key, pat) and (ys is None or y in ys) and kd in (kind, "*"):
matched.add((pat, ys, kd))
return r
return None
# Series statistics: medians and school counts per year; the moves between consecutive served years.
series = {}
for key, years in per_metric.items():
if len(years) < 3:
continue
ys = sorted(years)
meds = [statistics.median(years[y]) for y in ys]
cnts = [len(years[y]) for y in ys]
series[key] = (ys, meds, cnts, [meds[i + 1] - meds[i] for i in range(len(ys) - 1)], [cnts[i + 1] - cnts[i] for i in range(len(ys) - 1)])
# Pooled priors per unit, measured from the series long enough to estimate their own spread (≥ 4 moves): the
# median over those series of the MAD of their year-to-year moves. Used where a series is too short to speak.
pooled: dict[str, list[float]] = {}
pooled_rel: dict[str, list[float]] = {}
pooled_c: list[float] = []
iqr_of: dict[str, float] = {}
for key, years in per_metric.items():
iqrs = [statistics.quantiles(v, n=4)[2] - statistics.quantiles(v, n=4)[0] for v in years.values() if len(v) >= 4]
iqr_of[key] = statistics.median(iqrs) if iqrs else 0.0
for key, (ys, meds, cnts, moves, cmoves) in series.items():
if len(moves) >= 4:
u = defs.get(key, {}).get("unit") or "?"
pooled.setdefault(u, []).append(mad(moves))
pooled_c.append(mad(cmoves))
if iqr_of[key] > 0:
pooled_rel.setdefault(u, []).append(mad(moves) / iqr_of[key])
prior = {u: statistics.median(v) for u, v in pooled.items()}
# The typical year-to-year move of a series' median, in units of how much its schools differ (its pooled
# within-year IQR): measured across the long series of each unit (2026-09-16b: 0.06 of the IQR for percents),
# so a series' floor scales with its own cross-sectional spread instead of a fixed number of points.
prior_rel = {u: statistics.median(v) for u, v in pooled_rel.items()}
prior_c = statistics.median(pooled_c) if pooled_c else 2.0
for key, (ys, meds, cnts, moves, cmoves) in series.items():
unit = defs.get(key, {}).get("unit")
level = abs(statistics.median(meds))
floor = max(0.25 if unit == "PERCENT" else max(level * 0.025, 1e-9), prior_rel.get(unit or "?", 0.0) * iqr_of[key])
cfloor = max(0.01 * statistics.median(cnts), 2.0)
for i, y in enumerate(ys):
ref_i = i - 1 if i > 0 else 1
# Spread from the series' own moves that do not touch this year (a one-year defect must not inflate
# the spread it is tested against); where fewer than three remain, the unit's pooled prior.
own = [m for j, m in enumerate(moves) if j not in (i - 1, i)]
spread = mad(own) if len(own) >= 3 else prior.get(unit or "?", floor)
threshold = 3 * max(spread, floor)
d = meds[i] - meds[ref_i]
if abs(d) > threshold:
alarms.append({"metric_key": key, "year": y, "kind": "median", "value": round(meds[i], 3), "reference_year": ys[ref_i], "reference": round(meds[ref_i], 3),
"spread_of_moves": round(spread, 3), "spread_source": "own" if len(own) >= 3 else f"pooled {unit}", "floor": round(floor, 3), "threshold": round(threshold, 3),
"lift": round(abs(d) / max(spread, floor), 1), "explained": explain(key, y, "median")})
ownc = [m for j, m in enumerate(cmoves) if j not in (i - 1, i)]
cspread = mad(ownc) if len(ownc) >= 3 else prior_c
cthreshold = 3 * max(cspread, cfloor)
dc = cnts[i] - cnts[ref_i]
if abs(dc) > cthreshold:
alarms.append({"metric_key": key, "year": y, "kind": "count", "value": cnts[i], "reference_year": ys[ref_i], "reference": cnts[ref_i],
"spread_of_moves": round(cspread, 1), "spread_source": "own" if len(ownc) >= 3 else "pooled", "floor": round(cfloor, 1), "threshold": round(cthreshold, 1),
"lift": round(abs(dc) / max(cspread, cfloor), 1), "explained": explain(key, y, "count")})
stale_allow = sorted([k for k in allowed if k not in matched] + [k for k, _ in allowed_patterns if k not in matched], key=str)
unkeyed_allow = sorted([k for k in allowed if k[2] == "*"] + [k for k, _ in allowed_patterns if k[2] == "*"], key=str)
unexplained = [a for a in alarms if not a["explained"]]
hard_ok = not range_violations and not identity_violations and not stale_rows
verdict = "FAIL" if not hard_ok else ("REVIEW" if unexplained else "PASS")
out = {"verdict": verdict, "hard": {"I1_range_violations": len(range_violations), "I2_identity_violations": len(identity_violations), "I3_stale_rows": stale_rows},
"shift_alarms": {"n": len(alarms), "unexplained": len(unexplained), "rows": alarms, "allowlist_entries_matching_no_alarm": [list(k) for k in stale_allow], "allowlist_entries_without_kind": [list(k) for k in unkeyed_allow]},
"examples": {"range": range_violations[:30], "identity": identity_violations[:30]},
"pooled_priors": {"median_moves_by_unit": prior, "median_moves_relative_to_iqr_by_unit": prior_rel, "count_moves": prior_c},
"note": "Hard rules fail on any violation. Shift alarms compare each year's served median and school count with the previous served year's against 3 × the MAD of the series' own other year-to-year moves (the unit's pooled prior where fewer than three remain), floored at 0.25 pp / 2.5% of level, counts at max(1%, 2); an alarm is explained when served_invariants_allowlist.yaml records, for that metric (or metric pattern), year and kind, why the break is real."}
(common.QA_REPORTS_DIR / "__served_invariants.json").write_text(json.dumps(out, indent=2, default=str))
md = [f"# Served-value invariants\n", f"Verdict: **{verdict}**\n",
f"- I1 range by unit: {len(range_violations)} violations", f"- I2 identities: {len(identity_violations)} violations",
f"- I3 rows owned by superseded loads: {sum(r['rows'] for r in stale_rows)}" + (" (" + ", ".join(f"{r['source_id']}={r['rows']}" for r in stale_rows) + ")" if stale_rows else ""),
f"- S1 shift alarms: {len(alarms)} ({len(unexplained)} unexplained)", ""]
if range_violations:
md += ["## Range violations (first 30)"] + [f"- {r['school']} {r['year']} `{r['metric_key']}` = {r['value']} ({r['unit']})" for r in range_violations[:30]] + [""]
if identity_violations:
md += ["## Identity violations (first 30)"] + [f"- {r['school']} {r['year']} {r['rule']}: {r['detail']}" for r in identity_violations[:30]] + [""]
md += ["## Shift alarms", "", f"Pooled priors (median MAD of year-to-year moves over series with ≥ 4 moves): " + ", ".join(f"{u} {v:.3f}" for u, v in sorted(prior.items())) + "; relative to the series' within-year IQR: " + ", ".join(f"{u} {v:.3f}" for u, v in sorted(prior_rel.items())) + f"; school counts {prior_c:.1f}. A series' floor is the larger of the unit floor (0.25 pp; 2.5% of level) and this relative prior × its own IQR.", "",
"| metric | year | kind | value | reference (prev. year) | spread of moves (source) | floor | threshold | lift | explained |", "|---|---|---|---:|---:|---:|---:|---:|---:|---|"]
md += [f"| `{a['metric_key']}` | {a['year']} | {a['kind']} | {a['value']} | {a['reference']} ({a['reference_year']}) | {a['spread_of_moves']} ({a['spread_source']}) | {a['floor']} | {a['threshold']} | {a['lift']} | {a['explained'] or '**no**'} |" for a in alarms]
if stale_allow:
md += ["", "## Allowlist entries matching no current alarm", ""] + [f"- `{k}` {y} {kind}" for k, y, kind in stale_allow]
(common.QA_REPORTS_DIR / "__served_invariants.md").write_text("\n".join(md))
print(json.dumps({k: v for k, v in out.items() if k not in ("examples",)} | {"shift_alarms": {"n": len(alarms), "unexplained": len(unexplained)}}, indent=1, default=str))
for a in alarms:
print(f" {'ok ' if a['explained'] else 'REVIEW'} {a['metric_key']:36s} {a['year']} {a['kind']:6s} {a['value']!s:>10} vs {a['reference']!s:>10} ({a['reference_year']}) lift {a['lift']} {(a['explained'] or '')[:60]}")
for k in stale_allow:
print(f" STALE allowlist entry matches no alarm: {k}")
for r in range_violations[:10]:
print(f" RANGE {r['school']} {r['year']} {r['metric_key']} = {r['value']}")
for r in identity_violations[:10]:
print(f" IDENTITY {r['school']} {r['year']} {r['rule']}: {r['detail']}")
if __name__ == "__main__":
main()