diff --git a/packages/populace-build/src/populace/build/us/ecps_parity_known_gaps.json b/packages/populace-build/src/populace/build/us/ecps_parity_known_gaps.json index 6af51950..aef661b7 100644 --- a/packages/populace-build/src/populace/build/us/ecps_parity_known_gaps.json +++ b/packages/populace-build/src/populace/build/us/ecps_parity_known_gaps.json @@ -330,10 +330,6 @@ "reason": "Stochastic take-up flag the incumbent seeds so take-up-gated programs do not ship at 100% participation; Populace does not yet produce it.", "issue": "PolicyEngine/populace#312" }, - "takes_up_medicaid_if_eligible": { - "reason": "Stochastic take-up flag the incumbent seeds so take-up-gated programs do not ship at 100% participation; Populace does not yet produce it.", - "issue": "PolicyEngine/populace#312" - }, "takes_up_medicare_if_eligible": { "reason": "Stochastic take-up flag the incumbent seeds so take-up-gated programs do not ship at 100% participation; Populace does not yet produce it.", "issue": "PolicyEngine/populace#312" diff --git a/packages/populace-build/src/populace/build/us/source_stages.json b/packages/populace-build/src/populace/build/us/source_stages.json index 29313e62..4db052ff 100644 --- a/packages/populace-build/src/populace/build/us/source_stages.json +++ b/packages/populace-build/src/populace/build/us/source_stages.json @@ -922,6 +922,57 @@ "selected_marketplace_plan_benchmark_ratio" ], "notes": "When PolicyEngine-US separates observed Marketplace coverage from simulated take-up, this stage should write the simulated take-up input while reported coverage remains a CPS-carried person-level input." + }, + { + "stage": "medicaid_take_up", + "survey": "CPS ASEC reported coverage + CMS Medicaid monthly enrollment snapshot", + "source": "https://data.medicaid.gov/dataset/6165f45b-ca93-5bb5-9d06-db29c692a360", + "grain": "person", + "artifacts": [ + { + "kind": "public_microdata", + "format": "census_asec", + "vintage": "build_year", + "locator": "CPS ASEC current Medicaid coverage at interview (has_medicaid_health_coverage_at_interview)" + }, + { + "kind": "administrative_table", + "format": "cms_enrollment_snapshot", + "vintage": "month2024_12", + "locator": "CMS Medicaid and CHIP monthly enrollment, December 2024 state snapshot (cms_medicaid.month2024_12.state_enrollment.*.total_medicaid_enrollment)" + } + ], + "operations": [ + { + "kind": "read_table", + "table": "person", + "weight": "person_weight" + }, + { + "kind": "assign_binary_from_rate", + "output": "takes_up_medicaid_if_eligible", + "draw": "stable_person_draw", + "rate_key": "medicaid", + "rate_column": "medicaid_take_up_rate", + "reported_true_anchor": "has_medicaid_health_coverage_at_interview" + }, + { + "kind": "calibrate_binary_assignment", + "variable": "takes_up_medicaid_if_eligible", + "targets": [ + "cms_medicaid_enrollment_by_state" + ], + "preserve_true_anchors": true, + "preserve_true_anchor": "has_medicaid_health_coverage_at_interview", + "domain": "is_medicaid_eligible", + "weight": "person_weight", + "draw": "stable_person_draw" + } + ], + "outputs": [ + "takes_up_medicaid_if_eligible" + ], + "notes": "Anchored count-calibration (take-up contract treatment: count_calibrated; populace #331). Eligible reporters always take up; the fill draws at an in-build state rate (CMS count over weighted modeled eligibles, an assignment prior, never cited as provenance) and is greedily calibrated to the CMS December 2024 state snapshot among eligible non-anchored persons. Point-in-time (average-month) enrollment semantics per #332 - never ever-enrolled-in-year counts. The assign operation deliberately carries no eligibility mask: off-domain persons keep a draw-based propensity so eligibility-expanding reforms do not inherit a hard-coded zero response; the engine's is_medicaid_eligible gate hides the flag at baseline. States whose CMS count meets or exceeds modeled eligible weight saturate (all eligibles enroll) and are recorded as saturated in diagnostics, not failed (#170's eligibility-undercount shortfall is out of scope). CHIP deliberately excluded pending the #321 M-CHIP/separate-CHIP ledger concept split." } ] } diff --git a/packages/populace-build/src/populace/build/us/take_up_contract.json b/packages/populace-build/src/populace/build/us/take_up_contract.json index 5823da5e..3f587295 100644 --- a/packages/populace-build/src/populace/build/us/take_up_contract.json +++ b/packages/populace-build/src/populace/build/us/take_up_contract.json @@ -10,8 +10,9 @@ }, "doctrine": { "engine_class": "Derived from engine metadata. model_simulated = policyengine-us computes the flag itself (formula / adds / start-date formula); populace must NOT seed it. data_seeded = a default-True input leaf policyengine-us reads from data; without a dataset value every eligible unit takes up (mechanical 100% participation). dead = no downstream variable reads the flag; seeding it changes no output.", - "populace_treatment": "Curated per-program decision. seed = populace assigns the flag from an administrative participation rate with cited provenance. rate_unsourced = the flag is data_seeded but no administrative rate is traceable to a federal agency publication, so it is left unseeded and stays a #313 parity-gate exemption until the rate is sourced (ideally as a Ledger fact); copying us-data's unsourced value would recreate the fabricated-rate class ledger#77 removed. out_of_scope = handled by another workstream. near_universal = the administrative reality is ~100% take-up, so seeding sub-100% would be the error.", - "provenance_rule": "A rate may only drive a seed stage when its exact value is traceable to an administrative source (FNS/CMS/IRS/ACF/ASPE publication) at a stated vintage. Research-institute-only estimates (e.g. NIEER, unadjusted Urban microsimulation outputs) and model-relative ratios (admin numerator over a policyengine eligibility estimate) do not clear the bar and are recorded rate_unsourced." + "populace_treatment": "Curated per-program decision. seed = populace assigns the flag from an administrative participation rate with cited provenance. count_calibrated = no administrative participation rate exists, but administrative enrollment counts do: populace anchors the flag on survey-reported coverage and calibrates the fill to those count facts (the native ACA stage pattern), so no rate is ever cited and the model-relative-ratio class stays out of the inventory; the entry's calibration block names the anchor column and target fact families. rate_unsourced = the flag is data_seeded but no administrative rate is traceable to a federal agency publication, so it is left unseeded and stays a #313 parity-gate exemption until the rate is sourced (ideally as a Ledger fact); copying us-data's unsourced value would recreate the fabricated-rate class ledger#77 removed. out_of_scope = handled by another workstream. near_universal = the administrative reality is ~100% take-up, so seeding sub-100% would be the error.", + "provenance_rule": "A rate may only drive a seed stage when its exact value is traceable to an administrative source (FNS/CMS/IRS/ACF/ASPE publication) at a stated vintage. Research-institute-only estimates (e.g. NIEER, unadjusted Urban microsimulation outputs) and model-relative ratios (admin numerator over a policyengine eligibility estimate) do not clear the bar and are recorded rate_unsourced.", + "enrollment_semantics": "Enrollment take-up flags carry point-in-time (average-month) semantics: anchored on interview-point survey coverage and calibrated to month-tagged administrative enrollment snapshots, never to ever-enrolled-in-year counts (which run ~10M+ higher for Medicaid and would overstate spending against per-full-year-equivalent cost denominators). Precedent: Urban HIPSM targets the CMS point-in-time snapshot 'rather than an annual average' (HIPSM 2020 methodology). Within-year churn is out of scope for take-up stages (longitudinal frame direction, DESIGN.md). populace #332." }, "programs": [ { @@ -67,11 +68,18 @@ "entity": "person", "value_type": "bool", "default": true, - "populace_treatment": "rate_unsourced", + "populace_treatment": "count_calibrated", "consumed_via": "medicaid_enrolled (adds), gated by is_medicaid_eligible", - "rate": {"status": "model_relative", "us_data_value": "state-specific 0.53-0.99"}, - "followup": "Source a state Medicaid participation rate as a Ledger fact; #170 diagnostics motivate it.", - "notes": "This is the #170 symptom (enrollment == eligibility). us-data's state values are 'derived from MACPAC enrollment targets vs modeled eligibility' - an admin count over a model eligibility estimate, i.e. a calibration ratio internal to a microsimulation, not a published participation rate; ceiling values (0.99) are ratio artifacts. KFF's cited brief carries no state participation table. Left unseeded rather than recreate an unsourced rate." + "calibration": { + "anchor": "has_medicaid_health_coverage_at_interview", + "targets": ["cms_medicaid_enrollment_by_state"], + "fact_family": "cms_medicaid.month2024_12.state_enrollment.*.total_medicaid_enrollment", + "semantics": "point_in_time_monthly_snapshot (populace #332)", + "anchor_month_note": "The ASEC anchor is interview-point (~Feb-Apr) while the target is the December snapshot: both point-in-time stocks, months deliberately unaligned. With 2024 enrollment declining through the unwinding, spring reporters can exceed a December count; the stage gate's anchor-floor rule treats that as the reachable floor, not a miss." + }, + "rate": {"status": "no_administrative_rate_exists", "us_data_value": "state-specific 0.53-0.99 (rejected: model-relative ratio, MACPAC enrollment over modeled eligibility)"}, + "scope_owner": "populace#331 (medicaid_take_up stage)", + "notes": "Heals the #170 symptom (enrollment == eligibility). No administrative participation rate clears the provenance bar: us-data's state values are an admin count over a model eligibility estimate (ceiling values 0.99 are ratio artifacts) and KFF's cited brief carries no state participation table. The medicaid_take_up stage therefore cites no rate: eligible reporters of has_medicaid_health_coverage_at_interview always take up, and the fill is calibrated to CMS state monthly-snapshot enrollment counts (Ledger facts), the native ACA stage pattern. States where modeled eligible weight falls below the CMS count saturate (all eligibles enrolled) and are recorded as saturated in diagnostics, not treated as landmines - the shortfall is an eligibility-undercount question, out of scope here." }, { "variable": "takes_up_chip_if_eligible", @@ -82,8 +90,8 @@ "populace_treatment": "rate_unsourced", "consumed_via": "chip_enrolled (adds), gated by is_chip_eligible", "rate": {"status": "no_standalone_current_rate"}, - "followup": "Source a standalone CHIP participation rate (Ledger fact); current MACPAC/CMS publish enrollment counts and combined Medicaid+CHIP child coverage rates, not a standalone CHIP take-up rate.", - "notes": "us-data never carried a standalone CHIP rate. The nearest published figures (Medicaid+CHIP combined child participation, MACPAC/InsureKidsNow) are 2017-vintage, combined, and children-only - not a defensible standalone CHIP take-up rate today. Left unseeded." + "followup": "Blocked on populace#321: CMS total_chip_enrollment counts M-CHIP + separate CHIP while the model materializes separate CHIP only (20 states zero-support), so count-calibration against the current fact family would calibrate to a concept the model cannot reach. Once the ledger splits M-CHIP from separate-CHIP enrollment, CHIP follows the #331 medicaid_take_up count_calibrated pattern (anchor: has_other_means_tested_health_coverage_at_interview or the CHIP-specific reported column the split settles on).", + "notes": "us-data never carried a standalone CHIP rate. The nearest published figures (Medicaid+CHIP combined child participation, MACPAC/InsureKidsNow) are 2017-vintage, combined, and children-only - not a defensible standalone CHIP take-up rate today. Left unseeded pending the #321 concept split; deliberately NOT count_calibrated in the same pass as Medicaid." }, { "variable": "takes_up_basic_health_program_if_eligible", diff --git a/packages/populace-build/src/populace/build/us_runtime/__init__.py b/packages/populace-build/src/populace/build/us_runtime/__init__.py index 91ff5eb0..93bfaa91 100644 --- a/packages/populace-build/src/populace/build/us_runtime/__init__.py +++ b/packages/populace-build/src/populace/build/us_runtime/__init__.py @@ -147,6 +147,17 @@ from populace.build.us_runtime.input_mass import ( us_input_mass_totals, ) +from populace.build.us_runtime.medicaid_take_up import ( + US_MEDICAID_ENROLLMENT_TARGET_TABLE, + US_MEDICAID_ENROLLMENT_TOLERANCE, + US_MEDICAID_TAKE_UP_ANCHOR, + US_MEDICAID_TAKE_UP_STAGE, + US_MEDICAID_TAKE_UP_VARIABLE, + us_medicaid_take_up_diagnostics, + us_medicaid_take_up_gate, + with_us_medicaid_take_up, + write_us_medicaid_take_up_diagnostics, +) from populace.build.us_runtime.nonzero_shares import ( nonzero_share, us_nonzero_shares, @@ -243,6 +254,7 @@ TakeUpProgram, assert_take_up_contract_current, assert_take_up_treatments_consistent, + count_calibrated_take_up_programs, load_take_up_contract, seeded_take_up_programs, ) @@ -364,10 +376,20 @@ "with_us_immigration_inputs", "US_TAKE_UP_SHARE_BAND", "SeededTakeUpResult", + "US_MEDICAID_ENROLLMENT_TARGET_TABLE", + "US_MEDICAID_ENROLLMENT_TOLERANCE", + "US_MEDICAID_TAKE_UP_ANCHOR", + "US_MEDICAID_TAKE_UP_STAGE", + "US_MEDICAID_TAKE_UP_VARIABLE", + "count_calibrated_take_up_programs", + "us_medicaid_take_up_diagnostics", + "us_medicaid_take_up_gate", "us_take_up_participation_diagnostics", "us_take_up_signal_gate", "us_take_up_summary", + "with_us_medicaid_take_up", "with_us_take_up_inputs", + "write_us_medicaid_take_up_diagnostics", "write_us_take_up_participation_diagnostics", "TakeUpContract", "TakeUpProgram", @@ -532,6 +554,18 @@ def to_manifest(self) -> dict[str, object]: "calibration targets." ), ), + "medicaid_take_up": DonorSpec( + survey="CPS ASEC reported coverage + CMS Medicaid monthly enrollment snapshot", + source="https://data.medicaid.gov/dataset/6165f45b-ca93-5bb5-9d06-db29c692a360", + notes=( + "Medicaid take-up by anchored count-calibration (contract " + "treatment count_calibrated, populace #331): CPS-reported " + "Medicaid coverage at interview anchors the flag; the fill is " + "calibrated to CMS December 2024 state enrollment snapshots. " + "Point-in-time semantics per #332; heals the #170 " + "enrollment==eligibility degeneracy." + ), + ), "prior_year_income": DonorSpec( survey="CPS ASEC (prior year)", source="https://www.census.gov/programs-surveys/cps.html", @@ -666,6 +700,7 @@ def to_manifest(self) -> dict[str, object]: "vehicle_assets", "entity_placement", "aca_marketplace_inputs", + "medicaid_take_up", "export", ) diff --git a/packages/populace-build/src/populace/build/us_runtime/medicaid_take_up.py b/packages/populace-build/src/populace/build/us_runtime/medicaid_take_up.py new file mode 100644 index 00000000..9fb7f805 --- /dev/null +++ b/packages/populace-build/src/populace/build/us_runtime/medicaid_take_up.py @@ -0,0 +1,517 @@ +"""Medicaid take-up: anchored count-calibration to CMS state snapshots. + +policyengine-us gates ``medicaid_enrolled`` on ``is_medicaid_eligible`` and the +data-seeded ``takes_up_medicaid_if_eligible`` flag, which defaults to ``True``: +without a dataset value, enrollment mechanically equals eligibility (72.30M == +72.30M in the published 2024 release — populace #170). Worse, weight +calibration targets ``medicaid_enrolled`` against CMS counts, so an all-True +flag forces the solver to shrink the *eligible* population toward enrollment +counts, biasing weights for everything correlated with Medicaid eligibility. + +No administrative participation *rate* clears the take-up contract's +provenance bar (us-data's state rates were MACPAC enrollment over modeled +eligibility — a model-relative ratio), so this stage cites no rate. It follows +the native ACA stage pattern instead (``count_calibrated`` in the contract): + +1. **Anchor**: persons reporting Medicaid coverage at interview + (``has_medicaid_health_coverage_at_interview``, CPS ASEC) always take up. + The CPS undercounts Medicaid against administrative counts, so the anchor + is a floor, not the answer. +2. **Fill**: everyone else draws against an in-build state fill rate (CMS + state enrollment over weighted modeled eligibles — computed transparently + here, never cited as a sourced rate), then the assignment is greedily + calibrated to the CMS state enrollment counts among eligible non-anchored + persons (``calibrate_binary_assignment``), so unsaturated states hit their + administrative count exactly up to unit-weight granularity. + +Enrollment semantics are **point-in-time** (average month): the anchor is +interview-point coverage and the targets are month-tagged CMS snapshots +(``cms_medicaid.month2024_12.state_enrollment.*``), never ever-enrolled-in-year +counts — the Urban HIPSM convention (populace #332). Within-year churn is out +of scope. The anchor and target are different months of the same year (ASEC +reports coverage as of the ~February-April interview; the snapshot is +December): both are point-in-time stocks, and with 2024 enrollment declining +through the unwinding, spring reporters can exceed a December count in some +states — the gate's anchor-floor rule absorbs exactly that direction rather +than reading it as a calibration miss. + +Two deliberate asymmetries with the ACA stage: + +- **Off-domain persons keep their draw-based propensity** rather than being + forced ``False``: the assign operation runs without an ``eligibility`` mask, + so a person outside today's modeled eligibility carries a take-up value at + the state fill rate. The engine's eligibility gate makes this invisible at + baseline, but a reform that expands eligibility then enrolls newly eligible + people at a plausible propensity instead of a silent zero. (Not a behavioral + model — that is future utility-layer work; this only avoids hard-coding "no + reform response" into the data.) +- **Saturation is recorded, not failed**: where the CMS count exceeds the + modeled eligible weight, every eligible person enrolls and the state is + reported ``saturated`` — an eligibility-undercount symptom (#170's -2.4% + combined shortfall), distinct from the all-True landmine the gate exists to + catch. + +CHIP is deliberately not assigned here: CMS ``total_chip_enrollment`` mixes +M-CHIP into a concept the model does not materialize (populace #321); CHIP +follows this pattern once the ledger splits the concepts. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import numpy as np +import pandas as pd + +from populace.build.gates import GateResult +from populace.build.source_runtime import SourceRuntimeConfig, run_source_stage +from populace.build.us_runtime.take_up import ( + _SOURCE_IDENTITY_COLUMNS, + _stable_unit_draws, +) +from populace.frame import Frame +from populace.frame.units import US_SCHEMA + +__all__ = [ + "US_MEDICAID_ENROLLMENT_TARGET_TABLE", + "US_MEDICAID_ENROLLMENT_TOLERANCE", + "US_MEDICAID_TAKE_UP_ANCHOR", + "US_MEDICAID_TAKE_UP_STAGE", + "US_MEDICAID_TAKE_UP_VARIABLE", + "us_medicaid_source_person_table", + "us_medicaid_take_up_diagnostics", + "us_medicaid_take_up_gate", + "with_us_medicaid_take_up", + "with_us_medicaid_take_up_rate", + "write_us_medicaid_take_up_diagnostics", +] + +US_MEDICAID_TAKE_UP_STAGE = "medicaid_take_up" +US_MEDICAID_TAKE_UP_VARIABLE = "takes_up_medicaid_if_eligible" +US_MEDICAID_TAKE_UP_ANCHOR = "has_medicaid_health_coverage_at_interview" +US_MEDICAID_ELIGIBILITY_COLUMN = "is_medicaid_eligible" +US_MEDICAID_ENROLLMENT_TARGET_TABLE = "cms_medicaid_enrollment_by_state" + +#: Relative miss tolerance for an unsaturated state's calibrated enrollment +#: against its CMS count. Greedy calibration overshoots by at most one unit +#: weight per state; the gate widens this to the state's largest person +#: weight where that exceeds 2%, so small-state granularity never reads as a +#: real calibration failure. +US_MEDICAID_ENROLLMENT_TOLERANCE = 0.02 + +_DRAW_COLUMN = "stable_person_draw" +_RATE_COLUMN = "medicaid_take_up_rate" + + +def _stable_person_draws(person: pd.DataFrame, *, seed: int) -> np.ndarray: + """Seeded uniform draws keyed by stable source identity per person. + + Delegates to the seeded take-up stages' shared blake2b keying + (:func:`populace.build.us_runtime.take_up._stable_unit_draws` at person + grain) so support-channel clones of one source person draw together, + reruns are bit-reproducible, and a future keying change lands in one + place. + """ + return _stable_unit_draws( + person, + id_column="person_id", + seed=seed, + variable=US_MEDICAID_TAKE_UP_VARIABLE, + ) + + +def us_medicaid_source_person_table( + frame: Frame, + *, + is_medicaid_eligible: np.ndarray, + state_fips: np.ndarray, + seed: int, +) -> pd.DataFrame: + """Assemble the person-grain stage table the manifest operations consume. + + Engine-derived columns (eligibility, state) are supplied by the caller — + the builder owns the simulation, this module owns assembly — and the + reported-coverage anchor must already ride the person table (a CPS-carried + column). + + Args: + frame: A US-schema frame. + is_medicaid_eligible: Person-aligned boolean eligibility from the + engine (``is_medicaid_eligible`` materialized at person grain). + state_fips: Person-aligned state FIPS codes (any numeric/str form; + normalized to zero-padded two-char strings to match target rows). + seed: Build-wide imputation seed for the stable draws. + + Raises: + ValueError: If the frame is not US-schema, the anchor column is + missing, or the supplied arrays do not align with the person + table. + """ + if frame.schema != US_SCHEMA: + raise ValueError("US Medicaid take-up requires the US schema.") + person = frame.table("person") + count = len(person) + if len(is_medicaid_eligible) != count or len(state_fips) != count: + raise ValueError( + "US Medicaid take-up inputs must align with the person table: " + f"{count} persons, {len(is_medicaid_eligible)} eligibility values, " + f"{len(state_fips)} state codes." + ) + if US_MEDICAID_TAKE_UP_ANCHOR not in person.columns: + raise ValueError( + f"US Medicaid take-up requires the {US_MEDICAID_TAKE_UP_ANCHOR!r} " + "reported-coverage column on the person table." + ) + + columns = { + "person_id": person["person_id"].to_numpy(), + US_MEDICAID_TAKE_UP_ANCHOR: person[US_MEDICAID_TAKE_UP_ANCHOR] + .fillna(False) + .to_numpy(dtype=bool), + US_MEDICAID_ELIGIBILITY_COLUMN: np.asarray(is_medicaid_eligible, dtype=bool), + "state_fips": _normalize_state_fips(state_fips), + "person_weight": np.asarray( + frame.resolve_weights("person").values, dtype=np.float64 + ), + } + for column in _SOURCE_IDENTITY_COLUMNS: + if column in person.columns: + columns[column] = person[column].to_numpy() + table = pd.DataFrame(columns) + table[_DRAW_COLUMN] = _stable_person_draws(table, seed=seed) + return table + + +def _normalize_state_fips(values: np.ndarray) -> np.ndarray: + """Zero-padded two-char state codes from numeric or string input. + + Deliberately NOT named like the builder's ``_state_fips_text`` (which has + different semantics — int coercion, bytes handling) so the two cannot be + confused; missing values are refused rather than silently becoming the + string ``'nan'`` and surfacing later as a states-without-targets gate + failure. + """ + series = pd.Series(values) + if series.isna().any(): + raise ValueError("US Medicaid take-up state_fips contains missing values.") + text = series.astype(str).str.replace(r"\.0$", "", regex=True) + return text.str.zfill(2).to_numpy() + + +def with_us_medicaid_take_up_rate( + table: pd.DataFrame, state_targets: pd.DataFrame +) -> pd.DataFrame: + """Derive the in-build state fill rate: CMS count over weighted eligibles. + + This mirrors the ACA stage's rate derivation and is an assignment prior, + not a sourced participation rate — the calibration operation is what makes + unsaturated states hit their count. The ratio is computed transparently + in-build so the contract never records it as provenance (the doctrine's + model-relative-ratio rule). + """ + result = table.copy(deep=True) + if state_targets.empty: + result[_RATE_COLUMN] = 0.0 + return result + _require_target_columns(state_targets) + eligible = result[US_MEDICAID_ELIGIBILITY_COLUMN].fillna(False).astype(bool) + weighted_eligible = ( + result.loc[eligible].groupby("state_fips")["person_weight"].sum().astype(float) + ) + targets = state_targets.groupby("state_fips")["target"].sum().astype(float) + rates = (targets / weighted_eligible).replace([np.inf, -np.inf], np.nan).fillna(0.0) + result[_RATE_COLUMN] = ( + result["state_fips"].map(rates).fillna(0.0).clip(lower=0.0, upper=1.0) + ) + return result + + +def _require_target_columns(state_targets: pd.DataFrame) -> None: + missing = [ + column for column in ("state_fips", "target") if column not in state_targets + ] + if missing: + raise ValueError( + "US Medicaid enrollment targets require columns " + f"['state_fips', 'target']; missing {missing}." + ) + + +def with_us_medicaid_take_up( + frame: Frame, + *, + is_medicaid_eligible: np.ndarray, + state_fips: np.ndarray, + state_targets: pd.DataFrame, + seed: int, +) -> tuple[Frame, dict[str, object]]: + """Assign ``takes_up_medicaid_if_eligible`` onto the frame's person table. + + Runs the ``medicaid_take_up`` manifest stage (anchor OR draw-under-rate, + then greedy count-calibration among eligible non-anchored persons) and + writes the flag back. The column is always recomputed — this stage owns + it; a persisted landmine is repaired, not trusted. + + Returns: + The new frame and the release diagnostics payload + (:func:`us_medicaid_take_up_diagnostics` of the assigned table). + + Raises: + ValueError: If ``state_targets`` is empty or malformed — an empty + feed would silently revert every state to anchored-only + enrollment, so it refuses to run rather than degrade. + """ + from populace.build.us_runtime import ( # local: package init owns the manifest + US_SOURCE_MANIFEST, + us_source_operation_handlers, + ) + + if state_targets.empty: + raise ValueError( + "US Medicaid take-up requires non-empty CMS state enrollment " + "targets; an empty feed would ship anchored-only enrollment." + ) + _require_target_columns(state_targets) + table = us_medicaid_source_person_table( + frame, + is_medicaid_eligible=is_medicaid_eligible, + state_fips=state_fips, + seed=seed, + ) + table = with_us_medicaid_take_up_rate(table, state_targets) + + stage = US_SOURCE_MANIFEST.stage_map()[US_MEDICAID_TAKE_UP_STAGE] + output = run_source_stage( + stage, + tables={ + "person": table, + US_MEDICAID_ENROLLMENT_TARGET_TABLE: state_targets, + }, + operation_handlers=us_source_operation_handlers(), + config=SourceRuntimeConfig(seed=seed), + ) + + person = frame.table("person").copy() + assigned = ( + output.set_index("person_id")[US_MEDICAID_TAKE_UP_VARIABLE] + .reindex(person["person_id"]) + .to_numpy() + ) + if pd.isna(assigned).any(): + raise ValueError( + "US Medicaid take-up stage output does not cover every person." + ) + person[US_MEDICAID_TAKE_UP_VARIABLE] = assigned.astype(bool) + + tables = { + entity: person if entity == "person" else frame.table(entity).copy() + for entity in frame.entities + } + result = Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + frame.strata, + mass_log=frame.mass_log, + ) + diagnostics = us_medicaid_take_up_diagnostics(output, state_targets) + return result, diagnostics + + +def us_medicaid_take_up_diagnostics( + assigned: pd.DataFrame, state_targets: pd.DataFrame +) -> dict[str, object]: + """The #170 eligibility-to-enrollment surface, by state and nationally. + + ``assigned`` is the stage output table (person grain, carrying the flag, + eligibility, anchor, weight, and state). Each state row records eligible + weight, anchored-eligible weight, modeled enrollment (flag AND eligible — + the engine's gate), the CMS target, the enrollment/eligibility ratio, and + whether the state is ``saturated`` (target at or above eligible weight, so + full enrollment is the calibrated answer, not a landmine). + """ + eligible = assigned[US_MEDICAID_ELIGIBILITY_COLUMN].fillna(False).astype(bool) + anchored = assigned[US_MEDICAID_TAKE_UP_ANCHOR].fillna(False).astype(bool) + takes_up = assigned[US_MEDICAID_TAKE_UP_VARIABLE].fillna(False).astype(bool) + weights = pd.to_numeric(assigned["person_weight"], errors="coerce").fillna(0.0) + enrolled = takes_up & eligible + # The anchor is unmasked by design, so EVERY anchored reporter must carry + # the flag — a person-level invariant the gate checks even in saturated + # states, where aggregate comparisons cannot see a dropped anchor. + anchor_violated = anchored & ~takes_up + + targets_by_state: dict[str, float] = {} + if not state_targets.empty: + _require_target_columns(state_targets) + targets_by_state = ( + state_targets.groupby("state_fips")["target"].sum().astype(float).to_dict() + ) + + states = [] + for state, group in assigned.groupby("state_fips", sort=True): + index = group.index + eligible_weight = float(weights[index][eligible[index]].sum()) + enrolled_weight = float(weights[index][enrolled[index]].sum()) + anchored_eligible_weight = float( + weights[index][(anchored & eligible)[index]].sum() + ) + target = targets_by_state.get(str(state)) + saturated = target is not None and target >= eligible_weight + states.append( + { + "state_fips": str(state), + "target": target, + "eligible_weight": eligible_weight, + "anchored_eligible_weight": anchored_eligible_weight, + "enrolled_weight": enrolled_weight, + "enrollment_to_eligibility_ratio": ( + enrolled_weight / eligible_weight if eligible_weight > 0 else None + ), + "saturated": bool(saturated), + "anchored_not_taking_up_count": int(anchor_violated[index].sum()), + # Greedy calibration lands within one person of the count; + # the gate uses this to size the granularity allowance. + "max_person_weight": float(weights[index].max()) if len(index) else 0.0, + } + ) + + total_eligible = float(weights[eligible].sum()) + total_enrolled = float(weights[enrolled].sum()) + return { + "schema_version": 1, + "classification": "release_diagnostics", + "issues": ["populace#331", "populace#170", "populace#332"], + "variable": US_MEDICAID_TAKE_UP_VARIABLE, + "anchor": US_MEDICAID_TAKE_UP_ANCHOR, + "enrollment_semantics": "point_in_time_monthly_snapshot", + "target_table": US_MEDICAID_ENROLLMENT_TARGET_TABLE, + # Stage-time surface: weights are the pre-calibration design weights. + # Post-calibration weighted enrollment is pulled to the same CMS + # counts by the medicaid_enrollment weight-calibration targets, but + # eligible/anchored weights here have no post-calibration counterpart. + "weights_basis": "pre_calibration_design_weights", + "national": { + "eligible_weight": total_eligible, + "enrolled_weight": total_enrolled, + "enrollment_to_eligibility_ratio": ( + total_enrolled / total_eligible if total_eligible > 0 else None + ), + "target_total": float(sum(targets_by_state.values())) + if targets_by_state + else None, + "anchored_not_taking_up_count": int(anchor_violated.sum()), + }, + "states": states, + "states_without_targets": sorted( + {str(s) for s in assigned["state_fips"].unique()} - set(targets_by_state) + ), + "saturated_states": sorted( + row["state_fips"] for row in states if row["saturated"] + ), + } + + +def us_medicaid_take_up_gate( + diagnostics: dict[str, object], + *, + tolerance: float = US_MEDICAID_ENROLLMENT_TOLERANCE, +) -> GateResult: + """Require a non-degenerate, count-faithful Medicaid take-up surface. + + Fails when: + + - any state has no CMS enrollment target (a feed hole would silently + revert that state to anchored-only enrollment); + - any state carries a positive CMS count with zero modeled-eligible + weight — an eligibility-feed collapse. Such a state classifies as + ``saturated`` (the count trivially exceeds zero eligibility), so + without this check it would skip the count checks and ship silently + wrong; + - any anchored reporter lost the flag — a person-level invariant checked + in EVERY state, saturated or not (aggregate comparisons cannot see a + dropped anchor once calibration back-fills to the count); + - an unsaturated state fully enrolls its eligibles when the reachable + floor (CMS count or anchor mass, whichever is higher) sits meaningfully + below eligibility — the #170 landmine. Full enrollment explained by + anchor mass or by a floor within granularity of eligibility is a + legitimate calibrated outcome, not a landmine; + - an unsaturated state misses its floor by more than the granularity + allowance (``tolerance`` relative, widened to one person weight where + that is larger — greedy calibration lands within one person). + + Saturated states (CMS count at or above modeled eligible weight) skip the + count checks: there, full enrollment IS the calibrated answer and the + shortfall is an eligibility-undercount question outside this stage's + scope. + """ + failures: list[str] = [] + states = diagnostics.get("states", []) + missing = diagnostics.get("states_without_targets", []) + if missing: + failures.append( + f"states without CMS enrollment targets: {missing} — the feed is " + "incomplete; those states would ship anchored-only enrollment." + ) + for row in states: + state = row["state_fips"] + target = row["target"] + violations = int(row.get("anchored_not_taking_up_count", 0)) + if violations: + failures.append( + f"state {state}: {violations} anchored reporter(s) lost the " + "flag; the reported-coverage anchor was not preserved." + ) + if target is not None and target > 0 and float(row["eligible_weight"]) <= 0: + failures.append( + f"state {state}: CMS count {target:.0f} with zero " + "modeled-eligible weight — the eligibility feed collapsed; " + "its saturated classification is a division artifact, not a " + "calibrated answer." + ) + continue + if target is None or row["saturated"]: + continue + eligible_weight = float(row["eligible_weight"]) + enrolled_weight = float(row["enrolled_weight"]) + anchored_weight = float(row["anchored_eligible_weight"]) + max_person_weight = float(row.get("max_person_weight", 0.0)) + # An unsaturated state's floor is its anchor mass: when reported + # coverage already exceeds the CMS count, calibration cannot remove + # anchored persons (by design) and the count is unreachable downward. + floor = max(float(target), anchored_weight) + granularity = max(tolerance * eligible_weight, max_person_weight) + if ( + eligible_weight > 0 + and enrolled_weight >= eligible_weight + and floor < eligible_weight - granularity + ): + failures.append( + f"state {state}: enrollment equals eligibility while the " + f"reachable floor {floor:.0f} sits below eligible weight " + f"{eligible_weight:.0f} — the universal-take-up landmine " + "(#170)." + ) + continue + allowed_miss = max(tolerance * floor, max_person_weight) + if floor > 0 and abs(enrolled_weight - floor) > allowed_miss: + failures.append( + f"state {state}: enrolled weight {enrolled_weight:.0f} misses " + f"the CMS count {target:.0f} (floor {floor:.0f}) by more than " + f"the granularity allowance {allowed_miss:.0f}." + ) + return GateResult( + name="medicaid_take_up", + passed=not failures, + failures=tuple(failures), + details=diagnostics, + ) + + +def write_us_medicaid_take_up_diagnostics( + payload: dict[str, object], path: Path | str +) -> Path: + """Write the Medicaid take-up diagnostics artifact as strict JSON.""" + path = Path(path) + path.write_text(json.dumps(payload, indent=1, allow_nan=False)) + return path diff --git a/packages/populace-build/src/populace/build/us_runtime/take_up.py b/packages/populace-build/src/populace/build/us_runtime/take_up.py index 7a0b9968..baef1d7b 100644 --- a/packages/populace-build/src/populace/build/us_runtime/take_up.py +++ b/packages/populace-build/src/populace/build/us_runtime/take_up.py @@ -482,6 +482,47 @@ def us_take_up_participation_diagnostics( row["take_up_share"] = info["take_up_share"] row["administrative_rate"] = info["administrative_rate"] row["share_band"] = info["share_band"] + elif program.populace_treatment == "count_calibrated": + # Assigned by its own anchored count-calibration stage (e.g. the + # medicaid_take_up stage, populace #331); a non-constant column on + # the frame is the live surface, a missing or constant one means + # the stage did not run and the flag ships at the engine default. + row["seeded"] = False + table = frame.table(program.entity) + materialized = variable in table.columns and _column_carries_signal( + table, variable + ) + row["count_calibrated"] = materialized + row["ships_at_engine_default"] = not materialized + # The owning-issue pointer must survive in BOTH cases — it is the + # debt-ledger surface this payload exists for. + followup = program.raw.get("followup") + if followup is not None: + row["followup"] = followup + scope_owner = program.raw.get("scope_owner") + if scope_owner is not None: + row["scope_owner"] = scope_owner + if materialized: + weights = np.asarray( + frame.resolve_weights(program.entity).values, dtype=np.float64 + ) + takes_up = table[variable].to_numpy(dtype=bool) + total_weight = float(weights.sum()) + row["weighted_participation_count"] = float(weights[takes_up].sum()) + row["weighted_flag_universe"] = total_weight + row["take_up_share"] = ( + float(weights[takes_up].sum()) / total_weight + if total_weight > 0 + else 0.0 + ) + # Count-calibrated flags may deliberately carry off-domain + # (ineligible) True values as reform propensities, so this + # share is flag-True over ALL units — not enrollment over + # eligibility. The stage's own diagnostics artifact carries + # the eligibility-restricted surface. + row["share_universe"] = ( + f"all_{program.entity}s_including_off_domain_propensity" + ) else: row["seeded"] = False # Left at the engine's universal-take-up default: no seeded diff --git a/packages/populace-build/src/populace/build/us_runtime/take_up_contract.py b/packages/populace-build/src/populace/build/us_runtime/take_up_contract.py index 52307551..800e0bae 100644 --- a/packages/populace-build/src/populace/build/us_runtime/take_up_contract.py +++ b/packages/populace-build/src/populace/build/us_runtime/take_up_contract.py @@ -10,9 +10,9 @@ This module owns the *inventory*: a checked-in table (``populace/build/us/take_up_contract.json``) recording, per program, the engine facts that decide treatment (entity, default, and the derived ``engine_class``) -and the curated populace treatment (seed / rate_unsourced / model_simulated / -out_of_scope / near_universal) with the administrative provenance of any rate -used. The engine facts are asserted against the installed engine by +and the curated populace treatment (seed / count_calibrated / rate_unsourced / +model_simulated / out_of_scope / near_universal) with the administrative +provenance of any rate used. The engine facts are asserted against the installed engine by :func:`assert_take_up_contract_current`, so the classification tracks the pinned policyengine-us version instead of a remembered snapshot -- the same metadata-derivation doctrine as the #301 formula-owned guard. @@ -38,6 +38,7 @@ "TakeUpProgram", "assert_take_up_contract_current", "assert_take_up_treatments_consistent", + "count_calibrated_take_up_programs", "load_take_up_contract", "seeded_take_up_programs", ] @@ -56,6 +57,7 @@ _VALID_TREATMENTS = frozenset( { "seed", + "count_calibrated", "rate_unsourced", "model_simulated", "out_of_scope", @@ -164,6 +166,7 @@ def load_take_up_contract() -> TakeUpContract: f"program {variable!r} is missing engine fact key(s): {missing}." ) _validate_seed_provenance(variable, treatment, entry) + _validate_count_calibration(variable, treatment, entry) programs.append( TakeUpProgram( variable=variable, @@ -221,11 +224,63 @@ def _validate_seed_provenance( ) +def _validate_count_calibration( + variable: str, treatment: str, entry: Mapping[str, Any] +) -> None: + """A count-calibrated program must name its anchor and count targets. + + ``count_calibrated`` is the doctrine-compliant path when no administrative + participation *rate* exists but administrative enrollment *counts* do (the + native ACA stage precedent): assignment anchors on survey-reported coverage + and calibrates the fill to Ledger count facts, so no rate is ever cited. + The entry must therefore carry a ``calibration`` block naming the anchor + column and the target fact family — and must NOT carry a sourced rate, or + the treatment is misclassified (a sourced rate is what ``seed`` is for). + """ + if treatment != "count_calibrated": + return + calibration = entry.get("calibration") + if not isinstance(calibration, Mapping): + raise ValueError( + f"count_calibrated program {variable!r} requires a 'calibration' " + "block naming its anchor and count targets." + ) + targets = calibration.get("targets") + if not isinstance(targets, (list, tuple)) or not targets: + raise ValueError( + f"count_calibrated program {variable!r} calibration requires a " + "non-empty 'targets' list of Ledger count-fact families." + ) + if not str(calibration.get("anchor") or ""): + raise ValueError( + f"count_calibrated program {variable!r} calibration requires an " + "'anchor' reported-coverage column." + ) + rate = entry.get("rate") + if isinstance(rate, Mapping) and str(rate.get("status") or "").startswith( + "sourced" + ): + raise ValueError( + f"count_calibrated program {variable!r} carries a sourced rate; " + "a program with an administrative rate should be 'seed', not " + "'count_calibrated'." + ) + + def seeded_take_up_programs() -> tuple[TakeUpProgram, ...]: """The programs populace seeds, in table order.""" return load_take_up_contract().seeded() +def count_calibrated_take_up_programs() -> tuple[TakeUpProgram, ...]: + """The programs populace assigns via anchored count-calibration.""" + return tuple( + program + for program in load_take_up_contract().programs + if program.populace_treatment == "count_calibrated" + ) + + def assert_take_up_contract_current(engine: PolicyEngineUSEngine | None = None) -> None: """Fail when the checked-in inventory drifts from the installed engine. diff --git a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py index 829d5419..3257bd7a 100644 --- a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py +++ b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py @@ -1937,6 +1937,20 @@ def n(self, entity): details={"checked": True}, ), ) + monkeypatch.setattr( + builder, + "_with_medicaid_take_up_outputs", + lambda frame, specs, *, seed, maximum_microsim_batch_size=None: (frame, {}), + ) + monkeypatch.setattr( + builder, + "us_medicaid_take_up_gate", + lambda diagnostics: builder.GateResult( + name="medicaid_take_up", + passed=True, + details={"checked": True}, + ), + ) def fake_materialize_target_frame(frame, specs, **kwargs): captured["materialize_kwargs"] = kwargs diff --git a/packages/populace-build/tests/test_us_medicaid_take_up.py b/packages/populace-build/tests/test_us_medicaid_take_up.py new file mode 100644 index 00000000..6fa8eb02 --- /dev/null +++ b/packages/populace-build/tests/test_us_medicaid_take_up.py @@ -0,0 +1,605 @@ +"""Medicaid take-up stage tests (populace #331). + +Covers the count_calibrated contract treatment (validation that the doctrine +cannot be bypassed), the anchored count-calibration itself (anchors always +enroll, unsaturated states hit their CMS count to unit-weight granularity, +saturated states enroll every eligible and are classified saturated), +determinism and source-identity draw keying, the off-domain propensity +carried for eligibility-expanding reforms, the gate's failure modes (with +prove-it-can-find-something checks), and the builder helpers that compile the +CMS state target table and person state codes. +""" + +from __future__ import annotations + +import copy +import importlib.util +import json +from pathlib import Path +from types import SimpleNamespace + +import numpy as np +import pandas as pd +import pytest + +pytest.importorskip("policyengine_us") + +from populace.build.us_runtime.medicaid_take_up import ( # noqa: E402 + US_MEDICAID_TAKE_UP_ANCHOR, + US_MEDICAID_TAKE_UP_VARIABLE, + _stable_person_draws, + us_medicaid_take_up_diagnostics, + us_medicaid_take_up_gate, + with_us_medicaid_take_up, +) +from populace.build.us_runtime.take_up import ( # noqa: E402 + us_take_up_participation_diagnostics, +) +from populace.build.us_runtime.take_up_contract import ( # noqa: E402 + count_calibrated_take_up_programs, + load_take_up_contract, + seeded_take_up_programs, +) +from populace.frame import US_SCHEMA, Frame, WeightKind, Weights # noqa: E402 + +UNIT_WEIGHT = 100.0 + + +def _frame(n: int, *, anchor: np.ndarray, source_identity: bool = True) -> Frame: + """Synthetic single-person-household US frame with a reported anchor.""" + person = pd.DataFrame( + { + "person_id": range(n), + "age": [30 + (i % 40) for i in range(n)], + US_MEDICAID_TAKE_UP_ANCHOR: np.asarray(anchor, dtype=bool), + } + ) + for entity in ("household", "tax_unit", "spm_unit", "family"): + person[f"person_{entity}_id"] = person["person_id"] + person["person_marital_unit_id"] = person["person_id"] + if source_identity: + person["source_year"] = 2023 + person["source_household_id"] = person["person_id"] + person["source_person_id"] = person["person_id"] + ids = person["person_id"].to_numpy() + tables = { + "person": person, + "household": pd.DataFrame({"household_id": ids}), + "tax_unit": pd.DataFrame({"tax_unit_id": ids}), + "spm_unit": pd.DataFrame({"spm_unit_id": ids}), + "family": pd.DataFrame({"family_id": ids}), + "marital_unit": pd.DataFrame({"marital_unit_id": ids}), + } + return Frame( + tables, + US_SCHEMA, + {"household": Weights(values=np.full(n, UNIT_WEIGHT), kind=WeightKind.DESIGN)}, + ) + + +def _scenario(n: int = 400, seed: int = 7): + """Two-state scenario: state 01 unsaturated (0.7x), state 02 saturated.""" + rng = np.random.default_rng(0) + anchor = rng.random(n) < 0.30 + frame = _frame(n, anchor=anchor) + state = np.where(np.arange(n) % 2 == 0, "01", "02") + eligible = rng.random(n) < 0.60 + eligible_weight_01 = UNIT_WEIGHT * ((state == "01") & eligible).sum() + eligible_weight_02 = UNIT_WEIGHT * ((state == "02") & eligible).sum() + targets = pd.DataFrame( + { + "state_fips": ["01", "02"], + "target": [0.7 * eligible_weight_01, 1.2 * eligible_weight_02], + } + ) + result, diagnostics = with_us_medicaid_take_up( + frame, + is_medicaid_eligible=eligible, + state_fips=state, + state_targets=targets, + seed=seed, + ) + return frame, result, diagnostics, anchor, eligible, state, targets + + +class TestContractTreatment: + def test_medicaid_is_count_calibrated_not_seeded(self) -> None: + assert [p.variable for p in count_calibrated_take_up_programs()] == [ + "takes_up_medicaid_if_eligible" + ] + assert "takes_up_medicaid_if_eligible" not in { + p.variable for p in seeded_take_up_programs() + } + + def test_calibration_block_names_anchor_and_targets(self) -> None: + program = load_take_up_contract().program_map()["takes_up_medicaid_if_eligible"] + calibration = program.raw["calibration"] + assert calibration["anchor"] == US_MEDICAID_TAKE_UP_ANCHOR + assert calibration["targets"] + + def test_calibration_block_matches_the_stage_manifest(self) -> None: + # The contract must describe what the release actually calibrates to: + # reconcile its calibration block against the manifest stage's ops so + # the inventory cannot silently drift from the stage (#331 review). + from populace.build.us_runtime import US_SOURCE_MANIFEST + + program = load_take_up_contract().program_map()["takes_up_medicaid_if_eligible"] + calibration = program.raw["calibration"] + stage = US_SOURCE_MANIFEST.stage_map()["medicaid_take_up"] + ops = {op.kind: op.parameters for op in stage.operations} + assert list(calibration["targets"]) == list( + ops["calibrate_binary_assignment"]["targets"] + ) + assert ( + calibration["anchor"] + == ops["assign_binary_from_rate"]["reported_true_anchor"] + == ops["calibrate_binary_assignment"]["preserve_true_anchor"] + ) + + def _reload_with(self, monkeypatch, tmp_path: Path, mutated: dict) -> None: + load_take_up_contract.cache_clear() + path = tmp_path / "take_up_contract.json" + path.write_text(json.dumps(mutated)) + monkeypatch.setattr( + "populace.build.us_runtime.take_up_contract._contract_path", + lambda: path, + ) + + @pytest.fixture() + def base_table(self): + from populace.build.us_runtime.take_up_contract import _contract_path + + raw = json.loads(_contract_path().read_text()) + yield raw + load_take_up_contract.cache_clear() + + def test_count_calibrated_without_calibration_block_fails( + self, monkeypatch, tmp_path, base_table + ) -> None: + mutated = copy.deepcopy(base_table) + for program in mutated["programs"]: + if program["variable"] == "takes_up_medicaid_if_eligible": + program.pop("calibration") + self._reload_with(monkeypatch, tmp_path, mutated) + with pytest.raises(ValueError, match="calibration"): + load_take_up_contract() + + def test_count_calibrated_with_sourced_rate_fails( + self, monkeypatch, tmp_path, base_table + ) -> None: + # A sourced rate means the program belongs in `seed`; carrying one on a + # count_calibrated entry is a misclassification the loader must refuse. + mutated = copy.deepcopy(base_table) + for program in mutated["programs"]: + if program["variable"] == "takes_up_medicaid_if_eligible": + program["rate"] = { + "value": 0.8, + "source": "https://example.gov", + "status": "sourced_administrative", + } + self._reload_with(monkeypatch, tmp_path, mutated) + with pytest.raises(ValueError, match="sourced rate"): + load_take_up_contract() + + +class TestAssignment: + def test_anchored_reporters_always_take_up(self) -> None: + _, result, _, anchor, _, _, _ = _scenario() + flag = result.table("person")[US_MEDICAID_TAKE_UP_VARIABLE].to_numpy() + assert flag[anchor].all() + + def test_anchored_but_model_ineligible_reporters_keep_the_flag(self) -> None: + # The anchor is deliberately unmasked by eligibility: the engine's + # is_medicaid_eligible gate hides the flag at baseline, and forcing it + # False would erase the observation. + _, result, _, anchor, eligible, _, _ = _scenario() + flag = result.table("person")[US_MEDICAID_TAKE_UP_VARIABLE].to_numpy() + ineligible_reporters = anchor & ~eligible + assert ineligible_reporters.any() + assert flag[ineligible_reporters].all() + + def test_unsaturated_state_hits_cms_count_to_unit_weight(self) -> None: + _, result, _, _, eligible, state, targets = _scenario() + flag = result.table("person")[US_MEDICAID_TAKE_UP_VARIABLE].to_numpy() + enrolled_weight = UNIT_WEIGHT * (flag & eligible & (state == "01")).sum() + target = float(targets.loc[targets["state_fips"] == "01", "target"].iloc[0]) + assert abs(enrolled_weight - target) <= UNIT_WEIGHT + + def test_saturated_state_enrolls_every_eligible_and_is_classified(self) -> None: + _, result, diagnostics, _, eligible, state, _ = _scenario() + flag = result.table("person")[US_MEDICAID_TAKE_UP_VARIABLE].to_numpy() + assert flag[eligible & (state == "02")].all() + assert diagnostics["saturated_states"] == ["02"] + + def test_rerun_is_bit_reproducible(self) -> None: + frame, result, _, _, eligible, state, targets = _scenario(seed=7) + again, _ = with_us_medicaid_take_up( + frame, + is_medicaid_eligible=eligible, + state_fips=state, + state_targets=targets, + seed=7, + ) + assert ( + result.table("person")[US_MEDICAID_TAKE_UP_VARIABLE].to_numpy() + == again.table("person")[US_MEDICAID_TAKE_UP_VARIABLE].to_numpy() + ).all() + + def test_draws_key_on_source_identity(self) -> None: + # Support-channel clones of one source person share the draw, so the + # Bernoulli phase treats them identically and they are adjacent in the + # greedy fill order. + table = pd.DataFrame( + { + "person_id": [0, 1], + "source_year": [2023, 2023], + "source_household_id": [5, 5], + "source_person_id": [9, 9], + } + ) + draws = _stable_person_draws(table, seed=3) + assert draws[0] == draws[1] + + def test_off_domain_persons_carry_a_propensity(self) -> None: + # Non-anchored, model-ineligible persons keep a draw-based flag at the + # state fill rate so eligibility-expanding reforms do not inherit a + # hard-coded zero response. + _, result, _, anchor, eligible, _, _ = _scenario() + flag = result.table("person")[US_MEDICAID_TAKE_UP_VARIABLE].to_numpy() + off_domain = ~anchor & ~eligible + assert off_domain.any() + assert flag[off_domain].any() + assert not flag[off_domain].all() + + def test_missing_anchor_column_raises(self) -> None: + frame = _frame(10, anchor=np.zeros(10, dtype=bool)) + person = frame.table("person").drop(columns=[US_MEDICAID_TAKE_UP_ANCHOR]) + tables = {entity: frame.table(entity) for entity in frame.entities} + tables["person"] = person + stripped = Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + ) + with pytest.raises(ValueError, match=US_MEDICAID_TAKE_UP_ANCHOR): + with_us_medicaid_take_up( + stripped, + is_medicaid_eligible=np.ones(10, dtype=bool), + state_fips=np.full(10, "01"), + state_targets=pd.DataFrame({"state_fips": ["01"], "target": [500.0]}), + seed=0, + ) + + def test_misaligned_inputs_raise(self) -> None: + frame = _frame(10, anchor=np.zeros(10, dtype=bool)) + with pytest.raises(ValueError, match="align"): + with_us_medicaid_take_up( + frame, + is_medicaid_eligible=np.ones(4, dtype=bool), + state_fips=np.full(10, "01"), + state_targets=pd.DataFrame({"state_fips": ["01"], "target": [500.0]}), + seed=0, + ) + + def test_malformed_targets_raise(self) -> None: + frame = _frame(10, anchor=np.zeros(10, dtype=bool)) + with pytest.raises(ValueError, match="state_fips"): + with_us_medicaid_take_up( + frame, + is_medicaid_eligible=np.ones(10, dtype=bool), + state_fips=np.full(10, "01"), + state_targets=pd.DataFrame({"state": ["01"], "count": [500.0]}), + seed=0, + ) + + def test_empty_targets_raise_early(self) -> None: + # An empty feed must refuse to run, not silently ship anchored-only + # enrollment or surface as a deep missing-value-column error. + frame = _frame(10, anchor=np.zeros(10, dtype=bool)) + with pytest.raises(ValueError, match="non-empty"): + with_us_medicaid_take_up( + frame, + is_medicaid_eligible=np.ones(10, dtype=bool), + state_fips=np.full(10, "01"), + state_targets=pd.DataFrame(), + seed=0, + ) + + +class TestGate: + def test_passes_on_healthy_assignment(self) -> None: + _, _, diagnostics, _, _, _, _ = _scenario() + gate = us_medicaid_take_up_gate(diagnostics) + assert gate.passed, gate.failures + + def _assigned_table( + self, + *, + flag: np.ndarray, + eligible: np.ndarray, + anchor: np.ndarray, + state: np.ndarray, + ) -> pd.DataFrame: + n = len(flag) + return pd.DataFrame( + { + "person_id": range(n), + US_MEDICAID_TAKE_UP_VARIABLE: flag, + "is_medicaid_eligible": eligible, + US_MEDICAID_TAKE_UP_ANCHOR: anchor, + "person_weight": np.full(n, UNIT_WEIGHT), + "state_fips": state, + } + ) + + def test_fails_on_the_universal_take_up_landmine(self) -> None: + # All eligibles enrolled while the CMS count sits well below eligible + # weight: exactly the #170 degeneracy this stage exists to heal. + n = 100 + eligible = np.ones(n, dtype=bool) + table = self._assigned_table( + flag=np.ones(n, dtype=bool), + eligible=eligible, + anchor=np.zeros(n, dtype=bool), + state=np.full(n, "01"), + ) + targets = pd.DataFrame( + {"state_fips": ["01"], "target": [0.5 * n * UNIT_WEIGHT]} + ) + diagnostics = us_medicaid_take_up_diagnostics(table, targets) + gate = us_medicaid_take_up_gate(diagnostics) + assert not gate.passed + assert any("landmine" in failure for failure in gate.failures) + + def test_fails_when_a_state_has_no_target(self) -> None: + n = 40 + table = self._assigned_table( + flag=np.zeros(n, dtype=bool), + eligible=np.ones(n, dtype=bool), + anchor=np.zeros(n, dtype=bool), + state=np.where(np.arange(n) % 2 == 0, "01", "03"), + ) + targets = pd.DataFrame({"state_fips": ["01"], "target": [500.0]}) + diagnostics = us_medicaid_take_up_diagnostics(table, targets) + assert diagnostics["states_without_targets"] == ["03"] + gate = us_medicaid_take_up_gate(diagnostics) + assert not gate.passed + assert any("03" in failure for failure in gate.failures) + + def test_fails_when_a_state_has_a_count_but_zero_eligible_weight(self) -> None: + # An eligibility-feed collapse classifies as saturated (any positive + # count exceeds zero eligible weight), which would skip the count + # checks and ship anchored-only enrollment silently. The gate must + # treat that saturation as a division artifact and fail loudly. + n = 40 + table = self._assigned_table( + flag=np.zeros(n, dtype=bool), + eligible=np.zeros(n, dtype=bool), + anchor=np.zeros(n, dtype=bool), + state=np.full(n, "01"), + ) + targets = pd.DataFrame({"state_fips": ["01"], "target": [500.0]}) + diagnostics = us_medicaid_take_up_diagnostics(table, targets) + assert diagnostics["saturated_states"] == ["01"] + gate = us_medicaid_take_up_gate(diagnostics) + assert not gate.passed + assert any("eligibility feed collapsed" in failure for failure in gate.failures) + + def test_fails_when_the_anchor_is_not_preserved(self) -> None: + n = 100 + anchor = np.zeros(n, dtype=bool) + anchor[:30] = True + flag = np.zeros(n, dtype=bool) + flag[:10] = True # fewer enrolled than anchored: anchor was dropped + table = self._assigned_table( + flag=flag, + eligible=np.ones(n, dtype=bool), + anchor=anchor, + state=np.full(n, "01"), + ) + targets = pd.DataFrame({"state_fips": ["01"], "target": [1000.0]}) + diagnostics = us_medicaid_take_up_diagnostics(table, targets) + gate = us_medicaid_take_up_gate(diagnostics) + assert not gate.passed + assert any("anchor" in failure for failure in gate.failures) + + def test_fully_anchored_unsaturated_state_is_not_a_landmine(self) -> None: + # Every modeled-eligible person reports Medicaid: anchors cannot be + # removed, so enrollment == eligibility is the correct calibrated + # answer even though the CMS count sits below eligible weight. The + # landmine check must not abort the build for it (#331 review). + n = 100 + anchor = np.ones(n, dtype=bool) + table = self._assigned_table( + flag=np.ones(n, dtype=bool), + eligible=np.ones(n, dtype=bool), + anchor=anchor, + state=np.full(n, "01"), + ) + targets = pd.DataFrame({"state_fips": ["01"], "target": [60 * UNIT_WEIGHT]}) + diagnostics = us_medicaid_take_up_diagnostics(table, targets) + gate = us_medicaid_take_up_gate(diagnostics) + assert gate.passed, gate.failures + + def test_full_enrollment_within_one_unit_weight_of_the_count_passes(self) -> None: + # Greedy calibration overshoots by up to one person: a target within + # one unit weight of eligible weight legitimately fills every + # eligible person without being the landmine. + n = 100 + table = self._assigned_table( + flag=np.ones(n, dtype=bool), + eligible=np.ones(n, dtype=bool), + anchor=np.zeros(n, dtype=bool), + state=np.full(n, "01"), + ) + targets = pd.DataFrame( + {"state_fips": ["01"], "target": [(n - 0.5) * UNIT_WEIGHT]} + ) + diagnostics = us_medicaid_take_up_diagnostics(table, targets) + gate = us_medicaid_take_up_gate(diagnostics) + assert gate.passed, gate.failures + + def test_dropped_anchor_fails_even_in_a_saturated_state(self) -> None: + # The anchor invariant is person-level and unconditional: a saturated + # state must not hide a regression that flips anchored reporters off. + n = 100 + anchor = np.zeros(n, dtype=bool) + anchor[:30] = True + flag = np.ones(n, dtype=bool) + flag[:5] = False # five anchored reporters lost the flag + table = self._assigned_table( + flag=flag, + eligible=np.ones(n, dtype=bool), + anchor=anchor, + state=np.full(n, "01"), + ) + targets = pd.DataFrame( + {"state_fips": ["01"], "target": [1.5 * n * UNIT_WEIGHT]} # saturated + ) + diagnostics = us_medicaid_take_up_diagnostics(table, targets) + assert diagnostics["saturated_states"] == ["01"] + gate = us_medicaid_take_up_gate(diagnostics) + assert not gate.passed + assert any("anchor" in failure for failure in gate.failures) + + def test_anchor_mass_above_the_cms_count_is_the_floor_not_a_miss(self) -> None: + # CPS reporting above the CMS snapshot in a state must not fail the + # gate: anchors are preserved by design, so the anchor mass is the + # reachable floor. + n = 100 + anchor = np.zeros(n, dtype=bool) + anchor[:80] = True + flag = anchor.copy() + table = self._assigned_table( + flag=flag, + eligible=np.ones(n, dtype=bool), + anchor=anchor, + state=np.full(n, "01"), + ) + targets = pd.DataFrame({"state_fips": ["01"], "target": [60 * UNIT_WEIGHT]}) + diagnostics = us_medicaid_take_up_diagnostics(table, targets) + gate = us_medicaid_take_up_gate(diagnostics) + assert gate.passed, gate.failures + + +class TestParticipationDiagnosticsIntegration: + def test_materialized_column_reports_live_surface(self) -> None: + _, result, _, _, _, _, _ = _scenario() + payload = us_take_up_participation_diagnostics(result) + row = next( + row + for row in payload["programs"] + if row["variable"] == US_MEDICAID_TAKE_UP_VARIABLE + ) + assert row["count_calibrated"] is True + assert row["ships_at_engine_default"] is False + assert 0.0 < row["take_up_share"] < 1.0 + + def test_missing_column_reports_engine_default_with_owner(self) -> None: + frame = _frame(20, anchor=np.zeros(20, dtype=bool)) + payload = us_take_up_participation_diagnostics(frame) + row = next( + row + for row in payload["programs"] + if row["variable"] == US_MEDICAID_TAKE_UP_VARIABLE + ) + assert row["count_calibrated"] is False + assert row["ships_at_engine_default"] is True + # The debt-ledger pointer must survive the count_calibrated branch. + assert "populace#331" in row["scope_owner"] + + def test_materialized_share_is_labeled_with_its_universe(self) -> None: + # The flag deliberately carries off-domain propensity Trues, so the + # share over all persons must be labeled as such rather than read as + # an enrollment or participation rate. + _, result, _, _, _, _, _ = _scenario() + payload = us_take_up_participation_diagnostics(result) + row = next( + row + for row in payload["programs"] + if row["variable"] == US_MEDICAID_TAKE_UP_VARIABLE + ) + assert "off_domain_propensity" in row["share_universe"] + + +def _load_builder_module(): + root = Path(__file__).resolve().parents[3] + path = root / "tools" / "build_us_fiscal_refresh_release.py" + spec = importlib.util.spec_from_file_location( + "build_us_fiscal_refresh_release", path + ) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +class TestBuilderHelpers: + def test_medicaid_target_table_filters_role_and_geography(self) -> None: + builder = _load_builder_module() + specs = ( + SimpleNamespace( + name="cms_medicaid.month2024_12.state_enrollment.06.total_medicaid_enrollment@2024", + value=12_000_000.0, + metadata={ + "ledger_geography_level": "state", + "state_fips": "6", + "target_role": "medicaid_enrollment", + }, + ), + SimpleNamespace( + name="national-row", + value=72_000_000.0, + metadata={ + "ledger_geography_level": "national", + "target_role": "medicaid_enrollment", + }, + ), + SimpleNamespace( + name="chip-row", + value=100_000.0, + metadata={ + "ledger_geography_level": "state", + "state_fips": "06", + "target_role": "chip_enrollment", + }, + ), + ) + table = builder._medicaid_source_target_table(specs) + assert table["state_fips"].tolist() == ["06"] + assert table["target"].tolist() == [12_000_000.0] + + def test_duplicate_state_target_rows_refuse_to_compile(self) -> None: + # Duplicate state rows diverge downstream (calibration is last-row- + # wins; rate/diagnostics/gate sum), so the table must refuse them. + builder = _load_builder_module() + spec = { + "ledger_geography_level": "state", + "state_fips": "06", + "target_role": "medicaid_enrollment", + } + specs = ( + SimpleNamespace(name="row-1", value=12_000_000.0, metadata=dict(spec)), + SimpleNamespace(name="row-2", value=11_000_000.0, metadata=dict(spec)), + ) + with pytest.raises(RuntimeError, match="duplicate state rows"): + builder._medicaid_source_target_table(specs) + + def test_person_state_fips_maps_households(self) -> None: + builder = _load_builder_module() + frame = _frame(4, anchor=np.zeros(4, dtype=bool)) + household = frame.table("household").copy() + household["state_fips"] = [6, 6, 36, 36] + tables = {entity: frame.table(entity) for entity in frame.entities} + tables["household"] = household + with_states = Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + ) + assert builder._person_state_fips(with_states).tolist() == [ + "06", + "06", + "36", + "36", + ] diff --git a/packages/populace-build/tests/test_us_take_up_contract.py b/packages/populace-build/tests/test_us_take_up_contract.py index 16c836e5..89261fbf 100644 --- a/packages/populace-build/tests/test_us_take_up_contract.py +++ b/packages/populace-build/tests/test_us_take_up_contract.py @@ -43,6 +43,7 @@ def test_every_program_has_a_valid_treatment(self) -> None: for program in load_take_up_contract().programs: assert program.populace_treatment in { "seed", + "count_calibrated", "rate_unsourced", "model_simulated", "out_of_scope", diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index 57ea2044..6aa6883c 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -50,6 +50,7 @@ US_FISCAL_TARGET_COVERAGE_REQUIREMENTS, US_FISCAL_TARGET_SUPPORT_EXCLUSIONS, US_JCT_TAX_EXPENDITURE_REFORMS, + US_MEDICAID_ENROLLMENT_TARGET_TABLE, US_SOURCE_MANIFEST, assert_validation_leaf_registry_current, compile_us_fiscal_target_registry, @@ -58,6 +59,7 @@ us_eligibility_inputs_signal_gate, us_hours_worked_signal_gate, us_immigration_composition_gate, + us_medicaid_take_up_gate, us_pregnancy_signal_gate, us_snap_discretionary_exemption_signal_gate, us_snap_take_up_signal_gate, @@ -69,10 +71,12 @@ with_us_eligibility_inputs, with_us_hours_worked_inputs, with_us_immigration_inputs, + with_us_medicaid_take_up, with_us_pregnancy_inputs, with_us_snap_discretionary_exemption_inputs, with_us_snap_take_up_inputs, with_us_take_up_inputs, + write_us_medicaid_take_up_diagnostics, write_us_source_coverage_diagnostics, write_us_take_up_participation_diagnostics, ) @@ -148,7 +152,12 @@ "congressional_district_vintage_crosswalk_sha256", ) TARGET_FRAME_CHECKPOINT_SCHEMA_VERSION = 1 -TARGET_FRAME_CHECKPOINT_MATERIALIZER_VERSION = 1 +# 2: the medicaid_take_up stage (populace #331) changed base_frame's +# takes_up_medicaid_if_eligible before target-frame materialization, so +# medicaid_enrolled target columns differ from version-1 checkpoints; the +# checkpoint identity hashes the on-disk base dataset, not the staged frame, +# and would otherwise silently reuse pre-stage frames. +TARGET_FRAME_CHECKPOINT_MATERIALIZER_VERSION = 2 DEFAULT_MAXIMUM_MICROSIM_BATCH_SIZE = 5_000 DEFAULT_L0_REFIT_LAMBDA_SHARE = 0.8 DEFAULT_US_FISCAL_CALIBRATION_EPOCHS = 1_500 @@ -411,9 +420,6 @@ def _automatic_gc_suspended(): "takes_up_ssi_if_eligible": ( "SSI take-up imputation backlog; constant True forces 100% take-up." ), - "takes_up_medicaid_if_eligible": ( - "Medicaid take-up imputation backlog (PolicyEngine/populace#98)." - ), "takes_up_medicare_if_eligible": ( "Medicare take-up imputation backlog (PolicyEngine/populace#98)." ), @@ -529,6 +535,7 @@ def _automatic_gc_suspended(): "cms_aca_bronze_aptc_consumers_by_state", } ) +US_MEDICAID_ENROLLMENT_TARGET_ROLE = "medicaid_enrollment" FILING_STATUS_MAP = { "All": None, @@ -2188,6 +2195,149 @@ def _with_aca_marketplace_source_outputs( ) +def _medicaid_source_target_table(target_specs: tuple) -> pd.DataFrame: + """CMS state Medicaid enrollment counts as the take-up calibration table. + + Mirrors :func:`_aca_source_target_tables` for the ``medicaid_enrollment`` + target role: month-tagged December 2024 state snapshot facts + (``cms_medicaid.month2024_12.state_enrollment.*``), point-in-time + semantics per populace #332. + """ + rows: list[dict[str, object]] = [] + for spec in target_specs: + if spec.metadata.get("ledger_geography_level") != "state": + continue + groupby_dimension = spec.metadata.get("ledger_layout_groupby_dimension") + if ( + isinstance(groupby_dimension, str) + and "congressional_district" in groupby_dimension + ): + continue + state_fips = spec.metadata.get("state_fips") + if not state_fips: + continue + if spec.metadata.get("target_role") != US_MEDICAID_ENROLLMENT_TARGET_ROLE: + continue + rows.append( + { + "state_fips": str(state_fips).zfill(2), + "target": float(spec.value), + "source_record_id": spec.name, + } + ) + table = pd.DataFrame(rows, columns=["state_fips", "target", "source_record_id"]) + duplicated = table["state_fips"][table["state_fips"].duplicated()].unique() + if len(duplicated): + # The calibrate operation applies duplicate state rows sequentially + # (last row wins) while the rate prior, diagnostics, and gate SUM + # them — divergent semantics that must never reach the stage. + raise RuntimeError( + "Medicaid enrollment targets carry duplicate state rows for " + f"state_fips {sorted(duplicated)}; the ledger feed must supply " + "exactly one medicaid_enrollment count per state." + ) + return table + + +def _person_state_fips(frame: Frame) -> np.ndarray: + """Person-aligned state FIPS text codes via the frame's linkage.""" + # Frame.broadcast is the validated membership-mapping path (linkage is + # asserted at Frame construction), so no manual reindex/NaN handling. + return np.asarray(_state_fips_text(frame.broadcast("state_fips").to_numpy())) + + +def _medicaid_person_eligibility( + frame: Frame, + *, + simulation=None, + maximum_microsim_batch_size: int | None = DEFAULT_MAXIMUM_MICROSIM_BATCH_SIZE, +) -> np.ndarray: + """Engine-computed person-level ``is_medicaid_eligible``, batched like ACA.""" + if simulation is not None: + return _calculate_array(simulation, "is_medicaid_eligible", map_to="person") > 0 + from policyengine_us import Microsimulation + + person_ids = frame.table("person")["person_id"].to_numpy() + eligibility = np.zeros(len(person_ids), dtype=bool) + person_positions = pd.Series( + np.arange(len(person_ids), dtype=np.int64), index=person_ids + ) + n_households = frame.n("household") + batches = tuple( + _household_position_batches(n_households, maximum_microsim_batch_size) + ) + if len(batches) > 1: + print( + "Materializing Medicaid eligibility in " + f"{len(batches)} batches of up to " + f"{maximum_microsim_batch_size:,} households.", + flush=True, + ) + for household_positions in batches: + with _automatic_gc_suspended(): + full_batch = len(household_positions) == n_households + batch_frame = ( + frame + if full_batch + else _select_households_by_position(frame, household_positions) + ) + batch_simulation = Microsimulation( + dataset=_dataset_from_frame( + batch_frame, + assert_no_formula_owned_columns=False, + ) + ) + batch_eligible = ( + _calculate_array( + batch_simulation, "is_medicaid_eligible", map_to="person" + ) + > 0 + ) + positions = person_positions.reindex( + batch_frame.table("person")["person_id"].to_numpy() + ).to_numpy() + if np.isnan(positions).any(): + raise RuntimeError( + "Medicaid eligibility batch produced person_id values not " + "present in the full person table." + ) + eligibility[positions.astype(np.int64)] = batch_eligible + batch_simulation._invalidate_all_caches() + del batch_frame, batch_simulation + _collect_batch_garbage() + return eligibility + + +def _with_medicaid_take_up_outputs( + frame: Frame, + target_specs: tuple, + *, + seed: int, + simulation=None, + maximum_microsim_batch_size: int | None = DEFAULT_MAXIMUM_MICROSIM_BATCH_SIZE, +) -> tuple[Frame, dict[str, object]]: + """Assign Medicaid take-up (populace #331) and return frame + diagnostics.""" + target_table = _medicaid_source_target_table(target_specs) + if target_table.empty: + raise RuntimeError( + "Medicaid take-up requires CMS state enrollment targets " + f"({US_MEDICAID_ENROLLMENT_TARGET_TABLE}); none were compiled from " + "the ledger facts." + ) + eligibility = _medicaid_person_eligibility( + frame, + simulation=simulation, + maximum_microsim_batch_size=maximum_microsim_batch_size, + ) + return with_us_medicaid_take_up( + frame, + is_medicaid_eligible=eligibility, + state_fips=_person_state_fips(frame), + state_targets=target_table, + seed=seed, + ) + + def _assert_no_formula_owned_columns(frame: Frame) -> None: adapter = PolicyEngineUSEngine() tables = {entity: frame.table(entity) for entity in frame.entities} @@ -5738,6 +5888,37 @@ def main() -> None: for failure in health_input_gate.failures ) ) + if telemetry is not None: + telemetry.stage( + "medicaid_take_up", + message=( + "Assigning Medicaid take-up from reported coverage and CMS " + "state enrollment snapshots." + ), + ) + base_frame, medicaid_take_up_diagnostics = _with_medicaid_take_up_outputs( + base_frame, + target_specs, + seed=args.seed, + maximum_microsim_batch_size=args.maximum_microsim_batch_size, + ) + medicaid_take_up_gate = us_medicaid_take_up_gate(medicaid_take_up_diagnostics) + if not medicaid_take_up_gate.passed: + if telemetry is not None: + telemetry.stage( + "medicaid_take_up_gate", + status="failed", + message="Medicaid take-up gate failed.", + failures=list(medicaid_take_up_gate.failures), + force_upload=True, + ) + raise RuntimeError( + "Release gates failed: " + + "; ".join( + f"Medicaid take-up failed: {failure}" + for failure in medicaid_take_up_gate.failures + ) + ) if telemetry is not None and args.input_mass_reference_h5 is not None: telemetry.stage( "input_mass_reference_gate", @@ -6269,15 +6450,40 @@ def main() -> None: "take_up_participation", message="Writing take-up participation diagnostics.", ) + take_up_participation = us_take_up_participation_diagnostics(export_frame) write_us_take_up_participation_diagnostics( - us_take_up_participation_diagnostics(export_frame), + take_up_participation, release_dir / "us_take_up_participation.json", ) + write_us_medicaid_take_up_diagnostics( + medicaid_take_up_diagnostics, + release_dir / "us_medicaid_take_up.json", + ) + # The stage gate ran on the stage output; this re-checks the EXPORT frame + # so a downstream transform that drops or flattens a count-calibrated + # column cannot ship the engine-default landmine with only an + # observational JSON field recording it. + stale_count_calibrated = [ + str(row["variable"]) + for row in take_up_participation["programs"] + if row.get("populace_treatment") == "count_calibrated" + and row.get("ships_at_engine_default") + ] + if stale_count_calibrated: + raise RuntimeError( + "Release gates failed: count-calibrated take-up column(s) " + f"{stale_count_calibrated} ship at the engine default on the " + "export frame despite the stage having run." + ) if telemetry is not None: telemetry.attach_artifact( "us_take_up_participation", release_dir / "us_take_up_participation.json", ) + telemetry.attach_artifact( + "us_medicaid_take_up", + release_dir / "us_medicaid_take_up.json", + ) telemetry.stage("manifests", message="Writing release manifests.") timing["total_build_seconds"] = time.perf_counter() - build_started _build_manifests(