From 29921fb203a781e17969e7f5f5696ec01baff9bd Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Fri, 2 Oct 2026 07:42:49 -0400 Subject: [PATCH] Gate the ACS local release on district ESS collapse, report-only by default Finalize reads the calibration summary's Kish ESS by congressional district at the design and the calibrated weights (weight_origin), counts districts whose calibrated ESS is below 0.25 x their design ESS and below an absolute floor of 15, and records the result as the district_ess_collapse gate (gate_summary.json and the build manifest) and the district_effective_sample_size_gate reviewed limitation. Whether it blocks releases awaits review, so blocking is the explicit --district-ess-gate-blocking flag, off by default; under it finalize fails a collapse or missing evidence, and package refuses a report finalize did not block on at the same floors. The default solve is unchanged. Co-Authored-By: Claude Opus 5.5 --- README.md | 15 + .../us-acs-local-district-ess-gate.added.md | 1 + .../us/test_us_acs_local_district_ess_gate.py | 542 ++++++++++++++++++ .../us/test_us_acs_local_release_tool.py | 215 +++++++ tools/build_us_acs_local_release.py | 290 +++++++++- 5 files changed, 1062 insertions(+), 1 deletion(-) create mode 100644 changelog.d/us-acs-local-district-ess-gate.added.md create mode 100644 packages/microcosm-build/tests/engine_free/us/test_us_acs_local_district_ess_gate.py diff --git a/README.md b/README.md index e584a6a02..5e5ac6082 100644 --- a/README.md +++ b/README.md @@ -344,6 +344,21 @@ the calibrated dataset exists. If construction, validation, serialization, or writing fails, the release manifest records the failure and publication emits a warning without discarding the dataset release. +The ACS local-area chain's finalize stage also records a district ESS gate, +`district_ess_collapse`. It reads the calibration summary's Kish ESS by +congressional district at the design and the calibrated weights. It counts the +districts whose calibrated ESS is below a quarter of their design-weight ESS +(`--district-ess-relative-floor`, default 0.25) and those below 15 +(`--district-ess-floor`); a floor of 0 turns its check off. At the release's +ACS share of 0.5, the unpenalized release solve leaves 15-24 districts below a +quarter across its full-surface and two holdout solves, and every +chi-square-penalized solve leaves none (microcosm#1078). By default the gate +is report-only: it passes, and records what it measured in `gate_summary.json`, +the build manifest's `gates` and the reviewed limitation +`district_effective_sample_size_gate`. With `--district-ess-gate-blocking`, a +failing gate stops finalize, and package refuses a gate report that finalize +did not block on at the same floors. + Current UK national and rowwise builders use that same schema and writer. The UK extension is fully typed: weight summaries, zero-weight strata, geography-level pass rates, local fit summaries, and rotated holdout evidence diff --git a/changelog.d/us-acs-local-district-ess-gate.added.md b/changelog.d/us-acs-local-district-ess-gate.added.md new file mode 100644 index 000000000..3cb6346f5 --- /dev/null +++ b/changelog.d/us-acs-local-district-ess-gate.added.md @@ -0,0 +1 @@ +The ACS local-area release's finalize stage records a district ESS gate, `district_ess_collapse`, read from the calibration summary's Kish ESS by congressional district at the design and the calibrated weights. It counts the districts whose calibrated ESS is below 0.25 x their design-weight ESS (`--district-ess-relative-floor`) and those below 15 (`--district-ess-floor`); a floor of 0 turns its check off. The result is recorded in `gate_summary.json`, the build manifest's `gates` and the reviewed limitation `district_effective_sample_size_gate`. The gate is report-only and passes by default. `--district-ess-gate-blocking` makes a failing gate (or missing per-district evidence) stop finalize, and makes package refuse a gate report that finalize did not block on at the same floors. The default solve is unchanged. diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_district_ess_gate.py b/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_district_ess_gate.py new file mode 100644 index 000000000..910f64d86 --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_district_ess_gate.py @@ -0,0 +1,542 @@ +"""The ACS local release's district ESS gate (finalize stage). + +Invariants, each for every input Hypothesis draws: + +1. The collapsed districts are nested in the relative floor, and the districts + below the floor in the absolute floor, so both counts are monotone in their + threshold. +2. A design-weight solve (calibrated weights equal to the design weights, or a + uniform rescaling of them) has no collapsed district at any relative floor + in [0, 1]. +3. Both counts are at most the number of districts, which is the number of + distinct district codes. +4. ``blocking`` changes only ``passed``, ``blocking`` and ``report_only``: a + report-only entry always passes, and a blocking one passes exactly when no + district collapses or falls below the floor. +5. Differential: the collapsed districts the gate reads from + ``weight_origin_summary`` (the calibration summary's evidence) are the ones + an independent bincount computation finds from the raw weights, away from + floating-point ties. +""" + +from __future__ import annotations + +import importlib.util +import json +import math + +import numpy as np +import pytest + +from test_support.paths import paths_for + +hypothesis = pytest.importorskip("hypothesis") +from hypothesis import HealthCheck, given, settings # noqa: E402 +from hypothesis import strategies as st # noqa: E402 + +_TEST_PATHS = paths_for("microcosm-build") +_SETTINGS = settings( + max_examples=80, + deadline=None, + suppress_health_check=[HealthCheck.too_slow], +) + + +def _load_tool_module(): + path = _TEST_PATHS.repository / "tools" / "build_us_acs_local_release.py" + spec = importlib.util.spec_from_file_location( + "build_us_acs_local_release_district_ess_gate", path + ) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +_TOOL = _load_tool_module() +_CD = _TOOL.cd_surface +_GATE = _TOOL.district_ess_collapse_gate +_LIMITATION = _TOOL.district_ess_collapse_limitation + + +def _origin(design: dict, calibrated: dict) -> dict: + """A calibration summary's ``weight_origin`` with per-district ESS only.""" + + return { + "design": {"effective_sample_size_by_district": design}, + "calibrated": {"effective_sample_size_by_district": calibrated}, + } + + +def _districts(rows: list[dict]) -> list[str]: + return [row["district"] for row in rows] + + +# --------------------------------------------------------------------------- +# Examples +# --------------------------------------------------------------------------- + + +def test_the_floors_default_to_the_measured_separation() -> None: + assert _TOOL.DISTRICT_ESS_RELATIVE_FLOOR == 0.25 + assert _TOOL.DISTRICT_ESS_FLOOR == 15.0 + assert _TOOL.DISTRICT_ESS_GATE == "district_ess_collapse" + + +def test_the_gate_counts_collapsed_and_below_floor_districts() -> None: + gate = _GATE( + _origin( + {"0101": 100.0, "0102": 100.0, "0103": 40.0, "0104": 80.0, "0105": 30.0}, + {"0101": 20.0, "0102": 25.0, "0103": 12.0, "0104": 60.0, "0105": 15.0}, + ) + ) + # 0101 keeps a fifth of its design ESS; 0102 exactly a quarter, which is + # not below it; 0103 keeps 30% but ends below 15; 0105 sits on the floor. + assert gate["n_districts"] == 5 + assert _districts(gate["collapsed"]) == ["0101"] + assert _districts(gate["below_floor"]) == ["0103"] + assert (gate["n_collapsed"], gate["n_below_floor"]) == (1, 1) + assert gate["collapsed"][0] == { + "district": "0101", + "calibrated_ess": 20.0, + "design_ess": 100.0, + "ratio": 0.2, + } + assert gate["criteria_met"] is False + assert (gate["min_calibrated_ess"], gate["min_calibrated_ess_district"]) == ( + 12.0, + "0103", + ) + assert (gate["min_ess_ratio"], gate["min_ess_ratio_district"]) == (0.2, "0101") + assert gate["failures"] == [] + # Report-only by default: the entry passes and says what it measured. + assert gate["passed"] is True + assert (gate["blocking"], gate["report_only"]) == (False, True) + assert (gate["relative_floor"], gate["absolute_floor"]) == (0.25, 15.0) + + +def test_collapsed_districts_are_listed_worst_first() -> None: + gate = _GATE( + _origin( + {"a": 100.0, "b": 100.0, "c": 100.0}, + {"a": 20.0, "b": 10.0, "c": 20.0}, + ), + absolute_floor=0.0, + ) + assert _districts(gate["collapsed"]) == ["b", "a", "c"] + + +def test_a_blocking_gate_passes_only_when_no_district_collapses() -> None: + collapsed = _origin({"1": 100.0, "2": 50.0}, {"1": 10.0, "2": 40.0}) + clean = _origin({"1": 100.0, "2": 50.0}, {"1": 90.0, "2": 40.0}) + below_floor = _origin({"1": 100.0, "2": 12.0}, {"1": 90.0, "2": 11.0}) + + failing = _GATE(collapsed, blocking=True) + assert (failing["passed"], failing["criteria_met"]) == (False, False) + assert (failing["blocking"], failing["report_only"]) == (True, False) + assert _GATE(below_floor, blocking=True)["passed"] is False + passing = _GATE(clean, blocking=True) + assert (passing["passed"], passing["criteria_met"]) == (True, True) + + +def test_a_zero_floor_turns_its_check_off() -> None: + origin = _origin({"1": 100.0, "2": 10.0}, {"1": 1.0, "2": 9.0}) + + assert (_GATE(origin)["n_collapsed"], _GATE(origin)["n_below_floor"]) == (1, 2) + relative_off = _GATE(origin, relative_floor=0.0) + assert (relative_off["n_collapsed"], relative_off["n_below_floor"]) == (0, 2) + absolute_off = _GATE(origin, absolute_floor=0.0) + assert (absolute_off["n_collapsed"], absolute_off["n_below_floor"]) == (1, 0) + both_off = _GATE(origin, relative_floor=0.0, absolute_floor=0.0, blocking=True) + assert (both_off["criteria_met"], both_off["passed"]) == (True, True) + + +def test_a_district_with_no_design_ess_cannot_collapse() -> None: + gate = _GATE(_origin({"1": 0.0, "2": 50.0}, {"1": 0.0, "2": 40.0})) + + assert gate["collapsed"] == [] + assert _districts(gate["below_floor"]) == ["1"] + assert gate["below_floor"][0]["ratio"] is None + # The smallest ratio is taken over districts that have one. + assert (gate["min_ess_ratio"], gate["min_ess_ratio_district"]) == (0.8, "2") + + +def test_integer_ess_values_from_json_are_read_as_numbers() -> None: + gate = _GATE(_origin({"1": 100, "2": 40}, {"1": 20, "2": 40})) + + assert (gate["n_collapsed"], gate["n_below_floor"]) == (1, 0) + assert gate["collapsed"][0]["calibrated_ess"] == 20.0 + + +@pytest.mark.parametrize( + "origin", + [ + None, + {}, + "not a mapping", + # A summary written before weight_origin recorded districts. + {"design": {"rows": 3}, "calibrated": {"rows": 3}}, + {"design": {"effective_sample_size_by_district": {"1": 10.0}}}, + _origin({}, {}), + _origin({"1": 10.0}, {"2": 10.0}), + _origin({"1": 10.0}, {"1": math.nan}), + _origin({"1": math.inf}, {"1": 10.0}), + _origin({"1": 10.0}, {"1": -1.0}), + _origin({"1": 10.0}, {"1": "10"}), + _origin({"1": 10.0}, {"1": True}), + _origin({"1": 10.0}, {"1": None}), + _origin([("1", 10.0)], {"1": 10.0}), + {"design": [10.0], "calibrated": "10"}, + ], + ids=[ + "none", + "empty", + "not-a-mapping", + "legacy-summary", + "no-calibrated", + "no-districts", + "different-districts", + "nan", + "infinite", + "negative", + "string", + "bool", + "null", + "list", + "blocks-not-mappings", + ], +) +def test_missing_or_malformed_evidence_fails_only_a_blocking_gate(origin) -> None: + report_only = _GATE(origin) + assert report_only["criteria_met"] is None + assert report_only["failures"] + assert report_only["passed"] is True + assert "n_collapsed" not in report_only + + blocking = _GATE(origin, blocking=True) + assert blocking["criteria_met"] is None + assert blocking["passed"] is False + + +@pytest.mark.parametrize( + "floors", + [ + {"relative_floor": -0.1}, + {"relative_floor": 1.5}, + {"relative_floor": math.nan}, + {"absolute_floor": -1.0}, + {"absolute_floor": math.inf}, + {"absolute_floor": math.nan}, + ], +) +def test_floors_out_of_range_are_refused(floors) -> None: + with pytest.raises(ValueError, match="floor must be"): + _GATE(_origin({"1": 10.0}, {"1": 10.0}), **floors) + + +def test_the_limitation_states_the_result_and_whether_it_blocks() -> None: + collapsed = _origin({"0601": 100.0, "0602": 80.0}, {"0601": 20.0, "0602": 60.0}) + clean = _origin({"0601": 100.0, "0602": 80.0}, {"0601": 90.0, "0602": 60.0}) + + report_only = _LIMITATION(_GATE(collapsed)) + assert report_only["id"] == "district_effective_sample_size_gate" + assert report_only["status"] == "recorded_concentration" + assert report_only["gate"] == _TOOL.DISTRICT_ESS_GATE + assert report_only["reason"] == ( + "Congressional districts with a calibrated Kish ESS below 0.25 x their " + "design-weight Kish ESS: 1 of 2; below 15: 0 of 2. Smallest calibrated " + "district ESS: 20.0 (0601); smallest ratio to design ESS: 0.200 (0601). " + "The gate is report-only for this build; --district-ess-gate-blocking " + "makes it a hard failure." + ) + assert (report_only["n_collapsed"], report_only["n_below_floor"]) == (1, 0) + assert report_only["criteria_met"] is False + assert report_only["blocking"] is False + assert report_only["calibration_blocker"] is False + + blocked = _LIMITATION(_GATE(collapsed, blocking=True)) + assert blocked["reason"].endswith( + "The gate is blocking for this build and fails, so the build is not " + "simulation-ready." + ) + assert blocked["calibration_blocker"] is True + + passed = _LIMITATION(_GATE(clean, blocking=True)) + assert passed["reason"].endswith("The gate is blocking for this build and passed.") + assert passed["calibration_blocker"] is False + + unavailable = _LIMITATION(_GATE(None)) + assert unavailable["reason"].startswith( + "The district ESS gate could not be evaluated: calibration_summary.json " + "weight_origin records no" + ) + assert unavailable["criteria_met"] is None + assert unavailable["n_collapsed"] is None + assert unavailable["calibration_blocker"] is False + assert _LIMITATION(_GATE(None, blocking=True))["calibration_blocker"] is True + + unchecked = _LIMITATION(_GATE(collapsed, relative_floor=0.0, absolute_floor=0.0)) + assert unchecked["reason"].startswith( + "No district ESS check is on for the 2 districts. Smallest calibrated " + "district ESS: 20.0 (0601)" + ) + relative_only = _LIMITATION(_GATE(collapsed, absolute_floor=0.0)) + assert relative_only["reason"].startswith( + "Congressional districts with a calibrated Kish ESS below 0.25 x their " + "design-weight Kish ESS: 1 of 2. Smallest" + ) + + +def test_the_release_contract_publishes_a_report_only_gate_whatever_it_measured(): + """Every local-area gate must pass to publish, so report-only must pass.""" + + from microcosm.data import contract + + collapsed = _origin({"1": 100.0}, {"1": 10.0}) + for blocking, expected in ((False, 0), (True, 1)): + failures: list[str] = [] + gate = _GATE(collapsed, blocking=blocking) + contract._check_local_area_gates( + {"gates": {_TOOL.DISTRICT_ESS_GATE: gate}}, failures + ) + assert len(failures) == expected, failures + + +def _parse(tmp_path, *extra: str): + return _TOOL._parse_args( + [ + "--stage", + "finalize", + "--staging-h5", + str(tmp_path / "staging.h5"), + "--checkpoint-dir", + str(tmp_path / "ckpt"), + "--out-h5", + str(tmp_path / "out.h5"), + *extra, + ] + ) + + +def test_the_gate_is_report_only_unless_blocking_is_asked_for(tmp_path) -> None: + args = _parse(tmp_path) + assert args.district_ess_relative_floor == 0.25 + assert args.district_ess_floor == 15.0 + assert args.district_ess_gate_blocking is False + + args = _parse( + tmp_path, + "--district-ess-relative-floor", + "0.5", + "--district-ess-floor", + "0", + "--district-ess-gate-blocking", + ) + assert (args.district_ess_relative_floor, args.district_ess_floor) == (0.5, 0.0) + assert args.district_ess_gate_blocking is True + + +@pytest.mark.parametrize( + ("flag", "value"), + [ + ("--district-ess-relative-floor", "-0.1"), + ("--district-ess-relative-floor", "1.5"), + ("--district-ess-relative-floor", "nan"), + ("--district-ess-floor", "-1"), + ("--district-ess-floor", "inf"), + ("--district-ess-floor", "nan"), + ], +) +def test_floors_out_of_range_are_refused_at_parse_time( + tmp_path, capsys, flag, value +) -> None: + with pytest.raises(SystemExit): + _parse(tmp_path, flag, value) + assert flag in capsys.readouterr().err + + +# --------------------------------------------------------------------------- +# Properties +# --------------------------------------------------------------------------- + + +@st.composite +def _solves(draw): + """Design and calibrated household weights over a few districts. + + Calibrated weights are the design weights times a per-household factor in + [0, 5] (the release's weight cap), so a whole district can go to zero. + """ + + n = draw(st.integers(min_value=1, max_value=80)) + n_districts = draw(st.integers(min_value=1, max_value=6)) + codes = draw( + st.lists( + st.integers(min_value=0, max_value=n_districts - 1), + min_size=n, + max_size=n, + ) + ) + design = draw( + st.lists( + st.floats(min_value=0.5, max_value=1_000.0, allow_nan=False), + min_size=n, + max_size=n, + ) + ) + factors = draw( + st.lists( + st.floats(min_value=0.0, max_value=5.0, allow_nan=False), + min_size=n, + max_size=n, + ) + ) + design = np.asarray(design, dtype=np.float64) + calibrated = design * np.asarray(factors, dtype=np.float64) + return design, calibrated, np.asarray([f"{code:04d}" for code in codes]) + + +def _weight_origin(design, calibrated, codes) -> dict: + """``weight_origin`` exactly as ``calibration_evidence`` records it.""" + + return { + "design": _CD.weight_origin_summary( + design, spine=None, source_id=None, district=codes + ), + "calibrated": _CD.weight_origin_summary( + calibrated, spine=None, source_id=None, district=codes + ), + } + + +_FLOORS = st.floats(min_value=0.0, max_value=1.0, allow_nan=False) +_ABSOLUTE_FLOORS = st.floats(min_value=0.0, max_value=200.0, allow_nan=False) + + +@_SETTINGS +@given(solve=_solves(), floors=st.tuples(_FLOORS, _FLOORS)) +def test_collapsed_districts_are_monotone_in_the_relative_floor(solve, floors): + low, high = sorted(floors) + origin = _weight_origin(*solve) + at_low = _GATE(origin, relative_floor=low) + at_high = _GATE(origin, relative_floor=high) + assert set(_districts(at_low["collapsed"])) <= set(_districts(at_high["collapsed"])) + assert at_low["n_collapsed"] <= at_high["n_collapsed"] + + +@_SETTINGS +@given(solve=_solves(), floors=st.tuples(_ABSOLUTE_FLOORS, _ABSOLUTE_FLOORS)) +def test_districts_below_the_floor_are_monotone_in_the_floor(solve, floors): + low, high = sorted(floors) + origin = _weight_origin(*solve) + at_low = _GATE(origin, absolute_floor=low) + at_high = _GATE(origin, absolute_floor=high) + assert set(_districts(at_low["below_floor"])) <= set( + _districts(at_high["below_floor"]) + ) + assert at_low["n_below_floor"] <= at_high["n_below_floor"] + + +@_SETTINGS +@given(solve=_solves(), relative_floor=_FLOORS, floor=_ABSOLUTE_FLOORS) +def test_a_design_weight_solve_collapses_no_district(solve, relative_floor, floor): + design, _, codes = solve + gate = _GATE( + _weight_origin(design, design, codes), + relative_floor=relative_floor, + absolute_floor=floor, + ) + assert gate["n_collapsed"] == 0 + # The absolute floor is not relative: it flags the design's own thin + # districts, and only those. + design_ess = _CD.ess_by_group(design, codes) + assert set(_districts(gate["below_floor"])) == { + district for district, ess in design_ess.items() if ess < floor + } + + +@_SETTINGS +@given( + solve=_solves(), + scale=st.floats(min_value=0.01, max_value=100.0, allow_nan=False), + relative_floor=st.floats(min_value=0.0, max_value=0.99, allow_nan=False), +) +def test_a_uniform_rescaling_of_the_design_weights_collapses_no_district( + solve, scale, relative_floor +): + design, _, codes = solve + gate = _GATE( + _weight_origin(design, design * scale, codes), relative_floor=relative_floor + ) + assert gate["n_collapsed"] == 0 + + +@_SETTINGS +@given(solve=_solves(), relative_floor=_FLOORS, floor=_ABSOLUTE_FLOORS) +def test_counts_are_bounded_by_the_districts(solve, relative_floor, floor): + _, _, codes = solve + gate = _GATE( + _weight_origin(*solve), relative_floor=relative_floor, absolute_floor=floor + ) + assert gate["n_districts"] == len(set(codes.tolist())) + assert 0 <= gate["n_collapsed"] <= gate["n_districts"] + assert 0 <= gate["n_below_floor"] <= gate["n_districts"] + assert gate["n_collapsed"] == len(gate["collapsed"]) + assert gate["n_below_floor"] == len(gate["below_floor"]) + assert gate["criteria_met"] is ( + gate["n_collapsed"] == 0 and gate["n_below_floor"] == 0 + ) + # The entry is what gate_summary.json stores, unchanged by a round trip. + assert json.loads(json.dumps(gate)) == gate + + +@_SETTINGS +@given(solve=_solves(), relative_floor=_FLOORS, floor=_ABSOLUTE_FLOORS) +def test_blocking_changes_only_whether_the_gate_passes(solve, relative_floor, floor): + origin = _weight_origin(*solve) + report_only = _GATE(origin, relative_floor=relative_floor, absolute_floor=floor) + blocking = _GATE( + origin, relative_floor=relative_floor, absolute_floor=floor, blocking=True + ) + switched = {"passed", "blocking", "report_only"} + assert {k: v for k, v in report_only.items() if k not in switched} == { + k: v for k, v in blocking.items() if k not in switched + } + assert report_only["passed"] is True + assert blocking["passed"] is report_only["criteria_met"] + assert _LIMITATION(blocking)["calibration_blocker"] is not blocking["passed"] + + +def _reference_ess(weights: np.ndarray, index: np.ndarray, groups: int) -> np.ndarray: + total = np.bincount(index, weights=weights, minlength=groups) + square = np.bincount(index, weights=weights * weights, minlength=groups) + ess = np.zeros(groups) + positive = square > 0 + ess[positive] = total[positive] ** 2 / square[positive] + return ess + + +@_SETTINGS +@given(solve=_solves(), relative_floor=_FLOORS) +def test_the_gate_agrees_with_a_reference_from_the_raw_weights(solve, relative_floor): + design, calibrated, codes = solve + labels, index = np.unique(codes, return_inverse=True) + design_ess = _reference_ess(design, index, len(labels)) + calibrated_ess = _reference_ess(calibrated, index, len(labels)) + threshold = relative_floor * design_ess + expected = { + str(label) + for label, c, t in zip(labels, calibrated_ess, threshold, strict=True) + if c < t + } + # Districts within rounding of the threshold may fall either way: the two + # computations sum in a different order. + ties = { + str(label) + for label, c, t in zip(labels, calibrated_ess, threshold, strict=True) + if abs(c - t) <= 1e-9 * (1.0 + t) + } + gate = _GATE( + _weight_origin(design, calibrated, codes), relative_floor=relative_floor + ) + assert set(_districts(gate["collapsed"])) - ties == expected - ties diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_release_tool.py b/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_release_tool.py index a8c4eea70..1e07e4e6b 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_release_tool.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_release_tool.py @@ -1207,6 +1207,221 @@ def changed_artifact(frame, **kwargs): assert not args.out_summary.exists() +# --------------------------------------------------------------------------- +# District ESS gate: report-only by default, blocking on request +# --------------------------------------------------------------------------- + +#: One district calibrated down to a fifth of its design ESS, one healthy. +_COLLAPSED_DISTRICT_ESS = ( + {"0601": 100.0, "0602": 80.0}, + {"0601": 20.0, "0602": 60.0}, +) +_HEALTHY_DISTRICT_ESS = ( + {"0601": 100.0, "0602": 80.0}, + {"0601": 90.0, "0602": 60.0}, +) + + +def _ess_origin(district_ess) -> dict: + """A calibration summary's ``weight_origin`` carrying per-district ESS.""" + + design, calibrated = district_ess + return { + "design": {"effective_sample_size_by_district": design}, + "calibrated": {"effective_sample_size_by_district": calibrated}, + } + + +def _with_district_ess(args, district_ess) -> None: + """Record per-district ESS in the finalize fixture's calibration summary.""" + + path = args.checkpoint_dir / "calibration_summary.json" + summary = json.loads(path.read_text()) + summary["weight_origin"] = _ess_origin(district_ess) + path.write_text(json.dumps(summary)) + + +def _district_ess_limitation(report: dict) -> dict: + [entry] = [ + item + for item in report["reviewed_limitations"] + if item["id"] == "district_effective_sample_size_gate" + ] + return entry + + +def test_finalize_records_a_collapsed_district_without_blocking_by_default( + tmp_path, monkeypatch +) -> None: + module = _load_tool_module() + args = _finalize_args(module, tmp_path) + _with_district_ess(args, _COLLAPSED_DISTRICT_ESS) + # The fixture has no QA evidence, so finalize stops on consumer_ready + # after writing its report. + message, report = _run_finalize(module, monkeypatch, args, _plausible_hours_frame()) + assert module.DISTRICT_ESS_GATE not in message + gate = report["gates"][module.DISTRICT_ESS_GATE] + assert (gate["passed"], gate["report_only"], gate["criteria_met"]) == ( + True, + True, + False, + ) + assert [row["district"] for row in gate["collapsed"]] == ["0601"] + summary = json.loads(args.out_summary.read_text()) + assert module.DISTRICT_ESS_GATE not in summary["simulation_readiness_blockers"] + entry = _district_ess_limitation(report) + assert entry["n_collapsed"] == 1 + assert entry["calibration_blocker"] is False + assert "report-only" in entry["reason"] + assert summary["reviewed_limitations"] == report["reviewed_limitations"] + + +def test_blocking_finalize_fails_on_a_collapsed_district(tmp_path, monkeypatch) -> None: + module = _load_tool_module() + args = _finalize_args(module, tmp_path) + args.district_ess_gate_blocking = True + _with_district_ess(args, _COLLAPSED_DISTRICT_ESS) + message, report = _run_finalize(module, monkeypatch, args, _plausible_hours_frame()) + assert module.DISTRICT_ESS_GATE in message + gate = report["gates"][module.DISTRICT_ESS_GATE] + assert (gate["passed"], gate["blocking"]) == (False, True) + summary = json.loads(args.out_summary.read_text()) + assert summary["simulation_ready"] is False + assert module.DISTRICT_ESS_GATE in summary["simulation_readiness_blockers"] + assert _district_ess_limitation(report)["calibration_blocker"] is True + + +def test_blocking_finalize_fails_closed_without_district_evidence( + tmp_path, monkeypatch +) -> None: + """A summary that records no per-district ESS cannot pass a blocking gate.""" + + module = _load_tool_module() + args = _finalize_args(module, tmp_path) + args.district_ess_gate_blocking = True + message, report = _run_finalize(module, monkeypatch, args, _plausible_hours_frame()) + assert module.DISTRICT_ESS_GATE in message + gate = report["gates"][module.DISTRICT_ESS_GATE] + assert (gate["passed"], gate["criteria_met"]) == (False, None) + + +def _blocking_package_args(module, tmp_path: Path, gate: dict | None): + tmp_path.mkdir(parents=True, exist_ok=True) + args = _package_args_before_evidence(module, tmp_path, max_households=None) + args.district_ess_gate_blocking = True + gates = {} if gate is None else {module.DISTRICT_ESS_GATE: gate} + (args.checkpoint_dir / "gate_summary.json").write_text(json.dumps({"gates": gates})) + return args + + +@pytest.mark.parametrize( + "recorded", + ["absent", "report_only", "blocking_failed", "other_relative", "other_floor"], +) +def test_blocking_package_refuses_a_report_finalize_did_not_block_on( + tmp_path: Path, recorded: str +) -> None: + module = _load_tool_module() + gate = module.district_ess_collapse_gate + healthy = _ess_origin(_HEALTHY_DISTRICT_ESS) + entry = { + "absent": None, + # A report-only gate passes whatever it measured. + "report_only": gate(_ess_origin(_COLLAPSED_DISTRICT_ESS)), + "blocking_failed": gate(_ess_origin(_COLLAPSED_DISTRICT_ESS), blocking=True), + "other_relative": gate(healthy, relative_floor=0.1, blocking=True), + "other_floor": gate(healthy, absolute_floor=5.0, blocking=True), + }[recorded] + args = _blocking_package_args(module, tmp_path, entry) + with pytest.raises(SystemExit, match="--district-ess-gate-blocking"): + module.do_package(args) + assert not (args.out / "releases").exists(), "a refusal leaves no release" + + +def test_package_accepts_a_passing_blocking_gate_and_ignores_it_otherwise( + tmp_path: Path, +) -> None: + """Either way the run reaches the next check (missing QA evidence).""" + + module = _load_tool_module() + passing = module.district_ess_collapse_gate( + _ess_origin(_HEALTHY_DISTRICT_ESS), blocking=True + ) + args = _blocking_package_args(module, tmp_path / "blocking", passing) + with pytest.raises(SystemExit, match="spine_qa.json is missing"): + module.do_package(args) + + report_only = module.district_ess_collapse_gate( + _ess_origin(_COLLAPSED_DISTRICT_ESS) + ) + args = _blocking_package_args(module, tmp_path / "report-only", report_only) + args.district_ess_gate_blocking = False + with pytest.raises(SystemExit, match="spine_qa.json is missing"): + module.do_package(args) + + +@requires_pytables +@pytest.mark.parametrize( + ("blocking", "district_ess", "criteria_met"), + [ + (False, _COLLAPSED_DISTRICT_ESS, False), + (True, _HEALTHY_DISTRICT_ESS, True), + ], + ids=["report-only-collapse", "blocking-pass"], +) +def test_the_district_ess_gate_ships_in_both_manifests( + tmp_path, monkeypatch, blocking, district_ess, criteria_met +) -> None: + """finalize -> package records the gate in gate_summary.json and the build + manifest, its result in the release manifest's limitations, and the + release contract accepts the gate either way.""" + + from microcosm.data import contract + + module = _load_tool_module() + args = _finalize_args(module, tmp_path) + args.district_ess_gate_blocking = blocking + _with_district_ess(args, district_ess) + _write_frame_h5(args.out_h5, _plausible_hours_frame()) + artifact_sha = module._sha256(args.out_h5) + evidence = { + "run_identity.json": { + "staging_sha256": module._sha256(args.staging_h5), + "ladder_sha256": module._sha256(args.ladder), + "population_cells_dropped": [], + "sampling": _FULL_RUNG, + }, + "spine_qa.json": { + "plain_consumption": True, + "artifact_sha256": artifact_sha, + "per_spine": {}, + }, + "consumer_export.json": {"staging_sha256": artifact_sha}, + "consumer_reviewed_null_fills.json": {"columns_filled": []}, + } + for name, value in evidence.items(): + (args.checkpoint_dir / name).write_text(json.dumps(value)) + _patch_finalize_collaborators(module, monkeypatch, identity=False) + module.do_finalize(args) + + args.out = tmp_path / "release" + args.allow_dirty = True + release_dir = Path(module.do_package(args)["release_dir"]) + gate_summary = json.loads((release_dir / "gate_summary.json").read_text()) + build_manifest = json.loads((release_dir / "build_manifest.json").read_text()) + release_manifest = json.loads((release_dir / "release_manifest.json").read_text()) + shipped = gate_summary["gates"][module.DISTRICT_ESS_GATE] + assert build_manifest["gates"][module.DISTRICT_ESS_GATE] == shipped + assert (shipped["passed"], shipped["blocking"]) == (True, blocking) + assert shipped["criteria_met"] is criteria_met + entry = _district_ess_limitation(release_manifest) + assert entry["criteria_met"] is criteria_met + assert entry["calibration_blocker"] is False + failures: list[str] = [] + contract._check_local_area_gates(gate_summary, failures) + assert not [f for f in failures if module.DISTRICT_ESS_GATE in f], failures + + @requires_pytables def test_finalize_report_round_trips_into_package(tmp_path, monkeypatch) -> None: """A finalize-written report must satisfy the package stage's binding.""" diff --git a/tools/build_us_acs_local_release.py b/tools/build_us_acs_local_release.py index 7c3cf446c..1ae2796d4 100644 --- a/tools/build_us_acs_local_release.py +++ b/tools/build_us_acs_local_release.py @@ -37,7 +37,9 @@ per-spine SSI incidence and intensity (the microcosm#403 signature, measured rather than assumed). finalize : release-style gate report (PUMA-ladder gate, calibration - gate, spine-composition evidence) + the reviewed-limitations + gate, district ESS collapse gate -- report-only unless + ``--district-ess-gate-blocking`` --, spine-composition + evidence) + the reviewed-limitations register for this lineage (inherited #507 SSI aged-band collapse, #393 miscellaneous-income defect, CD marginal vintage, ESS concentration, sparse-selection donor, mixed @@ -2458,6 +2460,247 @@ def state_cd_reviewed_limitations(materialize_rss: dict) -> list[dict]: ] +#: Finalize's district ESS gate. A congressional district collapses when +#: calibration leaves its Kish ESS below this share of its Kish ESS at the +#: design weights. At the release's ACS share (0.5), microcosm#1078's solves +#: (full surface and two holdout folds) leave 15-24 districts below a quarter +#: unpenalized, and none with the chi-square design-weight penalty. +DISTRICT_ESS_RELATIVE_FLOOR = 0.25 +#: The absolute floor on a district's calibrated Kish ESS (0 turns it off). +#: Those solves' smallest district ESS is 7.8-11.7 unpenalized and at least +#: 18.6 penalized. +DISTRICT_ESS_FLOOR = 15.0 +DISTRICT_ESS_GATE = "district_ess_collapse" + + +def _district_ess(weight_summary) -> dict[str, float] | None: + """A weight summary's per-district Kish ESS as finite, nonnegative floats. + + None when the summary records none, or any value is not such a number. + """ + + by_district = ( + weight_summary.get("effective_sample_size_by_district") + if isinstance(weight_summary, dict) + else None + ) + if not isinstance(by_district, dict) or not by_district: + return None + values = {} + for district, ess in by_district.items(): + if isinstance(ess, bool) or not isinstance(ess, (int, float)): + return None + if not np.isfinite(ess) or ess < 0: + return None + values[str(district)] = float(ess) + return values + + +def district_ess_collapse_gate( + weight_origin: dict | None, + *, + relative_floor: float = DISTRICT_ESS_RELATIVE_FLOOR, + absolute_floor: float = DISTRICT_ESS_FLOOR, + blocking: bool = False, +) -> dict: + """The district ESS gate entry, from the calibration summary's weight origin. + + Reads ``effective_sample_size_by_district`` at the design and the + calibrated weights. A district collapses when its calibrated Kish ESS is + below ``relative_floor`` times its design-weight Kish ESS, and is below + the floor when its calibrated Kish ESS is below ``absolute_floor``; a + floor of 0 turns its check off. ``criteria_met`` says whether no district + does either (None when the evidence is missing or malformed). + + The release contract publishes only gates that pass, so a report-only + entry passes whatever it measured. Under ``blocking`` the entry passes + only when the criteria are met. + """ + + if not 0.0 <= relative_floor <= 1.0: + raise ValueError(f"relative_floor must be in [0, 1], got {relative_floor!r}.") + if not (np.isfinite(absolute_floor) and absolute_floor >= 0.0): + raise ValueError( + f"absolute_floor must be finite and >= 0, got {absolute_floor!r}." + ) + origin = weight_origin if isinstance(weight_origin, dict) else {} + design = _district_ess(origin.get("design")) + calibrated = _district_ess(origin.get("calibrated")) + gate = { + "blocking": bool(blocking), + "report_only": not blocking, + "relative_floor": float(relative_floor), + "absolute_floor": float(absolute_floor), + "note": ( + "A congressional district collapses when its calibrated Kish ESS " + "is below relative_floor x its design-weight Kish ESS, and is " + "below the floor when its calibrated Kish ESS is below " + "absolute_floor (0 turns a check off); from " + "calibration_summary.json weight_origin." + ), + } + failures = [] + if design is None or calibrated is None: + failures.append( + "calibration_summary.json weight_origin records no finite, " + "nonnegative effective_sample_size_by_district at both the design " + "and the calibrated weights" + ) + elif set(design) != set(calibrated): + failures.append( + "the design and calibrated per-district ESS cover different districts" + ) + if failures: + return { + **gate, + "passed": not blocking, + "criteria_met": None, + "failures": failures, + } + + rows = [ + { + "district": district, + "calibrated_ess": calibrated[district], + "design_ess": design[district], + "ratio": ( + calibrated[district] / design[district] + if design[district] > 0 + else None + ), + } + for district in sorted(design) + ] + collapsed = sorted( + ( + row + for row in rows + if row["calibrated_ess"] < relative_floor * row["design_ess"] + ), + key=lambda row: (row["ratio"], row["district"]), + ) + below_floor = sorted( + (row for row in rows if row["calibrated_ess"] < absolute_floor), + key=lambda row: (row["calibrated_ess"], row["district"]), + ) + smallest = min(rows, key=lambda row: (row["calibrated_ess"], row["district"])) + rated = [row for row in rows if row["ratio"] is not None] + lowest = ( + min(rated, key=lambda row: (row["ratio"], row["district"])) if rated else None + ) + criteria_met = not collapsed and not below_floor + return { + **gate, + "passed": criteria_met or not blocking, + "criteria_met": criteria_met, + "failures": [], + "n_districts": len(rows), + "n_collapsed": len(collapsed), + "n_below_floor": len(below_floor), + "min_calibrated_ess": smallest["calibrated_ess"], + "min_calibrated_ess_district": smallest["district"], + "min_ess_ratio": lowest["ratio"] if lowest else None, + "min_ess_ratio_district": lowest["district"] if lowest else None, + "collapsed": collapsed, + "below_floor": below_floor, + } + + +def district_ess_collapse_limitation(gate: dict) -> dict: + """The reviewed-limitations entry stating the district ESS gate's result.""" + + if gate["criteria_met"] is None: + measured = ( + "The district ESS gate could not be evaluated: " + + "; ".join(gate["failures"]) + + "." + ) + else: + n = gate["n_districts"] + checks = [] + if gate["relative_floor"] > 0: + checks.append( + f"below {gate['relative_floor']:g} x their design-weight Kish " + f"ESS: {gate['n_collapsed']} of {n}" + ) + if gate["absolute_floor"] > 0: + checks.append( + f"below {gate['absolute_floor']:g}: {gate['n_below_floor']} of {n}" + ) + ratio = gate["min_ess_ratio"] + measured = ( + ( + "Congressional districts with a calibrated Kish ESS " + + "; ".join(checks) + + "." + if checks + else f"No district ESS check is on for the {n} districts." + ) + + f" Smallest calibrated district ESS: " + f"{gate['min_calibrated_ess']:.1f} " + f"({gate['min_calibrated_ess_district']})" + + ( + f"; smallest ratio to design ESS: {ratio:.3f} " + f"({gate['min_ess_ratio_district']})." + if ratio is not None + else "." + ) + ) + if not gate["blocking"]: + mode = ( + " The gate is report-only for this build; " + "--district-ess-gate-blocking makes it a hard failure." + ) + elif gate["passed"]: + mode = " The gate is blocking for this build and passed." + else: + mode = ( + " The gate is blocking for this build and fails, so the build is " + "not simulation-ready." + ) + return { + "id": "district_effective_sample_size_gate", + "status": "recorded_concentration", + "reason": measured + mode, + "gate": DISTRICT_ESS_GATE, + "criteria_met": gate["criteria_met"], + "n_collapsed": gate.get("n_collapsed"), + "n_below_floor": gate.get("n_below_floor"), + "relative_floor": gate["relative_floor"], + "absolute_floor": gate["absolute_floor"], + "blocking": gate["blocking"], + "calibration_blocker": not gate["passed"], + } + + +def _require_blocking_district_ess_gate(gate_report: dict, args) -> None: + """Under ``--district-ess-gate-blocking``, refuse a report finalize did not block on. + + A finalize run without the flag (or at other floors, or from before the + gate existed) records a gate that passes whatever it measured, so + packaging its report under the flag would ship what the flag refuses. + """ + + if not args.district_ess_gate_blocking: + return + gates = gate_report.get("gates") + gate = gates.get(DISTRICT_ESS_GATE) if isinstance(gates, dict) else None + if not ( + isinstance(gate, dict) + and gate.get("blocking") is True + and gate.get("passed") is True + and gate.get("relative_floor") == args.district_ess_relative_floor + and gate.get("absolute_floor") == args.district_ess_floor + ): + raise SystemExit( + f"--district-ess-gate-blocking: the finalize gate report carries no " + f"passing blocking {DISTRICT_ESS_GATE} gate at relative floor " + f"{args.district_ess_relative_floor:g} and floor " + f"{args.district_ess_floor:g}; re-run --stage finalize with the " + "same flags." + ) + + def _local_hours_gate(frame, staging_summary: dict): audit = staging_summary.get("reviewed_engine_input_nulls") if not isinstance(audit, list) or not all(isinstance(item, dict) for item in audit): @@ -2707,8 +2950,16 @@ def do_finalize(args) -> None: "artifact_sha256": spine_qa.get("artifact_sha256"), } + gates[DISTRICT_ESS_GATE] = district_ess_collapse_gate( + diagnostics.get("weight_origin"), + relative_floor=args.district_ess_relative_floor, + absolute_floor=args.district_ess_floor, + blocking=args.district_ess_gate_blocking, + ) + limitations = finalize_reviewed_limitations(staging_summary, diagnostics, spine_qa) limitations += state_cd_reviewed_limitations(materialize_rss) + limitations.append(district_ess_collapse_limitation(gates[DISTRICT_ESS_GATE])) hard_failures = [ name for name in ( @@ -2720,6 +2971,9 @@ def do_finalize(args) -> None: ) if not gates[name]["passed"] ] + # Report-only unless --district-ess-gate-blocking, so it passes otherwise. + if not gates[DISTRICT_ESS_GATE]["passed"]: + hard_failures.append(DISTRICT_ESS_GATE) updated_summary = dict(staging_summary) updated_summary.update( { @@ -2897,6 +3151,7 @@ def do_package(args) -> dict: soi_mode = _require_recorded_soi_mode(materialize_rss) _require_full_rung(identity) _require_current_calibration(identity, diagnostics, stage="package") + _require_blocking_district_ess_gate(gate_report, args) calibrated_h5 = Path(args.out_h5) if not calibrated_h5.exists(): raise SystemExit(f"Calibrated H5 not found: {calibrated_h5}.") @@ -3382,6 +3637,35 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: "build: materialize hard-fails on dropped cells without this." ), ) + parser.add_argument( + "--district-ess-relative-floor", + type=float, + default=DISTRICT_ESS_RELATIVE_FLOOR, + help=( + "Finalize's district ESS gate: a congressional district collapses " + "when its calibrated Kish ESS is below this share of its " + "design-weight Kish ESS (0 to 1; 0 turns the check off)." + ), + ) + parser.add_argument( + "--district-ess-floor", + type=float, + default=DISTRICT_ESS_FLOOR, + help=( + "Finalize's district ESS gate: the floor on every congressional " + "district's calibrated Kish ESS (0 turns the check off)." + ), + ) + parser.add_argument( + "--district-ess-gate-blocking", + action="store_true", + help=( + "Make the district ESS gate a hard failure at finalize, and make " + "package refuse a gate report finalize did not block on. Off by " + "default: the gate is recorded in gate_summary.json, the build " + "manifest and the reviewed limitations, and passes." + ), + ) args = parser.parse_args(argv) stages = ( @@ -3412,6 +3696,10 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: "--cd-holdout-fraction must be in " f"[0, {cd_surface.MAX_CD_HOLDOUT_FRACTION}]." ) + if not 0.0 <= args.district_ess_relative_floor <= 1.0: + parser.error("--district-ess-relative-floor must be in [0, 1].") + if not (np.isfinite(args.district_ess_floor) and args.district_ess_floor >= 0.0): + parser.error("--district-ess-floor must be finite and >= 0.") args.stages = stages return args