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 ec992318..2d40e4cb 100644 --- a/packages/populace-build/src/populace/build/us/source_stages.json +++ b/packages/populace-build/src/populace/build/us/source_stages.json @@ -521,8 +521,8 @@ }, { "stage": "snap_take_up", - "survey": "Census CPS ASEC + USDA FNS participation rate estimates", - "source": "https://www.fns.usda.gov/snap/participation-rates", + "survey": "Census CPS ASEC + USDA FNS FY2022 state participation rate estimates", + "source": "https://www.fns.usda.gov/research/snap/state-participation-rates/2022", "grain": "spm_unit", "artifacts": [ { @@ -534,8 +534,8 @@ { "kind": "administrative_table", "format": "published_estimate", - "vintage": "latest_available", - "locator": "USDA FNS SNAP participation-rate estimates (Reaching Those in Need)" + "vintage": "fiscal_year_2022", + "locator": "USDA FNS Reaching Those in Need: Estimates of State SNAP Participation Rates in 2022, Table 1" } ], "operations": [ @@ -546,16 +546,70 @@ { "kind": "derive_snap_take_up", "seed_from_build_config": true, - "take_up_rate": { - "value": 0.82, - "source": "https://www.fns.usda.gov/snap/participation-rates" + "state_take_up_rates": { + "fiscal_year": 2022, + "measure": "share_of_eligible_people_participating_in_average_month", + "source": "https://www.fns.usda.gov/sites/default/files/resource-files/ear-snap-Reaching-Those-in-Need-2022.pdf", + "values": { + "01": 0.9, + "02": 0.73, + "04": 0.77, + "05": 0.59, + "06": 0.81, + "08": 1.0, + "09": 0.98, + "10": 0.91, + "11": 1.0, + "12": 0.81, + "13": 0.92, + "15": 0.81, + "16": 0.73, + "17": 1.0, + "18": 0.89, + "19": 0.98, + "20": 0.79, + "21": 0.75, + "22": 0.99, + "23": 0.94, + "24": 0.85, + "25": 1.0, + "26": 1.0, + "27": 0.93, + "28": 0.74, + "29": 0.92, + "30": 0.75, + "31": 0.93, + "32": 0.98, + "33": 0.82, + "34": 0.91, + "35": 1.0, + "36": 0.91, + "37": 0.95, + "38": 0.81, + "39": 0.99, + "40": 0.98, + "41": 1.0, + "42": 1.0, + "44": 1.0, + "45": 0.76, + "46": 0.84, + "47": 0.84, + "48": 0.74, + "49": 0.76, + "50": 0.99, + "51": 0.83, + "53": 1.0, + "54": 0.98, + "55": 1.0, + "56": 0.63 + } } } ], "outputs": [ "takes_up_snap_if_eligible" ], - "notes": "Reported recipients (SPM_SNAPSUB > 0) always take up \u2014 survey measurement first. Non-reporting units receive seeded draws at exactly the rate needed for the overall weighted take-up share to land on the cited FNS participation rate; when reporters alone exceed the rate the share is emergent. The flag is assigned across all SPM units and the engine intersects it with modeled eligibility, matching the retired enhanced-CPS pipeline. Draws are keyed by stable source identity so support-channel clones of one source unit always agree. Without this stage the engine defaults every unit to take-up (100% participation), overstating the reach of eligibility-side reforms (populace #243)." + "notes": "Reported recipients (SPM_SNAPSUB > 0) always take up \u2014 survey measurement first. In each state, eligible non-reporting SPM units are ordered by stable source-identity draw and topped up until weighted eligible people (household design weight times modeled snap_unit_size) are closest to the cited FY2022 FNS eligible-person participation rate; when eligible reporters alone exceed the target, the anchor is the floor. The calibrated state draw cutoff also applies to currently ineligible units, so eligibility-expanding reforms expose a plausible state-specific propensity instead of a hard-coded zero. Stable-draw ties are selected together so support-channel clones agree. FNS publishes rounded estimates and caps reported estimates at 100 percent; a reported 100 does not literally establish universal participation. Without this stage the engine defaults every unit to take-up, and the former national top-up preserved CPS state under-reporting (populace #243/#372)." }, { "stage": "eligibility_inputs", 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 3f587295..d48ca1e5 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 @@ -22,9 +22,9 @@ "value_type": "bool", "default": true, "populace_treatment": "out_of_scope", - "scope_owner": "snap_take_up source stage (populace#243, landed in PR #294)", - "rate": {"value": 0.82, "source": "https://www.fns.usda.gov/snap/participation-rates", "agency": "USDA FNS", "status": "sourced_administrative"}, - "notes": "SNAP take-up is seeded by the snap_take_up source stage, reported-receipt-anchored on raw ASEC SPM_SNAPSUB and gated by snap_take_up_signal. Not seeded here to avoid duplicating that work; out_of_scope means out of scope for this module's Bernoulli seeding, not unseeded." + "scope_owner": "snap_take_up source stage (populace#243/#372)", + "rate": {"grain": "state", "period": "FY2022", "manifest_parameter": "state_take_up_rates", "measure": "share_of_eligible_people_participating_in_average_month", "source": "https://www.fns.usda.gov/sites/default/files/resource-files/ear-snap-Reaching-Those-in-Need-2022.pdf", "agency": "USDA FNS", "status": "sourced_administrative"}, + "notes": "SNAP take-up is seeded by the snap_take_up source stage, reported-receipt-anchored on raw ASEC SPM_SNAPSUB, calibrated at state grain to FNS eligible-person participation rates, and gated by snap_take_up_signal. The state values live once in source_stages.json. Not seeded in the generic Bernoulli module to avoid duplicating that work; out_of_scope means out of scope for that module, not unseeded." }, { "variable": "takes_up_tanf_if_eligible", diff --git a/packages/populace-build/src/populace/build/us_runtime/__init__.py b/packages/populace-build/src/populace/build/us_runtime/__init__.py index 50142677..747f9b7a 100644 --- a/packages/populace-build/src/populace/build/us_runtime/__init__.py +++ b/packages/populace-build/src/populace/build/us_runtime/__init__.py @@ -247,10 +247,14 @@ with_us_snap_discretionary_exemption_inputs, ) from populace.build.us_runtime.snap_take_up import ( + US_SNAP_TAKE_UP_ELIGIBILITY_COLUMN, US_SNAP_TAKE_UP_OUTPUT_COLUMN, US_SNAP_TAKE_UP_RAW_COLUMN, US_SNAP_TAKE_UP_STAGE_NAME, + US_SNAP_TAKE_UP_TOLERANCE, + US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN, derive_us_snap_take_up_from_manifest, + us_snap_take_up_diagnostics, us_snap_take_up_signal_gate, us_snap_take_up_stage_spec, us_snap_take_up_summary, @@ -364,10 +368,14 @@ "us_hours_worked_stage_spec", "us_hours_worked_summary", "with_us_hours_worked_inputs", + "US_SNAP_TAKE_UP_ELIGIBILITY_COLUMN", "US_SNAP_TAKE_UP_OUTPUT_COLUMN", "US_SNAP_TAKE_UP_RAW_COLUMN", "US_SNAP_TAKE_UP_STAGE_NAME", + "US_SNAP_TAKE_UP_TOLERANCE", + "US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN", "derive_us_snap_take_up_from_manifest", + "us_snap_take_up_diagnostics", "us_snap_take_up_signal_gate", "us_snap_take_up_stage_spec", "us_snap_take_up_summary", @@ -649,13 +657,15 @@ def to_manifest(self) -> dict[str, object]: ), ), US_SNAP_TAKE_UP_STAGE_NAME: DonorSpec( - survey="Census CPS ASEC + USDA FNS participation rate estimates", - source="https://www.fns.usda.gov/snap/participation-rates", + survey=("Census CPS ASEC + USDA FNS FY2022 state participation rate estimates"), + source=( + "https://www.fns.usda.gov/research/snap/state-participation-rates/2022" + ), notes=( "SNAP take-up: reported recipients (SPM_SNAPSUB) always take " - "up; non-reporting units drawn to the cited FNS participation " - "rate. Without it the engine defaults every eligible unit to " - "100% take-up." + "up; non-reporting units are calibrated by state to cited FNS " + "eligible-person participation rates. Without it the engine " + "defaults every eligible unit to 100% take-up." ), ), US_ELIGIBILITY_INPUTS_STAGE_NAME: DonorSpec( diff --git a/packages/populace-build/src/populace/build/us_runtime/snap_take_up.py b/packages/populace-build/src/populace/build/us_runtime/snap_take_up.py index e8e5b126..206548ce 100644 --- a/packages/populace-build/src/populace/build/us_runtime/snap_take_up.py +++ b/packages/populace-build/src/populace/build/us_runtime/snap_take_up.py @@ -1,38 +1,36 @@ -"""SNAP take-up assignment from reported receipt and the FNS participation rate. - -Without this stage the published dataset stores no -``takes_up_snap_if_eligible`` (or stores it constant ``True`` — the engine -default), so PolicyEngine-US pays SNAP to 100% of eligible units. USDA FNS -measures participation among the eligible at roughly 82%, so universal -take-up misstates who receives SNAP and overstates the reach of any -eligibility-side reform (populace issue #243). - -The stage writes one PolicyEngine-US SPM-unit input column, -``takes_up_snap_if_eligible``, with the same semantics the retired -enhanced-CPS pipeline used: - -1. **Reported recipients always take up.** An SPM unit whose raw ASEC - ``SPM_SNAPSUB`` subsidy is positive reported receiving SNAP; the flag is - ``True`` for those units unconditionally (survey measurement first). -2. **Non-reporters are drawn to hit the published rate.** Among units with - no reported receipt, a seeded draw grants take-up at exactly the rate - needed for the overall weighted take-up share to land on the manifest's - FNS participation rate. When reporters alone exceed the rate, no - non-reporter is granted take-up and the share is emergent. - -The engine intersects this flag with modeled eligibility, so the flag is -assigned across all units (matching the retired pipeline) rather than only -modeled-eligible ones — eligibility is the rules engine's job, not the -label stage's. - -Selection draws are seeded blake2b hashes keyed by the unit's stable source -identity (``source_year`` / ``source_household_id`` / the unit's smallest -``source_person_id`` when present), so support-channel clones of one source -unit always receive the same flag and reruns are bit-reproducible. - -The participation rate is data, not code: it lives in the ``snap_take_up`` -stage of ``populace/build/us/source_stages.json`` with its FNS citation and -reaches this module as a manifest operation parameter. +"""State-conditional SNAP take-up from reported receipt and FNS rates. + +PolicyEngine-US gates SNAP receipt on modeled eligibility and the +``takes_up_snap_if_eligible`` input, which defaults to ``True``. The original +Populace source stage repaired that default by anchoring reported CPS receipt +and topping non-reporters up to one national FNS participation rate. A +national top-up, however, preserves the CPS's uneven state under-reporting and +leaves too little SNAP-recipient support in the states that need the largest +administrative calibration adjustment (populace #372). + +This stage instead uses USDA FNS's FY2022 state estimates of the share of +eligible *people* participating in SNAP: + +1. SPM units reporting ``SPM_SNAPSUB > 0`` always take up, including units the + current model does not classify as eligible. The survey anchor is a floor. +2. Within each state, eligible non-reporters are ordered by a stable seeded + draw. A draw cutoff is chosen so the weighted number of eligible people in + take-up units is closest to the FNS state rate, subject to the anchor. +3. The same cutoff is applied to currently ineligible non-reporters. They do + not receive SNAP at baseline, but an eligibility-expanding reform therefore + exposes a state-specific take-up propensity rather than a hard-coded zero. + +The calibration weight is household design weight times modeled +``snap_unit_size`` for currently eligible SPM units. This matches the FNS +estimate's eligible-person denominator while persisting one SPM-unit boolean. +Stable-draw ties are calibrated together, so support-channel clones with the +same source identity always receive the same flag. + +Rates and provenance live in the ``snap_take_up`` entry of +``populace/build/us/source_stages.json``. The release gate checks every state +against its cited rate up to source-unit granularity and rejects lost anchors, +missing rate rows, collapsed eligibility, and an unexplained universal-take-up +surface. """ from __future__ import annotations @@ -59,38 +57,43 @@ from populace.frame.units import US_SCHEMA __all__ = [ + "US_SNAP_TAKE_UP_ELIGIBILITY_COLUMN", "US_SNAP_TAKE_UP_OUTPUT_COLUMN", "US_SNAP_TAKE_UP_RAW_COLUMN", "US_SNAP_TAKE_UP_STAGE_NAME", + "US_SNAP_TAKE_UP_TOLERANCE", + "US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN", "derive_us_snap_take_up_from_manifest", + "us_snap_take_up_diagnostics", "us_snap_take_up_signal_gate", - "us_snap_take_up_summary", "us_snap_take_up_stage_spec", + "us_snap_take_up_summary", "with_us_snap_take_up_inputs", ] US_SNAP_TAKE_UP_STAGE_NAME = "snap_take_up" - -#: The PolicyEngine-US SPM-unit input column this stage owns. US_SNAP_TAKE_UP_OUTPUT_COLUMN = "takes_up_snap_if_eligible" - -#: Raw CPS ASEC person column carrying the SPM unit's reported SNAP subsidy. US_SNAP_TAKE_UP_RAW_COLUMN = "SPM_SNAPSUB" +US_SNAP_TAKE_UP_ELIGIBILITY_COLUMN = "is_snap_eligible" +US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN = "snap_unit_size" + +#: Relative rate tolerance. The gate widens this to the largest stable-draw +#: cluster in each state, because clone-consistent calibration cannot split a +#: source-identity cluster. +US_SNAP_TAKE_UP_TOLERANCE = 0.02 _PERSON_WEIGHT_COLUMN = "person_weight" _SPM_MEMBERSHIP_COLUMN = "person_spm_unit_id" +_STATE_COLUMN = "state_fips" +_REPORTED_COLUMN = "reported_snap_receipt" +_ELIGIBLE_PERSON_WEIGHT_COLUMN = "eligible_person_weight" +_DRAW_COLUMN = "stable_spm_unit_draw" +_TARGET_RATE_COLUMN = "snap_take_up_target_rate" _DERIVE_SNAP_TAKE_UP_PARAMETER_KEYS = frozenset( - {"take_up_rate", "seed_from_build_config"} + {"state_take_up_rates", "seed_from_build_config"} ) -#: Weighted take-up share must land in this band. FNS estimates ~82% of -#: eligible persons participate; the share here is over all SPM units (the -#: engine applies eligibility), so the assigned share tracks the manifest -#: rate directly. A share outside the band means the anchor or the draw -#: collapsed — or a constant-True surface (the published landmine). -_TAKE_UP_SHARE_BAND = (0.70, 0.95) - def us_snap_take_up_stage_spec() -> SourceStageSpec: """Load the packaged ``snap_take_up`` source-stage manifest entry.""" @@ -106,29 +109,65 @@ def us_snap_take_up_stage_spec() -> SourceStageSpec: return stage_map[US_SNAP_TAKE_UP_STAGE_NAME] -def _take_up_rate(operation: SourceOperationSpec) -> float: - declared = operation.parameters.get("take_up_rate") - if not isinstance(declared, dict) or "value" not in declared: +def _state_take_up_rate_declaration( + operation: SourceOperationSpec, +) -> tuple[dict[str, float], dict[str, object]]: + declared = operation.parameters.get("state_take_up_rates") + if not isinstance(declared, dict) or not isinstance(declared.get("values"), dict): raise SourceRuntimeError( - "SNAP take-up requires a take_up_rate parameter with a cited value." + "SNAP take-up requires state_take_up_rates with a values mapping." ) - if not str(declared.get("source") or ""): + source = str(declared.get("source") or "") + if not source: raise SourceRuntimeError( - "SNAP take-up rate requires a source citation in the manifest." + "SNAP state take-up rates require a source citation in the manifest." ) - rate = float(declared["value"]) - if not (0.0 < rate <= 1.0): - raise SourceRuntimeError(f"SNAP take-up rate must be in (0, 1], got {rate!r}.") - return rate + fiscal_year = declared.get("fiscal_year") + if not isinstance(fiscal_year, int) or fiscal_year < 1969: + raise SourceRuntimeError( + "SNAP state take-up rates require a valid fiscal_year vintage." + ) + measure = str(declared.get("measure") or "") + if not measure: + raise SourceRuntimeError( + "SNAP state take-up rates require a declared participation measure." + ) + rates: dict[str, float] = {} + for state, raw_rate in declared["values"].items(): + state_fips = str(state).zfill(2) + if state_fips in rates: + raise SourceRuntimeError( + "SNAP state take-up rates contain duplicate normalized state_fips " + f"{state_fips}." + ) + rate = float(raw_rate) + if not (0.0 < rate <= 1.0): + raise SourceRuntimeError( + "SNAP state take-up rates must be in (0, 1]; " + f"state {state_fips} has {rate!r}." + ) + rates[state_fips] = rate + if not rates: + raise SourceRuntimeError("SNAP state take-up rates cannot be empty.") + metadata = { + "source": source, + "fiscal_year": fiscal_year, + "measure": measure, + } + return rates, metadata -def _stable_unit_draws(units: pd.DataFrame, *, seed: int) -> np.ndarray: - """Seeded uniform draws keyed by stable source identity per SPM unit. +def _normalize_state_fips(values: pd.Series | np.ndarray) -> np.ndarray: + series = pd.Series(values) + if series.isna().any(): + raise SourceRuntimeError("SNAP take-up state_fips contains missing values.") + return ( + series.astype(str).str.replace(r"\.0$", "", regex=True).str.zfill(2).to_numpy() + ) - Support-channel clones share their source identity, so they always - receive the same draw; frames without source columns key on the SPM - unit id itself. - """ + +def _stable_unit_draws(units: pd.DataFrame, *, seed: int) -> np.ndarray: + """Seeded uniform draws keyed by stable source identity per SPM unit.""" if {"source_year", "source_household_id", "source_person_id"} <= set(units.columns): keys = ( @@ -158,17 +197,54 @@ def _stable_unit_draws(units: pd.DataFrame, *, seed: int) -> np.ndarray: ) +def _require_unit_constant_columns( + person: pd.DataFrame, columns: tuple[str, ...] +) -> None: + grouped = person.groupby(_SPM_MEMBERSHIP_COLUMN, sort=False) + for column in columns: + inconsistent = grouped[column].nunique(dropna=False).gt(1) + if inconsistent.any(): + examples = inconsistent[inconsistent].index[:5].tolist() + raise SourceRuntimeError( + f"SNAP take-up requires {column!r} to be constant within each " + f"SPM unit; inconsistent unit ids include {examples}." + ) + + +def _closest_draw_cutoff( + draws: np.ndarray, + weights: np.ndarray, + *, + target_weight: float, +) -> float: + """Choose a clone-preserving draw cutoff closest to ``target_weight``.""" + + if target_weight <= 0.0 or len(draws) == 0: + return -np.inf + by_draw = ( + pd.DataFrame({_DRAW_COLUMN: draws, "weight": weights}) + .groupby(_DRAW_COLUMN, sort=True)["weight"] + .sum() + ) + cumulative = by_draw.cumsum().to_numpy(dtype=np.float64) + reachable = np.concatenate(([0.0], cumulative)) + misses = np.abs(reachable - target_weight) + best_miss = float(misses.min()) + # Prefer the higher-weight prefix on an exact tie: this issue exists to + # restore missing SNAP-carrying support, and both prefixes are equally + # faithful to the rounded FNS rate. + best = np.flatnonzero(np.isclose(misses, best_miss, rtol=0.0, atol=1e-12))[-1] + if best == 0: + return -np.inf + return float(by_draw.index[best - 1]) + + def derive_us_snap_take_up_from_manifest( frame: pd.DataFrame | None, operation: SourceOperationSpec, context: SourceRuntimeContext, ) -> pd.DataFrame: - """Assign ``takes_up_snap_if_eligible`` at SPM-unit grain. - - The current frame must be the raw-column person table (from the stage's - ``read_table`` operation) carrying a ``person_weight`` column. Returns - one row per SPM unit. - """ + """Assign state-calibrated ``takes_up_snap_if_eligible`` at SPM grain.""" if operation.kind != "derive_snap_take_up": raise SourceRuntimeError( @@ -183,26 +259,64 @@ def derive_us_snap_take_up_from_manifest( raise SourceRuntimeError( f"SNAP take-up derivation received unsupported parameter(s): {unexpected}." ) - required = [US_SNAP_TAKE_UP_RAW_COLUMN, _SPM_MEMBERSHIP_COLUMN] + required = ( + US_SNAP_TAKE_UP_RAW_COLUMN, + _SPM_MEMBERSHIP_COLUMN, + _PERSON_WEIGHT_COLUMN, + _STATE_COLUMN, + US_SNAP_TAKE_UP_ELIGIBILITY_COLUMN, + US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN, + ) missing = [column for column in required if column not in frame.columns] if missing: raise SourceRuntimeError( f"SNAP take-up derivation requires person column(s): {missing}." ) - if _PERSON_WEIGHT_COLUMN not in frame.columns: - raise SourceRuntimeError( - "SNAP take-up derivation requires a person_weight column." - ) - rate = _take_up_rate(operation) + rates, _ = _state_take_up_rate_declaration(operation) subsidy = pd.to_numeric(frame[US_SNAP_TAKE_UP_RAW_COLUMN], errors="coerce").fillna( 0.0 ) - weight = pd.to_numeric(frame[_PERSON_WEIGHT_COLUMN], errors="coerce").fillna(0.0) - person = frame.assign(_subsidy=subsidy, _weight=weight) - aggregates = { + weight = pd.to_numeric(frame[_PERSON_WEIGHT_COLUMN], errors="coerce") + unit_size = pd.to_numeric(frame[US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN], errors="coerce") + if not np.isfinite(weight.to_numpy(dtype=np.float64)).all() or (weight < 0).any(): + raise SourceRuntimeError("SNAP take-up requires finite nonnegative weights.") + if ( + not np.isfinite(unit_size.to_numpy(dtype=np.float64)).all() + or (unit_size < 0).any() + ): + raise SourceRuntimeError( + "SNAP take-up requires finite nonnegative snap_unit_size values." + ) + person = frame.assign( + _subsidy=subsidy, + _weight=weight, + **{ + _STATE_COLUMN: _normalize_state_fips(frame[_STATE_COLUMN]), + US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN: unit_size, + }, + ) + _require_unit_constant_columns( + person, + ( + _PERSON_WEIGHT_COLUMN, + _STATE_COLUMN, + US_SNAP_TAKE_UP_ELIGIBILITY_COLUMN, + US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN, + ), + ) + aggregates: dict[str, tuple[str, str]] = { "_subsidy": ("_subsidy", "max"), "_weight": ("_weight", "first"), + _STATE_COLUMN: (_STATE_COLUMN, "first"), + US_SNAP_TAKE_UP_ELIGIBILITY_COLUMN: ( + US_SNAP_TAKE_UP_ELIGIBILITY_COLUMN, + "first", + ), + US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN: ( + US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN, + "first", + ), } for column in ("source_year", "source_household_id"): if column in person.columns: @@ -215,86 +329,121 @@ def derive_us_snap_take_up_from_manifest( .reset_index() ) + states = units[_STATE_COLUMN].astype(str).to_numpy() + missing_states = sorted(set(states) - set(rates)) + if missing_states: + raise SourceRuntimeError( + "SNAP take-up has no cited FNS participation rate for represented " + f"state_fips {missing_states}." + ) reported = units["_subsidy"].to_numpy(dtype=np.float64) > 0.0 - weights = units["_weight"].to_numpy(dtype=np.float64) - total_weight = float(weights.sum()) - reporter_weight = float(weights[reported].sum()) - non_reporter_weight = float(weights[~reported].sum()) - target_weight = rate * total_weight - non_reporter_rate = ( - max(0.0, target_weight - reporter_weight) / non_reporter_weight - if non_reporter_weight > 0.0 - else 0.0 + eligible = ( + units[US_SNAP_TAKE_UP_ELIGIBILITY_COLUMN].fillna(False).to_numpy(dtype=bool) ) + household_weights = units["_weight"].to_numpy(dtype=np.float64) + snap_unit_size = units[US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN].to_numpy(dtype=np.float64) + eligible_person_weights = household_weights * snap_unit_size * eligible + target_rates = np.asarray([rates[state] for state in states], dtype=np.float64) draws = _stable_unit_draws(units, seed=int(context.config.seed)) - takes_up = reported | (~reported & (draws < non_reporter_rate)) + takes_up = reported.copy() + + for state in sorted(set(states)): + in_state = states == state + eligible_weight = float(eligible_person_weights[in_state].sum()) + reporter_weight = float(eligible_person_weights[in_state & reported].sum()) + target_weight = rates[state] * eligible_weight + residual = max(0.0, target_weight - reporter_weight) + candidates = in_state & eligible & ~reported & (eligible_person_weights > 0) + cutoff = _closest_draw_cutoff( + draws[candidates], + eligible_person_weights[candidates], + target_weight=residual, + ) + takes_up[in_state & ~reported] = draws[in_state & ~reported] <= cutoff return pd.DataFrame( { "spm_unit_id": units[_SPM_MEMBERSHIP_COLUMN].to_numpy(), + _STATE_COLUMN: states, + _PERSON_WEIGHT_COLUMN: household_weights, + US_SNAP_TAKE_UP_ELIGIBILITY_COLUMN: eligible, + US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN: snap_unit_size, + _ELIGIBLE_PERSON_WEIGHT_COLUMN: eligible_person_weights, + _REPORTED_COLUMN: reported, + _DRAW_COLUMN: draws, + _TARGET_RATE_COLUMN: target_rates, US_SNAP_TAKE_UP_OUTPUT_COLUMN: takes_up, } ) -def _take_up_carries_signal(spm_unit: pd.DataFrame) -> bool: - """Whether the persisted take-up column is trustworthy as-is. - - A constant column (all ``True`` — the engine default — or all - ``False``) is the published landmine and must be recomputed. - """ - - values = spm_unit[US_SNAP_TAKE_UP_OUTPUT_COLUMN].dropna() - return values.nunique() > 1 +def _align_spm_values_to_person( + frame: Frame, + values: np.ndarray, + *, + name: str, +) -> np.ndarray: + spm_unit = frame.table("spm_unit") + if len(values) != len(spm_unit): + raise ValueError( + f"US SNAP take-up {name} must align with the spm_unit table: " + f"{len(spm_unit)} units, {len(values)} values." + ) + by_id = pd.Series(values, index=spm_unit["spm_unit_id"].to_numpy()) + aligned = by_id.reindex(frame.table("person")[_SPM_MEMBERSHIP_COLUMN]).to_numpy() + if pd.isna(aligned).any(): + raise ValueError( + f"US SNAP take-up could not align {name} to every person membership." + ) + return aligned def with_us_snap_take_up_inputs( frame: Frame, *, + is_snap_eligible: np.ndarray, + snap_unit_size: np.ndarray, + state_fips: np.ndarray, seed: int, time_period: int, -) -> Frame: - """Run the ``snap_take_up`` manifest stage over a US frame. - - A frame already carrying a non-constant take-up column passes through - untouched (idempotent). A missing or constant column is recomputed from - the raw ASEC reported subsidy. - - Args: - frame: A US-schema frame whose person table still carries the raw - ``SPM_SNAPSUB`` column (unless the output already carries - signal). - seed: Build-wide imputation seed for the non-reporter draws. - time_period: The dataset's time period. - - Returns: - A new frame whose spm_unit table carries - ``takes_up_snap_if_eligible``. - - Raises: - ValueError: If the frame is not US-schema or the stage output does - not cover every SPM unit. - SourceRuntimeError: If required raw columns are missing. +) -> tuple[Frame, dict[str, object]]: + """Assign state-conditional SNAP take-up and return release diagnostics. + + The three supplied arrays must align with ``frame.table("spm_unit")`` and + come from the PolicyEngine-US baseline simulation. The owned output column + is always recomputed; a persisted national assignment or constant engine + default is not trusted. """ if frame.schema != US_SCHEMA: raise ValueError("US SNAP take-up inputs require the US schema.") - spm_unit = frame.table("spm_unit") - if US_SNAP_TAKE_UP_OUTPUT_COLUMN in spm_unit.columns and _take_up_carries_signal( - spm_unit - ): - return frame - stage_person = frame.table("person").copy(deep=True) stage_person[_PERSON_WEIGHT_COLUMN] = frame.resolve_weights("person").values + stage_person[US_SNAP_TAKE_UP_ELIGIBILITY_COLUMN] = _align_spm_values_to_person( + frame, + np.asarray(is_snap_eligible, dtype=bool), + name=US_SNAP_TAKE_UP_ELIGIBILITY_COLUMN, + ) + stage_person[US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN] = _align_spm_values_to_person( + frame, + np.asarray(snap_unit_size, dtype=np.float64), + name=US_SNAP_TAKE_UP_UNIT_SIZE_COLUMN, + ) + stage_person[_STATE_COLUMN] = _align_spm_values_to_person( + frame, + np.asarray(state_fips), + name=_STATE_COLUMN, + ) + spec = us_snap_take_up_stage_spec() output = run_source_stage( - us_snap_take_up_stage_spec(), + spec, tables={"person": stage_person}, operation_handlers={ "derive_snap_take_up": derive_us_snap_take_up_from_manifest, }, config=SourceRuntimeConfig(seed=int(seed), target_year=int(time_period)), ) + spm_unit = frame.table("spm_unit") aligned = output.set_index("spm_unit_id").reindex(spm_unit["spm_unit_id"]) if aligned[US_SNAP_TAKE_UP_OUTPUT_COLUMN].isna().any(): raise ValueError("US SNAP take-up stage output does not cover every SPM unit.") @@ -303,97 +452,222 @@ def with_us_snap_take_up_inputs( tables["spm_unit"][US_SNAP_TAKE_UP_OUTPUT_COLUMN] = aligned[ US_SNAP_TAKE_UP_OUTPUT_COLUMN ].to_numpy(dtype=bool) - return Frame( + result = Frame( tables, frame.schema, {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, frame.strata, mass_log=frame.mass_log, ) + derive = next(op for op in spec.operations if op.kind == "derive_snap_take_up") + _, metadata = _state_take_up_rate_declaration(derive) + return result, us_snap_take_up_diagnostics(output, rate_metadata=metadata) -def _reported_by_spm_unit(frame: Frame) -> pd.Series | None: - person = frame.table("person") - if US_SNAP_TAKE_UP_RAW_COLUMN not in person.columns: - return None - subsidy = pd.to_numeric(person[US_SNAP_TAKE_UP_RAW_COLUMN], errors="coerce").fillna( - 0.0 - ) - return ( - person.assign(_subsidy=subsidy) - .groupby(_SPM_MEMBERSHIP_COLUMN)["_subsidy"] - .max() - .gt(0.0) +def us_snap_take_up_diagnostics( + assigned: pd.DataFrame, + *, + rate_metadata: dict[str, object] | None = None, +) -> dict[str, object]: + """State and national eligible-person participation diagnostics.""" + + required = ( + _STATE_COLUMN, + _ELIGIBLE_PERSON_WEIGHT_COLUMN, + _REPORTED_COLUMN, + _DRAW_COLUMN, + _TARGET_RATE_COLUMN, + US_SNAP_TAKE_UP_OUTPUT_COLUMN, ) + missing = [column for column in required if column not in assigned.columns] + if missing: + raise ValueError(f"SNAP take-up diagnostics require column(s): {missing}.") + takes_up = assigned[US_SNAP_TAKE_UP_OUTPUT_COLUMN].fillna(False).astype(bool) + reported = assigned[_REPORTED_COLUMN].fillna(False).astype(bool) + weights = pd.to_numeric( + assigned[_ELIGIBLE_PERSON_WEIGHT_COLUMN], errors="coerce" + ).fillna(0.0) + participating = takes_up & (weights > 0) + anchor_violated = reported & ~takes_up + + states: list[dict[str, object]] = [] + for state, group in assigned.groupby(_STATE_COLUMN, sort=True): + index = group.index + rates = pd.to_numeric(group[_TARGET_RATE_COLUMN], errors="coerce").dropna() + if rates.nunique() > 1: + raise ValueError( + f"SNAP take-up diagnostics found multiple target rates for state " + f"{state}." + ) + target_rate = float(rates.iloc[0]) if not rates.empty else None + eligible_weight = float(weights[index].sum()) + participating_weight = float(weights[index][participating[index]].sum()) + reported_eligible_weight = float(weights[index][reported[index]].sum()) + selectable = (~reported[index]) & (weights[index] > 0) + draw_clusters = ( + pd.DataFrame( + { + _DRAW_COLUMN: group.loc[selectable, _DRAW_COLUMN].to_numpy(), + "weight": weights[index][selectable].to_numpy(), + } + ) + .groupby(_DRAW_COLUMN)["weight"] + .sum() + ) + target_weight = ( + target_rate * eligible_weight if target_rate is not None else None + ) + states.append( + { + "state_fips": str(state), + "target_rate": target_rate, + "eligible_person_weight": eligible_weight, + "target_participating_person_weight": target_weight, + "reported_eligible_person_weight": reported_eligible_weight, + "participating_person_weight": participating_weight, + "modeled_participation_rate": ( + participating_weight / eligible_weight + if eligible_weight > 0 + else None + ), + "anchor_floor_exceeds_target": bool( + target_weight is not None + and reported_eligible_weight > target_weight + ), + "reported_not_taking_up_count": int(anchor_violated[index].sum()), + "max_selection_cluster_weight": ( + float(draw_clusters.max()) if len(draw_clusters) else 0.0 + ), + } + ) + + total_eligible = float(weights.sum()) + total_participating = float(weights[participating].sum()) + return { + "schema_version": 2, + "classification": "release_diagnostics", + "issues": ["populace#372", "populace#243"], + "variable": US_SNAP_TAKE_UP_OUTPUT_COLUMN, + "anchor": f"{US_SNAP_TAKE_UP_RAW_COLUMN} > 0", + "rate_semantics": "eligible_person_average_month_participation", + "weights_basis": "pre_calibration_design_weight_x_snap_unit_size", + "rate_source": dict(rate_metadata or {}), + "national": { + "eligible_person_weight": total_eligible, + "participating_person_weight": total_participating, + "modeled_participation_rate": ( + total_participating / total_eligible if total_eligible > 0 else None + ), + "reported_not_taking_up_count": int(anchor_violated.sum()), + "take_up_unique_count": int(takes_up.nunique()), + }, + "states": states, + "states_without_rates": sorted( + row["state_fips"] for row in states if row["target_rate"] is None + ), + } def us_snap_take_up_summary(frame: Frame) -> dict[str, object]: - """Weighted take-up summary for gates and release manifests.""" + """Backward-compatible output-column summary. + State-rate fidelity requires engine eligibility inputs and therefore lives + in :func:`us_snap_take_up_diagnostics`; this lightweight frame-only view is + retained for callers that inspect persisted take-up signal and anchors. + """ spm_unit = frame.table("spm_unit") + if US_SNAP_TAKE_UP_OUTPUT_COLUMN not in spm_unit: + return {"missing": [US_SNAP_TAKE_UP_OUTPUT_COLUMN]} weights = np.asarray(frame.resolve_weights("spm_unit").values, dtype=np.float64) takes_up = spm_unit[US_SNAP_TAKE_UP_OUTPUT_COLUMN].to_numpy(dtype=bool) total_weight = float(weights.sum()) - take_up_share = ( - float(weights[takes_up].sum()) / total_weight if total_weight > 0 else 0.0 - ) summary: dict[str, object] = { - "take_up_share": take_up_share, - "take_up_share_band": list(_TAKE_UP_SHARE_BAND), + "take_up_share": ( + float(weights[takes_up].sum()) / total_weight if total_weight > 0 else 0.0 + ), "unique_count": int(spm_unit[US_SNAP_TAKE_UP_OUTPUT_COLUMN].nunique()), } - reported = _reported_by_spm_unit(frame) - if reported is not None: - aligned = reported.reindex(spm_unit["spm_unit_id"]).fillna(False).to_numpy() - reporter_weight = float(weights[aligned].sum()) - summary["reported_share"] = ( - reporter_weight / total_weight if total_weight > 0 else 0.0 - ) - summary["reporters_not_taking_up"] = int(np.count_nonzero(aligned & ~takes_up)) + person = frame.table("person") + if US_SNAP_TAKE_UP_RAW_COLUMN not in person: + return summary + subsidy = pd.to_numeric(person[US_SNAP_TAKE_UP_RAW_COLUMN], errors="coerce").fillna( + 0.0 + ) + reported = ( + person.assign(_subsidy=subsidy) + .groupby(_SPM_MEMBERSHIP_COLUMN)["_subsidy"] + .max() + .gt(0.0) + .reindex(spm_unit["spm_unit_id"]) + .fillna(False) + .to_numpy(dtype=bool) + ) + summary["reported_share"] = ( + float(weights[reported].sum()) / total_weight if total_weight > 0 else 0.0 + ) + summary["reporters_not_taking_up"] = int(np.count_nonzero(reported & ~takes_up)) return summary -def us_snap_take_up_signal_gate(frame: Frame) -> GateResult: - """Require a plausible, anchor-respecting take-up surface. - - Fails when the column is missing or constant (the 100%-take-up - landmine), when any SPM unit that reported SNAP receipt is assigned no - take-up (the survey anchor must win), or when the weighted take-up - share leaves the plausibility band around the FNS participation rate. - """ - - spm_unit = frame.table("spm_unit") - if US_SNAP_TAKE_UP_OUTPUT_COLUMN not in spm_unit.columns: - return GateResult( - name="snap_take_up_signal", - passed=False, - failures=(f"spm_unit.{US_SNAP_TAKE_UP_OUTPUT_COLUMN}: missing.",), - details={"missing": [US_SNAP_TAKE_UP_OUTPUT_COLUMN]}, - ) +def us_snap_take_up_signal_gate( + diagnostics: dict[str, object], + *, + tolerance: float = US_SNAP_TAKE_UP_TOLERANCE, +) -> GateResult: + """Require state-rate-faithful, anchor-preserving SNAP take-up.""" - summary = us_snap_take_up_summary(frame) failures: list[str] = [] - if int(summary["unique_count"]) < 2: - failures.append( - f"{US_SNAP_TAKE_UP_OUTPUT_COLUMN}: constant column — universal " - "take-up is the engine-default landmine this stage exists to fix." - ) - reporters_not_taking_up = summary.get("reporters_not_taking_up") - if reporters_not_taking_up: + missing = diagnostics.get("states_without_rates", []) + if missing: failures.append( - f"{reporters_not_taking_up} SPM unit(s) reported SNAP receipt but " - "carry no take-up; reported recipients must always take up." + f"states without cited FNS participation rates: {missing}; the " + "state-conditional feed is incomplete." ) - take_up_share = float(summary["take_up_share"]) - low, high = _TAKE_UP_SHARE_BAND - if not (low <= take_up_share <= high): + unexplained_universal = False + for row in diagnostics.get("states", []): + state = row["state_fips"] + violations = int(row.get("reported_not_taking_up_count", 0)) + if violations: + failures.append( + f"state {state}: {violations} reported SNAP recipient unit(s) " + "lost the take-up flag; the survey anchor was not preserved." + ) + target_rate = row.get("target_rate") + eligible_weight = float(row["eligible_person_weight"]) + if target_rate is not None and float(target_rate) > 0 and eligible_weight <= 0: + failures.append( + f"state {state}: positive FNS participation rate with zero " + "modeled-eligible person weight; the eligibility surface collapsed." + ) + continue + if target_rate is None: + continue + target_weight = float(row["target_participating_person_weight"]) + anchor_weight = float(row["reported_eligible_person_weight"]) + participating_weight = float(row["participating_person_weight"]) + floor = max(target_weight, anchor_weight) + cluster_weight = float(row.get("max_selection_cluster_weight", 0.0)) + allowance = max(tolerance * max(floor, eligible_weight), cluster_weight) + if abs(participating_weight - floor) > allowance: + failures.append( + f"state {state}: participating eligible-person weight " + f"{participating_weight:.0f} misses the FNS/anchor floor " + f"{floor:.0f} by more than the granularity allowance " + f"{allowance:.0f}." + ) + if floor < eligible_weight - allowance: + unexplained_universal = True + + national = diagnostics.get("national", {}) + if int(national.get("take_up_unique_count", 0)) < 2 and unexplained_universal: failures.append( - f"take-up share {take_up_share:.3f} outside plausibility band " - f"[{low}, {high}]." + f"{US_SNAP_TAKE_UP_OUTPUT_COLUMN}: constant column despite at least " + "one state target materially below full participation — the " + "universal-take-up landmine remains." ) return GateResult( name="snap_take_up_signal", passed=not failures, failures=tuple(failures), - details=summary, + details=diagnostics, ) 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 dee2d2f0..2d60eb03 100644 --- a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py +++ b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py @@ -1833,7 +1833,12 @@ def n(self, entity): monkeypatch.setattr( builder, "with_us_snap_take_up_inputs", - lambda frame, *, seed, time_period: frame, + lambda frame, **kwargs: (frame, {}), + ) + monkeypatch.setattr( + builder, + "_with_snap_take_up_outputs", + lambda frame, *, seed, maximum_microsim_batch_size=None: (frame, {}), ) monkeypatch.setattr( builder, @@ -1881,7 +1886,7 @@ def n(self, entity): monkeypatch.setattr( builder, "us_snap_take_up_signal_gate", - lambda frame: builder.GateResult( + lambda diagnostics: builder.GateResult( name="snap_take_up_signal", passed=True, details={"checked": True}, @@ -2518,6 +2523,74 @@ def table(self, name): assert marketplace_takers["above_benchmark_count"] == 1 +def test_snap_take_up_builder_uses_spm_policy_inputs_and_state_fips( + monkeypatch, +) -> None: + builder = _load_builder_module() + count = 8 + household_ids = np.arange(1, count + 1, dtype="int64") + spm_unit_ids = household_ids + 100 + person = pd.DataFrame( + { + "person_id": household_ids, + "person_household_id": household_ids, + "person_tax_unit_id": household_ids + 1_000, + "person_spm_unit_id": spm_unit_ids, + "person_family_id": household_ids + 3_000, + "person_marital_unit_id": household_ids + 4_000, + "SPM_SNAPSUB": np.zeros(count), + } + ) + frame = Frame( + { + "person": person, + "household": pd.DataFrame( + { + "household_id": household_ids, + "state_fips": np.asarray([5] * 4 + [17] * 4), + } + ), + "tax_unit": pd.DataFrame({"tax_unit_id": household_ids + 1_000}), + "spm_unit": pd.DataFrame({"spm_unit_id": spm_unit_ids}), + "family": pd.DataFrame({"family_id": household_ids + 3_000}), + "marital_unit": pd.DataFrame({"marital_unit_id": household_ids + 4_000}), + }, + builder.US_SCHEMA, + { + "household": builder.Weights( + values=np.ones(count), + kind=WeightKind.DESIGN, + ) + }, + ) + fake_simulation = object() + + def fake_calculate_array( + simulation, variable, *, map_to=None, period=builder.PERIOD + ): + assert simulation is fake_simulation + assert map_to == "spm_unit" + assert period == f"{builder.PERIOD}-01" + if variable == "is_snap_eligible": + return np.ones(count) + assert variable == "snap_unit_size" + return np.ones(count) + + monkeypatch.setattr(builder, "_calculate_array", fake_calculate_array) + + result, diagnostics = builder._with_snap_take_up_outputs( + frame, + seed=42, + simulation=fake_simulation, + ) + + assigned = result.table("spm_unit")["takes_up_snap_if_eligible"] + assert assigned.iloc[4:].all() + assert assigned.iloc[:4].nunique() == 2 + assert [row["state_fips"] for row in diagnostics["states"]] == ["05", "17"] + assert builder.us_snap_take_up_signal_gate(diagnostics).passed + + def test_aca_source_runtime_refreshes_degenerate_release_inputs(monkeypatch) -> None: builder = _load_builder_module() person = pd.DataFrame( diff --git a/packages/populace-build/tests/test_us_snap_take_up.py b/packages/populace-build/tests/test_us_snap_take_up.py index f4877c6e..3b6d4241 100644 --- a/packages/populace-build/tests/test_us_snap_take_up.py +++ b/packages/populace-build/tests/test_us_snap_take_up.py @@ -1,4 +1,4 @@ -"""US SNAP take-up stage tests (populace #243).""" +"""US state-conditional SNAP take-up tests (populace #372).""" from __future__ import annotations @@ -16,9 +16,9 @@ US_SNAP_TAKE_UP_OUTPUT_COLUMN, US_SNAP_TAKE_UP_STAGE_NAME, derive_us_snap_take_up_from_manifest, + us_snap_take_up_diagnostics, us_snap_take_up_signal_gate, us_snap_take_up_stage_spec, - us_snap_take_up_summary, with_us_snap_take_up_inputs, ) from populace.build.us_runtime.source_runtime import us_source_operation_handlers @@ -28,7 +28,11 @@ _HANDLERS = {"derive_snap_take_up": derive_us_snap_take_up_from_manifest} -def _stage_spec(*, rate: float, source: str = "https://example.com") -> SourceStageSpec: +def _stage_spec( + *, + rates: dict[str, float], + source: str = "https://example.com/fns.pdf", +) -> SourceStageSpec: return SourceStageSpec.from_mapping( { "stage": US_SNAP_TAKE_UP_STAGE_NAME, @@ -40,7 +44,12 @@ def _stage_spec(*, rate: float, source: str = "https://example.com") -> SourceSt { "kind": "derive_snap_take_up", "seed_from_build_config": True, - "take_up_rate": {"value": rate, "source": source}, + "state_take_up_rates": { + "fiscal_year": 2022, + "measure": "eligible_person_share", + "source": source, + "values": rates, + }, }, ], "outputs": [US_SNAP_TAKE_UP_OUTPUT_COLUMN], @@ -49,7 +58,13 @@ def _stage_spec(*, rate: float, source: str = "https://example.com") -> SourceSt def _person_table(rows: list[dict]) -> pd.DataFrame: - baseline = {"SPM_SNAPSUB": 0.0, "person_weight": 1.0} + baseline = { + "SPM_SNAPSUB": 0.0, + "person_weight": 1.0, + "state_fips": "05", + "is_snap_eligible": True, + "snap_unit_size": 1, + } records = [] for index, row in enumerate(rows): record = dict(baseline) @@ -60,9 +75,14 @@ def _person_table(rows: list[dict]) -> pd.DataFrame: return pd.DataFrame(records) -def _run(rows: list[dict], *, rate: float, seed: int = 0) -> pd.DataFrame: +def _run( + rows: list[dict], + *, + rates: dict[str, float], + seed: int = 0, +) -> pd.DataFrame: return run_source_stage( - _stage_spec(rate=rate), + _stage_spec(rates=rates), tables={"person": _person_table(rows)}, operation_handlers=_HANDLERS, config=SourceRuntimeConfig(seed=seed, target_year=TIME_PERIOD), @@ -75,7 +95,9 @@ def _us_frame( household_weights: list[float] | None = None, spm_extra: dict | None = None, ) -> Frame: - person = _person_table(person_rows).drop(columns=["person_weight"]) + person = _person_table(person_rows).drop( + columns=["person_weight", "state_fips", "is_snap_eligible", "snap_unit_size"] + ) n = len(person) person["person_household_id"] = person["person_spm_unit_id"] household_ids = person["person_household_id"].to_numpy() @@ -109,15 +131,55 @@ def _us_frame( ) +def _with_inputs( + frame: Frame, + *, + state: str = "05", + seed: int = 0, +) -> tuple[Frame, dict[str, object]]: + count = frame.n("spm_unit") + return with_us_snap_take_up_inputs( + frame, + is_snap_eligible=np.ones(count, dtype=bool), + snap_unit_size=np.ones(count, dtype=np.int64), + state_fips=np.full(count, state), + seed=seed, + time_period=TIME_PERIOD, + ) + + +def _assigned_for_gate( + *, + takes_up: list[bool], + reported: list[bool] | None = None, + eligible_weights: list[float] | None = None, + target_rate: float | None = 0.59, +) -> pd.DataFrame: + count = len(takes_up) + return pd.DataFrame( + { + "state_fips": ["05"] * count, + "eligible_person_weight": eligible_weights or [1.0] * count, + "reported_snap_receipt": reported or [False] * count, + "stable_spm_unit_draw": np.linspace(0.01, 0.99, count), + "snap_take_up_target_rate": [target_rate] * count, + US_SNAP_TAKE_UP_OUTPUT_COLUMN: takes_up, + } + ) + + class TestManifestDeclaration: - def test_stage_is_declared_with_cited_rate(self) -> None: + def test_stage_declares_all_state_rates_with_official_source(self) -> None: spec = us_snap_take_up_stage_spec() assert spec.stage == US_SNAP_TAKE_UP_STAGE_NAME assert tuple(spec.outputs) == (US_SNAP_TAKE_UP_OUTPUT_COLUMN,) derive = next(op for op in spec.operations if op.kind == "derive_snap_take_up") - rate = derive.parameters["take_up_rate"] - assert 0.5 < float(rate["value"]) <= 1.0 - assert str(rate["source"]).startswith("https://") + declaration = derive.parameters["state_take_up_rates"] + assert declaration["fiscal_year"] == 2022 + assert len(declaration["values"]) == 51 + assert declaration["values"]["05"] == 0.59 + assert declaration["values"]["17"] == 1.0 + assert str(declaration["source"]).startswith("https://www.fns.usda.gov/") def test_handler_is_registered(self) -> None: handlers = us_source_operation_handlers() @@ -126,38 +188,46 @@ def test_handler_is_registered(self) -> None: class TestDerivation: def test_reported_recipients_always_take_up(self) -> None: - rows = [{"SPM_SNAPSUB": 3_000.0}] + [{"SPM_SNAPSUB": 0.0}] * 9 - output = _run(rows, rate=0.1) + rows = [{"SPM_SNAPSUB": 3_000.0}] + [{}] * 9 + output = _run(rows, rates={"05": 0.1}) by_unit = output.set_index("spm_unit_id")[US_SNAP_TAKE_UP_OUTPUT_COLUMN] assert bool(by_unit.loc[1]) - def test_non_reporters_fill_to_the_weighted_rate(self) -> None: - # 2 reporting units + 98 non-reporting, equal weights, rate 0.5: - # non-reporter fill rate is (50 - 2) / 98, so roughly half take up. - rows = [{"SPM_SNAPSUB": 1_200.0}] * 2 + [{"SPM_SNAPSUB": 0.0}] * 98 - output = _run(rows, rate=0.5) - share = output[US_SNAP_TAKE_UP_OUTPUT_COLUMN].mean() - assert 0.35 <= share <= 0.65 - - def test_reporters_beyond_the_rate_grant_no_non_reporters(self) -> None: - rows = [{"SPM_SNAPSUB": 1_200.0}] * 8 + [{"SPM_SNAPSUB": 0.0}] * 2 - output = _run(rows, rate=0.5) + def test_state_rates_change_the_amount_of_recipient_support(self) -> None: + rows = ([{"state_fips": "05"}] * 100) + ([{"state_fips": "17"}] * 100) + output = _run(rows, rates={"05": 0.59, "17": 1.0}) + shares = output.groupby("state_fips")[US_SNAP_TAKE_UP_OUTPUT_COLUMN].mean() + assert shares["05"] == 0.59 + assert shares["17"] == 1.0 + + def test_reporters_beyond_target_are_the_floor(self) -> None: + rows = [{"SPM_SNAPSUB": 1_200.0}] * 8 + [{}] * 2 + output = _run(rows, rates={"05": 0.5}) by_unit = output.set_index("spm_unit_id")[US_SNAP_TAKE_UP_OUTPUT_COLUMN] assert by_unit.iloc[:8].all() assert not by_unit.iloc[8:].any() + def test_calibration_uses_eligible_person_weight(self) -> None: + sizes = [1, 2, 3] * 20 + rows = [{"snap_unit_size": size} for size in sizes] + output = _run(rows, rates={"05": 0.59}) + modeled = output.loc[ + output[US_SNAP_TAKE_UP_OUTPUT_COLUMN], "eligible_person_weight" + ].sum() + target = 0.59 * output["eligible_person_weight"].sum() + assert abs(modeled - target) <= max(sizes) + def test_draws_are_seed_stable(self) -> None: - rows = [{"SPM_SNAPSUB": 0.0}] * 40 - first = _run(rows, rate=0.5, seed=7) - second = _run(rows, rate=0.5, seed=7) - assert first[US_SNAP_TAKE_UP_OUTPUT_COLUMN].tolist() == ( - second[US_SNAP_TAKE_UP_OUTPUT_COLUMN].tolist() + rows = [{}] * 40 + first = _run(rows, rates={"05": 0.5}, seed=7) + second = _run(rows, rates={"05": 0.5}, seed=7) + assert ( + first[US_SNAP_TAKE_UP_OUTPUT_COLUMN].tolist() + == second[US_SNAP_TAKE_UP_OUTPUT_COLUMN].tolist() ) - def test_source_identity_keys_clone_consistent_draws(self) -> None: - # Two "clones" of the same source unit must always agree. + def test_source_identity_clones_are_selected_together(self) -> None: clone = { - "SPM_SNAPSUB": 0.0, "source_year": 2023, "source_household_id": 555, "source_person_id": 7, @@ -165,37 +235,50 @@ def test_source_identity_keys_clone_consistent_draws(self) -> None: rows = [ {**clone, "person_spm_unit_id": 1}, {**clone, "person_spm_unit_id": 2}, - ] + [{"SPM_SNAPSUB": 0.0, "person_spm_unit_id": 3 + i} for i in range(20)] - output = _run(rows, rate=0.5) + ] + [{"person_spm_unit_id": 3 + index} for index in range(20)] + output = _run(rows, rates={"05": 0.5}) by_unit = output.set_index("spm_unit_id")[US_SNAP_TAKE_UP_OUTPUT_COLUMN] assert bool(by_unit.loc[1]) == bool(by_unit.loc[2]) - def test_multi_person_units_report_once(self) -> None: + def test_multi_person_units_are_emitted_once(self) -> None: rows = [ - {"person_spm_unit_id": 1, "SPM_SNAPSUB": 2_400.0}, - {"person_spm_unit_id": 1, "SPM_SNAPSUB": 2_400.0}, - {"person_spm_unit_id": 2, "SPM_SNAPSUB": 0.0}, + { + "person_spm_unit_id": 1, + "SPM_SNAPSUB": 2_400.0, + "snap_unit_size": 2, + }, + { + "person_spm_unit_id": 1, + "SPM_SNAPSUB": 2_400.0, + "snap_unit_size": 2, + }, + {"person_spm_unit_id": 2}, ] - output = _run(rows, rate=0.0001) + output = _run(rows, rates={"05": 0.1}) assert len(output) == 2 - by_unit = output.set_index("spm_unit_id")[US_SNAP_TAKE_UP_OUTPUT_COLUMN] - assert bool(by_unit.loc[1]) + assert bool( + output.set_index("spm_unit_id").loc[1, US_SNAP_TAKE_UP_OUTPUT_COLUMN] + ) def test_rate_without_citation_is_refused(self) -> None: - spec = _stage_spec(rate=0.82, source="") + spec = _stage_spec(rates={"05": 0.59}, source="") with pytest.raises(SourceRuntimeError, match="citation"): run_source_stage( spec, - tables={"person": _person_table([{"SPM_SNAPSUB": 0.0}])}, + tables={"person": _person_table([{}])}, operation_handlers=_HANDLERS, config=SourceRuntimeConfig(seed=0, target_year=TIME_PERIOD), ) - def test_missing_raw_column_is_named(self) -> None: - table = _person_table([{"SPM_SNAPSUB": 0.0}]).drop(columns=["SPM_SNAPSUB"]) - with pytest.raises(SourceRuntimeError, match="SPM_SNAPSUB"): + def test_represented_state_without_rate_is_refused(self) -> None: + with pytest.raises(SourceRuntimeError, match="no cited FNS"): + _run([{"state_fips": "06"}], rates={"05": 0.59}) + + def test_missing_policy_input_is_named(self) -> None: + table = _person_table([{}]).drop(columns=["is_snap_eligible"]) + with pytest.raises(SourceRuntimeError, match="is_snap_eligible"): run_source_stage( - _stage_spec(rate=0.82), + _stage_spec(rates={"05": 0.59}), tables={"person": table}, operation_handlers=_HANDLERS, config=SourceRuntimeConfig(seed=0, target_year=TIME_PERIOD), @@ -204,78 +287,102 @@ def test_missing_raw_column_is_named(self) -> None: class TestFrameIntegration: def _mixed_rows(self, n_reporters: int = 20, n_total: int = 100) -> list[dict]: - return [{"SPM_SNAPSUB": 2_000.0}] * n_reporters + [{"SPM_SNAPSUB": 0.0}] * ( - n_total - n_reporters - ) + return [{"SPM_SNAPSUB": 2_000.0}] * n_reporters + [{}] * (n_total - n_reporters) - def test_with_inputs_writes_the_spm_unit_column(self) -> None: - frame = with_us_snap_take_up_inputs( - _us_frame(self._mixed_rows()), seed=0, time_period=TIME_PERIOD - ) + def test_with_inputs_writes_the_spm_unit_column_and_diagnostics(self) -> None: + frame, diagnostics = _with_inputs(_us_frame(self._mixed_rows())) spm_unit = frame.table("spm_unit") assert US_SNAP_TAKE_UP_OUTPUT_COLUMN in spm_unit.columns assert spm_unit[US_SNAP_TAKE_UP_OUTPUT_COLUMN].iloc[:20].all() + assert diagnostics["states"][0]["target_rate"] == 0.59 - def test_frame_with_signal_passes_through_untouched(self) -> None: - derived = with_us_snap_take_up_inputs( - _us_frame(self._mixed_rows()), seed=0, time_period=TIME_PERIOD - ) - again = with_us_snap_take_up_inputs(derived, seed=9, time_period=TIME_PERIOD) - assert again is derived + def test_owned_column_is_always_recomputed(self) -> None: + derived, _ = _with_inputs(_us_frame(self._mixed_rows()), seed=0) + original = derived.table("spm_unit")[US_SNAP_TAKE_UP_OUTPUT_COLUMN].copy() + recomputed, _ = _with_inputs(derived, seed=9) + updated = recomputed.table("spm_unit")[US_SNAP_TAKE_UP_OUTPUT_COLUMN] + assert not original.equals(updated) def test_constant_true_landmine_is_recomputed(self) -> None: - # The published failure mode: takes_up constant True for every unit. frame = _us_frame( self._mixed_rows(n_reporters=2, n_total=10), spm_extra={US_SNAP_TAKE_UP_OUTPUT_COLUMN: [True] * 10}, ) - healed = with_us_snap_take_up_inputs(frame, seed=0, time_period=TIME_PERIOD) + healed, _ = _with_inputs(frame) values = healed.table("spm_unit")[US_SNAP_TAKE_UP_OUTPUT_COLUMN] assert values.iloc[:2].all() assert values.nunique() == 2 + def test_input_arrays_must_align_with_spm_units(self) -> None: + frame = _us_frame(self._mixed_rows(n_reporters=2, n_total=10)) + with pytest.raises(ValueError, match="10 units, 9 values"): + with_us_snap_take_up_inputs( + frame, + is_snap_eligible=np.ones(9, dtype=bool), + snap_unit_size=np.ones(10), + state_fips=np.full(10, "05"), + seed=0, + time_period=TIME_PERIOD, + ) + class TestGate: - def test_plausible_assignment_passes(self) -> None: - frame = with_us_snap_take_up_inputs( - _us_frame([{"SPM_SNAPSUB": 2_000.0}] * 20 + [{"SPM_SNAPSUB": 0.0}] * 80), - seed=0, - time_period=TIME_PERIOD, + def test_state_faithful_assignment_passes(self) -> None: + _, diagnostics = _with_inputs( + _us_frame([{"SPM_SNAPSUB": 2_000.0}] * 20 + [{}] * 80) ) - gate = us_snap_take_up_signal_gate(frame) + gate = us_snap_take_up_signal_gate(diagnostics) assert gate.passed - summary = us_snap_take_up_summary(frame) - assert 0.70 <= summary["take_up_share"] <= 0.95 - assert summary["reporters_not_taking_up"] == 0 + state = diagnostics["states"][0] + assert abs(state["modeled_participation_rate"] - 0.59) <= 0.01 + assert state["reported_not_taking_up_count"] == 0 - def test_missing_column_fails(self) -> None: - gate = us_snap_take_up_signal_gate(_us_frame([{"SPM_SNAPSUB": 0.0}] * 4)) + def test_missing_rate_fails(self) -> None: + diagnostics = us_snap_take_up_diagnostics( + _assigned_for_gate(takes_up=[True, False], target_rate=None) + ) + gate = us_snap_take_up_signal_gate(diagnostics) assert not gate.passed - assert "missing" in gate.failures[0] + assert "without cited FNS" in gate.failures[0] - def test_constant_true_fails(self) -> None: - frame = _us_frame( - [{"SPM_SNAPSUB": 0.0}] * 4, - spm_extra={US_SNAP_TAKE_UP_OUTPUT_COLUMN: [True] * 4}, + def test_constant_true_below_full_target_fails(self) -> None: + diagnostics = us_snap_take_up_diagnostics( + _assigned_for_gate(takes_up=[True] * 10) ) - gate = us_snap_take_up_signal_gate(frame) + gate = us_snap_take_up_signal_gate(diagnostics) assert not gate.passed - assert any("constant" in failure for failure in gate.failures) + assert any("constant column" in failure for failure in gate.failures) def test_reporter_without_take_up_fails(self) -> None: - frame = _us_frame( - [{"SPM_SNAPSUB": 3_000.0}] + [{"SPM_SNAPSUB": 0.0}] * 3, - spm_extra={US_SNAP_TAKE_UP_OUTPUT_COLUMN: [False, True, True, False]}, + diagnostics = us_snap_take_up_diagnostics( + _assigned_for_gate( + takes_up=[False, True, False, True], + reported=[True, False, False, False], + ) ) - gate = us_snap_take_up_signal_gate(frame) + gate = us_snap_take_up_signal_gate(diagnostics) assert not gate.passed - assert any("reported SNAP receipt but" in failure for failure in gate.failures) + assert any("reported SNAP recipient" in failure for failure in gate.failures) - def test_share_outside_band_fails(self) -> None: - frame = _us_frame( - [{"SPM_SNAPSUB": 0.0}] * 10, - spm_extra={US_SNAP_TAKE_UP_OUTPUT_COLUMN: [True] + [False] * 9}, + def test_collapsed_eligibility_fails(self) -> None: + diagnostics = us_snap_take_up_diagnostics( + _assigned_for_gate( + takes_up=[False, True], + eligible_weights=[0.0, 0.0], + ) ) - gate = us_snap_take_up_signal_gate(frame) + gate = us_snap_take_up_signal_gate(diagnostics) assert not gate.passed - assert any("take-up share" in failure for failure in gate.failures) + assert any( + "eligibility surface collapsed" in failure for failure in gate.failures + ) + + def test_rate_miss_outside_granularity_fails(self) -> None: + diagnostics = us_snap_take_up_diagnostics( + _assigned_for_gate(takes_up=[True] + [False] * 99) + ) + gate = us_snap_take_up_signal_gate(diagnostics) + assert not gate.passed + assert any( + "misses the FNS/anchor floor" in failure for failure in gate.failures + ) diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index 5158574e..fa41e061 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -2313,6 +2313,133 @@ def _person_state_fips(frame: Frame) -> np.ndarray: return np.asarray(_state_fips_text(frame.broadcast("state_fips").to_numpy())) +def _spm_unit_state_fips(frame: Frame) -> np.ndarray: + """SPM-unit-aligned state FIPS text codes via household nesting.""" + positions = _group_to_household_positions(frame, "spm_unit") + state_fips = frame.table("household")["state_fips"].to_numpy()[positions] + return np.asarray(_state_fips_text(state_fips)) + + +def _snap_spm_unit_policy_inputs( + frame: Frame, + *, + simulation=None, + maximum_microsim_batch_size: int | None = DEFAULT_MAXIMUM_MICROSIM_BATCH_SIZE, +) -> tuple[np.ndarray, np.ndarray]: + """Engine SNAP eligibility and eligible unit size at SPM-unit grain. + + The FNS rates are average-month eligible-person shares. Populace persists + one annual take-up flag, so the source stage uses January as the baseline + representative month and applies the resulting state cutoff off-domain as + well; units becoming eligible later in the year retain that propensity. + """ + period = f"{PERIOD}-01" + if simulation is not None: + eligible = ( + _calculate_array( + simulation, + "is_snap_eligible", + map_to="spm_unit", + period=period, + ) + > 0 + ) + unit_size = _calculate_array( + simulation, + "snap_unit_size", + map_to="spm_unit", + period=period, + ) + return eligible, unit_size + + from policyengine_us import Microsimulation + + spm_unit_ids = frame.table("spm_unit")["spm_unit_id"].to_numpy() + eligible = np.zeros(len(spm_unit_ids), dtype=bool) + unit_size = np.zeros(len(spm_unit_ids), dtype=np.float64) + spm_unit_positions = pd.Series( + np.arange(len(spm_unit_ids), dtype=np.int64), index=spm_unit_ids + ) + n_households = frame.n("household") + batches = tuple( + _household_position_batches(n_households, maximum_microsim_batch_size) + ) + if len(batches) > 1: + print( + "Materializing SNAP eligibility in " + f"{len(batches)} batches of up to " + f"{maximum_microsim_batch_size:,} households.", + flush=True, + ) + for household_positions in batches: + with _automatic_gc_suspended(): + full_batch = len(household_positions) == n_households + batch_frame = ( + frame + if full_batch + else _select_households_by_position(frame, household_positions) + ) + batch_simulation = Microsimulation( + dataset=_dataset_from_frame( + batch_frame, + assert_no_formula_owned_columns=False, + ) + ) + batch_eligible = ( + _calculate_array( + batch_simulation, + "is_snap_eligible", + map_to="spm_unit", + period=period, + ) + > 0 + ) + batch_unit_size = _calculate_array( + batch_simulation, + "snap_unit_size", + map_to="spm_unit", + period=period, + ) + positions = spm_unit_positions.reindex( + batch_frame.table("spm_unit")["spm_unit_id"].to_numpy() + ).to_numpy() + if np.isnan(positions).any(): + raise RuntimeError( + "SNAP eligibility batch produced spm_unit_id values not " + "present in the full SPM-unit table." + ) + positions = positions.astype(np.int64) + eligible[positions] = batch_eligible + unit_size[positions] = batch_unit_size + batch_simulation._invalidate_all_caches() + del batch_frame, batch_simulation + _collect_batch_garbage() + return eligible, unit_size + + +def _with_snap_take_up_outputs( + frame: Frame, + *, + seed: int, + simulation=None, + maximum_microsim_batch_size: int | None = DEFAULT_MAXIMUM_MICROSIM_BATCH_SIZE, +) -> tuple[Frame, dict[str, object]]: + """Assign state-conditional SNAP take-up and return diagnostics.""" + eligibility, unit_size = _snap_spm_unit_policy_inputs( + frame, + simulation=simulation, + maximum_microsim_batch_size=maximum_microsim_batch_size, + ) + return with_us_snap_take_up_inputs( + frame, + is_snap_eligible=eligibility, + snap_unit_size=unit_size, + state_fips=_spm_unit_state_fips(frame), + seed=seed, + time_period=PERIOD, + ) + + def _medicaid_person_eligibility( frame: Frame, *, @@ -2448,9 +2575,13 @@ def _dataset_from_frame( def _calculate_array( - simulation, variable: str, *, map_to: str | None = None + simulation, + variable: str, + *, + map_to: str | None = None, + period: int | str = PERIOD, ) -> np.ndarray: - kwargs: dict[str, Any] = {"period": PERIOD} + kwargs: dict[str, Any] = {"period": period} if map_to is not None: kwargs["map_to"] = map_to return np.asarray(simulation.calculate(variable, **kwargs)) @@ -5823,33 +5954,6 @@ def main() -> None: for failure in hours_worked_gate.failures ) ) - if telemetry is not None: - telemetry.stage( - "snap_take_up_inputs", - message="Assigning SNAP take-up from reported receipt.", - ) - base_frame = with_us_snap_take_up_inputs( - base_frame, - seed=args.seed, - time_period=PERIOD, - ) - snap_take_up_gate = us_snap_take_up_signal_gate(base_frame) - if not snap_take_up_gate.passed: - if telemetry is not None: - telemetry.stage( - "snap_take_up_gate", - status="failed", - message="SNAP take-up signal gate failed.", - failures=list(snap_take_up_gate.failures), - force_upload=True, - ) - raise RuntimeError( - "Release gates failed: " - + "; ".join( - f"SNAP take-up signal failed: {failure}" - for failure in snap_take_up_gate.failures - ) - ) if telemetry is not None: telemetry.stage( "eligibility_inputs", @@ -5976,6 +6080,36 @@ def main() -> None: for failure in scf_wealth_gate.failures ) ) + if telemetry is not None: + telemetry.stage( + "snap_take_up_inputs", + message=( + "Assigning SNAP take-up from reported receipt and FNS state " + "participation rates." + ), + ) + base_frame, snap_take_up_diagnostics = _with_snap_take_up_outputs( + base_frame, + seed=args.seed, + maximum_microsim_batch_size=args.maximum_microsim_batch_size, + ) + snap_take_up_gate = us_snap_take_up_signal_gate(snap_take_up_diagnostics) + if not snap_take_up_gate.passed: + if telemetry is not None: + telemetry.stage( + "snap_take_up_gate", + status="failed", + message="SNAP take-up signal gate failed.", + failures=list(snap_take_up_gate.failures), + force_upload=True, + ) + raise RuntimeError( + "Release gates failed: " + + "; ".join( + f"SNAP take-up signal failed: {failure}" + for failure in snap_take_up_gate.failures + ) + ) if telemetry is not None: telemetry.stage( "source_inputs",