From 0e4bb8c0a1d6bb67b695dcd228a8c9889da194b5 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sat, 22 Aug 2026 18:19:45 +0200 Subject: [PATCH 1/4] Add per-vintage reported-coverage signal gate (#720) Every pooled ASEC source_year must carry at least one reporter for each of the nine reported-coverage person inputs. On the certified Build P artifact the 2022/2023 vintage inputs lacked the NOW_* at-interview recodes (except NOW_GRP/NOW_MRK), so seven of the nine flags were False for two-thirds of the weighted population and reported Medicaid at interview thinned to 24.6M under 65 against ~58M survey. Presence-style checks cannot see a partial-vintage hole; this gate ships red against current bases by design and turns green with the #720 re-extraction. Co-Authored-By: Claude Fable 5 --- ...20-reported-coverage-vintage-gate.added.md | 1 + .../microcosm/build/us_runtime/__init__.py | 6 + .../microcosm/build/us_runtime/cps_carried.py | 100 +++++++ .../tests/test_us_fiscal_refresh_builder.py | 30 +++ .../test_us_reported_coverage_vintage_gate.py | 254 ++++++++++++++++++ tools/build_us_fiscal_refresh_release.py | 65 +++++ 6 files changed, 456 insertions(+) create mode 100644 changelog.d/720-reported-coverage-vintage-gate.added.md create mode 100644 packages/microcosm-build/tests/test_us_reported_coverage_vintage_gate.py diff --git a/changelog.d/720-reported-coverage-vintage-gate.added.md b/changelog.d/720-reported-coverage-vintage-gate.added.md new file mode 100644 index 000000000..b20e9888b --- /dev/null +++ b/changelog.d/720-reported-coverage-vintage-gate.added.md @@ -0,0 +1 @@ +Add `us_reported_coverage_vintage_signal_gate`, a US release gate requiring every pooled ASEC `source_year` to carry at least one reporter for each of the nine reported-coverage person inputs; it closes the partial-vintage hole behind the certified Build P artifact's `has_medicaid_health_coverage_at_interview` thinning (24.6M vs ~58M survey under 65, #720), which presence-style checks cannot see. Wired into `tools/build_us_fiscal_refresh_release.py` as a base-frame hard gate and manifest entry. diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py b/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py index e55690cb5..ee39d8259 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py @@ -162,10 +162,13 @@ CPS_REPORTED_TANF_AMOUNT_RAW_COLUMN, CPS_REPORTED_TANF_TYPE_RAW_COLUMN, CPS_REPORTED_WIC_RAW_COLUMN, + US_REPORTED_COVERAGE_PERSON_INPUTS, + US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS, WIC_CARRIER_ADJUDICATION_URL, derive_us_cps_carried_inputs, reported_tanf_enrollment_by_spm_unit, reported_wic_receipt_carrier, + us_reported_coverage_vintage_signal_gate, ) from microcosm.build.us_runtime.demographics import ( AGE_BANDS, @@ -1095,6 +1098,8 @@ "CPS_REPORTED_TANF_AMOUNT_RAW_COLUMN", "CPS_REPORTED_TANF_TYPE_RAW_COLUMN", "CPS_REPORTED_WIC_RAW_COLUMN", + "US_REPORTED_COVERAGE_PERSON_INPUTS", + "US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS", "WIC_CARRIER_ADJUDICATION_URL", "SimpleTaxExpenditureReform", "ReformValidationSpec", @@ -1852,6 +1857,7 @@ "derive_us_cps_carried_inputs", "reported_tanf_enrollment_by_spm_unit", "reported_wic_receipt_carrier", + "us_reported_coverage_vintage_signal_gate", "disaggregate_us_puf_aggregate_records_from_manifest", "finalize_us_puf_tax_detail_predictions", "hard_target_package_aliases", diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py b/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py index 08d761260..23766a3ac 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py @@ -13,6 +13,7 @@ import numpy as np import pandas as pd +from microcosm.build.gates import GateResult from microcosm.build.us_runtime.alimony import ( US_ASEC_OTHER_INCOME_OUTPUT_COLUMNS, derive_us_alimony_from_asec, @@ -31,11 +32,14 @@ "CPS_REPORTED_TANF_AMOUNT_RAW_COLUMN", "CPS_REPORTED_TANF_TYPE_RAW_COLUMN", "CPS_REPORTED_WIC_RAW_COLUMN", + "US_REPORTED_COVERAGE_PERSON_INPUTS", + "US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS", "WIC_CARRIER_ADJUDICATION_URL", "derive_us_cps_carried_inputs", "reported_snap_receipt_by_spm_unit", "reported_tanf_enrollment_by_spm_unit", "reported_wic_receipt_carrier", + "us_reported_coverage_vintage_signal_gate", ] TAXABLE_INTEREST_FRACTION = 0.680 @@ -50,6 +54,26 @@ "https://github.com/PolicyEngine/microcosm/issues/591#issuecomment-5160668979" ) +# The nine reported-coverage person inputs _fill_health_coverage_inputs derives +# from the ASEC NOW_* at-interview recodes (microcosm #720). +US_REPORTED_COVERAGE_PERSON_INPUTS: tuple[str, ...] = ( + "has_champva_health_coverage_at_interview", + "has_esi", + "has_indian_health_service_coverage_at_interview", + "has_marketplace_health_coverage_at_interview", + "has_medicaid_health_coverage_at_interview", + "has_non_marketplace_direct_purchase_health_coverage_at_interview", + "has_other_means_tested_health_coverage_at_interview", + "has_tricare_health_coverage_at_interview", + "has_va_health_coverage_at_interview", +) + +# Below this row count a vintage is a smoke pool and zero reporters for a rare +# flag is sampling noise; at or above it, a zero is a missing-source symptom +# (the rarest flag, CHAMPVA at ~0.3% of persons, is expected ~15 times in +# 5,000 rows). +US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS = 5_000 + CPS_CARRIED_FORMULA_OWNED_COLUMNS = frozenset( { "capital_gains", @@ -502,3 +526,79 @@ def _integer_source(person: pd.DataFrame, column: str) -> np.ndarray: def _yes_code(person: pd.DataFrame, column: str) -> np.ndarray: return _integer_source(person, column) == 1 + + +def us_reported_coverage_vintage_signal_gate( + frame: Frame, + *, + min_vintage_rows: int = US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS, +) -> GateResult: + """Require every pooled ASEC vintage to carry reported-coverage signal. + + Microcosm #720: the pooled 2022/2023 ASEC source inputs lacked the + ``NOW_*`` at-interview recodes (except ``NOW_GRP``/``NOW_MRK``), so + :func:`derive_us_cps_carried_inputs` silently mapped every 2022/2023 + -vintage person to ``False`` for seven of the nine reported-coverage + flags and the certified Build P artifact reported 24.6M under-65 + Medicaid at interview against ~58M survey. A flag populated for one + vintage passes the presence-style checks (``release_input_coverage``, + ``degenerate_input_signal``); this gate enforces the per-vintage + invariant those checks cannot see: every ``source_year`` with at least + ``min_vintage_rows`` person rows must have at least one reporter for + every reported-coverage input. + + Vintages below ``min_vintage_rows`` (smoke pools) are recorded in the + details but not enforced. A frame without a ``source_year`` column is + checked as a single vintage. + """ + + person = frame.table("person") + missing = [ + column + for column in US_REPORTED_COVERAGE_PERSON_INPUTS + if column not in person.columns + ] + if missing: + return GateResult( + name="reported_coverage_vintage_signal", + passed=False, + failures=tuple(f"person column missing: {column}." for column in missing), + details={"missing": missing}, + ) + if "source_year" in person.columns: + vintage_labels = person["source_year"] + else: + vintage_labels = pd.Series("all", index=person.index) + failures: list[str] = [] + vintages: dict[str, dict[str, object]] = {} + for vintage, group in person.groupby(vintage_labels, sort=True, dropna=False): + label = str(vintage) + rows = int(len(group)) + reporter_counts = { + column: int(group[column].fillna(False).astype(bool).sum()) + for column in US_REPORTED_COVERAGE_PERSON_INPUTS + } + enforced = rows >= min_vintage_rows + vintages[label] = { + "rows": rows, + "enforced": enforced, + "reporter_counts": reporter_counts, + } + if not enforced: + continue + for column, count in reporter_counts.items(): + if count == 0: + failures.append( + f"{column}: source_year {label} has 0 reporters over " + f"{rows} person rows — the vintage source lacks the " + "at-interview recode (microcosm #720)." + ) + return GateResult( + name="reported_coverage_vintage_signal", + passed=not failures, + failures=tuple(failures), + details={ + "min_vintage_rows": int(min_vintage_rows), + "vintages": vintages, + }, + ) diff --git a/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py b/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py index 1c4c4a623..fa6dc5113 100644 --- a/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py +++ b/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py @@ -2237,6 +2237,36 @@ def test_release_gate_failures_include_health_input_signal() -> None: ] +def test_release_gate_failures_include_reported_coverage_vintage_signal() -> None: + builder = _load_builder_module() + result = SimpleNamespace( + skipped=(), + diagnostics=_passing_critical_diagnostics(builder), + initial_loss=10.0, + final_loss=5.0, + ) + reported_coverage_vintage_gate = builder.GateResult( + name="reported_coverage_vintage_signal", + passed=False, + failures=( + "has_medicaid_health_coverage_at_interview: source_year 2022 has 0 " + "reporters over 54464 person rows — the vintage source lacks the " + "at-interview recode (microcosm #720).", + ), + ) + + assert builder._release_gate_failures( + result, + {"dropped_target_names": []}, + reported_coverage_vintage_gate=reported_coverage_vintage_gate, + ) == [ + "Reported-coverage vintage signal failed: " + "has_medicaid_health_coverage_at_interview: source_year 2022 has 0 " + "reporters over 54464 person rows — the vintage source lacks the " + "at-interview recode (microcosm #720).", + ] + + def test_base_population_scale_gate_rejects_underweighted_base(small_frame) -> None: builder = _load_builder_module() diff --git a/packages/microcosm-build/tests/test_us_reported_coverage_vintage_gate.py b/packages/microcosm-build/tests/test_us_reported_coverage_vintage_gate.py new file mode 100644 index 000000000..8dbc7db01 --- /dev/null +++ b/packages/microcosm-build/tests/test_us_reported_coverage_vintage_gate.py @@ -0,0 +1,254 @@ +"""Per-vintage reported-coverage signal gate tests (microcosm #720).""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + +from microcosm.build.us_runtime import ( + CPS_CARRIED_PERSON_INPUTS, + US_REPORTED_COVERAGE_PERSON_INPUTS, + US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS, + us_reported_coverage_vintage_signal_gate, +) +from microcosm.build.us_runtime.cps_carried import _fill_health_coverage_inputs +from microcosm.frame import US_SCHEMA, Frame, WeightKind, Weights + +_ALL_FALSE = {column: False for column in US_REPORTED_COVERAGE_PERSON_INPUTS} + + +def _person_table(rows: list[dict]) -> pd.DataFrame: + records = [] + for index, row in enumerate(rows): + record = {"A_AGE": 40} + record.update(row) + record.setdefault("person_id", index + 1) + record.setdefault("person_household_id", index + 1) + records.append(record) + return pd.DataFrame(records) + + +def _us_frame(person_rows: list[dict]) -> Frame: + return _us_frame_from_person(_person_table(person_rows)) + + +def _us_frame_from_person(person: pd.DataFrame) -> Frame: + person = person.copy() + n = len(person) + household_ids = person["person_household_id"].to_numpy() + unique_households = np.unique(household_ids) + person["person_tax_unit_id"] = person["person_household_id"] + 1_000 + person["person_spm_unit_id"] = person["person_household_id"] + 2_000 + person["person_family_id"] = person["person_household_id"] + 3_000 + person["person_marital_unit_id"] = np.arange(n, dtype="int64") + 4_000 + tables = { + "person": person, + "household": pd.DataFrame({"household_id": unique_households}), + "tax_unit": pd.DataFrame({"tax_unit_id": unique_households + 1_000}), + "spm_unit": pd.DataFrame({"spm_unit_id": unique_households + 2_000}), + "family": pd.DataFrame({"family_id": unique_households + 3_000}), + "marital_unit": pd.DataFrame( + {"marital_unit_id": np.arange(n, dtype="int64") + 4_000} + ), + } + return Frame( + tables, + US_SCHEMA, + { + "household": Weights( + values=np.ones(len(unique_households), dtype=np.float64), + kind=WeightKind.DESIGN, + ) + }, + ) + + +def _vintage_rows( + source_year: int, + *, + reporters: tuple[str, ...] = US_REPORTED_COVERAGE_PERSON_INPUTS, + rows: int = 3, +) -> list[dict]: + """``rows`` persons of one vintage; the first row reports every column in + ``reporters``, the rest report nothing.""" + + out: list[dict] = [] + for index in range(rows): + record = {"source_year": source_year, **_ALL_FALSE} + if index == 0: + record.update({column: True for column in reporters}) + out.append(record) + return out + + +class TestContract: + def test_the_nine_inputs_are_cps_carried_flags(self) -> None: + assert len(US_REPORTED_COVERAGE_PERSON_INPUTS) == 9 + assert set(US_REPORTED_COVERAGE_PERSON_INPUTS) <= CPS_CARRIED_PERSON_INPUTS + assert all( + column == "has_esi" or column.endswith("_coverage_at_interview") + for column in US_REPORTED_COVERAGE_PERSON_INPUTS + ) + + def test_the_nine_inputs_are_exactly_the_derived_health_coverage_flags( + self, + ) -> None: + person = pd.DataFrame({"NOW_GRP": [1, 2]}) + _fill_health_coverage_inputs(person) + derived = {column for column in person.columns if column.startswith("has_")} + assert derived == set(US_REPORTED_COVERAGE_PERSON_INPUTS) + + def test_default_threshold_is_a_release_pool_scale(self) -> None: + assert US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS == 5_000 + + +class TestGate: + def test_every_vintage_with_signal_passes(self) -> None: + frame = _us_frame(_vintage_rows(2023) + _vintage_rows(2024)) + gate = us_reported_coverage_vintage_signal_gate(frame, min_vintage_rows=3) + assert gate.passed, gate.failures + assert gate.name == "reported_coverage_vintage_signal" + vintages = gate.details["vintages"] + assert set(vintages) == {"2023", "2024"} + for summary in vintages.values(): + assert summary["rows"] == 3 + assert summary["enforced"] is True + assert ( + summary["reporter_counts"]["has_medicaid_health_coverage_at_interview"] + == 1 + ) + + def test_vintage_without_medicaid_reporters_fails(self) -> None: + without_medicaid = tuple( + column + for column in US_REPORTED_COVERAGE_PERSON_INPUTS + if column != "has_medicaid_health_coverage_at_interview" + ) + frame = _us_frame( + _vintage_rows(2022, reporters=without_medicaid) + _vintage_rows(2024) + ) + gate = us_reported_coverage_vintage_signal_gate(frame, min_vintage_rows=3) + assert not gate.passed + assert len(gate.failures) == 1 + assert "has_medicaid_health_coverage_at_interview" in gate.failures[0] + assert "source_year 2022" in gate.failures[0] + assert "#720" in gate.failures[0] + + def test_issue_720_shape_fails_seven_flags_for_two_vintages(self) -> None: + # Build P: ASEC 2022/2023 inputs carried only NOW_GRP and NOW_MRK, so + # every other at-interview flag was False for those vintages. + carried = ("has_esi", "has_marketplace_health_coverage_at_interview") + frame = _us_frame( + _vintage_rows(2022, reporters=carried) + + _vintage_rows(2023, reporters=carried) + + _vintage_rows(2024) + ) + gate = us_reported_coverage_vintage_signal_gate(frame, min_vintage_rows=3) + assert not gate.passed + assert len(gate.failures) == 14 + failed_vintages = { + failure.split("source_year ")[1].split(" ")[0] for failure in gate.failures + } + assert failed_vintages == {"2022", "2023"} + assert not any("has_esi" in failure for failure in gate.failures) + assert ( + gate.details["vintages"]["2024"]["reporter_counts"][ + "has_medicaid_health_coverage_at_interview" + ] + == 1 + ) + assert ( + gate.details["vintages"]["2022"]["reporter_counts"][ + "has_medicaid_health_coverage_at_interview" + ] + == 0 + ) + + def test_small_vintage_is_recorded_but_not_enforced(self) -> None: + frame = _us_frame( + _vintage_rows(2022, reporters=(), rows=2) + _vintage_rows(2024) + ) + gate = us_reported_coverage_vintage_signal_gate(frame, min_vintage_rows=3) + assert gate.passed, gate.failures + assert gate.details["vintages"]["2022"]["enforced"] is False + assert gate.details["vintages"]["2024"]["enforced"] is True + assert gate.details["min_vintage_rows"] == 3 + + def test_default_threshold_skips_unit_sized_frames(self) -> None: + frame = _us_frame(_vintage_rows(2022, reporters=())) + gate = us_reported_coverage_vintage_signal_gate(frame) + assert gate.passed, gate.failures + assert gate.details["vintages"]["2022"]["enforced"] is False + + def test_frame_without_source_year_is_one_vintage(self) -> None: + rows = [{**row} for row in _vintage_rows(2024)] + for row in rows: + del row["source_year"] + gate = us_reported_coverage_vintage_signal_gate( + _us_frame(rows), min_vintage_rows=3 + ) + assert gate.passed, gate.failures + assert set(gate.details["vintages"]) == {"all"} + + for row in rows: + row["has_tricare_health_coverage_at_interview"] = False + gate = us_reported_coverage_vintage_signal_gate( + _us_frame(rows), min_vintage_rows=3 + ) + assert not gate.passed + assert "source_year all" in gate.failures[0] + assert "has_tricare_health_coverage_at_interview" in gate.failures[0] + + def test_missing_column_fails(self) -> None: + rows = [{**row} for row in _vintage_rows(2024)] + for row in rows: + del row["has_esi"] + gate = us_reported_coverage_vintage_signal_gate( + _us_frame(rows), min_vintage_rows=3 + ) + assert not gate.passed + assert gate.failures == ("person column missing: has_esi.",) + assert gate.details == {"missing": ["has_esi"]} + + +class TestMechanism: + def test_nan_source_recode_becomes_false_and_fails_the_vintage(self) -> None: + """The #720 chain: a vintage whose source lacks ``NOW_MCAID`` derives + ``False`` for every person, and only this gate sees it.""" + + person = _person_table( + [ + {"source_year": 2023, "NOW_GRP": 1, "NOW_MRK": 2, "NOW_MCAID": np.nan}, + {"source_year": 2023, "NOW_GRP": 2, "NOW_MRK": 1, "NOW_MCAID": np.nan}, + {"source_year": 2023, "NOW_GRP": 1, "NOW_MRK": 2, "NOW_MCAID": np.nan}, + {"source_year": 2024, "NOW_GRP": 1, "NOW_MRK": 2, "NOW_MCAID": 1}, + {"source_year": 2024, "NOW_GRP": 2, "NOW_MRK": 1, "NOW_MCAID": 2}, + {"source_year": 2024, "NOW_GRP": 1, "NOW_MRK": 2, "NOW_MCAID": 1}, + ] + ) + for column in ( + "NOW_NONM", + "NOW_CHAMPVA", + "NOW_MIL", + "NOW_VACARE", + "NOW_OTHMT", + "NOW_IHSFLG", + ): + person[column] = [np.nan] * 3 + [1, 2, 1] + _fill_health_coverage_inputs(person) + + medicaid = person["has_medicaid_health_coverage_at_interview"] + assert not medicaid[person["source_year"] == 2023].any() + assert medicaid[person["source_year"] == 2024].sum() == 2 + assert person["has_esi"].sum() == 4 # NOW_GRP carried for both vintages + + gate = us_reported_coverage_vintage_signal_gate( + _us_frame_from_person(person), min_vintage_rows=3 + ) + assert not gate.passed + failed = {failure.split(":")[0] for failure in gate.failures} + assert failed == set(US_REPORTED_COVERAGE_PERSON_INPUTS) - { + "has_esi", + "has_marketplace_health_coverage_at_interview", + } + assert all("source_year 2023" in failure for failure in gate.failures) diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index 1a1f73cba..11b208320 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -147,6 +147,7 @@ us_relationship_inputs_signal_gate, us_release_input_coverage_gate, us_release_target_parity_gate, + us_reported_coverage_vintage_signal_gate, us_retirement_contributions_signal_gate, us_retirement_distributions_signal_gate, us_salt_refund_income_signal_gate, @@ -6395,6 +6396,7 @@ def _release_gate_failures( snap_take_up_gate: GateResult | None = None, eligibility_inputs_gate: GateResult | None = None, pregnancy_gate: GateResult | None = None, + reported_coverage_vintage_gate: GateResult | None = None, snap_discretionary_exemption_gate: GateResult | None = None, target_registry: TargetRegistry | None = None, ) -> list[str]: @@ -6438,6 +6440,14 @@ def _release_gate_failures( failures.extend( f"Pregnancy signal failed: {failure}" for failure in pregnancy_gate.failures ) + if ( + reported_coverage_vintage_gate is not None + and not reported_coverage_vintage_gate.passed + ): + failures.extend( + f"Reported-coverage vintage signal failed: {failure}" + for failure in reported_coverage_vintage_gate.failures + ) if ( snap_discretionary_exemption_gate is not None and not snap_discretionary_exemption_gate.passed @@ -6942,6 +6952,7 @@ def _write_release_calibration_diagnostics( snap_take_up_gate: GateResult | None = None, eligibility_inputs_gate: GateResult | None = None, pregnancy_gate: GateResult | None = None, + reported_coverage_vintage_gate: GateResult | None = None, snap_discretionary_exemption_gate: GateResult | None = None, gate_failures: Iterable[str], timing: Mapping[str, object] | None = None, @@ -7070,6 +7081,15 @@ def _write_release_calibration_diagnostics( if pregnancy_gate is not None else None ), + "reported_coverage_vintage_signal": ( + { + "passed": reported_coverage_vintage_gate.passed, + "failures": list(reported_coverage_vintage_gate.failures), + "details": dict(reported_coverage_vintage_gate.details), + } + if reported_coverage_vintage_gate is not None + else None + ), "snap_discretionary_exemption_signal": ( { "passed": snap_discretionary_exemption_gate.passed, @@ -7336,6 +7356,7 @@ def _build_manifests( snap_take_up_gate: GateResult | None = None, eligibility_inputs_gate: GateResult | None = None, pregnancy_gate: GateResult | None = None, + reported_coverage_vintage_gate: GateResult | None = None, snap_discretionary_exemption_gate: GateResult | None = None, timing: Mapping[str, object] | None = None, warm_start_calibration: Mapping[str, object] | None = None, @@ -7374,6 +7395,7 @@ def _build_manifests( snap_take_up_gate=snap_take_up_gate, eligibility_inputs_gate=eligibility_inputs_gate, pregnancy_gate=pregnancy_gate, + reported_coverage_vintage_gate=reported_coverage_vintage_gate, snap_discretionary_exemption_gate=snap_discretionary_exemption_gate, target_registry=registry, ) @@ -7582,6 +7604,17 @@ def _build_manifests( if pregnancy_gate is not None else {} ), + **( + { + "reported_coverage_vintage_signal": { + "passed": reported_coverage_vintage_gate.passed, + "failures": list(reported_coverage_vintage_gate.failures), + "details": dict(reported_coverage_vintage_gate.details), + } + } + if reported_coverage_vintage_gate is not None + else {} + ), **( { "snap_discretionary_exemption_signal": { @@ -7713,6 +7746,16 @@ def _build_manifests( if pregnancy_gate is not None else {} ), + **( + { + "reported_coverage_vintage_signal": { + "passed": reported_coverage_vintage_gate.passed, + "details": dict(reported_coverage_vintage_gate.details), + } + } + if reported_coverage_vintage_gate is not None + else {} + ), **( { "snap_discretionary_exemption_signal": { @@ -9302,6 +9345,25 @@ def main(argv: Sequence[str] | None = None) -> None: for failure in pregnancy_gate.failures ) ) + reported_coverage_vintage_gate = us_reported_coverage_vintage_signal_gate( + base_frame + ) + if not reported_coverage_vintage_gate.passed: + if telemetry is not None: + telemetry.stage( + "reported_coverage_vintage_gate", + status="failed", + message="Reported-coverage vintage signal gate failed.", + failures=list(reported_coverage_vintage_gate.failures), + force_upload=True, + ) + raise RuntimeError( + "Release gates failed: " + + "; ".join( + f"Reported-coverage vintage signal failed: {failure}" + for failure in reported_coverage_vintage_gate.failures + ) + ) if telemetry is not None: telemetry.stage( "wic_claim_input", @@ -10668,6 +10730,7 @@ def main(argv: Sequence[str] | None = None) -> None: snap_take_up_gate=snap_take_up_gate, eligibility_inputs_gate=eligibility_inputs_gate, pregnancy_gate=pregnancy_gate, + reported_coverage_vintage_gate=reported_coverage_vintage_gate, snap_discretionary_exemption_gate=snap_discretionary_exemption_gate, target_registry=registry, ) @@ -10714,6 +10777,7 @@ def main(argv: Sequence[str] | None = None) -> None: snap_take_up_gate=snap_take_up_gate, eligibility_inputs_gate=eligibility_inputs_gate, pregnancy_gate=pregnancy_gate, + reported_coverage_vintage_gate=reported_coverage_vintage_gate, snap_discretionary_exemption_gate=snap_discretionary_exemption_gate, support_value_repairs={ "social_security_components": social_security_component_repair, @@ -11301,6 +11365,7 @@ def main(argv: Sequence[str] | None = None) -> None: snap_take_up_gate=snap_take_up_gate, eligibility_inputs_gate=eligibility_inputs_gate, pregnancy_gate=pregnancy_gate, + reported_coverage_vintage_gate=reported_coverage_vintage_gate, snap_discretionary_exemption_gate=snap_discretionary_exemption_gate, timing=timing, warm_start_calibration=warm_start_calibration, From 142c6b8c245b9de9cb967fe71332369f34f1d3f0 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 23 Aug 2026 10:33:46 +0200 Subject: [PATCH 2/4] Stub the reported-coverage vintage gate in the terminal-batch builder test The #547 degraded-mode test drives main() against a fake frame whose table() only serves the household table; every base-frame signal gate is monkeypatched there, and the new gate must be too. Co-Authored-By: Claude Fable 5 --- .../tests/test_us_fiscal_refresh_builder.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py b/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py index fa6dc5113..0c4b05d42 100644 --- a/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py +++ b/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py @@ -4614,6 +4614,15 @@ def fake_retirement_distributions_signal_gate(frame): details={"checked": True}, ), ) + monkeypatch.setattr( + builder, + "us_reported_coverage_vintage_signal_gate", + lambda frame: builder.GateResult( + name="reported_coverage_vintage_signal", + passed=True, + details={"checked": True}, + ), + ) monkeypatch.setattr( builder, "us_wic_claim_signal_gate", From 8e61c75baae47b4f3d2f26b3377f6699f549b471 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 23 Aug 2026 11:10:38 +0200 Subject: [PATCH 3/4] Harden the reported-coverage vintage gate per cross-family review - Key groups by person_support_channel x source_year so ACS-spine signal cannot mask a missing ASEC recode (both carry source_year). - Fail closed on provenance: missing source_year column, null source years, or an empty person table fail instead of collapsing to one group. - Require full population and boolean dtype per vintage, not just one truthy value; failure text is observational (zero reporters), with the #720 source-input reading as the documented cause. - Re-run the gate on the export frame (sparse selection can drop a vintage's reporters) under the #547 degraded-mode guard. - Persist a failure receipt in the release directory before the early raise so RED is durable without staging telemetry; the receipt does not mark the directory certified (#568). Co-Authored-By: Claude Fable 5 --- ...20-reported-coverage-vintage-gate.added.md | 2 +- .../microcosm/build/us_runtime/cps_carried.py | 103 ++++++++++++++---- .../tests/test_us_fiscal_refresh_builder.py | 27 +++++ .../test_us_reported_coverage_vintage_gate.py | 69 ++++++++++-- tools/build_us_fiscal_refresh_release.py | 51 +++++++++ 5 files changed, 217 insertions(+), 35 deletions(-) diff --git a/changelog.d/720-reported-coverage-vintage-gate.added.md b/changelog.d/720-reported-coverage-vintage-gate.added.md index b20e9888b..35a0fdb4e 100644 --- a/changelog.d/720-reported-coverage-vintage-gate.added.md +++ b/changelog.d/720-reported-coverage-vintage-gate.added.md @@ -1 +1 @@ -Add `us_reported_coverage_vintage_signal_gate`, a US release gate requiring every pooled ASEC `source_year` to carry at least one reporter for each of the nine reported-coverage person inputs; it closes the partial-vintage hole behind the certified Build P artifact's `has_medicaid_health_coverage_at_interview` thinning (24.6M vs ~58M survey under 65, #720), which presence-style checks cannot see. Wired into `tools/build_us_fiscal_refresh_release.py` as a base-frame hard gate and manifest entry. +Add `us_reported_coverage_vintage_signal_gate`, a US release gate requiring every pooled source vintage (`person_support_channel` × `source_year`) to carry a fully populated, boolean, non-zero-signal column for each of the nine reported-coverage person inputs; it closes the partial-vintage hole behind the certified Build P artifact's `has_medicaid_health_coverage_at_interview` thinning (24.6M vs ~58M survey under 65, #720), which presence-style checks cannot see. Wired into `tools/build_us_fiscal_refresh_release.py` as a base-frame hard gate with a persisted failure receipt, re-run on the export frame, and a `reported_coverage_vintage_signal` manifest entry. diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py b/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py index 23766a3ac..9a8401627 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py @@ -533,23 +533,31 @@ def us_reported_coverage_vintage_signal_gate( *, min_vintage_rows: int = US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS, ) -> GateResult: - """Require every pooled ASEC vintage to carry reported-coverage signal. + """Require every pooled source vintage to carry reported-coverage signal. - Microcosm #720: the pooled 2022/2023 ASEC source inputs lacked the - ``NOW_*`` at-interview recodes (except ``NOW_GRP``/``NOW_MRK``), so + Microcosm #720: the pooled income-year 2022/2023 ASEC inputs carried only + ``NOW_GRP``/``NOW_MRK`` of the 18 ``NOW_*`` at-interview recodes, so :func:`derive_us_cps_carried_inputs` silently mapped every 2022/2023 -vintage person to ``False`` for seven of the nine reported-coverage flags and the certified Build P artifact reported 24.6M under-65 Medicaid at interview against ~58M survey. A flag populated for one vintage passes the presence-style checks (``release_input_coverage``, ``degenerate_input_signal``); this gate enforces the per-vintage - invariant those checks cannot see: every ``source_year`` with at least - ``min_vintage_rows`` person rows must have at least one reporter for - every reported-coverage input. - - Vintages below ``min_vintage_rows`` (smoke pools) are recorded in the - details but not enforced. A frame without a ``source_year`` column is - checked as a single vintage. + invariant those checks cannot see. + + Groups are ``person_support_channel`` x ``source_year`` when the channel + column is present (ACS-spine rows also carry ``source_year``, so a + year-only key would let ACS signal mask a missing ASEC recode), else + ``source_year`` alone. Every group with at least ``min_vintage_rows`` + person rows must have, for every reported-coverage input, a boolean-like + column with no nulls and at least one reporter. Provenance must be + present: a missing ``source_year`` column, null source years, or an + empty person table fail the gate rather than collapsing into one group. + + Groups below ``min_vintage_rows`` (smoke pools) are recorded in the + details but not enforced. This is a zero-signal sentinel, not a survey + mass check: it observes that a vintage has no reporters; the #720 + cause (a source input lacking the recode) is the documented reading. """ person = frame.table("person") @@ -558,6 +566,8 @@ def us_reported_coverage_vintage_signal_gate( for column in US_REPORTED_COVERAGE_PERSON_INPUTS if column not in person.columns ] + if "source_year" not in person.columns: + missing.append("source_year") if missing: return GateResult( name="reported_coverage_vintage_signal", @@ -565,33 +575,73 @@ def us_reported_coverage_vintage_signal_gate( failures=tuple(f"person column missing: {column}." for column in missing), details={"missing": missing}, ) - if "source_year" in person.columns: - vintage_labels = person["source_year"] - else: - vintage_labels = pd.Series("all", index=person.index) + if len(person) == 0: + # Unreachable through a valid Frame (weights cannot be empty); kept + # so a direct caller cannot pass an empty table as "no failures". + return GateResult( + name="reported_coverage_vintage_signal", + passed=False, + failures=("person table is empty: no vintage carries any signal.",), + details={"rows": 0}, + ) failures: list[str] = [] + null_years = int(person["source_year"].isna().sum()) + if null_years: + failures.append( + f"source_year: {null_years} person rows have no source year; the " + "per-vintage invariant cannot be proven for unprovenanced rows." + ) + keys: list[pd.Series] = [] + if "person_support_channel" in person.columns: + keys.append(person["person_support_channel"].astype(str)) + keys.append(person["source_year"]) vintages: dict[str, dict[str, object]] = {} - for vintage, group in person.groupby(vintage_labels, sort=True, dropna=False): - label = str(vintage) + for key, group in person.groupby(keys, sort=True, dropna=True): + parts = key if isinstance(key, tuple) else (key,) + label = "/".join(str(part) for part in parts) rows = int(len(group)) - reporter_counts = { - column: int(group[column].fillna(False).astype(bool).sum()) - for column in US_REPORTED_COVERAGE_PERSON_INPUTS - } + reporter_counts: dict[str, int] = {} + null_counts: dict[str, int] = {} + dtype_failures: list[str] = [] + for column in US_REPORTED_COVERAGE_PERSON_INPUTS: + values = group[column] + if not ( + pd.api.types.is_bool_dtype(values) + or pd.api.types.is_numeric_dtype(values) + ): + dtype_failures.append(column) + reporter_counts[column] = 0 + null_counts[column] = int(values.isna().sum()) + continue + null_counts[column] = int(values.isna().sum()) + reporter_counts[column] = int(values.fillna(False).astype(bool).sum()) enforced = rows >= min_vintage_rows vintages[label] = { "rows": rows, "enforced": enforced, "reporter_counts": reporter_counts, + "null_counts": null_counts, } if not enforced: continue + for column in dtype_failures: + failures.append( + f"{column}: vintage {label} stores a non-boolean dtype " + f"({group[column].dtype}); reported-coverage inputs must be " + "boolean." + ) + for column, count in null_counts.items(): + if count and column not in dtype_failures: + failures.append( + f"{column}: vintage {label} has {count} null values over " + f"{rows} person rows; the flag must be fully populated." + ) for column, count in reporter_counts.items(): - if count == 0: + if count == 0 and column not in dtype_failures: failures.append( - f"{column}: source_year {label} has 0 reporters over " - f"{rows} person rows — the vintage source lacks the " - "at-interview recode (microcosm #720)." + f"{column}: vintage {label} has 0 reporters over {rows} " + "person rows (consistent with a source input lacking the " + "at-interview recode, microcosm #720)." ) return GateResult( name="reported_coverage_vintage_signal", @@ -599,6 +649,11 @@ def us_reported_coverage_vintage_signal_gate( failures=tuple(failures), details={ "min_vintage_rows": int(min_vintage_rows), + "grouping": ( + ["person_support_channel", "source_year"] + if "person_support_channel" in person.columns + else ["source_year"] + ), "vintages": vintages, }, ) diff --git a/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py b/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py index 0c4b05d42..ebb7d3999 100644 --- a/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py +++ b/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py @@ -2267,6 +2267,33 @@ def test_release_gate_failures_include_reported_coverage_vintage_signal() -> Non ] +def test_reported_coverage_vintage_gate_receipt_is_written(tmp_path) -> None: + builder = _load_builder_module() + gate = builder.GateResult( + name="reported_coverage_vintage_signal", + passed=False, + failures=( + "has_medicaid_health_coverage_at_interview: vintage asec/2022 has 0 " + "reporters over 54464 person rows (consistent with a source input " + "lacking the at-interview recode, microcosm #720).", + ), + details={"min_vintage_rows": 5000, "vintages": {"asec/2022": {"rows": 54464}}}, + ) + release_dir = tmp_path / "releases" / "rid" + + path = builder._write_reported_coverage_vintage_gate_receipt(release_dir, gate) + + assert path == release_dir / builder.US_REPORTED_COVERAGE_VINTAGE_GATE_RECEIPT + payload = json.loads(path.read_text()) + assert payload["gate"] == "reported_coverage_vintage_signal" + assert payload["passed"] is False + assert payload["failures"] == list(gate.failures) + assert payload["details"]["vintages"]["asec/2022"]["rows"] == 54464 + # A receipt-only directory is not a certified release (#568): reruns under + # the same id stay allowed. + builder._refuse_certified_release_dir_reuse(release_dir) + + def test_base_population_scale_gate_rejects_underweighted_base(small_frame) -> None: builder = _load_builder_module() diff --git a/packages/microcosm-build/tests/test_us_reported_coverage_vintage_gate.py b/packages/microcosm-build/tests/test_us_reported_coverage_vintage_gate.py index 8dbc7db01..670f4c008 100644 --- a/packages/microcosm-build/tests/test_us_reported_coverage_vintage_gate.py +++ b/packages/microcosm-build/tests/test_us_reported_coverage_vintage_gate.py @@ -131,7 +131,7 @@ def test_vintage_without_medicaid_reporters_fails(self) -> None: assert not gate.passed assert len(gate.failures) == 1 assert "has_medicaid_health_coverage_at_interview" in gate.failures[0] - assert "source_year 2022" in gate.failures[0] + assert "vintage 2022 has 0 reporters" in gate.failures[0] assert "#720" in gate.failures[0] def test_issue_720_shape_fails_seven_flags_for_two_vintages(self) -> None: @@ -147,7 +147,7 @@ def test_issue_720_shape_fails_seven_flags_for_two_vintages(self) -> None: assert not gate.passed assert len(gate.failures) == 14 failed_vintages = { - failure.split("source_year ")[1].split(" ")[0] for failure in gate.failures + failure.split("vintage ")[1].split(" ")[0] for failure in gate.failures } assert failed_vintages == {"2022", "2023"} assert not any("has_esi" in failure for failure in gate.failures) @@ -180,24 +180,71 @@ def test_default_threshold_skips_unit_sized_frames(self) -> None: assert gate.passed, gate.failures assert gate.details["vintages"]["2022"]["enforced"] is False - def test_frame_without_source_year_is_one_vintage(self) -> None: + def test_missing_source_year_fails_closed(self) -> None: rows = [{**row} for row in _vintage_rows(2024)] for row in rows: del row["source_year"] gate = us_reported_coverage_vintage_signal_gate( _us_frame(rows), min_vintage_rows=3 ) - assert gate.passed, gate.failures - assert set(gate.details["vintages"]) == {"all"} + assert not gate.passed + assert gate.failures == ("person column missing: source_year.",) - for row in rows: - row["has_tricare_health_coverage_at_interview"] = False + def test_null_source_year_rows_fail(self) -> None: + rows = _vintage_rows(2024) + [{**_vintage_rows(2024)[0], "source_year": None}] gate = us_reported_coverage_vintage_signal_gate( _us_frame(rows), min_vintage_rows=3 ) assert not gate.passed - assert "source_year all" in gate.failures[0] - assert "has_tricare_health_coverage_at_interview" in gate.failures[0] + assert any("1 person rows have no source year" in f for f in gate.failures) + + def test_support_channel_separates_acs_from_asec(self) -> None: + # An ACS-spine vintage with signal must not mask an ASEC vintage + # whose source lacked the recode: both carry source_year 2024. + asec = [ + {**row, "person_support_channel": "asec"} + for row in _vintage_rows( + 2024, + reporters=("has_esi", "has_marketplace_health_coverage_at_interview"), + ) + ] + acs = [{**row, "person_support_channel": "acs"} for row in _vintage_rows(2024)] + gate = us_reported_coverage_vintage_signal_gate( + _us_frame(asec + acs), min_vintage_rows=3 + ) + assert not gate.passed + assert gate.details["grouping"] == ["person_support_channel", "source_year"] + assert set(gate.details["vintages"]) == {"acs/2024", "asec/2024"} + assert len(gate.failures) == 7 + assert all("vintage asec/2024" in failure for failure in gate.failures) + + def test_null_flag_values_fail_completeness(self) -> None: + person = _person_table(_vintage_rows(2024)) + person["has_medicaid_health_coverage_at_interview"] = pd.array( + [True, None, False], dtype="boolean" + ) + gate = us_reported_coverage_vintage_signal_gate( + _us_frame_from_person(person), min_vintage_rows=3 + ) + assert not gate.passed + assert len(gate.failures) == 1 + assert "1 null values" in gate.failures[0] + assert ( + gate.details["vintages"]["2024"]["null_counts"][ + "has_medicaid_health_coverage_at_interview" + ] + == 1 + ) + + def test_non_boolean_dtype_fails(self) -> None: + person = _person_table(_vintage_rows(2024)) + person["has_esi"] = ["True", "False", "False"] + gate = us_reported_coverage_vintage_signal_gate( + _us_frame_from_person(person), min_vintage_rows=3 + ) + assert not gate.passed + assert len(gate.failures) == 1 + assert "non-boolean dtype" in gate.failures[0] def test_missing_column_fails(self) -> None: rows = [{**row} for row in _vintage_rows(2024)] @@ -251,4 +298,6 @@ def test_nan_source_recode_becomes_false_and_fails_the_vintage(self) -> None: "has_esi", "has_marketplace_health_coverage_at_interview", } - assert all("source_year 2023" in failure for failure in gate.failures) + assert all( + "vintage 2023 has 0 reporters" in failure for failure in gate.failures + ) diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index 11b208320..048f6a535 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -5021,6 +5021,41 @@ def _health_input_signal_gate(frame: Frame) -> GateResult: ) +US_REPORTED_COVERAGE_VINTAGE_GATE_RECEIPT = ( + "reported_coverage_vintage_gate_failure.json" +) + + +def _write_reported_coverage_vintage_gate_receipt( + release_dir: Path, gate: GateResult +) -> Path: + """Persist a RED reported-coverage vintage gate before the early raise. + + The base-frame gate raises before the collector, diagnostics, and + manifests run, so without this receipt a failure would leave only + optional staging telemetry and an exception (microcosm #720 review). + The receipt never marks a directory certified: certification is keyed + on ``release_manifest.json`` (#568), so a rerun under the same id is + not blocked by it. + """ + + release_dir.mkdir(parents=True, exist_ok=True) + path = release_dir / US_REPORTED_COVERAGE_VINTAGE_GATE_RECEIPT + path.write_text( + json.dumps( + { + "gate": gate.name, + "passed": gate.passed, + "failures": list(gate.failures), + "details": dict(gate.details), + }, + indent=1, + allow_nan=False, + ) + ) + return path + + def _engine_input_variables() -> tuple[str, ...]: """Persistable PolicyEngine input variables (formula-owned excluded).""" return tuple(PolicyEngineUSEngine().variables()) @@ -9349,6 +9384,9 @@ def main(argv: Sequence[str] | None = None) -> None: base_frame ) if not reported_coverage_vintage_gate.passed: + receipt_path = _write_reported_coverage_vintage_gate_receipt( + release_dir, reported_coverage_vintage_gate + ) if telemetry is not None: telemetry.stage( "reported_coverage_vintage_gate", @@ -9363,6 +9401,7 @@ def main(argv: Sequence[str] | None = None) -> None: f"Reported-coverage vintage signal failed: {failure}" for failure in reported_coverage_vintage_gate.failures ) + + f" (receipt: {receipt_path})" ) if telemetry is not None: telemetry.stage( @@ -10556,6 +10595,18 @@ def main(argv: Sequence[str] | None = None) -> None: "Health-input signal evaluation crashed in degraded mode; " f"recorded instead of masking earlier failures: {error}" ) + try: + reported_coverage_vintage_gate = us_reported_coverage_vintage_signal_gate( + export_frame + ) + except Exception as error: + if not early_terminal_gate_failures: + raise + reported_coverage_vintage_gate = None + early_terminal_gate_failures.append( + "Reported-coverage vintage signal evaluation crashed in degraded " + f"mode; recorded instead of masking earlier failures: {error}" + ) try: other_health_insurance_gate = us_other_health_insurance_signal_gate( export_frame From 093daef3d6cb805c8c64b0705542c52ccb965811 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sun, 23 Aug 2026 12:28:25 +0200 Subject: [PATCH 4/4] Move the reported-coverage vintage gate to the provenance owner module test_us_spine_blindness requires runtime population operators to stay source-spine blind: only the reviewed provenance-owner modules may read person_support_channel. The gate is origin-aware by charter (it keys vintages by support channel so ACS signal cannot mask a missing ASEC recode), so it lives in support_provenance.py; cps_carried.py keeps the input list, the threshold, and the derivation. The inputs import is lazy because cps_carried -> alimony -> support_provenance would otherwise cycle. Co-Authored-By: Claude Fable 5 --- .../microcosm/build/us_runtime/__init__.py | 4 +- .../microcosm/build/us_runtime/cps_carried.py | 133 ---------------- .../build/us_runtime/support_provenance.py | 144 ++++++++++++++++++ 3 files changed, 147 insertions(+), 134 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py b/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py index ee39d8259..e9e756473 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py @@ -168,7 +168,6 @@ derive_us_cps_carried_inputs, reported_tanf_enrollment_by_spm_unit, reported_wic_receipt_carrier, - us_reported_coverage_vintage_signal_gate, ) from microcosm.build.us_runtime.demographics import ( AGE_BANDS, @@ -967,6 +966,9 @@ with_us_ssi_take_up, write_us_ssi_take_up_diagnostics, ) +from microcosm.build.us_runtime.support_provenance import ( + us_reported_coverage_vintage_signal_gate, +) from microcosm.build.us_runtime.take_up import ( US_TAKE_UP_SHARE_BAND, SeededTakeUpResult, diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py b/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py index 9a8401627..0c93ad543 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py @@ -13,7 +13,6 @@ import numpy as np import pandas as pd -from microcosm.build.gates import GateResult from microcosm.build.us_runtime.alimony import ( US_ASEC_OTHER_INCOME_OUTPUT_COLUMNS, derive_us_alimony_from_asec, @@ -39,7 +38,6 @@ "reported_snap_receipt_by_spm_unit", "reported_tanf_enrollment_by_spm_unit", "reported_wic_receipt_carrier", - "us_reported_coverage_vintage_signal_gate", ] TAXABLE_INTEREST_FRACTION = 0.680 @@ -526,134 +524,3 @@ def _integer_source(person: pd.DataFrame, column: str) -> np.ndarray: def _yes_code(person: pd.DataFrame, column: str) -> np.ndarray: return _integer_source(person, column) == 1 - - -def us_reported_coverage_vintage_signal_gate( - frame: Frame, - *, - min_vintage_rows: int = US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS, -) -> GateResult: - """Require every pooled source vintage to carry reported-coverage signal. - - Microcosm #720: the pooled income-year 2022/2023 ASEC inputs carried only - ``NOW_GRP``/``NOW_MRK`` of the 18 ``NOW_*`` at-interview recodes, so - :func:`derive_us_cps_carried_inputs` silently mapped every 2022/2023 - -vintage person to ``False`` for seven of the nine reported-coverage - flags and the certified Build P artifact reported 24.6M under-65 - Medicaid at interview against ~58M survey. A flag populated for one - vintage passes the presence-style checks (``release_input_coverage``, - ``degenerate_input_signal``); this gate enforces the per-vintage - invariant those checks cannot see. - - Groups are ``person_support_channel`` x ``source_year`` when the channel - column is present (ACS-spine rows also carry ``source_year``, so a - year-only key would let ACS signal mask a missing ASEC recode), else - ``source_year`` alone. Every group with at least ``min_vintage_rows`` - person rows must have, for every reported-coverage input, a boolean-like - column with no nulls and at least one reporter. Provenance must be - present: a missing ``source_year`` column, null source years, or an - empty person table fail the gate rather than collapsing into one group. - - Groups below ``min_vintage_rows`` (smoke pools) are recorded in the - details but not enforced. This is a zero-signal sentinel, not a survey - mass check: it observes that a vintage has no reporters; the #720 - cause (a source input lacking the recode) is the documented reading. - """ - - person = frame.table("person") - missing = [ - column - for column in US_REPORTED_COVERAGE_PERSON_INPUTS - if column not in person.columns - ] - if "source_year" not in person.columns: - missing.append("source_year") - if missing: - return GateResult( - name="reported_coverage_vintage_signal", - passed=False, - failures=tuple(f"person column missing: {column}." for column in missing), - details={"missing": missing}, - ) - if len(person) == 0: - # Unreachable through a valid Frame (weights cannot be empty); kept - # so a direct caller cannot pass an empty table as "no failures". - return GateResult( - name="reported_coverage_vintage_signal", - passed=False, - failures=("person table is empty: no vintage carries any signal.",), - details={"rows": 0}, - ) - failures: list[str] = [] - null_years = int(person["source_year"].isna().sum()) - if null_years: - failures.append( - f"source_year: {null_years} person rows have no source year; the " - "per-vintage invariant cannot be proven for unprovenanced rows." - ) - keys: list[pd.Series] = [] - if "person_support_channel" in person.columns: - keys.append(person["person_support_channel"].astype(str)) - keys.append(person["source_year"]) - vintages: dict[str, dict[str, object]] = {} - for key, group in person.groupby(keys, sort=True, dropna=True): - parts = key if isinstance(key, tuple) else (key,) - label = "/".join(str(part) for part in parts) - rows = int(len(group)) - reporter_counts: dict[str, int] = {} - null_counts: dict[str, int] = {} - dtype_failures: list[str] = [] - for column in US_REPORTED_COVERAGE_PERSON_INPUTS: - values = group[column] - if not ( - pd.api.types.is_bool_dtype(values) - or pd.api.types.is_numeric_dtype(values) - ): - dtype_failures.append(column) - reporter_counts[column] = 0 - null_counts[column] = int(values.isna().sum()) - continue - null_counts[column] = int(values.isna().sum()) - reporter_counts[column] = int(values.fillna(False).astype(bool).sum()) - enforced = rows >= min_vintage_rows - vintages[label] = { - "rows": rows, - "enforced": enforced, - "reporter_counts": reporter_counts, - "null_counts": null_counts, - } - if not enforced: - continue - for column in dtype_failures: - failures.append( - f"{column}: vintage {label} stores a non-boolean dtype " - f"({group[column].dtype}); reported-coverage inputs must be " - "boolean." - ) - for column, count in null_counts.items(): - if count and column not in dtype_failures: - failures.append( - f"{column}: vintage {label} has {count} null values over " - f"{rows} person rows; the flag must be fully populated." - ) - for column, count in reporter_counts.items(): - if count == 0 and column not in dtype_failures: - failures.append( - f"{column}: vintage {label} has 0 reporters over {rows} " - "person rows (consistent with a source input lacking the " - "at-interview recode, microcosm #720)." - ) - return GateResult( - name="reported_coverage_vintage_signal", - passed=not failures, - failures=tuple(failures), - details={ - "min_vintage_rows": int(min_vintage_rows), - "grouping": ( - ["person_support_channel", "source_year"] - if "person_support_channel" in person.columns - else ["source_year"] - ), - "vintages": vintages, - }, - ) diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/support_provenance.py b/packages/microcosm-build/src/microcosm/build/us_runtime/support_provenance.py index e0f8d40fc..da5bf9e08 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/support_provenance.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/support_provenance.py @@ -8,6 +8,8 @@ import numpy as np import pandas as pd +from microcosm.build.gates import GateResult + __all__ = [ "BASE_ASEC_SUPPORT_CHANNEL", "PERSON_SUPPORT_CHANNEL_COLUMN", @@ -24,6 +26,7 @@ "support_clone_index_column", "support_role_series", "support_source_id_column", + "us_reported_coverage_vintage_signal_gate", "validate_assembly_provenance", "without_support_role_metadata", ] @@ -511,3 +514,144 @@ def thaw(item: Any) -> Any: return item return {str(key): thaw(item) for key, item in value.items()} + + +def us_reported_coverage_vintage_signal_gate( + frame: _ProvenanceFrame, + *, + min_vintage_rows: int | None = None, +) -> GateResult: + """Require every pooled source vintage to carry reported-coverage signal. + + Microcosm #720: the pooled income-year 2022/2023 ASEC inputs carried only + ``NOW_GRP``/``NOW_MRK`` of the 18 ``NOW_*`` at-interview recodes, so + :func:`derive_us_cps_carried_inputs` silently mapped every 2022/2023 + -vintage person to ``False`` for seven of the nine reported-coverage + flags and the certified Build P artifact reported 24.6M under-65 + Medicaid at interview against ~58M survey. A flag populated for one + vintage passes the presence-style checks (``release_input_coverage``, + ``degenerate_input_signal``); this gate enforces the per-vintage + invariant those checks cannot see. + + Groups are ``person_support_channel`` x ``source_year`` when the channel + column is present (ACS-spine rows also carry ``source_year``, so a + year-only key would let ACS signal mask a missing ASEC recode), else + ``source_year`` alone. Every group with at least ``min_vintage_rows`` + person rows must have, for every reported-coverage input, a boolean-like + column with no nulls and at least one reporter. Provenance must be + present: a missing ``source_year`` column, null source years, or an + empty person table fail the gate rather than collapsing into one group. + + Groups below ``min_vintage_rows`` (smoke pools) are recorded in the + details but not enforced. This is a zero-signal sentinel, not a survey + mass check: it observes that a vintage has no reporters; the #720 + cause (a source input lacking the recode) is the documented reading. + """ + + # Lazy import: cps_carried (the derivation owner) imports alimony, which + # imports this module; the gate lives here because it is origin-aware by + # charter (provenance owners only may read the support channel). + from microcosm.build.us_runtime.cps_carried import ( + US_REPORTED_COVERAGE_PERSON_INPUTS, + US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS, + ) + + if min_vintage_rows is None: + min_vintage_rows = US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS + person = frame.table("person") + missing = [ + column + for column in US_REPORTED_COVERAGE_PERSON_INPUTS + if column not in person.columns + ] + if "source_year" not in person.columns: + missing.append("source_year") + if missing: + return GateResult( + name="reported_coverage_vintage_signal", + passed=False, + failures=tuple(f"person column missing: {column}." for column in missing), + details={"missing": missing}, + ) + if len(person) == 0: + # Unreachable through a valid Frame (weights cannot be empty); kept + # so a direct caller cannot pass an empty table as "no failures". + return GateResult( + name="reported_coverage_vintage_signal", + passed=False, + failures=("person table is empty: no vintage carries any signal.",), + details={"rows": 0}, + ) + failures: list[str] = [] + null_years = int(person["source_year"].isna().sum()) + if null_years: + failures.append( + f"source_year: {null_years} person rows have no source year; the " + "per-vintage invariant cannot be proven for unprovenanced rows." + ) + keys: list[pd.Series] = [] + if PERSON_SUPPORT_CHANNEL_COLUMN in person.columns: + keys.append(person[PERSON_SUPPORT_CHANNEL_COLUMN].astype(str)) + keys.append(person["source_year"]) + vintages: dict[str, dict[str, object]] = {} + for key, group in person.groupby(keys, sort=True, dropna=True): + parts = key if isinstance(key, tuple) else (key,) + label = "/".join(str(part) for part in parts) + rows = int(len(group)) + reporter_counts: dict[str, int] = {} + null_counts: dict[str, int] = {} + dtype_failures: list[str] = [] + for column in US_REPORTED_COVERAGE_PERSON_INPUTS: + values = group[column] + if not ( + pd.api.types.is_bool_dtype(values) + or pd.api.types.is_numeric_dtype(values) + ): + dtype_failures.append(column) + reporter_counts[column] = 0 + null_counts[column] = int(values.isna().sum()) + continue + null_counts[column] = int(values.isna().sum()) + reporter_counts[column] = int(values.fillna(False).astype(bool).sum()) + enforced = rows >= min_vintage_rows + vintages[label] = { + "rows": rows, + "enforced": enforced, + "reporter_counts": reporter_counts, + "null_counts": null_counts, + } + if not enforced: + continue + for column in dtype_failures: + failures.append( + f"{column}: vintage {label} stores a non-boolean dtype " + f"({group[column].dtype}); reported-coverage inputs must be " + "boolean." + ) + for column, count in null_counts.items(): + if count and column not in dtype_failures: + failures.append( + f"{column}: vintage {label} has {count} null values over " + f"{rows} person rows; the flag must be fully populated." + ) + for column, count in reporter_counts.items(): + if count == 0 and column not in dtype_failures: + failures.append( + f"{column}: vintage {label} has 0 reporters over {rows} " + "person rows (consistent with a source input lacking the " + "at-interview recode, microcosm #720)." + ) + return GateResult( + name="reported_coverage_vintage_signal", + passed=not failures, + failures=tuple(failures), + details={ + "min_vintage_rows": int(min_vintage_rows), + "grouping": ( + [PERSON_SUPPORT_CHANNEL_COLUMN, "source_year"] + if PERSON_SUPPORT_CHANNEL_COLUMN in person.columns + else ["source_year"] + ), + "vintages": vintages, + }, + )