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 dbf4be55..262c48ea 100644 --- a/packages/populace-build/src/populace/build/us/source_stages.json +++ b/packages/populace-build/src/populace/build/us/source_stages.json @@ -1229,13 +1229,7 @@ "vintage": "2024-12", "locator": "SSA SSI Monthly Statistics, December 2024, Table 1, row Total with—Federal payment, by age", "source": "https://www.ssa.gov/policy/docs/statcomps/ssi_monthly/2024-12/table01.html", - "measure": "Total with—Federal payment", - "target_values": { - "under_18": 1001922, - "18_64": 3905779, - "65_plus": 2382142, - "total": 7289843 - } + "measure": "Total with—Federal payment" }, { "kind": "archived_derivation_evidence", @@ -1284,40 +1278,23 @@ "rate_column": "ssi_take_up_assignment_prior", "reported_true_anchor": "SSI_VAL > 0", "assignment_unit": "person_source_id", - "fan_to_support_clones": true - }, - { - "kind": "calibrate_binary_assignment", - "variable": "takes_up_ssi_if_eligible", - "targets": [ - "ssa_ssi_federal_payment_recipients_by_age" - ], - "preserve_true_anchors": true, - "preserve_true_anchor": "SSI_VAL > 0", - "domain": "uncapped_ssi > 0", - "weight": "person_weight", - "draw": "stable_source_person_draw", - "calibration_unit": "person_source_id", + "fan_to_support_clones": true, "age_bands": { "under_18": "age < 18", "18_64": "18 <= age < 65", "65_plus": "age >= 65" }, + "rate_derivation": "band_target / weighted_candidate_capacity(uncapped_ssi > 0); min(reporter_candidate_floor / capacity, 1) once that ratio reaches one", + "rate_target_role": "ssa_ssi_age_band_recipients", "target_source": "https://www.ssa.gov/policy/docs/statcomps/ssi_monthly/2024-12/table01.html", "target_period": "2024-12", - "target_measure": "Total with—Federal payment", - "target_values": { - "under_18": 1001922, - "18_64": 3905779, - "65_plus": 2382142 - }, - "aggregate_target": 7289843 + "target_measure": "Total with—Federal payment" } ], "outputs": [ "takes_up_ssi_if_eligible" ], - "notes": "Restores the final retired eCPS SSI take-up export from source-backed evidence at archived commit 42ed5d45c56df80d754fbe24cce21cfeb8d05cbe. datasets/cps/cps.py line 584 maps the direct ASEC SSI_VAL reporter signal, lines 650-657 derive the take-up leaf, and lines 1497-1499 export it; datasets/cps/takeup.py lines 10-35 define the retired stable random assignment. The retired all-age 0.50 rate has no full-population administrative source and is not reused. Instead, utils/ssi_targets.py lines 41-74 preserve the archived age-target method and cite SSA SSI Monthly Statistics, December 2024, Table 1, row Total with—Federal payment: 1,001,922 recipients under 18, 3,905,779 ages 18-64, and 2,382,142 ages 65+, totaling 7,289,843. Direct SSI_VAL reporters remain true. Remaining source-person identities are assigned deterministically within the uncapped_ssi > 0 domain and fanned identically to both support clones; an age band whose modeled eligible support cannot reach its SSA count saturates without assigning outside the eligibility domain, and the diagnostic records the irreducible shortfall." + "notes": "Restores the final retired eCPS SSI take-up export from source-backed evidence at archived commit 42ed5d45c56df80d754fbe24cce21cfeb8d05cbe. datasets/cps/cps.py line 584 maps the direct ASEC SSI_VAL reporter signal, lines 650-657 derive the take-up leaf, and lines 1497-1499 export it; datasets/cps/takeup.py lines 10-35 define the retired stable random assignment. The retired all-age 0.50 rate has no full-population administrative source and is not reused. utils/ssi_targets.py lines 41-74 preserve the archived age-target method and cite SSA SSI Monthly Statistics, December 2024, Table 1, row Total with—Federal payment, by the three published age bands; those recipient counts enter the build only as ledger-fed calibration registry targets (role ssa_ssi_age_band_recipients, populace#469 and populace#470), never as count-matching goals in this stage. Direct SSI_VAL reporters remain true. Every other source-person identity draws once against the documented band prior — the band target over weighted uncapped_ssi > 0 candidate capacity, falling back to the observed reporter share of capacity when that ratio reaches one — seeded and stable per person_source_id, with the decision fanned identically to both support clones. Any gap between selected recipient mass and the SSA counts is calibration's residual and ships in the release scorecard." }, { "stage": "sipp_tips", 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 0c75e49a..8da7c8b3 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 @@ -132,22 +132,17 @@ "target_source": "https://www.ssa.gov/policy/docs/statcomps/ssi_monthly/2024-12/table01.html", "target_period": "2024-12", "target_measure": "Total with—Federal payment", - "target_values": { - "under_18": 1001922, - "18_64": 3905779, - "65_plus": 2382142 - }, - "aggregate_target": 7289843, + "target_role": "ssa_ssi_age_band_recipients", "age_bands": { "under_18": "age < 18", "18_64": "18 <= age < 65", "65_plus": "age >= 65" }, - "semantics": "SSA SSI Monthly Statistics December 2024 Table 1 recipients in the Total with—Federal payment row, calibrated within uncapped_ssi > 0 by source-person identity; unreachable age bands saturate without assigning outside modeled eligibility" + "semantics": "SSA SSI Monthly Statistics December 2024 Table 1 recipients in the Total with—Federal payment row, by age band. The counts live in the ledger and bind as ordinary weight-calibration targets (registry role ssa_ssi_age_band_recipients, populace#469/#470); the take-up stage derives its seeded Bernoulli priors from the same registry values (band target over weighted uncapped_ssi > 0 candidate capacity, the observed reporter share of capacity under saturation) and never count-matches the flags." }, "rate": {"status": "rejected_scope_mismatch", "retired_value": 0.50, "verified_source_value": {"value": 0.49, "scope": "adults 65+ only", "citation": "Urban Institute Giannarelli et al. 2024 (ATTIS, 2018), 49.0% aged SSI participation"}}, "scope_owner": "ssi_take_up source stage (eCPS exported-input coverage)", - "notes": "The retired 0.50 rate was applied to all SSI eligibles even though its cited 49% estimate covers adults 65+ only; it is rejected, not copied. The ssi_take_up stage instead preserves every direct ASEC SSI_VAL reporter and calibrates eligible source-person identities to SSA SSI Monthly Statistics December 2024 Table 1 counts in the Total with—Federal payment row (under 18: 1,001,922; ages 18-64: 3,905,779; ages 65+: 2,382,142; total: 7,289,843). This is administrative count calibration, not a participation-rate claim. If modeled eligible support cannot reach a band target, the stage saturates that band and reports the shortfall rather than assigning an ineligible person." + "notes": "The retired 0.50 rate was applied to all SSI eligibles even though its cited 49% estimate covers adults 65+ only; it is rejected, not copied. The ssi_take_up stage preserves every direct ASEC SSI_VAL reporter and gives every other source-person identity one seeded Bernoulli draw at the documented band prior derived from the ledger-fed SSA December 2024 Table 1 counts (populace#469). Those counts bind as ordinary calibration registry targets (populace#470), so recipient totals are calibration's to hit and any post-calibration miss ships in the release scorecard like every other program's — no count matching, no saturation fill, no reconcile loop." }, { "variable": "takes_up_dc_ptc", diff --git a/packages/populace-build/src/populace/build/us_runtime/ssi_take_up.py b/packages/populace-build/src/populace/build/us_runtime/ssi_take_up.py index ba80304b..e50811a3 100644 --- a/packages/populace-build/src/populace/build/us_runtime/ssi_take_up.py +++ b/packages/populace-build/src/populace/build/us_runtime/ssi_take_up.py @@ -1,4 +1,4 @@ -"""Reporter-anchored SSI take-up calibrated to SSA recipient counts. +"""Reporter-anchored Bernoulli SSI take-up at documented age-band priors. The retired eCPS exported ``takes_up_ssi_if_eligible`` after preserving every CPS ASEC ``SSI_VAL > 0`` reporter and filling additional people to a scalar @@ -7,21 +7,24 @@ to children and working-age disabled people too. This stage keeps the source-backed half of that method (the reporter anchor) -and replaces the scope-invalid rate with SSA's December 2024 counts of people -receiving a *federal payment*, split into the three published age bands. It -uses PolicyEngine-US ``uncapped_ssi > 0`` as the current-benefit candidate -domain, makes one stable decision per ``person_source_id``, and fans that -decision to every actual support row for the source person. A target above -modeled capacity saturates at capacity and records the shortfall; eligibility -is never broadened to manufacture recipients. +and replaces the scope-invalid rate with per-band priors derived from SSA's +December 2024 counts of people receiving a *federal payment*, split into the +three published age bands. Each prior is the band target over the weighted +PolicyEngine-US ``uncapped_ssi > 0`` candidate capacity — falling back to the +observed reporter share of capacity when that ratio reaches one, so the flag +never degenerates to a constant. Every source person draws once against the +band prior, seeded and stable per ``person_source_id``, with the decision +fanned to every actual support row; direct ASEC reporters stay true +unconditionally. There is no count matching here (populace#469): the SSA +band counts bind only as ordinary calibration registry targets +(populace#470), the caller passes those same registry values in as +``targets``, and any post-calibration miss ships in the release scorecard +like every other program's. The caller supplies December ``uncapped_ssi`` because the fiscal builder owns -PolicyEngine simulations and batching. Assignment is always recomputed from -the supplied frame weights. The builder fixes it on an initial calibrated -weight surface, replays dependent health inputs and fiscal targets, then -refits on the unchanged support. Returned release weights are diagnosed -without rewriting the persisted assignment; only those final diagnostics are -published. +PolicyEngine simulations and batching. Assignment is recomputed from the +supplied frame weights, fixed once before target materialization, and later +diagnosed on release weights without rewriting the persisted decisions. """ from __future__ import annotations @@ -102,17 +105,26 @@ _TARGET_MEASURE = "Total with—Federal payment" _CANDIDATE_DEFINITION = "uncapped_ssi > 0 at 2024-12" _WEIGHTS_BASIS = "current_frame_resolved_person_weights" -_DIAGNOSTICS_SCHEMA_VERSION = 1 +# Version 2 (populace#469): count-matching-era fields (reachable_goal) are +# gone, band rows split the prior into assignment_prior (the value that +# generated the frozen flags) and prior_recomputed_from_current_weights, and +# the top level carries bernoulli_law_violation_count. +_DIAGNOSTICS_SCHEMA_VERSION = 2 @dataclass(frozen=True) class SSITakeUpAgeTarget: - """One disjoint SSA federal-payment recipient age stratum.""" + """One disjoint SSA federal-payment recipient age stratum. + + Band structure only: the recipient counts themselves live in the ledger + (``ssa_ssi...by_age`` facts) and reach this stage as caller-supplied + ``targets`` read from the calibration registry (populace#469/#470), never + as module constants. + """ key: str minimum_age: int | None maximum_age: int | None - person_count: float label: str def contains(self, age: np.ndarray) -> np.ndarray: @@ -125,12 +137,10 @@ def contains(self, age: np.ndarray) -> np.ndarray: US_SSI_TAKE_UP_AGE_TARGETS: tuple[SSITakeUpAgeTarget, ...] = ( - SSITakeUpAgeTarget("under_18", None, 17, 1_001_922.0, "Under 18"), - SSITakeUpAgeTarget("18_64", 18, 64, 3_905_779.0, "Ages 18–64"), - SSITakeUpAgeTarget("65_plus", 65, None, 2_382_142.0, "Ages 65+"), + SSITakeUpAgeTarget("under_18", None, 17, "Under 18"), + SSITakeUpAgeTarget("18_64", 18, 64, "Ages 18–64"), + SSITakeUpAgeTarget("65_plus", 65, None, "Ages 65+"), ) -_TARGETS_BY_KEY = {target.key: target for target in US_SSI_TAKE_UP_AGE_TARGETS} -_TARGET_TOTAL = float(sum(target.person_count for target in US_SSI_TAKE_UP_AGE_TARGETS)) _READ_PARAMETERS: dict[str, object] = { "table": "person", @@ -144,28 +154,20 @@ def contains(self, age: np.ndarray) -> np.ndarray: "reported_true_anchor": "SSI_VAL > 0", "assignment_unit": _SOURCE_ID, "fan_to_support_clones": True, -} -_CALIBRATE_PARAMETERS: dict[str, object] = { - "variable": _OUTPUT, - "targets": [US_SSI_TAKE_UP_TARGET_TABLE_NAME], - "preserve_true_anchors": True, - "preserve_true_anchor": "SSI_VAL > 0", - "domain": "uncapped_ssi > 0", - "weight": "person_weight", - "draw": "stable_source_person_draw", - "calibration_unit": _SOURCE_ID, "age_bands": { "under_18": "age < 18", "18_64": "18 <= age < 65", "65_plus": "age >= 65", }, + "rate_derivation": ( + "band_target / weighted_candidate_capacity(uncapped_ssi > 0); " + "min(reporter_candidate_floor / capacity, 1) once that ratio " + "reaches one" + ), + "rate_target_role": "ssa_ssi_age_band_recipients", "target_source": SSI_TAKE_UP_SSA_SOURCE_URL, "target_period": _TARGET_PERIOD, "target_measure": _TARGET_MEASURE, - "target_values": { - target.key: int(target.person_count) for target in US_SSI_TAKE_UP_AGE_TARGETS - }, - "aggregate_target": int(_TARGET_TOTAL), } @@ -188,17 +190,16 @@ def us_ssi_take_up_stage_spec() -> SourceStageSpec: expected_kinds = [ "read_table", "assign_binary_from_rate", - "calibrate_binary_assignment", ] if [operation.kind for operation in spec.operations] != expected_kinds: raise ValueError( - "US SSI take-up stage must declare read, assignment, then age-count " - "calibration." + "US SSI take-up stage must declare read then one seeded Bernoulli " + "assignment; the SSA band counts bind only as calibration registry " + "targets (populace#469/#470)." ) expected_parameters = ( _READ_PARAMETERS, _ASSIGN_PARAMETERS, - _CALIBRATE_PARAMETERS, ) for operation, expected in zip(spec.operations, expected_parameters, strict=True): if dict(operation.parameters) != expected: @@ -211,17 +212,14 @@ def us_ssi_take_up_stage_spec() -> SourceStageSpec: if artifact.get("source") == SSI_TAKE_UP_SSA_SOURCE_URL ] if len(target_artifacts) != 1: - raise ValueError("US SSI take-up stage must pin exactly one SSA target table.") - values = target_artifacts[0].get("target_values") - expected_values = { - **{ - target.key: int(target.person_count) - for target in US_SSI_TAKE_UP_AGE_TARGETS - }, - "total": int(_TARGET_TOTAL), - } - if values != expected_values: - raise ValueError("US SSI take-up SSA target cells drifted from runtime.") + raise ValueError("US SSI take-up stage must pin exactly one SSA table.") + if target_artifacts[0].get("vintage") != _TARGET_PERIOD: + raise ValueError("US SSI take-up SSA table vintage drifted from runtime.") + if "target_values" in target_artifacts[0]: + raise ValueError( + "US SSI take-up must not hardcode SSA recipient counts; they enter " + "as ledger-fed calibration registry targets (populace#469/#470)." + ) return spec @@ -242,13 +240,8 @@ def _age_band_values(age: np.ndarray) -> np.ndarray: return bands -def _normalize_targets(targets: Mapping[str, float] | None) -> dict[str, float]: +def _normalize_targets(targets: Mapping[str, float]) -> dict[str, float]: expected_keys = tuple(target.key for target in US_SSI_TAKE_UP_AGE_TARGETS) - if targets is None: - return { - target.key: float(target.person_count) - for target in US_SSI_TAKE_UP_AGE_TARGETS - } actual = {str(key): float(value) for key, value in targets.items()} if set(actual) != set(expected_keys): raise ValueError( @@ -442,13 +435,40 @@ def _source_table( return rows, source +def _band_prior(target: float, capacity: float, reporter_floor: float) -> float: + """Return the documented Bernoulli prior for one SSA age band. + + The target/capacity ratio is a meaningful take-up propensity only while + it subsamples (capacity > target). Once it reaches one it would flag the + whole band — a constant, signal-free output — so the prior falls back to + the observed take-up rate among today's candidates: reporter mass over + candidate capacity. Reform-created eligibles then take up at the rate + today's modeled eligibles are observed reporting. + """ + + if capacity <= 0: + return 0.0 + count_ratio = target / capacity + if count_ratio < 1.0: + return count_ratio + return min(reporter_floor / capacity, 1.0) + + def _age_band_diagnostics( source: pd.DataFrame, selected: pd.Series, *, targets: Mapping[str, float], + assignment_priors: Mapping[str, float], ) -> list[dict[str, object]]: - """Summarize one existing source-grain assignment on current weights.""" + """Summarize one existing source-grain assignment on current weights. + + ``assignment_priors`` are the Bernoulli priors that generated the + persisted flags. Each band row publishes them verbatim next to + ``prior_recomputed_from_current_weights`` so release-weight measurements + never misdocument the one-shot assignment (populace#469; PR #477 review + finding 4). + """ bands: list[dict[str, object]] = [] for target_definition in US_SSI_TAKE_UP_AGE_TARGETS: @@ -461,7 +481,6 @@ def _age_band_diagnostics( reporter_floor = float( source.loc[candidate & anchored, "candidate_weight"].sum() ) - reachable_goal = min(max(target, reporter_floor), capacity) selected_recipient_weight = float( source.loc[candidate & selected, "candidate_weight"].sum() ) @@ -480,14 +499,14 @@ def _age_band_diagnostics( "reporter_source_identity_count": int(anchored.sum()), "candidate_capacity": capacity, "reporter_candidate_floor": reporter_floor, - "reachable_goal": reachable_goal, "selected_recipient_weight": selected_recipient_weight, "signed_target_error": selected_recipient_weight - target, "target_shortfall": max(target - selected_recipient_weight, 0.0), "anchor_excess": max(reporter_floor - target, 0.0), "saturated": bool(capacity < target), - "assignment_prior": ( - min(target / capacity, 1.0) if capacity > 0 else 0.0 + "assignment_prior": float(assignment_priors[key]), + "prior_recomputed_from_current_weights": _band_prior( + target, capacity, reporter_floor ), "max_source_candidate_weight": max_source_weight, } @@ -495,14 +514,58 @@ def _age_band_diagnostics( return bands +def _bernoulli_law_violations( + source: pd.DataFrame, + selected: pd.Series, + assignment_priors: Mapping[str, float], +) -> int: + """Count source identities whose flag breaks the seeded Bernoulli law. + + The law is exact and weight-free: a source person is selected iff + anchored or its stable draw fell below the band's assignment-time prior. + Recomputing it against persisted flags catches any post-assignment + corruption of the frozen decisions (populace#469; PR #477 review + finding 3). + """ + + priors = source["age_band"].map(dict(assignment_priors)).to_numpy(dtype=np.float64) + expected = source["anchor"].to_numpy(dtype=bool) | ( + source["draw"].to_numpy(dtype=np.float64) < priors + ) + return int(np.count_nonzero(expected != selected.to_numpy(dtype=bool))) + + +def _normalize_assignment_priors( + assignment_priors: Mapping[str, float], +) -> dict[str, float]: + expected_keys = tuple(band.key for band in US_SSI_TAKE_UP_AGE_TARGETS) + actual = {str(key): float(value) for key, value in assignment_priors.items()} + if set(actual) != set(expected_keys): + raise ValueError( + "US SSI take-up assignment priors require exactly age bands " + f"{list(expected_keys)}; got {sorted(actual)}." + ) + invalid = { + key: value + for key, value in actual.items() + if not np.isfinite(value) or not 0.0 <= value <= 1.0 + } + if invalid: + raise ValueError( + f"US SSI take-up assignment priors must lie in [0, 1]: {invalid}." + ) + return {key: actual[key] for key in expected_keys} + + def _assign_sources( source: pd.DataFrame, *, targets: Mapping[str, float], -) -> tuple[pd.Series, list[dict[str, object]]]: - """Assign one flag per source identity and return age-band diagnostics.""" +) -> tuple[pd.Series, list[dict[str, object]], dict[str, float]]: + """Assign one flag per source identity; return diagnostics and priors.""" selected = pd.Series(False, index=source.index, dtype=bool) + priors: dict[str, float] = {} for target_definition in US_SSI_TAKE_UP_AGE_TARGETS: key = target_definition.key target = float(targets[key]) @@ -513,55 +576,22 @@ def _assign_sources( reporter_floor = float( source.loc[candidate & anchored, "candidate_weight"].sum() ) - reachable_goal = min(max(target, reporter_floor), capacity) - # The count-matching ratio is a meaningful reform propensity only - # while it subsamples (capacity >= target). Under saturation it - # degenerates to 1.0 and Bernoulli(1.0) flags the entire band — with - # candidates in every band (the restored disability battery), that is - # a constant, signal-free output and the take-up gate fails. Fall - # back to the observed take-up rate among today's candidates - # (reporter mass over candidate capacity): if a reform makes a - # household eligible, it takes up at the rate today's modeled - # eligibles are observed reporting. Current-law recipiency is - # unchanged either way — only the candidate domain below is ever - # paid, and the saturated branch still flags every candidate. - if capacity > 0: - count_ratio = target / capacity - prior = ( - count_ratio - if count_ratio < 1.0 - else min(reporter_floor / capacity, 1.0) - ) - else: - prior = 0.0 - - # Keep reform propensities off today's candidate domain. Candidate - # decisions are then greedily count-calibrated below. + prior = _band_prior(target, capacity, reporter_floor) + priors[key] = prior + + # Seeded Bernoulli at the documented band prior for everyone in the + # band — candidates and reform-created eligibles alike — with survey + # reporters anchored unconditionally (populace#469). No count + # matching: the SSA band counts are ordinary calibration targets + # (populace#470) and the post-calibration miss ships in the + # scorecard like every other program's. selected.loc[in_band] = source.loc[in_band, "draw"].to_numpy() < prior selected.loc[anchored] = True - selectable = candidate & ~anchored - selected.loc[selectable] = False - - current = reporter_floor - if capacity < target: - # Select the saturated domain explicitly instead of depending on - # floating-point accumulation reaching the same capacity sum. - selected.loc[candidate] = True - else: - ordered = sorted( - source.index[selectable], - key=lambda source_id: ( - float(source.at[source_id, "draw"]), - str(source_id), - ), - ) - for source_id in ordered: - if current >= reachable_goal: - break - selected.at[source_id] = True - current += float(source.at[source_id, "candidate_weight"]) - return selected, _age_band_diagnostics(source, selected, targets=targets) + bands = _age_band_diagnostics( + source, selected, targets=targets, assignment_priors=priors + ) + return selected, bands, priors def _diagnostics( @@ -572,6 +602,7 @@ def _diagnostics( assigned: np.ndarray, bands: list[dict[str, object]], targets: Mapping[str, float], + law_violation_count: int, ) -> dict[str, object]: weights = np.asarray(frame.resolve_weights("person").values, dtype=np.float64) reporter_lost = int(np.count_nonzero(rows["anchor"].to_numpy() & ~assigned)) @@ -627,6 +658,7 @@ def _diagnostics( "missing_or_invalid_count": 0, "reporter_anchor_lost_count": reporter_lost, "source_identity_mismatch_count": mismatches, + "bernoulli_law_violation_count": int(law_violation_count), "channel_diagnostics": channel_diagnostics, "age_bands": bands, } @@ -637,10 +669,16 @@ def with_us_ssi_take_up( *, uncapped_ssi: np.ndarray, seed: int, - targets: Mapping[str, float] | None = None, + targets: Mapping[str, float], reporter_source_ids: Collection[str] | None = None, ) -> tuple[Frame, dict[str, object]]: - """Recompute SSI take-up and return the frame plus count diagnostics.""" + """Recompute SSI take-up and return the frame plus band diagnostics. + + ``targets`` carries the SSA band recipient counts the caller read from + the calibration registry (role ``ssa_ssi_age_band_recipients``); they set + the Bernoulli priors here and bind as ordinary calibration targets + downstream (populace#469/#470). + """ us_ssi_take_up_stage_spec() normalized_targets = _normalize_targets(targets) @@ -650,7 +688,7 @@ def with_us_ssi_take_up( seed=int(seed), reporter_source_ids=reporter_source_ids, ) - selected, bands = _assign_sources(source, targets=normalized_targets) + selected, bands, priors = _assign_sources(source, targets=normalized_targets) assigned = rows["source_id"].map(selected).to_numpy(dtype=bool) diagnostics = _diagnostics( frame, @@ -659,6 +697,7 @@ def with_us_ssi_take_up( assigned=assigned, bands=bands, targets=normalized_targets, + law_violation_count=_bernoulli_law_violations(source, selected, priors), ) person = frame.table("person") @@ -692,21 +731,26 @@ def us_ssi_take_up_diagnostics( *, uncapped_ssi: np.ndarray, seed: int, - targets: Mapping[str, float] | None = None, + targets: Mapping[str, float], + assignment_priors: Mapping[str, float], reporter_source_ids: Collection[str] | None = None, ) -> dict[str, object]: """Diagnose a persisted assignment without changing its decisions. - The fiscal builder uses this to measure the pre-refit (frozen) flags on the - returned weights for the reconciliation swap delta. Count-faithfulness on the - returned weights is no longer assumed by freezing the flags: the builder - re-assigns take-up under the returned weights and verifies by measurement - that the fresh pair moved aggregate recipient mass within one source-identity - weight per age band before publishing it. + The fiscal builder assigns take-up once before target materialization and + publishes these measurements of the frozen flags on the final release + weights. ``assignment_priors`` are the per-band priors the assignment + stage documented (its diagnostics' ``assignment_prior`` fields): they are + republished verbatim and every persisted flag is re-verified against the + seeded Bernoulli law they define, so silent post-assignment corruption + fails the gate. Any gap between the measured recipient mass and the SSA + band targets is calibration's residual and ships in the scorecard — it + is reported here, never corrected here. """ us_ssi_take_up_stage_spec() normalized_targets = _normalize_targets(targets) + normalized_priors = _normalize_assignment_priors(assignment_priors) person = frame.table("person") if _OUTPUT not in person: raise ValueError(f"US SSI take-up diagnostics require person.{_OUTPUT}.") @@ -731,6 +775,7 @@ def us_ssi_take_up_diagnostics( source, source_assignment, targets=normalized_targets, + assignment_priors=normalized_priors, ) return _diagnostics( frame, @@ -739,15 +784,18 @@ def us_ssi_take_up_diagnostics( assigned=assigned, bands=bands, targets=normalized_targets, + law_violation_count=_bernoulli_law_violations( + source, source_assignment, normalized_priors + ), ) def us_ssi_take_up_gate( diagnostics: Mapping[str, object], *, - targets: Mapping[str, float] | None = None, + targets: Mapping[str, float], ) -> GateResult: - """Require source-faithful anchors and count calibration by age.""" + """Require source-faithful anchors and documented Bernoulli priors by age.""" expected_targets = _normalize_targets(targets) failures: list[str] = [] @@ -781,6 +829,11 @@ def us_ssi_take_up_gate( failures.append("SSI take-up lost one or more direct ASEC reporter anchors.") if int(diagnostics.get("source_identity_mismatch_count", -1)) != 0: failures.append("SSI take-up support rows disagree within a source identity.") + if int(diagnostics.get("bernoulli_law_violation_count", -1)) != 0: + failures.append( + "SSI take-up persisted flags violate the seeded Bernoulli law " + "(anchored, or draw below the documented assignment prior)." + ) channels = diagnostics.get("channel_diagnostics") if not isinstance(channels, Mapping): @@ -835,22 +888,24 @@ def us_ssi_take_up_gate( continue capacity = float(row.get("candidate_capacity", np.nan)) floor = float(row.get("reporter_candidate_floor", np.nan)) - reachable = float(row.get("reachable_goal", np.nan)) selected = float(row.get("selected_recipient_weight", np.nan)) signed_error = float(row.get("signed_target_error", np.nan)) shortfall = float(row.get("target_shortfall", np.nan)) anchor_excess = float(row.get("anchor_excess", np.nan)) prior = float(row.get("assignment_prior", np.nan)) + recomputed_prior = float( + row.get("prior_recomputed_from_current_weights", np.nan) + ) max_weight = float(row.get("max_source_candidate_weight", np.nan)) numeric_values = ( capacity, floor, - reachable, selected, signed_error, shortfall, anchor_excess, prior, + recomputed_prior, max_weight, ) if not all(np.isfinite(value) for value in numeric_values): @@ -866,22 +921,13 @@ def us_ssi_take_up_gate( f"SSI take-up age band {key!r} has an invalid anchor floor." ) continue - goal = min(max(target, floor), capacity) - epsilon = max(1e-6, np.finfo(np.float64).eps * max(goal, 1.0) * 16.0) - allowance = max(max_weight, epsilon) - if abs(reachable - goal) > epsilon: - failures.append( - f"SSI take-up age band {key!r} carries the wrong reachable goal." - ) - if selected < -epsilon or selected > capacity + epsilon: + epsilon = max(1e-6, np.finfo(np.float64).eps * max(capacity, 1.0) * 16.0) + if selected < floor - epsilon or selected > capacity + epsilon: + # Anchored reporters are always selected, so the selected weight + # can never fall below the reporter floor nor exceed capacity. failures.append( - f"SSI take-up age band {key!r} selected count is outside capacity." - ) - if abs(selected - goal) > allowance + epsilon: - failures.append( - f"SSI take-up age band {key!r} selected weight {selected:.3f} " - f"misses reachable goal {goal:.3f} by more than one source-" - f"identity weight ({allowance:.3f})." + f"SSI take-up age band {key!r} selected weight is outside " + "the [reporter floor, capacity] envelope." ) if abs(signed_error - (selected - target)) > epsilon: failures.append( @@ -895,26 +941,27 @@ def us_ssi_take_up_gate( failures.append( f"SSI take-up age band {key!r} carries the wrong anchor excess." ) - expected_prior = min(target / capacity, 1.0) - if abs(prior - expected_prior) > epsilon: + # The assignment prior is the value that generated the frozen flags; + # on release weights it will differ from target/capacity and its + # integrity is enforced by the Bernoulli-law recheck above, so the + # gate only requires a valid probability here. The recomputed prior + # is defined on THIS row's capacity/floor and must match that + # arithmetic exactly. The band-count MISS is calibration's to close + # (#470 targets) and ships in the scorecard — never a failure here. + if not 0.0 <= prior <= 1.0: + failures.append( + f"SSI take-up age band {key!r} assignment prior {prior} is " + "outside [0, 1]." + ) + if abs(recomputed_prior - _band_prior(target, capacity, floor)) > epsilon: failures.append( - f"SSI take-up age band {key!r} carries the wrong assignment prior." + f"SSI take-up age band {key!r} carries the wrong recomputed prior." ) saturated = bool(row.get("saturated")) if saturated != (capacity < target): failures.append( f"SSI take-up age band {key!r} saturation status is inconsistent." ) - if saturated: - if abs(selected - capacity) > epsilon: - failures.append( - f"SSI take-up saturated age band {key!r} did not select " - "every candidate." - ) - if shortfall <= 0: - failures.append( - f"SSI take-up saturated age band {key!r} hides its shortfall." - ) expected_total = float(sum(expected_targets.values())) recorded_total = float(diagnostics.get("target_total", np.nan)) 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 7787718c..b28dc6bc 100644 --- a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py +++ b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py @@ -116,7 +116,9 @@ def test__given_target_frame_checkpoint__then_builder_round_trips_frame( weeks_unemployed_source_sha256="weeks-source-sha", congressional_district_vintage_crosswalk_sha256="crosswalk-sha", ) - assert identity["materializer_version"] == 6 + # 7 = one-shot Bernoulli SSI take-up (populace#469): checkpoints + # materialized from count-matched flags must not survive the cutover. + assert identity["materializer_version"] == 7 assert identity["weeks_unemployed_source_sha256"] == "weeks-source-sha" path = tmp_path / "target_frame_checkpoint.h5" @@ -182,489 +184,85 @@ def calculate(self, variable, *, period, map_to): np.testing.assert_array_equal(values, np.asarray([0.0, 125.0, -2.0])) -def _ssi_diag_with_bands(selected, allowance): - """Minimal SSI take-up diagnostics carrying the swap-delta fields per band.""" - return { - "age_bands": [ - { - "age_band": key, - "selected_recipient_weight": float(selected[key]), - "max_source_candidate_weight": float(allowance), - } - for key in ("under_18", "18_64", "65_plus") - ] +def _band_spec(value, lower, upper, name, *, role=None, extra=None): + """A minimal registry target spec carrying first-class age bounds.""" + metadata = { + "target_role": role, + "age_lower_bound": lower, + "age_upper_bound": upper, + **(extra or {}), } + return SimpleNamespace(value=value, metadata=metadata, name=name) -def _count_faithful_ssi_gate(builder): - """Gate fake implementing the real per-band count-faithful band on fakes. - - Fails a band when the selected recipient weight misses its (rescaled) goal - by more than one source-identity weight -- the same test the release gate - applies, so the stale pair (which overshoots) is rejected while the fresh - re-assigned pair (which hits the goal) passes. - """ - - def gate(diagnostics, *, targets): - failures = [] - for band in diagnostics["age_bands"]: - key = band["age_band"] - selected = float(band["selected_recipient_weight"]) - allowance = float(band["max_source_candidate_weight"]) - goal = float(targets[key]) - if abs(selected - goal) > allowance: - failures.append(f"{key} selected {selected:.0f} misses {goal:.0f}") - return builder.GateResult( - name="ssi", passed=not failures, failures=tuple(failures) - ) - - return gate - - -@pytest.mark.parametrize( - ("sparse_selection_arm", "expected_cap_ratio"), - [ - pytest.param(False, 0.12, id="dense-arm"), - # The leak regression: the sparse rmloss100 script ALSO passes - # --dense-default-dataset (an export-mode flag), so the arm - # discriminator must be the frozen-selection identity — a sparse-arm - # reconcile gates at 0.10 even in dense export mode. - pytest.param(True, 0.10, id="sparse-arm-dense-export-mode"), - ], -) -def test_ssi_reconciliation_returns_fresh_pair_the_stale_gate_would_reject( - monkeypatch, - small_frame, - sparse_selection_arm, - expected_cap_ratio, -) -> None: - """The fresh pair is published where the stale pair the old loop gated fails. - - The refit drifts the weights the pass-head flags were fixed on, shifting - recipient mass between age bands (the calibration target is age-blind) while - preserving the national total. The frozen flags (the stale pair) overshoot a - band by more than one source-identity weight, so the retired stale-pair gate - would reject the run. Re-assigning under the returned weights restores - per-band count-faithfulness at the same national total, so the swap delta is - within bound and the fresh pair -- not the stale pair -- is published. - """ +def test_ssi_band_targets_from_registry_read_the_ledger_band_specs() -> None: builder = _load_builder_module() - national_spec = SimpleNamespace( - value=7_404_820.0, - metadata={"target_role": builder.SSA_SSI_RECIPIENTS_TARGET_ROLE}, - name="ssa-ssi-recipients-national", - ) - target_specs = (national_spec,) - band_targets, _ = builder._aligned_ssi_take_up_band_targets(target_specs) - - initial_result = SimpleNamespace( - weights=np.asarray([1_100.0, 1_900.0]), - initial_weights=np.asarray([1_000.0, 2_000.0]), - ) - reconciled_result = SimpleNamespace( - weights=np.asarray([1_250.0, 1_750.0]), - initial_weights=np.asarray([1_000.0, 2_000.0]), - final_loss=0.25, - ) - - allowance = 100.0 - fresh_selected = dict(band_targets) - # The refit shifted 1,000 of recipient mass from under_18 into 18_64: the - # national total is unchanged, but 18_64 now overshoots its goal by 1,000. - stale_selected = { - "under_18": band_targets["under_18"] - 1_000.0, - "18_64": band_targets["18_64"] + 1_000.0, - "65_plus": band_targets["65_plus"], - } - - monkeypatch.setattr(builder, "_assert_no_formula_owned_columns", lambda frame: None) - monkeypatch.setattr( - builder, "us_ssi_take_up_reporter_source_ids", lambda frame: frozenset({"r"}) - ) - monkeypatch.setattr( - builder, - "_ssi_person_uncapped_amount", - lambda frame, **kwargs: np.zeros(frame.n("person")), - ) - - def fake_assign(frame, *, uncapped_ssi, seed, targets, reporter_source_ids): - assert targets == band_targets - return frame, _ssi_diag_with_bands(fresh_selected, allowance) - - monkeypatch.setattr(builder, "with_us_ssi_take_up", fake_assign) - - def fake_stale_diagnostics( - frame, *, uncapped_ssi, seed, targets, reporter_source_ids - ): - assert targets == band_targets - return _ssi_diag_with_bands(stale_selected, allowance) - - monkeypatch.setattr(builder, "us_ssi_take_up_diagnostics", fake_stale_diagnostics) - monkeypatch.setattr( - builder, "us_ssi_take_up_gate", _count_faithful_ssi_gate(builder) - ) - - monkeypatch.setattr( - builder, "_with_aca_marketplace_source_outputs", lambda frame, *a, **k: frame - ) - monkeypatch.setattr( - builder, - "_health_input_signal_gate", - lambda frame: builder.GateResult(name="health", passed=True), - ) - monkeypatch.setattr( - builder, - "_with_medicaid_take_up_outputs", - lambda frame, *a, **k: ( - frame, - {"states": [{"state_fips": "06", "enrolled_weight": 10.0}]}, - ), - ) - monkeypatch.setattr( - builder, - "_medicaid_diagnostics_for_existing_output", - lambda *a, **k: {"states": [{"state_fips": "06", "enrolled_weight": 12.0}]}, - ) - monkeypatch.setattr( - builder, - "us_medicaid_take_up_gate", - lambda diagnostics: builder.GateResult(name="medicaid", passed=True), - ) - monkeypatch.setattr( - builder, "with_us_other_health_insurance_inputs", lambda frame, **k: frame - ) - monkeypatch.setattr( - builder, - "us_other_health_insurance_signal_gate", - lambda frame: builder.GateResult(name="other_health", passed=True), - ) - - class FakeRegistry: - specs = (SimpleNamespace(),) - - def to_target_set(self): - return "targets" - - monkeypatch.setattr( - builder, - "_materialize_target_frame", - lambda frame, *a, **k: (frame, FakeRegistry(), {"declared_targets": 1}), - ) - monkeypatch.setattr( - builder, "_fiscal_target_loss_weights", lambda registry: np.ones(1) - ) - monkeypatch.setattr(builder, "calibrate", lambda *a, **k: reconciled_result) - - reconciliation = builder._reconcile_ssi_take_up_and_refit( - small_frame, - initial_result, - target_specs, - dense_default_dataset=True, - sparse_selection_arm=sparse_selection_arm, - seed=3, - epochs=5, - learning_rate=0.01, - max_weight_ratio=10.0, - l2_lambda=0.0, - target_loss_cap=1.0, - ) - - # Teeth: the stale pair the retired loop gated FAILS that gate, yet the - # reconciliation succeeds by publishing the fresh re-assigned pair. - stale_diagnostics = _ssi_diag_with_bands(stale_selected, allowance) - fresh_diagnostics = _ssi_diag_with_bands(fresh_selected, allowance) - assert not builder.us_ssi_take_up_gate( - stale_diagnostics, targets=band_targets - ).passed - assert builder.us_ssi_take_up_gate(fresh_diagnostics, targets=band_targets).passed - - assert reconciliation.passes == 1 - assert reconciliation.ssi_diagnostics == fresh_diagnostics - np.testing.assert_array_equal( - reconciliation.export_frame.weights_for("household").values, - reconciled_result.weights, - ) - record = reconciliation.compilation["ssi_take_up_reconciliation"] - assert record["exit_policy"] == "fresh_pair_under_returned_weights" - assert ( - record["ssi_swap_delta"]["national_swap_sanity_cap_ratio"] == expected_cap_ratio - ) - assert [entry["pass"] for entry in record["pass_history"]] == list( - range(1, len(record["pass_history"]) + 1) - ) - assert all( - "national_swap_delta" in entry and "within_bound" in entry - for entry in record["pass_history"] - ) - assert record["target_alignment"][ - "registry_national_recipients_total" - ] == pytest.approx(7_404_820.0) - swap = record["ssi_swap_delta"] - assert swap["within_bound"] is True - # The refit preserved the national total, so the national swap delta is ~0 - # even though the per-band re-assignment moved 1,000 of mass in two bands. - assert swap["national_swap_delta"] == pytest.approx(0.0, abs=1e-6) - assert swap["age_bands"]["under_18"]["swap_delta"] == pytest.approx(1_000.0) - assert swap["age_bands"]["18_64"]["swap_delta"] == pytest.approx(-1_000.0) - # Medicaid enrolled-mass swap recorded per state (fresh 10 - stale 12). - med_state = record["medicaid_enrollment_swap_delta"]["states"]["06"] - assert med_state["swap_delta"] == pytest.approx(-2.0) - - -def test_ssi_reconciliation_fails_closed_when_reassignment_swap_exceeds_bound( - monkeypatch, - small_frame, -) -> None: - """A runaway swap delta still fails closed after the bounded passes. - - When re-assigning under the returned weights moves the aggregate recipient - mass by more than a tenth of the national total (the solve effectively - abandoned the SSI family), the fresh pair's own gates pass but the sanity - cap does not, so the pass fails and the loop raises after ``max_passes`` - with the swap value in the message. - """ - builder = _load_builder_module() - national_spec = SimpleNamespace( - value=7_404_820.0, - metadata={"target_role": builder.SSA_SSI_RECIPIENTS_TARGET_ROLE}, - name="ssa-ssi-recipients-national", - ) - target_specs = (national_spec,) - band_targets, _ = builder._aligned_ssi_take_up_band_targets(target_specs) - - initial_result = SimpleNamespace( - weights=np.asarray([1_100.0, 1_900.0]), - initial_weights=np.asarray([1_000.0, 2_000.0]), - ) - returned_result = SimpleNamespace( - weights=np.asarray([1_100.0, 1_900.0]), - initial_weights=np.asarray([1_000.0, 2_000.0]), - final_loss=1.0, - ) - - counts = {"assign": 0, "calibrate": 0, "stale": 0} - allowance = 100.0 - fresh_selected = dict(band_targets) - # The frozen flags overshoot the national total by 1,100,000 (in under_18); - # the fresh re-assignment removes it, so the national swap delta exceeds - # the runaway sanity cap on BOTH arms (0.10 sparse / 0.12 dense of the - # ~7.4M fresh total; populace#447) — the solve - # abandoned the SSI family, which must still fail closed. - stale_selected = { - "under_18": band_targets["under_18"] + 1_100_000.0, - "18_64": band_targets["18_64"], - "65_plus": band_targets["65_plus"], - } - - monkeypatch.setattr(builder, "_assert_no_formula_owned_columns", lambda frame: None) - monkeypatch.setattr( - builder, "us_ssi_take_up_reporter_source_ids", lambda frame: frozenset({"r"}) - ) - monkeypatch.setattr( - builder, - "_ssi_person_uncapped_amount", - lambda frame, **kwargs: np.zeros(frame.n("person")), - ) - - def fake_assign(frame, **kwargs): - counts["assign"] += 1 - return frame, _ssi_diag_with_bands(fresh_selected, allowance) - - monkeypatch.setattr(builder, "with_us_ssi_take_up", fake_assign) - - def fake_stale(*args, **kwargs): - counts["stale"] += 1 - return _ssi_diag_with_bands(stale_selected, allowance) - - monkeypatch.setattr(builder, "us_ssi_take_up_diagnostics", fake_stale) - monkeypatch.setattr( - builder, "us_ssi_take_up_gate", _count_faithful_ssi_gate(builder) - ) - - monkeypatch.setattr( - builder, "_with_aca_marketplace_source_outputs", lambda frame, *a, **k: frame - ) - monkeypatch.setattr( - builder, - "_health_input_signal_gate", - lambda frame: builder.GateResult(name="health", passed=True), - ) - monkeypatch.setattr( - builder, - "_with_medicaid_take_up_outputs", - lambda frame, *a, **k: (frame, {"states": []}), - ) - monkeypatch.setattr( - builder, - "_medicaid_diagnostics_for_existing_output", - lambda *a, **k: {"states": []}, - ) - monkeypatch.setattr( - builder, - "us_medicaid_take_up_gate", - lambda diagnostics: builder.GateResult(name="medicaid", passed=True), - ) - monkeypatch.setattr( - builder, "with_us_other_health_insurance_inputs", lambda frame, **k: frame - ) - monkeypatch.setattr( - builder, - "us_other_health_insurance_signal_gate", - lambda frame: builder.GateResult(name="other_health", passed=True), - ) - - class FakeRegistry: - specs = (SimpleNamespace(),) - - def to_target_set(self): - return "targets" - - monkeypatch.setattr( - builder, - "_materialize_target_frame", - lambda frame, *a, **k: (frame, FakeRegistry(), {}), - ) - monkeypatch.setattr( - builder, "_fiscal_target_loss_weights", lambda registry: np.ones(1) - ) - - def fake_calibrate(*args, **kwargs): - counts["calibrate"] += 1 - return returned_result - - monkeypatch.setattr(builder, "calibrate", fake_calibrate) - - with pytest.raises(RuntimeError, match="after 2 pass") as excinfo: - builder._reconcile_ssi_take_up_and_refit( - small_frame, - initial_result, - target_specs, - dense_default_dataset=True, - seed=0, - epochs=1, - learning_rate=0.01, - max_weight_ratio=10.0, - l2_lambda=0.0, - target_loss_cap=1.0, - max_passes=2, - ) - - message = str(excinfo.value) - assert "swap delta" in message - assert "1100000.000" in message - # populace#447: the per-pass trajectory must survive the terminal raise — - # converging-but-over-cap vs oscillating is the adjudication evidence. - assert "Pass trajectory: pass 1: delta=" in message - assert "pass 2: delta=" in message - assert "within_bound=False" in message - # Two passes: a stage assign and an exit assign each pass, one stale diag - # each pass, one refit each pass. - assert counts == {"assign": 4, "calibrate": 2, "stale": 2} - - -def test_registry_national_ssi_recipients_total_sums_national_specs() -> None: - builder = _load_builder_module() - role = builder.SSA_SSI_RECIPIENTS_TARGET_ROLE + role = builder.SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE + # The real feed's under-18 fact carries an explicit "age >= 0" + # constraint, so its lower bound compiles as "0", not "-inf"; ages are + # nonnegative, so both mean the same stratum (PR #477 review finding 1). specs = ( - SimpleNamespace( - value=7_000_000.0, metadata={"target_role": role}, name="nat-a" - ), - SimpleNamespace(value=404_820.0, metadata={"target_role": role}, name="nat-b"), - SimpleNamespace( - value=123.0, - metadata={"target_role": role, "state_fips": "06"}, - name="state-row", - ), + _band_spec(1_001_922.0, "0", "18", "under-18", role=role), + _band_spec(3_905_779.0, "18", "65", "18-64", role=role), + _band_spec(2_382_142.0, "65", "inf", "65-plus", role=role), SimpleNamespace(value=9.0, metadata={"target_role": "other"}, name="unrelated"), ) - assert builder._registry_national_ssi_recipients_total(specs) == pytest.approx( - 7_404_820.0 - ) + assert builder._ssi_take_up_band_targets_from_registry(specs) == { + "under_18": pytest.approx(1_001_922.0), + "18_64": pytest.approx(3_905_779.0), + "65_plus": pytest.approx(2_382_142.0), + } -def test_registry_national_ssi_recipients_total_fails_closed_without_national() -> None: +def test_ssi_band_targets_accept_unbounded_lower_edge_spelling() -> None: builder = _load_builder_module() - role = builder.SSA_SSI_RECIPIENTS_TARGET_ROLE + role = builder.SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE specs = ( - SimpleNamespace( - value=123.0, - metadata={"target_role": role, "state_fips": "06"}, - name="state-row", - ), - SimpleNamespace( - value=9.0, metadata={"target_role": "medicaid_enrollment"}, name="medicaid" - ), + _band_spec(1_001_922.0, "-inf", "18", "under-18", role=role), + _band_spec(3_905_779.0, "18", "65", "18-64", role=role), + _band_spec(2_382_142.0, "65", "inf", "65-plus", role=role), ) - with pytest.raises(RuntimeError, match="could not find a national"): - builder._registry_national_ssi_recipients_total(specs) + assert builder._ssi_take_up_band_targets_from_registry(specs)[ + "under_18" + ] == pytest.approx(1_001_922.0) -def test_aligned_ssi_take_up_band_targets_applies_shares_to_national_total() -> None: +def test_ssi_band_targets_fail_closed_when_a_band_is_missing() -> None: builder = _load_builder_module() - role = builder.SSA_SSI_RECIPIENTS_TARGET_ROLE - national = 7_404_820.0 + role = builder.SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE specs = ( - SimpleNamespace( - value=national, metadata={"target_role": role}, name="national" - ), + _band_spec(1_001_922.0, "-inf", "18", "under-18", role=role), + _band_spec(2_382_142.0, "65", "inf", "65-plus", role=role), ) - band_targets, record = builder._aligned_ssi_take_up_band_targets(specs) + with pytest.raises(RuntimeError, match=r"missing band\(s\) \['18_64'\]"): + builder._ssi_take_up_band_targets_from_registry(specs) - ssa_total = sum(t.person_count for t in builder.US_SSI_TAKE_UP_AGE_TARGETS) - for target in builder.US_SSI_TAKE_UP_AGE_TARGETS: - expected = target.person_count / ssa_total * national - assert band_targets[target.key] == pytest.approx(expected) - # The rescaled band goals sum to the national total the refit enforces -- - # not to the ~115k-larger SSA federal-payment-by-age band total. - assert sum(band_targets.values()) == pytest.approx(national) - assert record["registry_national_recipients_total"] == pytest.approx(national) - assert record["ssa_federal_payment_recipient_band_total"] == pytest.approx( - ssa_total - ) - assert sum(record["band_shares"].values()) == pytest.approx(1.0) +def test_ssi_band_targets_reject_unrecognized_bounds() -> None: + builder = _load_builder_module() + role = builder.SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE + specs = (_band_spec(1_001_922.0, "-inf", "19", "off-by-one", role=role),) + with pytest.raises(RuntimeError, match="unrecognized age bounds"): + builder._ssi_take_up_band_targets_from_registry(specs) -def test_ssi_take_up_swap_delta_records_solve_residual_within_sanity_cap() -> None: - """The delta is the solve's SSI-family residual: recorded, sanity-capped. - - Attempt 8 proved the delta tracks the age-blind solve's equilibrium miss - on the SSI-recipient family (~419k on 7.4M), not assignment granularity — - gating it at one source-identity weight per band demanded solve precision - no other target faces. A residual well beyond granularity but under a - tenth of the fresh national total is therefore recorded and passes; the - granularity sum ships as a reference quantity only. - """ +def test_ssi_band_targets_reject_duplicate_bands() -> None: builder = _load_builder_module() - stale = _ssi_diag_with_bands( - {"under_18": 1_000.0, "18_64": 4_000.0, "65_plus": 2_000.0}, allowance=50.0 - ) - # Fresh restores +400 nationally: far beyond the 150 granularity sum, - # within the 10% sanity cap (703 on a 7,030 fresh total). - fresh = _ssi_diag_with_bands( - {"under_18": 1_030.0, "18_64": 4_300.0, "65_plus": 1_700.0}, allowance=50.0 + role = builder.SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE + specs = ( + _band_spec(1_001_922.0, "-inf", "18", "under-18", role=role), + _band_spec(999_999.0, "-inf", "18", "under-18-again", role=role), ) - swap = builder._ssi_take_up_swap_delta(stale, fresh) - assert swap["national_swap_delta"] == pytest.approx(30.0) - assert swap["assignment_granularity_reference"] == pytest.approx(150.0) - assert swap["national_swap_sanity_cap"] == pytest.approx(703.0) - assert swap["within_bound"] is True - assert swap["age_bands"]["18_64"]["swap_delta"] == pytest.approx(300.0) - assert swap["age_bands"]["65_plus"]["swap_delta"] == pytest.approx(-300.0) + with pytest.raises(RuntimeError, match="duplicate registry targets"): + builder._ssi_take_up_band_targets_from_registry(specs) -def test_ssi_take_up_swap_delta_flags_runaway_beyond_sanity_cap() -> None: +def test_ssi_band_targets_reject_nonpositive_values() -> None: builder = _load_builder_module() - stale = _ssi_diag_with_bands( - {"under_18": 1_000.0, "18_64": 4_000.0, "65_plus": 2_000.0}, allowance=50.0 - ) - fresh = _ssi_diag_with_bands( - {"under_18": 2_000.0, "18_64": 4_000.0, "65_plus": 2_000.0}, allowance=50.0 - ) - swap = builder._ssi_take_up_swap_delta(stale, fresh) - assert swap["national_swap_delta"] == pytest.approx(1_000.0) - assert swap["national_swap_sanity_cap"] == pytest.approx(800.0) - assert swap["within_bound"] is False + role = builder.SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE + specs = (_band_spec(0.0, "-inf", "18", "under-18", role=role),) + with pytest.raises(RuntimeError, match="finite and positive"): + builder._ssi_take_up_band_targets_from_registry(specs) def test__given_stale_target_frame_checkpoint__then_builder_ignores_it( @@ -3226,28 +2824,56 @@ def fake_ssi_uncapped_amount( captured["ssi_uncapped_batch_size"] = maximum_microsim_batch_size return np.zeros(4, dtype=np.float64) + fake_band_targets = { + "under_18": 1_001_922.0, + "18_64": 3_905_779.0, + "65_plus": 2_382_142.0, + } + + def fake_band_targets_from_registry(specs): + captured["ssi_band_targets_specs"] = specs + return dict(fake_band_targets) + + fake_stage_priors = {"under_18": 0.3, "18_64": 0.4, "65_plus": 0.5} + fake_stage_diagnostics = { + "checked": True, + "age_bands": [ + {"age_band": key, "assignment_prior": prior} + for key, prior in fake_stage_priors.items() + ], + } + def fake_with_ssi_take_up( frame, *, uncapped_ssi, seed, + targets, reporter_source_ids, ): captured["ssi_take_up_stage_called"] = True captured["ssi_take_up_seed"] = seed captured["ssi_take_up_uncapped"] = np.asarray(uncapped_ssi) + captured["ssi_take_up_targets"] = dict(targets) captured["ssi_reporter_source_ids"] = reporter_source_ids - return frame, {"checked": True} + return frame, dict(fake_stage_diagnostics) - def fake_ssi_take_up_gate(diagnostics): + def fake_ssi_take_up_gate(diagnostics, *, targets): captured["ssi_take_up_gate_called"] = True - captured["ssi_take_up_gate_diagnostics"] = diagnostics + captured.setdefault("ssi_take_up_gate_calls", []).append( + {"diagnostics": diagnostics, "targets": dict(targets)} + ) return builder.GateResult( name="ssi_take_up", passed=True, details=diagnostics, ) + monkeypatch.setattr( + builder, + "_ssi_take_up_band_targets_from_registry", + fake_band_targets_from_registry, + ) monkeypatch.setattr( builder, "_ssi_person_uncapped_amount", @@ -3505,35 +3131,46 @@ def fake_write_diagnostics(**kwargs): monkeypatch.setattr(builder, "calibrate_l0_refit", fake_calibrate_l0_refit) - def fake_reconcile(frame, initial_result, specs, **kwargs): - captured["ssi_reconciliation_called"] = True - captured["ssi_reconciliation_reporter_source_ids"] = kwargs[ - "reporter_source_ids" - ] - return builder._SSITakeUpReconciliationResult( - export_frame=frame, - calibration_result=initial_result, - registry=registry, - compilation={"dropped_target_names": []}, - ssi_diagnostics={"checked": True}, - medicaid_diagnostics={}, - health_input_gate=builder.GateResult( - name="health_input_signal", - passed=True, - details={"checked": True}, - ), - other_health_insurance_gate=builder.GateResult( - name="other_health_insurance_premiums_signal", - passed=True, - details={"checked": True}, - ), - passes=1, - ) + def fake_l0_refit_weights(frame, refit_result): + captured["export_frame_from_l0_refit"] = True + return frame + + def fake_final_ssi_diagnostics( + frame, + *, + uncapped_ssi, + seed, + targets, + assignment_priors, + reporter_source_ids, + ): + captured["final_ssi_diagnostics_called"] = True + captured["final_ssi_diagnostics_targets"] = dict(targets) + captured["final_ssi_diagnostics_assignment_priors"] = dict(assignment_priors) + captured["final_ssi_diagnostics_reporter_source_ids"] = reporter_source_ids + return {"checked": True} + + def fake_final_medicaid_diagnostics( + frame, + specs, + *, + seed, + substitutions, + maximum_microsim_batch_size=None, + ): + captured["final_medicaid_diagnostics_called"] = True + return {} + monkeypatch.setattr(builder, "_with_l0_refit_weights", fake_l0_refit_weights) monkeypatch.setattr( builder, - "_reconcile_ssi_take_up_and_refit", - fake_reconcile, + "us_ssi_take_up_diagnostics", + fake_final_ssi_diagnostics, + ) + monkeypatch.setattr( + builder, + "_medicaid_diagnostics_for_existing_output", + fake_final_medicaid_diagnostics, ) monkeypatch.setattr( builder, @@ -3581,8 +3218,6 @@ def fake_reconcile(frame, initial_result, specs, **kwargs): "selection_final_loss": 1.5, "refit_initial_loss": 2.0, "refit_final_loss": 1.0, - "pre_ssi_reconciliation_final_loss": 1.0, - "ssi_take_up_reconciliation_passes": 1, "final_loss": 1.0, } assert captured["l0_kwargs"]["l0_lambda"] == 0.2 @@ -3665,12 +3300,29 @@ def fake_reconcile(frame, initial_result, specs, **kwargs): assert captured["ssi_take_up_stage_called"] is True assert captured["ssi_take_up_seed"] == 0 assert captured["ssi_take_up_uncapped"].shape == (4,) + assert captured["ssi_take_up_targets"] == fake_band_targets assert captured["ssi_reporter_source_ids"] == frozenset({"asec-reporter"}) - assert captured["ssi_reconciliation_reporter_source_ids"] == frozenset( + assert captured["ssi_take_up_gate_called"] is True + # The gate binds twice: the fresh stage diagnostics at assignment time, + # then the persisted-flag measurement on the export frame (PR #477 + # review finding 3) — both against the registry band targets. + gate_calls = captured["ssi_take_up_gate_calls"] + assert [call["diagnostics"] for call in gate_calls] == [ + fake_stage_diagnostics, + {"checked": True}, + ] + assert all(call["targets"] == fake_band_targets for call in gate_calls) + # One-shot regime (populace#469): the frozen flags are measured on the + # release weights, never reassigned or reconciled, and the final + # measurement republishes the stage's assignment priors verbatim. + assert captured["export_frame_from_l0_refit"] is True + assert captured["final_ssi_diagnostics_called"] is True + assert captured["final_ssi_diagnostics_targets"] == fake_band_targets + assert captured["final_ssi_diagnostics_assignment_priors"] == fake_stage_priors + assert captured["final_ssi_diagnostics_reporter_source_ids"] == frozenset( {"asec-reporter"} ) - assert captured["ssi_take_up_gate_called"] is True - assert captured["ssi_take_up_gate_diagnostics"] == {"checked": True} + assert captured["final_medicaid_diagnostics_called"] is True assert captured["voluntary_filing_donor_path"] == Path("pu2023.csv") assert ( captured["voluntary_filing_donor_sha256"] @@ -3711,6 +3363,8 @@ def fake_reconcile(frame, initial_result, specs, **kwargs): "medicaid_gate", "other_health", "other_health_gate", + # The export-frame signal re-check (one-shot regime, populace#469). + "other_health_gate", ] @@ -7327,45 +6981,6 @@ def test_checkpoint_identity_protection_key_and_stale_checkpoint_miss( ) -def test_ssi_swap_delta_dense_cap_ratio_admits_measured_dense_equilibrium() -> None: - """populace#447: the dense arm's reconcile equilibrium (measured trajectory - 11.97% -> 11.64% -> 11.38%, decelerating toward ~10.6%) sits above the - sparse 10% runaway cap. The dense-specific 0.12 ratio admits the measured - equilibrium while both ratios still refuse a genuine runaway, and the - ratio used is recorded in the payload.""" - builder = _load_builder_module() - fresh = {"under_18": 1_000_000.0, "18_64": 4_000_000.0, "65_plus": 2_400_000.0} - stale = dict(fresh) - # 11.42% of the 7.4M fresh national: inside 0.12, outside 0.10. - stale["18_64"] += 845_000.0 - - sparse = builder._ssi_take_up_swap_delta( - _ssi_diag_with_bands(stale, 40_000.0), - _ssi_diag_with_bands(fresh, 40_000.0), - ) - dense = builder._ssi_take_up_swap_delta( - _ssi_diag_with_bands(stale, 40_000.0), - _ssi_diag_with_bands(fresh, 40_000.0), - sanity_cap_ratio=builder.SSI_TAKE_UP_SWAP_SANITY_CAP_RATIO_DENSE, - ) - - assert sparse["within_bound"] is False - assert sparse["national_swap_sanity_cap_ratio"] == 0.10 - assert dense["within_bound"] is True - assert dense["national_swap_sanity_cap_ratio"] == 0.12 - # A genuine runaway (>12%) is still refused on the dense ratio. - runaway = dict(fresh) - runaway["18_64"] += 1_000_000.0 - assert ( - builder._ssi_take_up_swap_delta( - _ssi_diag_with_bands(runaway, 40_000.0), - _ssi_diag_with_bands(fresh, 40_000.0), - sanity_cap_ratio=builder.SSI_TAKE_UP_SWAP_SANITY_CAP_RATIO_DENSE, - )["within_bound"] - is False - ) - - def _table_1_4_diagnostic(builder, name: str, target: float, final: float): return SimpleNamespace( name=f"{name}@{builder.PERIOD}", diff --git a/packages/populace-build/tests/test_us_plan.py b/packages/populace-build/tests/test_us_plan.py index 41ddbf3f..f793ad7f 100644 --- a/packages/populace-build/tests/test_us_plan.py +++ b/packages/populace-build/tests/test_us_plan.py @@ -43,7 +43,6 @@ US_SSI_DISABILITY_CRITERIA_STAGE_NAME, US_SSI_TAKE_UP_ANCHOR, US_SSI_TAKE_UP_STAGE_NAME, - US_SSI_TAKE_UP_TARGET_TABLE_NAME, US_STAGE_NAMES, US_SUPPORT_SPINE_MANIFEST, US_SUPPORT_SPINE_SPEC, @@ -331,7 +330,9 @@ def test_head_start_stage_pins_measured_sipp_response_contract(self) -> None: assert "AEDHEADST" in stage.notes assert "measured" in stage.notes.lower() - def test_ssi_take_up_stage_pins_reporter_and_ssa_count_contract(self) -> None: + def test_ssi_take_up_stage_pins_reporter_and_bernoulli_prior_contract( + self, + ) -> None: stage = US_SOURCE_MANIFEST.stage_map()[US_SSI_TAKE_UP_STAGE_NAME] donor = US_DONORS[US_SSI_TAKE_UP_STAGE_NAME] @@ -347,7 +348,6 @@ def test_ssi_take_up_stage_pins_reporter_and_ssa_count_contract(self) -> None: assert [operation.kind for operation in stage.operations] == [ "read_table", "assign_binary_from_rate", - "calibrate_binary_assignment", ] assert dict(stage.operations[0].parameters) == { "table": "person", @@ -361,30 +361,20 @@ def test_ssi_take_up_stage_pins_reporter_and_ssa_count_contract(self) -> None: "reported_true_anchor": f"{US_SSI_TAKE_UP_ANCHOR} > 0", "assignment_unit": "person_source_id", "fan_to_support_clones": True, - } - assert dict(stage.operations[2].parameters) == { - "variable": "takes_up_ssi_if_eligible", - "targets": [US_SSI_TAKE_UP_TARGET_TABLE_NAME], - "preserve_true_anchors": True, - "preserve_true_anchor": f"{US_SSI_TAKE_UP_ANCHOR} > 0", - "domain": "uncapped_ssi > 0", - "weight": "person_weight", - "draw": "stable_source_person_draw", - "calibration_unit": "person_source_id", "age_bands": { "under_18": "age < 18", "18_64": "18 <= age < 65", "65_plus": "age >= 65", }, + "rate_derivation": ( + "band_target / weighted_candidate_capacity(uncapped_ssi > 0); " + "min(reporter_candidate_floor / capacity, 1) once that ratio " + "reaches one" + ), + "rate_target_role": "ssa_ssi_age_band_recipients", "target_source": SSI_TAKE_UP_SSA_SOURCE_URL, "target_period": "2024-12", "target_measure": "Total with—Federal payment", - "target_values": { - "under_18": 1_001_922, - "18_64": 3_905_779, - "65_plus": 2_382_142, - }, - "aggregate_target": 7_289_843, } ssa_artifacts = [ artifact @@ -392,6 +382,9 @@ def test_ssi_take_up_stage_pins_reporter_and_ssa_count_contract(self) -> None: if artifact.get("source") == SSI_TAKE_UP_SSA_SOURCE_URL ] assert len(ssa_artifacts) == 1 + # The SSA recipient counts bind only through the ledger-fed + # calibration registry (populace#469/#470) — never hardcoded here. + assert all("target_values" not in artifact for artifact in stage.artifacts) evidence = next( artifact for artifact in stage.artifacts diff --git a/packages/populace-build/tests/test_us_ssi_take_up.py b/packages/populace-build/tests/test_us_ssi_take_up.py index 0a792479..f19efd11 100644 --- a/packages/populace-build/tests/test_us_ssi_take_up.py +++ b/packages/populace-build/tests/test_us_ssi_take_up.py @@ -1,4 +1,4 @@ -"""Reporter-anchored, SSA count-calibrated SSI take-up tests.""" +"""Reporter-anchored, Bernoulli-at-documented-prior SSI take-up tests.""" from __future__ import annotations @@ -12,6 +12,9 @@ import pandas as pd import pytest +from populace.build.us_runtime.fiscal_targets import ( + SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE, +) from populace.build.us_runtime.ssi_take_up import ( SSI_TAKE_UP_ARCHIVED_DERIVATION_URL, SSI_TAKE_UP_ARCHIVED_RANDOMNESS_URL, @@ -21,7 +24,7 @@ US_SSI_TAKE_UP_ANCHOR, US_SSI_TAKE_UP_OUTPUT_COLUMNS, US_SSI_TAKE_UP_STAGE_NAME, - US_SSI_TAKE_UP_TARGET_TABLE_NAME, + _stable_source_draw, us_ssi_take_up_diagnostics, us_ssi_take_up_gate, us_ssi_take_up_reporter_source_ids, @@ -34,6 +37,22 @@ _OUTPUT = US_SSI_TAKE_UP_OUTPUT_COLUMNS[0] _TARGETS = {target.key: 50.0 for target in US_SSI_TAKE_UP_AGE_TARGETS} _AGES = {"under_18": 12.0, "18_64": 40.0, "65_plus": 72.0} +# Fixture arithmetic per band: six dual-channel candidates weigh 20.0 each +# and the PUF-only candidate weighs 10.0 (capacity 130.0); the sole anchored +# candidate is source 0 (reporter floor 20.0). +_BAND_CAPACITY = 130.0 +_REPORTER_FLOOR = 20.0 +_ANCHORED_SOURCE_NUMBERS = {"0", "6"} + + +def _expected_bernoulli_flag(source_id: str, prior: float, *, seed: int = 17) -> bool: + """The selection law: anchors unconditionally, else draw below prior.""" + + if source_id.split(":")[1] in _ANCHORED_SOURCE_NUMBERS: + return True + return _stable_source_draw(source_id, seed=seed) < prior + + _policyengine_us_installed = importlib.util.find_spec("policyengine_us") is not None requires_us = pytest.mark.skipif( not _policyengine_us_installed, @@ -129,21 +148,29 @@ def _assigned( return frame, result, potential, diagnostics -def test_stage_contract_pins_archived_method_and_official_age_targets() -> None: +def test_stage_contract_pins_archived_method_and_band_structure() -> None: spec = us_ssi_take_up_stage_spec() assert spec.stage == US_SSI_TAKE_UP_STAGE_NAME assert spec.source == SSI_TAKE_UP_SSA_SOURCE_URL assert spec.outputs == (_OUTPUT,) - assert US_SSI_TAKE_UP_TARGET_TABLE_NAME in spec.operations[2].parameters["targets"] + assert [operation.kind for operation in spec.operations] == [ + "read_table", + "assign_binary_from_rate", + ] + assignment = dict(spec.operations[1].parameters) + assert assignment["rate_target_role"] == SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE + assert "uncapped_ssi > 0" in assignment["rate_derivation"] + # Recipient counts live in the ledger and bind via the calibration + # registry (populace#469/#470) — the stage may never hardcode them. + assert all("target_values" not in artifact for artifact in spec.artifacts) assert "42ed5d45" in SSI_TAKE_UP_ARCHIVED_DERIVATION_URL assert "cps.py#L650-L657" in SSI_TAKE_UP_ARCHIVED_DERIVATION_URL assert "takeup.py#L10-L35" in SSI_TAKE_UP_ARCHIVED_RANDOMNESS_URL assert "ssi_targets.py#L41-L74" in SSI_TAKE_UP_ARCHIVED_TARGETS_URL - assert [target.person_count for target in US_SSI_TAKE_UP_AGE_TARGETS] == [ - 1_001_922.0, - 3_905_779.0, - 2_382_142.0, - ] + assert [ + (band.key, band.minimum_age, band.maximum_age) + for band in US_SSI_TAKE_UP_AGE_TARGETS + ] == [("under_18", None, 17), ("18_64", 18, 64), ("65_plus", 65, None)] def test_assignment_preserves_asec_reporters_and_fans_source_decisions() -> None: @@ -215,29 +242,48 @@ def test_non_candidate_reporter_remains_anchored_but_not_in_recipient_count() -> assert band["reporter_candidate_floor"] == 20.0 -def test_reachable_targets_hit_within_one_source_identity_weight() -> None: - _, _, _, diagnostics = _assigned() +def test_selection_is_anchors_union_of_seeded_draws_below_the_band_prior() -> None: + _, result, _, diagnostics = _assigned() + priors = { + band["age_band"]: band["assignment_prior"] for band in diagnostics["age_bands"] + } + by_source = result.table("person").groupby("person_source_id")[_OUTPUT].first() + for source_id, flagged in by_source.items(): + prior = priors[source_id.split(":")[0]] + assert bool(flagged) == _expected_bernoulli_flag(source_id, prior) for band in diagnostics["age_bands"]: assert not band["saturated"] - assert ( - abs(band["selected_recipient_weight"] - band["target"]) - <= band["max_source_candidate_weight"] + assert band["assignment_prior"] == pytest.approx(50.0 / _BAND_CAPACITY) + # At assignment time the stored prior and the current-weight + # recomputation coincide by construction. + assert band["prior_recomputed_from_current_weights"] == pytest.approx( + band["assignment_prior"] ) + assert diagnostics["bernoulli_law_violation_count"] == 0 + # Schema 2 = the populace#469 shape (assignment/recomputed prior split, + # law count, no reachable_goal); pin the literal so reverting the + # constant alone cannot pass. + assert diagnostics["schema_version"] == 2 assert us_ssi_take_up_gate(diagnostics, targets=_TARGETS).passed -def test_unreachable_target_saturates_only_the_counted_candidate_domain() -> None: +def test_saturated_band_prior_falls_back_to_the_observed_reporter_rate() -> None: targets = {"under_18": 1_000.0, "18_64": 50.0, "65_plus": 50.0} - _, result, potential, diagnostics = _assigned(targets=targets) - flag = result.table("person")[_OUTPUT].to_numpy(dtype=bool) - child_rows = result.table("person")["age"].lt(18).to_numpy() - assert flag[child_rows & (potential > 0)].all() + _, result, _, diagnostics = _assigned(targets=targets) child, adult, aged = diagnostics["age_bands"] assert child["saturated"] - assert child["selected_recipient_weight"] == child["candidate_capacity"] + assert child["assignment_prior"] == pytest.approx(_REPORTER_FLOOR / _BAND_CAPACITY) assert child["target_shortfall"] > 0 assert not adult["saturated"] + assert adult["assignment_prior"] == pytest.approx(50.0 / _BAND_CAPACITY) assert not aged["saturated"] + by_source = result.table("person").groupby("person_source_id")[_OUTPUT].first() + for source_id, flagged in by_source.items(): + if not source_id.startswith("under_18:"): + continue + assert bool(flagged) == _expected_bernoulli_flag( + source_id, _REPORTER_FLOOR / _BAND_CAPACITY + ) assert us_ssi_take_up_gate(diagnostics, targets=targets).passed @@ -246,23 +292,36 @@ def test_every_band_saturated_stays_nonconstant_and_passes_the_gate() -> None: Build M's first sparse run died here: the restored disability battery put SSI candidates in every age band, every band's candidate capacity fell - short of its SSA target, the count-matching ratio degenerated to 1.0, and - Bernoulli(1.0) flagged the entire pool — a constant output the gate - rejects. Under saturation the reform-domain propensity now falls back to - the observed take-up rate among candidates (reporter mass over capacity), - so candidates stay fully selected (current-law recipiency unchanged) - while the pool-wide flag keeps signal. + short of its SSA target, and a raw target/capacity prior degenerates to + Bernoulli(1.0) — a constant, signal-free flag. The prior therefore falls + back to the observed reporter rate (reporter mass over capacity) and the + pool keeps signal. Candidates are no longer force-selected to chase the + count (populace#469): the SSA-count miss is calibration's to close + (populace#470) and ships in the scorecard. """ targets = {"under_18": 1e6, "18_64": 1e6, "65_plus": 1e6} _, result, potential, diagnostics = _assigned(targets=targets) person = result.table("person") flag = person[_OUTPUT].to_numpy(dtype=bool) - assert flag[potential > 0].all() + anchored = ( + person["person_source_id"] + .str.split(":") + .str[1] + .isin(_ANCHORED_SOURCE_NUMBERS) + .to_numpy() + ) + assert flag[anchored].all() + assert not flag[potential > 0].all() assert not flag.all() + fallback = _REPORTER_FLOOR / _BAND_CAPACITY + by_source = person.groupby("person_source_id")[_OUTPUT].first() + for source_id, flagged in by_source.items(): + assert bool(flagged) == _expected_bernoulli_flag(source_id, fallback) for band in diagnostics["age_bands"]: assert band["saturated"] - assert band["selected_recipient_weight"] == band["candidate_capacity"] + assert band["assignment_prior"] == pytest.approx(fallback) + assert band["target_shortfall"] > 0 assert us_ssi_take_up_gate(diagnostics, targets=targets).passed @@ -274,8 +333,11 @@ def test_reporter_floor_above_target_never_drops_an_anchor() -> None: US_SSI_TAKE_UP_ANCHOR ].gt(0) assert person.loc[direct_reporter, _OUTPUT].all() + assert diagnostics["reporter_anchor_lost_count"] == 0 for band in diagnostics["age_bands"]: - assert band["reachable_goal"] == band["reporter_candidate_floor"] + assert band["anchor_excess"] == pytest.approx(_REPORTER_FLOOR - 5.0) + assert band["selected_recipient_weight"] >= band["reporter_candidate_floor"] + assert us_ssi_take_up_gate(diagnostics, targets=targets).passed def test_assignment_is_deterministic_source_keyed_and_seed_sensitive() -> None: @@ -365,6 +427,7 @@ def test_source_provenance_failures_are_rejected(mutation: str, message: str) -> ("unique_count", 1, "constant"), ("reporter_anchor_lost_count", 1, "reporter anchors"), ("source_identity_mismatch_count", 1, "source identity"), + ("bernoulli_law_violation_count", 2, "Bernoulli law"), ], ) def test_gate_rejects_tampered_top_level_diagnostics( @@ -378,7 +441,7 @@ def test_gate_rejects_tampered_top_level_diagnostics( assert any(failure_fragment in failure for failure in gate.failures) -def test_gate_rejects_hidden_saturation_and_large_reachable_miss() -> None: +def test_gate_rejects_hidden_saturation_and_corrupted_band_arithmetic() -> None: _, _, _, diagnostics = _assigned() hidden = copy.deepcopy(diagnostics) hidden["age_bands"][0]["saturated"] = True @@ -386,11 +449,23 @@ def test_gate_rejects_hidden_saturation_and_large_reachable_miss() -> None: assert not hidden_gate.passed assert any("saturation status" in failure for failure in hidden_gate.failures) - missed = copy.deepcopy(diagnostics) - missed["age_bands"][0]["selected_recipient_weight"] = 0.0 - missed_gate = us_ssi_take_up_gate(missed, targets=_TARGETS) - assert not missed_gate.passed - assert any("misses reachable goal" in failure for failure in missed_gate.failures) + escaped = copy.deepcopy(diagnostics) + escaped["age_bands"][0]["selected_recipient_weight"] = 0.0 + escaped_gate = us_ssi_take_up_gate(escaped, targets=_TARGETS) + assert not escaped_gate.passed + assert any("envelope" in failure for failure in escaped_gate.failures) + + miscomputed = copy.deepcopy(diagnostics) + miscomputed["age_bands"][0]["prior_recomputed_from_current_weights"] = 0.9 + miscomputed_gate = us_ssi_take_up_gate(miscomputed, targets=_TARGETS) + assert not miscomputed_gate.passed + assert any("recomputed prior" in failure for failure in miscomputed_gate.failures) + + invalid_probability = copy.deepcopy(diagnostics) + invalid_probability["age_bands"][0]["assignment_prior"] = 1.5 + invalid_gate = us_ssi_take_up_gate(invalid_probability, targets=_TARGETS) + assert not invalid_gate.passed + assert any("outside [0, 1]" in failure for failure in invalid_gate.failures) def test_gate_rejects_duplicate_age_band_diagnostics() -> None: @@ -403,7 +478,11 @@ def test_gate_rejects_duplicate_age_band_diagnostics() -> None: def test_existing_assignment_diagnostics_do_not_reassign_flags() -> None: - _, result, potential, _ = _assigned() + _, result, potential, stage_diagnostics = _assigned() + stage_priors = { + band["age_band"]: band["assignment_prior"] + for band in stage_diagnostics["age_bands"] + } original = result.table("person")[_OUTPUT].to_numpy(dtype=bool).copy() candidate_selected = original & (potential > 0) drifted_weights = np.where(candidate_selected, 100.0, 1.0) @@ -424,9 +503,24 @@ def test_existing_assignment_diagnostics_do_not_reassign_flags() -> None: uncapped_ssi=potential, seed=17, targets=_TARGETS, + assignment_priors=stage_priors, ) np.testing.assert_array_equal(reweighted.table("person")[_OUTPUT], original) - assert not us_ssi_take_up_gate(diagnostics, targets=_TARGETS).passed + # Weight drift pushes the measured recipient mass far off target, and the + # gate still passes: the count miss is calibration's residual + # (populace#469/#470), reported in the scorecard, never a module failure. + # The published assignment prior stays the one that generated the frozen + # flags — never recomputed from the drifted weights — while the + # current-weight recomputation is reported separately and the flags + # re-verify against the seeded law exactly. + assert diagnostics["bernoulli_law_violation_count"] == 0 + for band in diagnostics["age_bands"]: + assert band["selected_recipient_weight"] > band["target"] + assert band["assignment_prior"] == pytest.approx(stage_priors[band["age_band"]]) + assert band["prior_recomputed_from_current_weights"] != pytest.approx( + band["assignment_prior"] + ) + assert us_ssi_take_up_gate(diagnostics, targets=_TARGETS).passed recomputed, _ = with_us_ssi_take_up( reweighted, @@ -437,6 +531,35 @@ def test_existing_assignment_diagnostics_do_not_reassign_flags() -> None: assert not np.array_equal(recomputed.table("person")[_OUTPUT], original) +def test_gate_rejects_persisted_flags_that_break_the_bernoulli_law() -> None: + _, result, potential, stage_diagnostics = _assigned() + stage_priors = { + band["age_band"]: band["assignment_prior"] + for band in stage_diagnostics["age_bands"] + } + person = result.table("person").copy() + # A non-anchored, non-candidate source with both support clones: flipping + # its flag keeps source-identity consistency and every anchor intact, + # so only the seeded-law recheck can catch the corruption. + flipped = person["person_source_id"].eq("18_64:8") + assert flipped.sum() == 2 + person.loc[flipped, _OUTPUT] = ~person.loc[flipped, _OUTPUT].astype(bool) + corrupted = _replace_person(result, person) + diagnostics = us_ssi_take_up_diagnostics( + corrupted, + uncapped_ssi=potential, + seed=17, + targets=_TARGETS, + assignment_priors=stage_priors, + ) + assert diagnostics["bernoulli_law_violation_count"] == 1 + assert diagnostics["reporter_anchor_lost_count"] == 0 + assert diagnostics["source_identity_mismatch_count"] == 0 + gate = us_ssi_take_up_gate(diagnostics, targets=_TARGETS) + assert not gate.passed + assert any("Bernoulli law" in failure for failure in gate.failures) + + def test_writer_emits_strict_json_and_refuses_nan(tmp_path: Path) -> None: _, _, _, diagnostics = _assigned() path = write_us_ssi_take_up_diagnostics(diagnostics, tmp_path / "ssi.json") 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 ce90fdbc..e05440d9 100644 --- a/packages/populace-build/tests/test_us_take_up_contract.py +++ b/packages/populace-build/tests/test_us_take_up_contract.py @@ -51,39 +51,34 @@ def test_every_program_has_a_valid_treatment(self) -> None: "near_universal", } - def test_ssi_uses_reporter_anchored_ssa_age_count_calibration(self) -> None: + def test_ssi_uses_reporter_anchored_registry_band_targets(self) -> None: program = load_take_up_contract().program_map()["takes_up_ssi_if_eligible"] calibration = program.raw["calibration"] assert program in count_calibrated_take_up_programs() - assert calibration == { - "anchor": "SSI_VAL", - "targets": ["ssa_ssi_federal_payment_recipients_by_age"], - "target_table": "ssa_ssi_federal_payment_recipients_by_age", - "target_source": ( - "https://www.ssa.gov/policy/docs/statcomps/" - "ssi_monthly/2024-12/table01.html" - ), - "target_period": "2024-12", - "target_measure": "Total with—Federal payment", - "target_values": { - "under_18": 1_001_922, - "18_64": 3_905_779, - "65_plus": 2_382_142, - }, - "aggregate_target": 7_289_843, - "age_bands": { - "under_18": "age < 18", - "18_64": "18 <= age < 65", - "65_plus": "age >= 65", - }, - "semantics": ( - "SSA SSI Monthly Statistics December 2024 Table 1 recipients " - "in the Total with—Federal payment row, calibrated within " - "uncapped_ssi > 0 by source-person identity; unreachable age " - "bands saturate without assigning outside modeled eligibility" - ), + assert calibration["anchor"] == "SSI_VAL" + assert calibration["targets"] == ["ssa_ssi_federal_payment_recipients_by_age"] + assert calibration["target_source"] == ( + "https://www.ssa.gov/policy/docs/statcomps/ssi_monthly/2024-12/table01.html" + ) + assert calibration["target_period"] == "2024-12" + assert calibration["target_measure"] == "Total with—Federal payment" + assert calibration["target_role"] == "ssa_ssi_age_band_recipients" + assert calibration["age_bands"] == { + "under_18": "age < 18", + "18_64": "18 <= age < 65", + "65_plus": "age >= 65", } + # The SSA recipient counts live in the ledger and bind through the + # calibration registry (populace#469/#470) — the contract must not + # carry a hardcoded copy, and the semantics must describe seeded + # Bernoulli priors, not flag count-matching. + assert "target_values" not in calibration + assert "aggregate_target" not in calibration + semantics = calibration["semantics"] + assert "populace#469" in semantics + assert "never count-matches" in semantics + assert "saturate" not in semantics assert program.raw["scope_owner"] == ( "ssi_take_up source stage (eCPS exported-input coverage)" ) diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index c23c0439..0bc367f5 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -41,9 +41,8 @@ import sys import time import tomllib -from collections.abc import Collection, Iterable, Mapping, Sequence +from collections.abc import Iterable, Mapping, Sequence from contextlib import contextmanager -from dataclasses import replace from datetime import UTC, datetime from pathlib import Path from typing import Any @@ -210,7 +209,9 @@ write_demographics, ) from populace.build.us_runtime.engine_lifecycle import release_engine_simulation -from populace.build.us_runtime.fiscal_targets import SSA_SSI_RECIPIENTS_TARGET_ROLE +from populace.build.us_runtime.fiscal_targets import ( + SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE, +) from populace.build.us_runtime.input_mass import us_input_mass_totals from populace.build.us_runtime.l0_refit_export import ( attach_l0_refit_entity_weights, @@ -300,30 +301,14 @@ # 6: the official-ASEC sidecar restores 2022 LKWEEKS before target # materialization. The source is external to the on-disk base hash, so old # checkpoints must not survive the new measured input. -TARGET_FRAME_CHECKPOINT_MATERIALIZER_VERSION = 6 +# 7: SSI take-up became a one-shot seeded Bernoulli at registry band priors +# (populace#469) — checkpoints materialized from count-matched flags must +# not survive, or the solve would run on old SSI rows while the frame +# carries the new assignment (PR #477 review finding 2). +TARGET_FRAME_CHECKPOINT_MATERIALIZER_VERSION = 7 DEFAULT_MAXIMUM_MICROSIM_BATCH_SIZE = 5_000 DEFAULT_L0_REFIT_LAMBDA_SHARE = 0.8 DEFAULT_US_FISCAL_CALIBRATION_EPOCHS = 1_500 -SSI_TAKE_UP_RECONCILIATION_MAX_PASSES = 3 - -#: Runaway sanity caps on the SSI swap delta, as a share of the fresh national -#: recipient total (#431: the delta is the solve's equilibrium residual on the -#: SSI family, recorded in the manifest; the cap exists only to catch a solve -#: that abandoned the family). The dense arm's equilibrium is structurally -#: larger than the sparse arm's on the same #424-undercounted candidate -#: universe: measured sparse deltas run 7.1-8.0% (attempts 9-16) while the -#: dense trajectory converges monotonically 11.97% -> 11.64% -> 11.38% with -#: decelerating decrements (asymptote ~10.6%), so 0.10 is unreachable there -#: while remaining the right bar for sparse (populace#447 adjudication, -#: 2026-07-18; trajectory recorded in the reconciliation pass_history). Both -#: caps converge back to one number when #424 restores the candidate universe. -#: ARM IDENTITY: the dense arm is the run WITHOUT a frozen selection manifest. -#: ``--dense-default-dataset`` is an EXPORT-MODE flag that the sparse -#: rmloss100 script also sets, and must never be used as the arm -#: discriminator (a 32605c9-era sparse rerun gated at 0.12 through exactly -#: that confusion; caught by its recorded cap ratio before anything shipped). -SSI_TAKE_UP_SWAP_SANITY_CAP_RATIO = 0.10 -SSI_TAKE_UP_SWAP_SANITY_CAP_RATIO_DENSE = 0.12 def _collect_batch_garbage() -> None: @@ -5169,711 +5154,108 @@ def _fiscal_target_loss_weights( return weights / weights.mean() -class _SSITakeUpReconciliationResult: - """Fixed assignments and calibration state after SSI dependency replay.""" - - __slots__ = ( - "export_frame", - "calibration_result", - "registry", - "compilation", - "ssi_diagnostics", - "medicaid_diagnostics", - "health_input_gate", - "other_health_insurance_gate", - "passes", - ) +def _ssi_take_up_band_targets_from_registry(target_specs: tuple) -> dict[str, float]: + """SSA age-band recipient counts as compiled into the calibration registry. - def __init__( - self, - *, - export_frame: Frame, - calibration_result: Any, - registry: TargetRegistry, - compilation: Mapping[str, object], - ssi_diagnostics: Mapping[str, object], - medicaid_diagnostics: Mapping[str, object], - health_input_gate: GateResult, - other_health_insurance_gate: GateResult, - passes: int, - ) -> None: - self.export_frame = export_frame - self.calibration_result = calibration_result - self.registry = registry - self.compilation = compilation - self.ssi_diagnostics = ssi_diagnostics - self.medicaid_diagnostics = medicaid_diagnostics - self.health_input_gate = health_input_gate - self.other_health_insurance_gate = other_health_insurance_gate - self.passes = passes - - -def _frame_with_reconciliation_weight_basis( - frame: Frame, - household_weights: np.ndarray, -) -> Frame: - """Copy input tables onto the fixed pre-refit household weight basis.""" + The SSI take-up stage derives its Bernoulli priors from the same + ledger-fed band targets the weight solve enforces (role + :data:`SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE`, populace#469/#470): one + official measure, bound once, never hardcoded in the module. Bands are + matched on the facts' first-class age constraints. Fails closed when the + feed does not carry all three SSA age bands. + """ - values = np.asarray(household_weights, dtype=np.float64) - if values.shape != (frame.n("household"),): - raise ValueError( - "SSI take-up reconciliation weight basis must align to households: " - f"{values.shape} != {(frame.n('household'),)}." + expected_bounds: dict[tuple[float, float], str] = {} + for band in US_SSI_TAKE_UP_AGE_TARGETS: + lower = float("-inf") if band.minimum_age is None else float(band.minimum_age) + upper = ( + float("inf") if band.maximum_age is None else float(band.maximum_age) + 1.0 ) - if not np.isfinite(values).all() or (values <= 0).any(): - raise ValueError( - "SSI take-up reconciliation requires finite strictly positive " - "prior weights." + expected_bounds[(lower, upper)] = band.key + band_targets: dict[str, float] = {} + for spec in target_specs: + if spec.metadata.get("target_role") != SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE: + continue + raw_bounds = ( + spec.metadata.get("age_lower_bound"), + spec.metadata.get("age_upper_bound"), ) - weights = {entity: frame.weights_for(entity) for entity in frame.weighted_entities} - weights["household"] = Weights(values, WeightKind.CALIBRATED) - return Frame( - {entity: frame.table(entity).copy() for entity in frame.entities}, - frame.schema, - weights, - frame.strata, - mass_log=frame.mass_log, - ) - - -def _assert_reconciliation_support_unchanged( - reference: Frame, - candidate: Frame, -) -> None: - """Fail if a dependency replay changes selected entity IDs or order.""" - - if reference.schema != candidate.schema or reference.entities != candidate.entities: - raise RuntimeError("SSI take-up reconciliation changed the support schema.") - for entity in reference.entities: - id_column = ( - reference.schema.person_id_column - if entity == reference.schema.person_entity - else reference.schema.id_column(entity) - ) - expected = reference.table(entity)[id_column].to_numpy() - actual = candidate.table(entity)[id_column].to_numpy() - if not np.array_equal(expected, actual): + key = None + try: + lower_value = float(raw_bounds[0]) + upper_value = float(raw_bounds[1]) + except (TypeError, ValueError): + lower_value = upper_value = float("nan") + else: + # Ages are nonnegative, so an explicit "age >= 0" floor is the + # same stratum as an unbounded lower edge — the real feed's + # under-18 fact carries one (PR #477 review finding 1). + if lower_value <= 0: + lower_value = float("-inf") + key = expected_bounds.get((lower_value, upper_value)) + if key is None: raise RuntimeError( - "SSI take-up reconciliation changed selected support IDs or " - f"order for {entity!r}." + "SSI take-up found a registry " + f"{SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE!r} target with " + f"unrecognized age bounds {raw_bounds!r}; expected one of " + f"{sorted(expected_bounds)}." ) - - -def _registry_national_ssi_recipients_total(target_specs: tuple) -> float: - """Sum the registry's national ``ssi_recipients`` calibration target(s). - - The reconciliation refit drives household weights to the SSA - ``ssi_recipients`` administrative family (an indicator sum of engine ``ssi`` - receipt; role :data:`SSA_SSI_RECIPIENTS_TARGET_ROLE`, national and state - grain). The take-up assignment must aim at the *same* recipient measure the - solve enforces, so its age-band goals are the SSA recipient age shares - applied to this national total rather than the raw SSA - federal-payment-by-age counts — a distinct official measure roughly 115k - larger. Enforced simultaneously to one-unit precision the two totals have no - common fixed point, which is what stalled the pre-alignment reconciliation. - - National specs carry the role with no ``state_fips`` (state rows carry one). - Only the all-category ``total`` compiles upstream - (``fiscal_targets._ssa_ssi_reference_from_fact``); summing is defensive - against a future per-category national split. Fails closed when the registry - carries no national ``ssi_recipients`` target. - """ - - national_values = [ - float(spec.value) - for spec in target_specs - if spec.metadata.get("target_role") == SSA_SSI_RECIPIENTS_TARGET_ROLE - and not spec.metadata.get("state_fips") + if key in band_targets: + raise RuntimeError( + "SSI take-up found duplicate registry targets for SSA age " + f"band {key!r}." + ) + value = float(spec.value) + if not np.isfinite(value) or value <= 0: + raise RuntimeError( + f"SSI take-up registry target for age band {key!r} must be " + f"finite and positive; got {value!r}." + ) + band_targets[key] = value + missing = [ + band.key for band in US_SSI_TAKE_UP_AGE_TARGETS if band.key not in band_targets ] - if not national_values: - raise RuntimeError( - "SSI take-up reconciliation could not find a national " - f"{SSA_SSI_RECIPIENTS_TARGET_ROLE!r} target in the fiscal registry; " - "the age-band take-up goals cannot be aligned to the calibration " - "total the refit enforces." - ) - total = float(sum(national_values)) - if not np.isfinite(total) or total <= 0: + if missing: raise RuntimeError( - "SSI take-up reconciliation found a nonpositive national " - f"{SSA_SSI_RECIPIENTS_TARGET_ROLE!r} target ({total!r})." + "SSI take-up requires the ledger-fed SSA age-band recipient " + f"targets (role {SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE!r}) in " + f"the fiscal registry; missing band(s) {missing}. The consumer " + "facts feed must carry the ssa ssi_federal_payment_recipients " + "by_age rows (populace#470)." ) - return total - - -def _aligned_ssi_take_up_band_targets( - target_specs: tuple, -) -> tuple[dict[str, float], dict[str, object]]: - """Age-band take-up goals rescaled onto the registry's national SSI total. - - Returns ``(band_targets, alignment_record)``. ``band_targets`` applies the - SSA federal-payment recipient age *shares* to the registry's national - ``ssi_recipients`` total, so the greedy count-match and the reconciliation - refit pursue one shared recipient measure and a fixed point exists. The band - counts are federal-payment recipients by age; the registry total is the - recipient measure the solve enforces; the shares reconcile the two official - measures. ``alignment_record`` carries both raw totals and the rescale for - the reconciliation compilation. - """ - - registry_national = _registry_national_ssi_recipients_total(target_specs) - ssa_band_total = float( - sum(target.person_count for target in US_SSI_TAKE_UP_AGE_TARGETS) - ) - band_shares = { - target.key: target.person_count / ssa_band_total - for target in US_SSI_TAKE_UP_AGE_TARGETS - } - band_targets = { - key: share * registry_national for key, share in band_shares.items() - } - alignment_record = { - "ssa_federal_payment_recipient_band_total": ssa_band_total, - "registry_national_recipients_total": registry_national, - "band_shares": band_shares, - "rescaled_band_targets": { - key: float(value) for key, value in band_targets.items() - }, - } - return band_targets, alignment_record - - -def _replay_ssi_dependent_inputs( - support: Frame, - target_specs: tuple, - *, - seed: int, - medicaid_enrollment_substitutions: Sequence[Mapping[str, object]], - maximum_microsim_batch_size: int | None, - selected_support: Frame, -) -> tuple[Frame, dict[str, object]]: - """Replay the ACA/Medicaid/other-health inputs that depend on SSI receipt. - - SSI recipient status can alter Marketplace eligibility, Medicaid - eligibility/take-up, and the ASEC private-premium residual. Both the - pre-calibration pass head (before any target vector is materialized) and the - post-refit fresh-pair exit replay this identical dependency tail, then assert - the selected support IDs are unchanged. Returns the replayed support and the - Medicaid take-up diagnostics; callers gate the results. - """ - - support = _with_aca_marketplace_source_outputs( - support, - target_specs, - seed=seed, - maximum_microsim_batch_size=maximum_microsim_batch_size, - ) - support, medicaid_diagnostics = _with_medicaid_take_up_outputs( - support, - target_specs, - seed=seed, - substitutions=medicaid_enrollment_substitutions, - maximum_microsim_batch_size=maximum_microsim_batch_size, - ) - support = with_us_other_health_insurance_inputs( - support, - seed=seed, - time_period=PERIOD, - maximum_microsim_batch_size=maximum_microsim_batch_size, - ) - _assert_reconciliation_support_unchanged(selected_support, support) - return support, dict(medicaid_diagnostics) - - -def _ssi_take_up_swap_delta( - stale_diagnostics: Mapping[str, object], - fresh_diagnostics: Mapping[str, object], - *, - sanity_cap_ratio: float = SSI_TAKE_UP_SWAP_SANITY_CAP_RATIO, -) -> dict[str, object]: - """National SSI-recipient mass moved by the post-refit re-assignment. - - The retired freeze invariant assumed the flags materialized into the SSI - and Medicaid target vectors must not change after optimization. The - fresh-pair exit replaces that assumption with a measurement — and what the - measurement turns out to capture is the SOLVE'S residual on the - SSI-recipient target family: the age-blind, loss-balanced calibration - leaves an equilibrium miss on that one family (attempt 8: ~419k on 7.4M, - ~5.7%), and the fresh assignment resolves recipiency back to the official - counts. That residual is already published as the family's target error in - the calibration diagnostics; demanding it fit within assignment - granularity (one source-identity weight per age band, ~112k) was a - calibration-quality bar no other target faces, dressed as a consistency - check. The delta is therefore RECORDED for the manifest — alongside the - granularity reference — and gated only against runaway at one tenth of the - fresh national total, which still catches a solve that abandoned the - family entirely. Per-band deltas are recorded, never gated: per-band - faithfulness of the shipped pair holds by construction. - """ - - stale_bands = { - str(band["age_band"]): band for band in stale_diagnostics["age_bands"] - } - per_band: dict[str, dict[str, float]] = {} - stale_total = 0.0 - fresh_total = 0.0 - national_bound = 0.0 - for fresh_band in fresh_diagnostics["age_bands"]: - key = str(fresh_band["age_band"]) - fresh_selected = float(fresh_band["selected_recipient_weight"]) - stale_selected = float(stale_bands[key]["selected_recipient_weight"]) - band_allowance = float(fresh_band["max_source_candidate_weight"]) - per_band[key] = { - "stale_selected_recipient_weight": stale_selected, - "fresh_selected_recipient_weight": fresh_selected, - "swap_delta": fresh_selected - stale_selected, - "max_source_candidate_weight": band_allowance, - } - stale_total += stale_selected - fresh_total += fresh_selected - national_bound += band_allowance - national_delta = abs(fresh_total - stale_total) - sanity_cap = sanity_cap_ratio * fresh_total - return { - "national_swap_sanity_cap_ratio": sanity_cap_ratio, - "stale_selected_recipient_weight_total": stale_total, - "fresh_selected_recipient_weight_total": fresh_total, - "national_swap_delta": national_delta, - "assignment_granularity_reference": national_bound, - "national_swap_sanity_cap": sanity_cap, - "within_bound": bool(national_delta <= sanity_cap), - "age_bands": per_band, - } - - -def _medicaid_enrollment_swap_deltas( - stale_diagnostics: Mapping[str, object], - fresh_diagnostics: Mapping[str, object], -) -> dict[str, object]: - """Per-state Medicaid enrolled-mass moved by the fresh replay under W'. - - Recorded, not gated: the fresh Medicaid diagnostics are gated directly. This - captures how much enrolled mass each state's re-assignment moved so the - reconciliation record stays auditable alongside the SSI swap delta. - """ - - def _by_state(diagnostics: Mapping[str, object]) -> dict[str, float]: - return { - str(row["state_fips"]): float(row["enrolled_weight"]) - for row in diagnostics.get("states", []) - } - - stale_states = _by_state(stale_diagnostics) - fresh_states = _by_state(fresh_diagnostics) - per_state: dict[str, dict[str, float]] = {} - for state in sorted(set(stale_states) | set(fresh_states)): - stale_enrolled = stale_states.get(state, 0.0) - fresh_enrolled = fresh_states.get(state, 0.0) - per_state[state] = { - "stale_enrolled_weight": stale_enrolled, - "fresh_enrolled_weight": fresh_enrolled, - "swap_delta": fresh_enrolled - stale_enrolled, - } - return {"states": per_state} - + return band_targets -class _SSITakeUpFreshPairExit: - """Fresh post-refit SSI/health assignment plus its gates and swap deltas.""" - - __slots__ = ( - "support", - "ssi_diagnostics", - "medicaid_diagnostics", - "ssi_gate", - "health_gate", - "medicaid_gate", - "other_health_gate", - "ssi_swap_delta", - "medicaid_swap_delta", - ) - def __init__( - self, - *, - support: Frame, - ssi_diagnostics: dict[str, object], - medicaid_diagnostics: dict[str, object], - ssi_gate: GateResult, - health_gate: GateResult, - medicaid_gate: GateResult, - other_health_gate: GateResult, - ssi_swap_delta: dict[str, object], - medicaid_swap_delta: dict[str, object], - ) -> None: - self.support = support - self.ssi_diagnostics = ssi_diagnostics - self.medicaid_diagnostics = medicaid_diagnostics - self.ssi_gate = ssi_gate - self.health_gate = health_gate - self.medicaid_gate = medicaid_gate - self.other_health_gate = other_health_gate - self.ssi_swap_delta = ssi_swap_delta - self.medicaid_swap_delta = medicaid_swap_delta - - @property - def gates_passed(self) -> bool: - """All fresh-pair gates pass and the national swap delta is in bound.""" +def _ssi_assignment_priors_from_diagnostics( + diagnostics: Mapping[str, object], +) -> dict[str, float]: + """The per-band Bernoulli priors the gated assignment stage documented. - return ( - self.ssi_gate.passed - and self.health_gate.passed - and self.medicaid_gate.passed - and self.other_health_gate.passed - and bool(self.ssi_swap_delta["within_bound"]) - ) - - def failures(self) -> tuple[str, ...]: - """Prefixed gate failures plus a swap-delta breach message when over.""" - - failures = ( - [f"SSI: {failure}" for failure in self.ssi_gate.failures] - + [f"ACA: {failure}" for failure in self.health_gate.failures] - + [f"Medicaid: {failure}" for failure in self.medicaid_gate.failures] - + [ - f"Other health: {failure}" - for failure in self.other_health_gate.failures - ] - ) - if not self.ssi_swap_delta["within_bound"]: - failures.append( - "SSI take-up swap delta " - f"{float(self.ssi_swap_delta['national_swap_delta']):.3f} exceeds " - "the runaway sanity cap " - f"{float(self.ssi_swap_delta['national_swap_sanity_cap']):.3f} " - "(the solve moved recipient mass more than a tenth of the " - "national total away from the official counts)." - ) - return tuple(failures) - - -def _ssi_take_up_fresh_pair_exit( - export_frame: Frame, - target_specs: tuple, - *, - band_targets: Mapping[str, float], - seed: int, - reporter_source_ids: Collection[str], - final_uncapped_ssi: np.ndarray, - medicaid_enrollment_substitutions: Sequence[Mapping[str, object]], - maximum_microsim_batch_size: int | None, - selected_support: Frame, - sanity_cap_ratio: float = SSI_TAKE_UP_SWAP_SANITY_CAP_RATIO, -) -> _SSITakeUpFreshPairExit: - """Re-assign SSI take-up under the returned weights and gate the fresh pair. - - ``export_frame`` carries the pass head's frozen flags on the refit weights - (the stale pair the pre-alignment loop gated). This measures that stale pair - for the swap delta, then re-runs the assignment under the returned weights so - the published flags are count-faithful by construction, replays the SSI - dependency tail on the fresh flags, and gates the fresh pair. The SSI swap - delta bounds how far the re-assignment moved aggregate recipient mass; the - Medicaid enrolled-mass deltas are recorded per state. + The final release-weight measurement republishes these verbatim and + re-verifies every frozen flag against the seeded law they define + (populace#469) — recomputing priors from release weights would + misdocument the one-shot assignment. """ - stale_ssi_diagnostics = us_ssi_take_up_diagnostics( - export_frame, - uncapped_ssi=final_uncapped_ssi, - seed=seed, - targets=band_targets, - reporter_source_ids=reporter_source_ids, - ) - stale_medicaid_diagnostics = _medicaid_diagnostics_for_existing_output( - export_frame, - target_specs, - seed=seed, - substitutions=medicaid_enrollment_substitutions, - maximum_microsim_batch_size=maximum_microsim_batch_size, - ) - exit_support, exit_ssi_diagnostics = with_us_ssi_take_up( - export_frame, - uncapped_ssi=final_uncapped_ssi, - seed=seed, - targets=band_targets, - reporter_source_ids=reporter_source_ids, - ) - exit_support, exit_medicaid_diagnostics = _replay_ssi_dependent_inputs( - exit_support, - target_specs, - seed=seed, - medicaid_enrollment_substitutions=medicaid_enrollment_substitutions, - maximum_microsim_batch_size=maximum_microsim_batch_size, - selected_support=selected_support, - ) - return _SSITakeUpFreshPairExit( - support=exit_support, - ssi_diagnostics=exit_ssi_diagnostics, - medicaid_diagnostics=exit_medicaid_diagnostics, - ssi_gate=us_ssi_take_up_gate(exit_ssi_diagnostics, targets=band_targets), - health_gate=_health_input_signal_gate(exit_support), - medicaid_gate=us_medicaid_take_up_gate(dict(exit_medicaid_diagnostics)), - other_health_gate=us_other_health_insurance_signal_gate(exit_support), - ssi_swap_delta=_ssi_take_up_swap_delta( - stale_ssi_diagnostics, - exit_ssi_diagnostics, - sanity_cap_ratio=sanity_cap_ratio, - ), - medicaid_swap_delta=_medicaid_enrollment_swap_deltas( - stale_medicaid_diagnostics, exit_medicaid_diagnostics - ), - ) - - -def _reconcile_ssi_take_up_and_refit( - base_frame: Frame, - initial_result, - target_specs: tuple, - *, - dense_default_dataset: bool, - sparse_selection_arm: bool = True, - seed: int, - epochs: int, - learning_rate: float, - max_weight_ratio: float | None, - l2_lambda: float, - target_loss_cap: float, - reporter_source_ids: Collection[str] | None = None, - medicaid_enrollment_substitutions: Sequence[Mapping[str, object]] = (), - maximum_microsim_batch_size: int | None = DEFAULT_MAXIMUM_MICROSIM_BATCH_SIZE, - gate_congressional_district_targets: bool = True, - progress_callback=None, - max_passes: int = SSI_TAKE_UP_RECONCILIATION_MAX_PASSES, -) -> _SSITakeUpReconciliationResult: - """Reconcile SSI on final weights before replaying dependent target inputs. - - The initial dense/L0 fit supplies an actual release-weight surface. Each - bounded pass fixes SSI on those weights, replays ACA, Medicaid, and - other-health inputs that can depend on SSI, rematerializes every fiscal - target from those exact fixed inputs, and performs an ordinary refit on the - already-selected support. - - Two facts make the fixed point reachable. First, the age-band take-up goals - are rescaled onto the registry's national ``ssi_recipients`` total (the - measure the refit enforces), so the greedy count-match and the calibration - pursue one recipient total instead of two official measures ~115k apart. - Second, the refit moves the weights the assignment was fixed on, so rather - than gate the frozen flags on the drifted weights (a stale pair the loop can - never make self-consistent) each pass re-assigns take-up under the returned - weights — the fresh pair, count-faithful by construction — and a bounded - swap delta caps how far the re-assignment moved aggregate recipient mass from - the pre-refit flags. That measured bound is the honest replacement for the - retired "never rewrite the flags after optimization" freeze invariant. - """ - - if max_passes <= 0: - raise ValueError("SSI take-up reconciliation requires at least one pass.") - reporter_source_ids = ( - us_ssi_take_up_reporter_source_ids(base_frame) - if reporter_source_ids is None - else frozenset(str(value) for value in reporter_source_ids) - ) - if dense_default_dataset: - current_support = _with_calibrated_weights( - base_frame, - np.asarray(initial_result.weights, dtype=np.float64), - ) - else: - current_support = _with_l0_refit_weights(base_frame, initial_result) - prior_weights = np.asarray(initial_result.initial_weights, dtype=np.float64) - if prior_weights.shape != (current_support.n("household"),): + bands = diagnostics.get("age_bands") + if not isinstance(bands, list) or not bands: raise RuntimeError( - "SSI take-up reconciliation prior weights do not align to the " - "selected release support." - ) - selected_support = current_support - - band_targets, ssi_target_alignment = _aligned_ssi_take_up_band_targets(target_specs) - - last_failures: tuple[str, ...] = () - # populace#447: the per-pass swap-delta trajectory must survive a terminal - # raise (converging-but-over-cap vs oscillating is the whole adjudication) - # and ride the success record. - pass_history: list[dict[str, object]] = [] - for pass_number in range(1, max_passes + 1): - uncapped_ssi = _ssi_person_uncapped_amount( - current_support, - maximum_microsim_batch_size=maximum_microsim_batch_size, - ) - assigned_support, stage_diagnostics = with_us_ssi_take_up( - current_support, - uncapped_ssi=uncapped_ssi, - seed=seed, - targets=band_targets, - reporter_source_ids=reporter_source_ids, - ) - stage_gate = us_ssi_take_up_gate(stage_diagnostics, targets=band_targets) - if not stage_gate.passed: - raise RuntimeError( - "SSI take-up reconciliation assignment failed: " - + "; ".join(stage_gate.failures) - ) - - # SSI recipient status can alter Marketplace eligibility, Medicaid - # eligibility/take-up, and the ASEC private-premium residual. Replay - # that full dependency tail before any target vector is materialized. - assigned_support, medicaid_diagnostics = _replay_ssi_dependent_inputs( - assigned_support, - target_specs, - seed=seed, - medicaid_enrollment_substitutions=medicaid_enrollment_substitutions, - maximum_microsim_batch_size=maximum_microsim_batch_size, - selected_support=selected_support, - ) - health_gate = _health_input_signal_gate(assigned_support) - if not health_gate.passed: - raise RuntimeError( - "SSI take-up reconciliation ACA input gate failed: " - + "; ".join(health_gate.failures) - ) - medicaid_gate = us_medicaid_take_up_gate(dict(medicaid_diagnostics)) - if not medicaid_gate.passed: - raise RuntimeError( - "SSI take-up reconciliation Medicaid gate failed: " - + "; ".join(medicaid_gate.failures) - ) - other_health_gate = us_other_health_insurance_signal_gate(assigned_support) - if not other_health_gate.passed: + "SSI take-up stage diagnostics carry no age-band rows to read " + "assignment priors from." + ) + priors: dict[str, float] = {} + for row in bands: + if not isinstance(row, Mapping): + raise RuntimeError("SSI take-up stage diagnostics band row is invalid.") + key = str(row.get("age_band")) + prior = float(row.get("assignment_prior", np.nan)) + if not np.isfinite(prior) or not 0.0 <= prior <= 1.0: raise RuntimeError( - "SSI take-up reconciliation other-health gate failed: " - + "; ".join(other_health_gate.failures) + f"SSI take-up stage diagnostics band {key!r} carries an " + f"invalid assignment prior {prior!r}." ) - - calibration_input = _frame_with_reconciliation_weight_basis( - assigned_support, - prior_weights, - ) - target_frame, registry, compilation = _materialize_target_frame( - calibration_input, - target_specs, - maximum_microsim_batch_size=maximum_microsim_batch_size, - gate_congressional_district_targets=gate_congressional_district_targets, - ) - current_weights = np.asarray( - assigned_support.weights_for("household").values, - dtype=np.float64, - ) - reconciled_result = calibrate( - target_frame, - registry.to_target_set(), - epochs=epochs, - learning_rate=learning_rate, - max_weight_ratio=max_weight_ratio, - seed=seed, - mass="conserve", - l2_lambda=l2_lambda, - target_loss_weights=_fiscal_target_loss_weights(registry), - target_loss_cap=target_loss_cap, - warm_start_weights=current_weights, - progress_callback=progress_callback, - ) - export_frame = _with_calibrated_weights( - calibration_input, - np.asarray(reconciled_result.weights, dtype=np.float64), - ) - _assert_reconciliation_support_unchanged(selected_support, export_frame) - final_uncapped_ssi = _ssi_person_uncapped_amount( - export_frame, - maximum_microsim_batch_size=maximum_microsim_batch_size, - ) - - # The refit moves the household weights the assignment was fixed on, so - # the frozen flags on the returned weights (the stale pair) are not - # count-faithful and the loop can never make them so. Re-assign take-up - # under the returned weights (the fresh pair, count-faithful by - # construction), gate it, and bound how far the re-assignment moved - # aggregate recipient mass with the swap delta. - exit_result = _ssi_take_up_fresh_pair_exit( - export_frame, - target_specs, - band_targets=band_targets, - seed=seed, - reporter_source_ids=reporter_source_ids, - final_uncapped_ssi=final_uncapped_ssi, - medicaid_enrollment_substitutions=medicaid_enrollment_substitutions, - maximum_microsim_batch_size=maximum_microsim_batch_size, - selected_support=selected_support, - sanity_cap_ratio=( - SSI_TAKE_UP_SWAP_SANITY_CAP_RATIO - if sparse_selection_arm - else SSI_TAKE_UP_SWAP_SANITY_CAP_RATIO_DENSE - ), - ) - pass_history.append( - { - "pass": pass_number, - "national_swap_delta": exit_result.ssi_swap_delta.get( - "national_swap_delta" - ), - "national_swap_sanity_cap": exit_result.ssi_swap_delta.get( - "national_swap_sanity_cap" - ), - "within_bound": exit_result.ssi_swap_delta.get("within_bound"), - } - ) - if exit_result.gates_passed: - reconciliation_compilation = { - **dict(compilation), - "target_frame_checkpoint": { - "enabled": False, - "status": "recomputed_after_ssi_take_up_reconciliation", - }, - "ssi_take_up_reconciliation": { - "passes": pass_number, - "max_passes": max_passes, - "reporter_source_identity_count": len(reporter_source_ids), - "exit_policy": "fresh_pair_under_returned_weights", - "dependency_replay": [ - "aca_marketplace", - "medicaid_take_up", - "other_health_insurance", - "fiscal_target_materialization", - "ordinary_refit", - ], - "target_alignment": ssi_target_alignment, - "ssi_swap_delta": exit_result.ssi_swap_delta, - "medicaid_enrollment_swap_delta": exit_result.medicaid_swap_delta, - "pass_history": pass_history, - }, - } - calibration_result = ( - reconciled_result - if dense_default_dataset - else replace(initial_result, refit=reconciled_result) - ) - return _SSITakeUpReconciliationResult( - export_frame=exit_result.support, - calibration_result=calibration_result, - registry=registry, - compilation=reconciliation_compilation, - ssi_diagnostics=exit_result.ssi_diagnostics, - medicaid_diagnostics=exit_result.medicaid_diagnostics, - health_input_gate=exit_result.health_gate, - other_health_insurance_gate=exit_result.other_health_gate, - passes=pass_number, - ) - last_failures = exit_result.failures() - current_support = export_frame - # populace#456: each pass rematerializes every fiscal target through - # the batched engine paths above; sweep the pass's cyclic residue so - # a multi-pass reconcile stays at single-pass footprint. - del target_frame, calibration_input, export_frame, exit_result - _collect_family_garbage() - - trajectory = "; ".join( - "pass {pass_number}: delta={delta:,.3f} cap={cap:,.3f} " - "within_bound={within}".format( - pass_number=entry["pass"], - delta=float(entry["national_swap_delta"] or 0.0), - cap=float(entry["national_swap_sanity_cap"] or 0.0), - within=entry["within_bound"], - ) - for entry in pass_history - ) - raise RuntimeError( - "SSI take-up reconciliation did not remain count-faithful on returned " - f"weights after {max_passes} pass(es): " - + "; ".join(last_failures) - + (f" Pass trajectory: {trajectory}." if pass_history else "") - ) + priors[key] = prior + return priors def _fiscal_target_concept_budget_weights(registry: TargetRegistry) -> np.ndarray: @@ -8391,10 +7773,16 @@ def main() -> None: telemetry.stage( "ssi_take_up", message=( - "Assigning SSI take-up from ASEC reporter anchors and SSA " - "December 2024 federal-payment recipient counts by age." + "Assigning SSI take-up once (populace#469): ASEC reporter " + "anchors plus a seeded Bernoulli draw at the registry's SSA " + "age-band priors." ), ) + # One-shot assignment before any target materialization (populace#469). + # The priors derive from the same ledger-fed SSA band counts the weight + # solve enforces as ordinary targets (populace#470); the flags are frozen + # from here and the post-calibration count miss ships in the scorecard. + ssi_band_targets = _ssi_take_up_band_targets_from_registry(target_specs) ssi_uncapped_amount = _ssi_person_uncapped_amount( base_frame, maximum_microsim_batch_size=args.maximum_microsim_batch_size, @@ -8403,15 +7791,18 @@ def main() -> None: base_frame, uncapped_ssi=ssi_uncapped_amount, seed=args.seed, + targets=ssi_band_targets, reporter_source_ids=ssi_reporter_source_ids, ) - ssi_take_up_gate = us_ssi_take_up_gate(ssi_take_up_stage_diagnostics) + ssi_take_up_gate = us_ssi_take_up_gate( + ssi_take_up_stage_diagnostics, targets=ssi_band_targets + ) if not ssi_take_up_gate.passed: if telemetry is not None: telemetry.stage( "ssi_take_up_gate", status="failed", - message="SSI take-up count-calibration gate failed.", + message="SSI take-up assignment gate failed.", failures=list(ssi_take_up_gate.failures), force_upload=True, ) @@ -8422,6 +7813,9 @@ def main() -> None: for failure in ssi_take_up_gate.failures ) ) + ssi_assignment_priors = _ssi_assignment_priors_from_diagnostics( + ssi_take_up_stage_diagnostics + ) if telemetry is not None: telemetry.stage( "scf_auto_loan_inputs", @@ -8996,47 +8390,85 @@ def main() -> None: } if telemetry is not None: telemetry.stage( - "ssi_take_up_reconciliation", + "take_up_final_diagnostics", message=( - "Reconciling SSI on release weights, replaying dependent health " - "inputs, and rematerializing fiscal targets before final refit." + "Applying release weights and measuring the frozen take-up " + "assignments (report-only; populace#469)." ), - max_passes=SSI_TAKE_UP_RECONCILIATION_MAX_PASSES, ) - pre_reconciliation_final_loss = float(result.final_loss) - reconciliation_l2_lambda = float( - args.l2_lambda - if args.dense_default_dataset or args.refit_l2_lambda is None - else args.refit_l2_lambda - ) - reconciliation = _reconcile_ssi_take_up_and_refit( - base_frame, - result, - target_specs, - dense_default_dataset=bool(args.dense_default_dataset), - sparse_selection_arm=args.selection_source_manifest is not None, - seed=args.seed, - epochs=args.epochs, - learning_rate=args.learning_rate, - max_weight_ratio=args.max_weight_ratio, - l2_lambda=reconciliation_l2_lambda, - target_loss_cap=US_FISCAL_TARGET_LOSS_CAP, - reporter_source_ids=ssi_reporter_source_ids, - medicaid_enrollment_substitutions=medicaid_enrollment_substitutions, + # SSI take-up was assigned once before target materialization + # (populace#469): apply the release weights to the same support, measure + # the frozen flags for the published diagnostics, and let the gap to the + # SSA band counts ship in the scorecard as calibration's residual on the + # #470 registry targets — like every other program's take-up miss. + if args.dense_default_dataset: + export_frame = _with_calibrated_weights( + base_frame, + np.asarray(result.weights, dtype=np.float64), + ) + else: + export_frame = _with_l0_refit_weights(base_frame, result) + compilation = dict(compilation) + final_uncapped_ssi = _ssi_person_uncapped_amount( + export_frame, maximum_microsim_batch_size=args.maximum_microsim_batch_size, - gate_congressional_district_targets=args.gate_congressional_district_targets, - progress_callback=( - telemetry.calibration_progress if telemetry is not None else None - ), ) - export_frame = reconciliation.export_frame - result = reconciliation.calibration_result - registry = reconciliation.registry - compilation = dict(reconciliation.compilation) - ssi_take_up_diagnostics = dict(reconciliation.ssi_diagnostics) - medicaid_take_up_diagnostics = dict(reconciliation.medicaid_diagnostics) - health_input_gate = reconciliation.health_input_gate - other_health_insurance_gate = reconciliation.other_health_insurance_gate + ssi_take_up_diagnostics = dict( + us_ssi_take_up_diagnostics( + export_frame, + uncapped_ssi=final_uncapped_ssi, + seed=args.seed, + targets=ssi_band_targets, + assignment_priors=ssi_assignment_priors, + reporter_source_ids=ssi_reporter_source_ids, + ) + ) + # Gate the persisted flags on the export frame, not just the stage + # output: the Bernoulli-law recheck and anchor/envelope laws are + # weight-safe, so any downstream transform that corrupted the frozen + # decisions fails the build here instead of shipping (PR #477 review + # finding 3). The SSA-count miss itself stays scorecard-only. + final_ssi_take_up_gate = us_ssi_take_up_gate( + ssi_take_up_diagnostics, targets=ssi_band_targets + ) + if not final_ssi_take_up_gate.passed: + if telemetry is not None: + telemetry.stage( + "ssi_take_up_final_gate", + status="failed", + message="SSI take-up final export-frame gate failed.", + failures=list(final_ssi_take_up_gate.failures), + force_upload=True, + ) + raise RuntimeError( + "Release gates failed: " + + "; ".join( + f"SSI take-up final measurement failed: {failure}" + for failure in final_ssi_take_up_gate.failures + ) + ) + medicaid_take_up_diagnostics = dict( + _medicaid_diagnostics_for_existing_output( + export_frame, + target_specs, + seed=args.seed, + substitutions=medicaid_enrollment_substitutions, + maximum_microsim_batch_size=args.maximum_microsim_batch_size, + ) + ) + # Signal gates re-check the exported support: sparse selection can drop + # rows, and a column nonconstant on the candidate base can flatten on the + # selected support. + health_input_gate = _health_input_signal_gate(export_frame) + other_health_insurance_gate = us_other_health_insurance_signal_gate(export_frame) + if not other_health_insurance_gate.passed: + raise RuntimeError( + "Release gates failed: " + + "; ".join( + f"Other health insurance signal failed on the export frame: {failure}" + for failure in other_health_insurance_gate.failures + ) + ) if congressional_district_vintage_crosswalk_metadata is not None: compilation = { **compilation, @@ -9046,13 +8478,8 @@ def main() -> None: } default_dataset = { **default_dataset, - "pre_ssi_reconciliation_final_loss": pre_reconciliation_final_loss, - "ssi_take_up_reconciliation_passes": reconciliation.passes, "final_loss": float(result.final_loss), } - if default_dataset["sparse"]: - default_dataset["refit_initial_loss"] = float(result.initial_loss) - default_dataset["refit_final_loss"] = float(result.final_loss) timing["calibration_seconds"] = time.perf_counter() - calibration_started timing["elapsed_through_calibration_seconds"] = time.perf_counter() - build_started if telemetry is not None: