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 c91ca604..c947a3b8 100644 --- a/packages/populace-build/src/populace/build/us_runtime/__init__.py +++ b/packages/populace-build/src/populace/build/us_runtime/__init__.py @@ -147,6 +147,11 @@ reform_validation_payload, write_reform_validation, ) +from populace.build.us_runtime.snap_local_proxy import ( + SNAP_LOCAL_PROXY_SCHEMA_VERSION, + snap_local_proxy_diagnostics, + write_snap_local_proxy_diagnostics, +) from populace.build.us_runtime.source_coverage import ( LEDGER_US_SOURCE_COVERAGE_CONTRACT_COMMIT, US_SOURCE_COVERAGE, @@ -177,6 +182,7 @@ "AgeBand", "AGE_BANDS", "DEMOGRAPHICS_SCHEMA_VERSION", + "SNAP_LOCAL_PROXY_SCHEMA_VERSION", "compute_age_distribution", "CONGRESSIONAL_DISTRICT_GEOID_COLUMN", "CONGRESSIONAL_DISTRICT_VINTAGE_CROSSWALK_SHA256_ATTR", @@ -260,9 +266,11 @@ "pool_asec_sources", "reform_validation_payload", "source_gap_family_ids", + "snap_local_proxy_diagnostics", "us_input_mass_totals", "us_plan", "us_source_operation_handlers", + "write_snap_local_proxy_diagnostics", "write_reform_validation", "us_source_coverage_diagnostics", "us_source_coverage_gate", diff --git a/packages/populace-build/src/populace/build/us_runtime/snap_local_proxy.py b/packages/populace-build/src/populace/build/us_runtime/snap_local_proxy.py new file mode 100644 index 00000000..1429d5dd --- /dev/null +++ b/packages/populace-build/src/populace/build/us_runtime/snap_local_proxy.py @@ -0,0 +1,372 @@ +"""US SNAP local proxy release diagnostics. + +ACS S2201 congressional-district SNAP household estimates are validation-only +references. These helpers compare district-level Populace SNAP household +receipt/support against those references without promoting them into calibration +targets. +""" + +from __future__ import annotations + +import json +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +import numpy as np +import pandas as pd + +__all__ = [ + "SNAP_LOCAL_PROXY_SCHEMA_VERSION", + "snap_local_proxy_diagnostics", + "write_snap_local_proxy_diagnostics", +] + +SNAP_LOCAL_PROXY_SCHEMA_VERSION = 1 +SNAP_LOCAL_PROXY_FAMILY = "snap_local_proxy" + + +def _finite(value: float | int | np.floating | None) -> float | None: + if value is None: + return None + value = float(value) + return value if np.isfinite(value) else None + + +def _effective_sample_size(weights: np.ndarray) -> float | None: + if weights.size == 0: + return None + total = float(weights.sum()) + squared = float(np.square(weights).sum()) + if total <= 0 or squared <= 0: + return None + return total * total / squared + + +def _require_columns( + frame: pd.DataFrame, + columns: tuple[str | None, ...], + *, + label: str, +) -> None: + missing = [column for column in columns if column and column not in frame.columns] + if missing: + raise ValueError(f"{label} missing required columns {missing}.") + + +def _reference_by_district( + acs_reference: pd.DataFrame | None, + *, + district_column: str, + snap_households_column: str, + snap_households_moe_column: str | None, +) -> dict[str, dict[str, float | None]] | None: + if acs_reference is None: + return None + _require_columns( + acs_reference, + (district_column, snap_households_column, snap_households_moe_column), + label="ACS SNAP local proxy reference", + ) + duplicated = acs_reference[district_column].duplicated(keep=False) + if duplicated.any(): + examples = acs_reference.loc[duplicated, district_column].head(5).tolist() + raise ValueError(f"ACS reference has duplicate district rows: {examples}.") + + references: dict[str, dict[str, float | None]] = {} + for _, row in acs_reference.iterrows(): + district = str(row[district_column]) + estimate = pd.to_numeric(row[snap_households_column], errors="coerce") + moe = ( + pd.to_numeric(row[snap_households_moe_column], errors="coerce") + if snap_households_moe_column + else None + ) + references[district] = { + "acs_snap_households": _finite(estimate), + "acs_snap_households_moe": _finite(moe), + } + return references + + +def _group_state( + group: pd.DataFrame, + *, + state_column: str | None, + flags: list[str], +) -> str | None: + if state_column is None: + return None + values = tuple(sorted(str(value) for value in group[state_column].dropna().unique())) + if not values: + return None + if len(values) > 1: + flags.append("mixed_state") + return None + return values[0] + + +def _group_state_error( + group: pd.DataFrame, + *, + state_snap_relative_error_column: str | None, +) -> float | None: + if state_snap_relative_error_column is None: + return None + values = pd.to_numeric( + group[state_snap_relative_error_column], errors="coerce" + ).dropna() + if values.empty: + return None + return _finite(values.iloc[0]) + + +def snap_local_proxy_diagnostics( + household_frame: pd.DataFrame, + *, + district_column: str, + weight_column: str, + snap_receipt_column: str, + snap_amount_column: str | None = None, + state_column: str | None = None, + state_snap_relative_error_column: str | None = None, + acs_reference: pd.DataFrame | None = None, + acs_district_column: str | None = None, + acs_snap_households_column: str = "acs_snap_households", + acs_snap_households_moe_column: str | None = None, + low_positive_sample_threshold: int = 30, + low_positive_ess_threshold: float = 10.0, + state_outlier_abs_relative_error: float = 0.10, + validation_source: Mapping[str, object] | None = None, +) -> dict[str, Any]: + """Return validation-only SNAP local proxy diagnostics. + + Args: + household_frame: One row per household or household-like record. + district_column: Congressional district GEOID/display column. + weight_column: Household analysis weight. + snap_receipt_column: Boolean/numeric SNAP receipt indicator. + snap_amount_column: Optional household SNAP dollar amount. + state_column: Optional state label repeated on household rows. + state_snap_relative_error_column: Optional state SNAP spending fit error. + acs_reference: Optional ACS S2201 reference frame, one row per district. + acs_district_column: District column in ``acs_reference``. Defaults to + ``district_column``. + acs_snap_households_column: ACS SNAP household estimate column. + acs_snap_households_moe_column: Optional ACS margin-of-error column. + low_positive_sample_threshold: Raw positive household count flag. + low_positive_ess_threshold: Positive-recipient ESS flag. + state_outlier_abs_relative_error: State SNAP fit flag threshold. + validation_source: Metadata about the ACS/reference source. + """ + if low_positive_sample_threshold < 0: + raise ValueError("low_positive_sample_threshold must be nonnegative.") + if low_positive_ess_threshold < 0: + raise ValueError("low_positive_ess_threshold must be nonnegative.") + if state_outlier_abs_relative_error < 0: + raise ValueError("state_outlier_abs_relative_error must be nonnegative.") + + _require_columns( + household_frame, + ( + district_column, + weight_column, + snap_receipt_column, + snap_amount_column, + state_column, + state_snap_relative_error_column, + ), + label="SNAP local proxy household frame", + ) + + acs_district_column = acs_district_column or district_column + references = _reference_by_district( + acs_reference, + district_column=acs_district_column, + snap_households_column=acs_snap_households_column, + snap_households_moe_column=acs_snap_households_moe_column, + ) + + frame = household_frame.copy() + weights = pd.to_numeric(frame[weight_column], errors="coerce") + valid = weights.notna() & np.isfinite(weights) & (weights > 0) + valid &= frame[district_column].notna() + if not valid.any(): + raise ValueError("SNAP local proxy frame has no positive-weight district rows.") + + frame = frame.loc[valid].copy() + frame["_snap_local_proxy_weight"] = weights.loc[valid].astype(float) + frame["_snap_local_proxy_receipt"] = ( + pd.to_numeric(frame[snap_receipt_column], errors="coerce") + .fillna(0.0) + .to_numpy(dtype=float) + ) + if snap_amount_column is not None: + frame["_snap_local_proxy_amount"] = ( + pd.to_numeric(frame[snap_amount_column], errors="coerce") + .fillna(0.0) + .to_numpy(dtype=float) + ) + + rows: list[dict[str, Any]] = [] + for district, group in frame.groupby(district_column, dropna=False, sort=True): + flags: list[str] = [] + district_id = str(district) + group_weights = group["_snap_local_proxy_weight"].to_numpy(dtype=float) + receipt = group["_snap_local_proxy_receipt"].to_numpy(dtype=float) > 0 + positive_weights = group_weights[receipt] + weighted_households = float(group_weights.sum()) + weighted_snap_households = float(positive_weights.sum()) + raw_positive_sample_households = int(receipt.sum()) + positive_ess = _effective_sample_size(positive_weights) + if raw_positive_sample_households < low_positive_sample_threshold: + flags.append("low_positive_sample") + if positive_ess is not None and positive_ess < low_positive_ess_threshold: + flags.append("low_positive_ess") + + state_snap_relative_error = _group_state_error( + group, + state_snap_relative_error_column=state_snap_relative_error_column, + ) + if ( + state_snap_relative_error is not None + and abs(state_snap_relative_error) >= state_outlier_abs_relative_error + ): + flags.append("state_snap_outlier") + + reference = references.get(district_id) if references is not None else None + acs_snap_households = None + acs_snap_households_moe = None + snap_household_difference = None + snap_household_relative_error = None + outside_acs_moe = None + if references is not None and reference is None: + flags.append("missing_acs_reference") + elif reference is not None: + acs_snap_households = reference["acs_snap_households"] + acs_snap_households_moe = reference["acs_snap_households_moe"] + if acs_snap_households is not None: + snap_household_difference = ( + weighted_snap_households - acs_snap_households + ) + if acs_snap_households != 0: + snap_household_relative_error = ( + snap_household_difference / acs_snap_households + ) + if acs_snap_households_moe is not None: + outside_acs_moe = ( + abs(snap_household_difference) > acs_snap_households_moe + ) + if outside_acs_moe: + flags.append("outside_acs_moe") + + snap_dollars = None + if snap_amount_column is not None: + amounts = group["_snap_local_proxy_amount"].to_numpy(dtype=float) + snap_dollars = float(np.sum(amounts * group_weights)) + + mean_positive_weight = ( + float(positive_weights.mean()) if positive_weights.size else None + ) + max_to_mean = ( + float(positive_weights.max() / mean_positive_weight) + if mean_positive_weight and mean_positive_weight > 0 + else None + ) + rows.append( + { + "congressional_district": district_id, + "state": _group_state(group, state_column=state_column, flags=flags), + "weighted_households": _finite(weighted_households), + "weighted_snap_households": _finite(weighted_snap_households), + "snap_household_share": _finite( + weighted_snap_households / weighted_households + if weighted_households + else None + ), + "snap_dollars": _finite(snap_dollars), + "raw_positive_snap_households": raw_positive_sample_households, + "positive_snap_ess": _finite(positive_ess), + "positive_snap_max_to_mean_weight_ratio": _finite(max_to_mean), + "state_snap_relative_error": _finite(state_snap_relative_error), + "acs_snap_households": _finite(acs_snap_households), + "acs_snap_households_moe": _finite(acs_snap_households_moe), + "snap_household_difference": _finite(snap_household_difference), + "snap_household_relative_error": _finite( + snap_household_relative_error + ), + "outside_acs_moe": outside_acs_moe, + "flags": tuple(sorted(flags)), + } + ) + + total_weighted_households = sum(row["weighted_households"] or 0.0 for row in rows) + total_weighted_snap_households = sum( + row["weighted_snap_households"] or 0.0 for row in rows + ) + total_snap_dollars = ( + sum(row["snap_dollars"] or 0.0 for row in rows) + if snap_amount_column is not None + else None + ) + total_acs_snap_households = ( + sum(row["acs_snap_households"] or 0.0 for row in rows) + if references is not None + else None + ) + outside_moe_values = [ + row["outside_acs_moe"] for row in rows if row["outside_acs_moe"] is not None + ] + payload = { + "schema_version": SNAP_LOCAL_PROXY_SCHEMA_VERSION, + "classification": "validation_only", + "source_family": SNAP_LOCAL_PROXY_FAMILY, + "validation_source": dict(validation_source or {}), + "thresholds": { + "low_positive_sample": int(low_positive_sample_threshold), + "low_positive_ess": float(low_positive_ess_threshold), + "state_outlier_abs_relative_error": float( + state_outlier_abs_relative_error + ), + }, + "summary": { + "districts": len(rows), + "districts_with_acs_reference": sum( + row["acs_snap_households"] is not None for row in rows + ), + "low_positive_sample_districts": sum( + "low_positive_sample" in row["flags"] for row in rows + ), + "low_positive_ess_districts": sum( + "low_positive_ess" in row["flags"] for row in rows + ), + "state_snap_outlier_districts": sum( + "state_snap_outlier" in row["flags"] for row in rows + ), + "outside_acs_moe_districts": ( + sum(bool(value) for value in outside_moe_values) + if outside_moe_values + else None + ), + "weighted_households": _finite(total_weighted_households), + "weighted_snap_households": _finite(total_weighted_snap_households), + "snap_household_share": _finite( + total_weighted_snap_households / total_weighted_households + if total_weighted_households + else None + ), + "snap_dollars": _finite(total_snap_dollars), + "acs_snap_households": _finite(total_acs_snap_households), + }, + "districts": rows, + } + return json.loads(json.dumps(payload, allow_nan=False)) + + +def write_snap_local_proxy_diagnostics( + payload: Mapping[str, object], path: Path | str +) -> Path: + """Write a SNAP local proxy diagnostics payload as strict JSON.""" + path = Path(path) + path.write_text(json.dumps(payload, indent=1, allow_nan=False)) + return path diff --git a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py index 1b9d5788..a6d492e4 100644 --- a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py +++ b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py @@ -1619,6 +1619,186 @@ def test_fiscal_target_value_basis_uses_only_amount_and_count() -> None: assert builder._fiscal_target_value_basis(bronze_count) == "count" +def _snap_local_proxy_frame(builder, *, include_cd: bool = True) -> Frame: + person = pd.DataFrame( + { + "person_id": np.asarray([1, 2, 3], dtype="int64"), + "person_household_id": np.asarray([1, 1, 2], dtype="int64"), + "person_tax_unit_id": np.asarray([10, 10, 20], dtype="int64"), + "person_spm_unit_id": np.asarray([100, 100, 200], dtype="int64"), + "person_family_id": np.asarray([1000, 1000, 2000], dtype="int64"), + "person_marital_unit_id": np.asarray([10000, 10000, 20000], dtype="int64"), + "SPM_SNAPSUB": np.asarray([100.0, 100.0, 0.0]), + } + ) + household = pd.DataFrame( + { + "household_id": np.asarray([1, 2], dtype="int64"), + "state_fips": np.asarray([6, 36], dtype="int64"), + } + ) + if include_cd: + # Bare 4-digit SSDD geoids, as the #277 geography spine persists them. + household["congressional_district_geoid"] = ["0601", "3601"] + return Frame( + { + "person": person, + "household": household, + "tax_unit": pd.DataFrame( + {"tax_unit_id": np.asarray([10, 20], dtype="int64")} + ), + "spm_unit": pd.DataFrame( + {"spm_unit_id": np.asarray([100, 200], dtype="int64")} + ), + "family": pd.DataFrame( + {"family_id": np.asarray([1000, 2000], dtype="int64")} + ), + "marital_unit": pd.DataFrame( + {"marital_unit_id": np.asarray([10000, 20000], dtype="int64")} + ), + }, + builder.US_SCHEMA, + { + "household": builder.Weights( + values=np.asarray([10.0, 5.0]), + kind=WeightKind.CALIBRATED, + ) + }, + ) + + +def _snap_local_proxy_facts() -> tuple[dict[str, object], ...]: + base = { + # Census summary-level prefixed geography id; the builder must + # normalize this to the spine's bare "0601" before joining. + "geography": {"level": "congressional_district", "id": "5001700US0601"}, + "source": { + "source_name": "census_acs", + "source_table": "ACS S2201 congressional district SNAP", + "source_file": ("census-acs-s2201-congressional-district-snap-2024"), + }, + "observed_measure": { + "source_name": "census_acs", + "source_measure_id": "snap_households", + "source_concept": "acs.s2201.snap_households", + }, + "layout": {"measure_id": "snap_households"}, + } + estimate = { + **base, + "lineage": { + "source_record_id": "census_acs.s2201.ca_01.snap_households.estimate" + }, + "value": 8.0, + } + margin = { + **base, + "lineage": { + "source_record_id": ( + "census_acs.s2201.ca_01.snap_households.margin_of_error" + ) + }, + "layout": {"measure_id": "snap_households_moe"}, + "value": 1.0, + } + return (estimate, margin) + + +def test__given_raw_spm_snap_and_acs_reference__then_builder_writes_snap_proxy( + tmp_path, +) -> None: + builder = _load_builder_module() + california_target_name = "usda_snap.fy2024.state_benefits.wro.ca.total_benefits" + new_york_target_name = "usda_snap.fy2024.state_benefits.nero.ny.total_benefits" + # Diagnostics deliberately arrive out of spec order (and with an extra + # unrelated row): state errors must be joined by name, not position. + result = SimpleNamespace( + diagnostics=( + SimpleNamespace(name="irs_soi.unrelated@2024", relative_error=0.99), + SimpleNamespace(name=f"{new_york_target_name}@2024", relative_error=0.01), + SimpleNamespace(name=f"{california_target_name}@2024", relative_error=0.12), + ) + ) + registry = SimpleNamespace( + specs=( + SimpleNamespace( + name=california_target_name, + metadata={"target_role": "snap_total", "state_fips": "06"}, + ), + SimpleNamespace( + name=new_york_target_name, + metadata={"target_role": "snap_total", "state_fips": "36"}, + ), + ) + ) + + result = builder._write_snap_local_proxy( + release_dir=tmp_path, + frame=_snap_local_proxy_frame(builder), + result=result, + registry=registry, + ledger_facts=_snap_local_proxy_facts(), + release_id="populace-us-2024-fixture", + ) + + path = result.path + assert path == tmp_path / "snap_local_proxy.json" + assert result.skipped_reason is None + assert result.missing_inputs == () + payload = json.loads(path.read_text()) + assert payload["classification"] == "validation_only" + assert payload["release_id"] == "populace-us-2024-fixture" + assert payload["validation_source"] == { + "package_alias": "census-acs-s2201-congressional-district-snap-2024", + "source_family": "snap_local_proxy", + "measure": "ACS S2201 congressional-district SNAP household estimates", + } + assert payload["summary"]["districts"] == 2 + assert payload["summary"]["districts_with_acs_reference"] == 1 + assert payload["summary"]["weighted_snap_households"] == 10.0 + assert payload["summary"]["snap_dollars"] == 1000.0 + assert payload["summary"]["outside_acs_moe_districts"] == 1 + + districts = {row["congressional_district"]: row for row in payload["districts"]} + ca = districts["0601"] + assert ca["state"] == "06" + assert ca["weighted_snap_households"] == 10.0 + assert ca["raw_positive_snap_households"] == 1 + assert ca["snap_dollars"] == 1000.0 + assert ca["state_snap_relative_error"] == 0.12 + assert ca["acs_snap_households"] == 8.0 + assert ca["acs_snap_households_moe"] == 1.0 + assert ca["snap_household_difference"] == 2.0 + assert ca["outside_acs_moe"] is True + assert ca["flags"] == [ + "low_positive_ess", + "low_positive_sample", + "outside_acs_moe", + "state_snap_outlier", + ] + assert districts["3601"]["flags"] == [ + "low_positive_sample", + "missing_acs_reference", + ] + + +def test__given_missing_cd_column__then_builder_skips_snap_proxy(tmp_path) -> None: + builder = _load_builder_module() + + result = builder._write_snap_local_proxy( + release_dir=tmp_path, + frame=_snap_local_proxy_frame(builder, include_cd=False), + result=SimpleNamespace(diagnostics=()), + registry=SimpleNamespace(specs=()), + ledger_facts=_snap_local_proxy_facts(), + release_id="populace-us-2024-fixture", + ) + + assert result.path is None + assert result.skipped_reason == "missing required input columns" + assert result.missing_inputs == ("household.congressional_district_geoid",) + + def test_release_calibration_diagnostics_include_gate_failures( monkeypatch, tmp_path ) -> None: @@ -3864,6 +4044,9 @@ def test_build_manifests_emits_policyengine_certifiable_release_manifest( (artifact_root / builder.CALIBRATION_FILENAME).write_bytes(b"npz") (release_dir / "calibration_diagnostics.json").write_text("{}") (release_dir / "us_source_coverage.json").write_text("{}") + (release_dir / builder.SNAP_LOCAL_PROXY_FILENAME).write_text( + json.dumps({"schema_version": 1, "classification": "validation_only"}) + ) monkeypatch.setattr( builder, @@ -3993,6 +4176,11 @@ def __len__(self): ) assert manifest["data_package"] == {"name": "populace-data", "version": "0.1.0"} assert manifest["default_datasets"] == {"national": "populace_us_2024"} + assert manifest["artifacts"]["snap_local_proxy"]["kind"] == "diagnostics" + assert ( + manifest["artifacts"]["snap_local_proxy"]["path"] + == builder.SNAP_LOCAL_PROXY_FILENAME + ) assert manifest["build"]["built_with_model_package"] == { "name": "policyengine-us", "version": "1.729.0", diff --git a/packages/populace-build/tests/test_us_snap_local_proxy.py b/packages/populace-build/tests/test_us_snap_local_proxy.py new file mode 100644 index 00000000..22f73463 --- /dev/null +++ b/packages/populace-build/tests/test_us_snap_local_proxy.py @@ -0,0 +1,118 @@ +import json + +import pandas as pd +import pytest + +from populace.build.us_runtime.snap_local_proxy import ( + snap_local_proxy_diagnostics, + write_snap_local_proxy_diagnostics, +) + + +def test_snap_local_proxy_diagnostics_reports_validation_only_payload() -> None: + household_frame = pd.DataFrame( + { + "district": ["CA-01", "CA-01", "CA-01", "NY-01"], + "state": ["CA", "CA", "CA", "NY"], + "weight": [10.0, 20.0, 30.0, 5.0], + "snap_receipt": [1, 0, 1, 1], + "snap_amount": [100.0, 0.0, 200.0, 50.0], + "state_snap_relative_error": [0.12, 0.12, 0.12, 0.01], + } + ) + acs_reference = pd.DataFrame( + { + "district": ["CA-01", "NY-01"], + "acs_snap_households": [30.0, 5.0], + "acs_snap_households_moe": [5.0, 10.0], + } + ) + + diagnostics = snap_local_proxy_diagnostics( + household_frame, + district_column="district", + weight_column="weight", + snap_receipt_column="snap_receipt", + snap_amount_column="snap_amount", + state_column="state", + state_snap_relative_error_column="state_snap_relative_error", + acs_reference=acs_reference, + acs_snap_households_moe_column="acs_snap_households_moe", + low_positive_sample_threshold=3, + low_positive_ess_threshold=2.0, + state_outlier_abs_relative_error=0.10, + validation_source={ + "package_alias": "census-acs-s2201-congressional-district-snap-2024" + }, + ) + + assert diagnostics["classification"] == "validation_only" + assert diagnostics["source_family"] == "snap_local_proxy" + assert diagnostics["summary"]["districts"] == 2 + assert diagnostics["summary"]["districts_with_acs_reference"] == 2 + assert diagnostics["summary"]["outside_acs_moe_districts"] == 1 + assert diagnostics["summary"]["weighted_snap_households"] == 45.0 + assert diagnostics["summary"]["snap_dollars"] == 7_250.0 + assert diagnostics["validation_source"] == { + "package_alias": "census-acs-s2201-congressional-district-snap-2024" + } + + ca = diagnostics["districts"][0] + assert ca["congressional_district"] == "CA-01" + assert ca["weighted_households"] == 60.0 + assert ca["weighted_snap_households"] == 40.0 + assert ca["raw_positive_snap_households"] == 2 + assert ca["positive_snap_ess"] == pytest.approx(1.6) + assert ca["positive_snap_max_to_mean_weight_ratio"] == pytest.approx(1.5) + assert ca["snap_household_difference"] == 10.0 + assert ca["snap_household_relative_error"] == pytest.approx(1 / 3) + assert ca["outside_acs_moe"] is True + assert ca["flags"] == [ + "low_positive_ess", + "low_positive_sample", + "outside_acs_moe", + "state_snap_outlier", + ] + + +def test_snap_local_proxy_diagnostics_requires_positive_weight_district_rows() -> None: + household_frame = pd.DataFrame( + {"district": ["CA-01"], "weight": [0.0], "snap_receipt": [1]} + ) + + with pytest.raises(ValueError, match="positive-weight district rows"): + snap_local_proxy_diagnostics( + household_frame, + district_column="district", + weight_column="weight", + snap_receipt_column="snap_receipt", + ) + + +def test_snap_local_proxy_diagnostics_rejects_duplicate_acs_rows() -> None: + household_frame = pd.DataFrame( + {"district": ["CA-01"], "weight": [1.0], "snap_receipt": [1]} + ) + acs_reference = pd.DataFrame( + { + "district": ["CA-01", "CA-01"], + "acs_snap_households": [1.0, 2.0], + } + ) + + with pytest.raises(ValueError, match="duplicate district rows"): + snap_local_proxy_diagnostics( + household_frame, + district_column="district", + weight_column="weight", + snap_receipt_column="snap_receipt", + acs_reference=acs_reference, + ) + + +def test_write_snap_local_proxy_diagnostics_writes_strict_json(tmp_path) -> None: + payload = {"schema_version": 1, "classification": "validation_only"} + path = write_snap_local_proxy_diagnostics( + payload, tmp_path / "snap_local_proxy.json" + ) + assert json.loads(path.read_text()) == payload diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index 21dfd9af..7f6753e0 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -51,12 +51,18 @@ compile_us_fiscal_target_registry, hard_target_package_aliases, load_congressional_district_vintage_crosswalk, + snap_local_proxy_diagnostics, us_immigration_composition_gate, us_source_coverage_diagnostics, us_source_operation_handlers, with_us_immigration_inputs, + write_snap_local_proxy_diagnostics, write_us_source_coverage_diagnostics, ) +from populace.build.us_runtime.congressional_district_geography import ( + CONGRESSIONAL_DISTRICT_GEOID_COLUMN, + _parse_congressional_district_geoid, +) from populace.build.us_runtime.demographics import ( CENSUS_NATIONAL_AGE_BENCHMARK, demographics_payload, @@ -88,6 +94,9 @@ REPO_ID = "policyengine/populace-us" DATASET_FILENAME = "populace_us_2024.h5" CALIBRATION_FILENAME = "populace_us_2024_calibration.npz" +SNAP_LOCAL_PROXY_FILENAME = "snap_local_proxy.json" +SNAP_LOCAL_PROXY_PACKAGE_ALIAS = "census-acs-s2201-congressional-district-snap-2024" +RAW_SPM_SNAP_COLUMN = "SPM_SNAPSUB" POST_EXPORT_ABSOLUTE_TOLERANCE = 1_000_000.0 POST_EXPORT_RELATIVE_TOLERANCE = 5e-4 US_FISCAL_TARGET_LOSS_WEIGHTING = ( @@ -272,6 +281,22 @@ def _automatic_gc_suspended(): }, ) + +class _SnapLocalProxyWriteResult: + __slots__ = ("missing_inputs", "path", "skipped_reason") + + def __init__( + self, + *, + path: Path | None = None, + skipped_reason: str | None = None, + missing_inputs: tuple[str, ...] = (), + ) -> None: + self.path = path + self.skipped_reason = skipped_reason + self.missing_inputs = missing_inputs + + DIRECT_ACTIVE_ALIASES = ( "census-pep-2024-national-age-sex", "census-pep-2024-state-age-sex", @@ -620,6 +645,14 @@ def _parse_args() -> argparse.Namespace: action="store_true", help="Do not emit demographics.json (weighted population by age) for this release.", ) + parser.add_argument( + "--skip-snap-local-proxy", + action="store_true", + help=( + "Do not emit snap_local_proxy.json (validation-only congressional " + "district SNAP support diagnostics) for this release." + ), + ) parser.add_argument( "--staging-dir", type=Path, @@ -4067,6 +4100,316 @@ def _write_demographics( write_demographics(payload, release_dir / "demographics.json") +def _text_value(value: object) -> str: + if isinstance(value, bytes): + return value.decode().strip() + if pd.isna(value): + return "" + return str(value).strip() + + +def _fact_at(fact: object, *path: str) -> object: + current = fact + for key in path: + if current is None: + return None + if isinstance(current, Mapping): + current = current.get(key) + else: + current = getattr(current, key, None) + return current + + +def _fact_text(fact: object, *path: str) -> str: + return _text_value(_fact_at(fact, *path)) + + +def _fact_numeric_value(fact: object) -> float | None: + try: + value = float(_fact_at(fact, "value")) + except (TypeError, ValueError): + return None + return value if math.isfinite(value) else None + + +def _raw_spm_snap_by_household(frame: Frame) -> np.ndarray | None: + person = frame.table("person") + household = frame.table("household") + required = {"person_household_id", RAW_SPM_SNAP_COLUMN} + if not required.issubset(person.columns): + return None + + snap = ( + pd.to_numeric(person[RAW_SPM_SNAP_COLUMN], errors="coerce") + .fillna(0.0) + .clip(lower=0.0) + .to_numpy(dtype=np.float64) + ) + household_ids = household["household_id"].to_numpy() + if "person_spm_unit_id" in person.columns: + spm = pd.DataFrame( + { + "spm_unit_id": person["person_spm_unit_id"].to_numpy(), + "household_id": person["person_household_id"].to_numpy(), + "snap": snap, + } + ) + household_counts = spm.groupby("spm_unit_id")["household_id"].nunique() + ambiguous = household_counts[household_counts != 1] + if not ambiguous.empty: + raise ValueError( + "SPM units must be nested in households for SNAP local proxy " + f"diagnostics; examples: {ambiguous.index[:5].tolist()}." + ) + by_unit = spm.groupby("spm_unit_id", sort=False).agg( + household_id=("household_id", "first"), + snap=("snap", "max"), + ) + by_household = by_unit.groupby("household_id")["snap"].sum() + else: + by_household = ( + pd.DataFrame( + { + "household_id": person["person_household_id"].to_numpy(), + "snap": snap, + } + ) + .groupby("household_id")["snap"] + .sum() + ) + return by_household.reindex(household_ids).fillna(0.0).to_numpy(dtype=np.float64) + + +def _state_snap_relative_errors(result, registry: TargetRegistry) -> dict[str, float]: + # Diagnostics are keyed by name (``{spec.name}@{PERIOD}``), never by + # position: calibration can skip targets, so a positional zip would + # silently misattribute every state error after the first skip. + diagnostics_by_name = { + getattr(diagnostic, "name", None): diagnostic + for diagnostic in getattr(result, "diagnostics", ()) + } + errors: dict[str, float] = {} + for spec in registry.specs: + metadata = getattr(spec, "metadata", {}) + if not isinstance(metadata, Mapping): + continue + if metadata.get("target_role") != "snap_total": + continue + state_fips = metadata.get("state_fips") + if not state_fips: + continue + diagnostic = diagnostics_by_name.get(f"{getattr(spec, 'name', None)}@{PERIOD}") + if diagnostic is None: + continue + relative_error = getattr(diagnostic, "relative_error", None) + if relative_error is None: + target = getattr(diagnostic, "target", None) + final_estimate = getattr(diagnostic, "final_estimate", None) + if target in (None, 0) or final_estimate is None: + continue + relative_error = (float(final_estimate) - float(target)) / float(target) + errors[str(state_fips).zfill(2)] = float(relative_error) + return errors + + +def _missing_snap_local_proxy_inputs(frame: Frame) -> tuple[str, ...]: + person = frame.table("person") + household = frame.table("household") + required = { + "household": (CONGRESSIONAL_DISTRICT_GEOID_COLUMN, "state_fips"), + "person": ("person_household_id", RAW_SPM_SNAP_COLUMN), + } + missing: list[str] = [] + for entity, columns in required.items(): + table = household if entity == "household" else person + missing.extend( + f"{entity}.{column}" for column in columns if column not in table.columns + ) + return tuple(missing) + + +def _snap_local_proxy_household_frame( + frame: Frame, + *, + result, + registry: TargetRegistry, +) -> pd.DataFrame | None: + if _missing_snap_local_proxy_inputs(frame): + return None + household = frame.table("household") + snap_amount = _raw_spm_snap_by_household(frame) + if snap_amount is None: + return None + + state_fips = _state_fips_text(household["state_fips"]) + state_errors = _state_snap_relative_errors(result, registry) + district = [ + (_normalize_congressional_district_geoid(value) or value or None) + for value in ( + _text_value(value) + for value in household[CONGRESSIONAL_DISTRICT_GEOID_COLUMN].to_numpy() + ) + ] + return pd.DataFrame( + { + "congressional_district": district, + "state_fips": state_fips, + "household_weight": frame.weights_for("household").values, + "raw_spm_snap_receipt": snap_amount > 0, + "raw_spm_snap_amount": snap_amount, + "state_snap_relative_error": [ + state_errors.get(state_fip) for state_fip in state_fips + ], + } + ) + + +def _is_snap_local_proxy_fact(fact: object) -> bool: + source_record_id = _fact_text(fact, "lineage", "source_record_id") + source_name = _fact_text(fact, "observed_measure", "source_name") or _fact_text( + fact, "source", "source_name" + ) + source_file = _fact_text(fact, "source", "source_file") + source_table = _fact_text(fact, "source", "source_table") + geography_level = _fact_text(fact, "geography", "level") + geography_id = _fact_text(fact, "geography", "id") + source_haystack = " ".join( + ( + source_record_id, + source_name, + source_file, + source_table, + geography_level, + geography_id, + ) + ).lower() + measure_haystack = " ".join( + ( + source_record_id, + _fact_text(fact, "observed_measure", "source_measure_id"), + _fact_text(fact, "observed_measure", "source_concept"), + _fact_text(fact, "layout", "measure_id"), + _fact_text(fact, "layout", "groupby_dimension"), + _fact_text(fact, "layout", "groupby_value_id"), + ) + ).lower() + return ( + ( + "s2201" in source_haystack + or SNAP_LOCAL_PROXY_PACKAGE_ALIAS in source_haystack + ) + and ("congressional" in source_haystack or "district" in source_haystack) + and ("snap" in measure_haystack or "food_stamp" in measure_haystack) + and "household" in measure_haystack + ) + + +def _is_acs_margin_of_error_fact(fact: object) -> bool: + haystack = " ".join( + ( + _fact_text(fact, "lineage", "source_record_id"), + _fact_text(fact, "observed_measure", "source_measure_id"), + _fact_text(fact, "observed_measure", "source_concept"), + _fact_text(fact, "layout", "measure_id"), + ) + ).lower() + return "margin_of_error" in haystack or "moe" in haystack + + +def _normalize_congressional_district_geoid(value: str) -> str | None: + """Normalize a CD identifier to the spine's bare 4-digit SSDD geoid. + + Household spine columns carry bare geoids (``"0601"``); Census fact + geography ids carry summary-level prefixes (``"5001700US0601"`` / + ``"5001900US0601"``). Both must land in the same vocabulary or every + district flags ``missing_acs_reference``. + """ + geoid = _parse_congressional_district_geoid(value) + if geoid is not None: + return geoid + if len(value) == 4 and value.isdigit(): + return value + return None + + +def _acs_s2201_snap_reference(ledger_facts: Iterable[object]) -> pd.DataFrame | None: + rows: dict[str, dict[str, float | str | None]] = {} + for fact in ledger_facts: + if not _is_snap_local_proxy_fact(fact): + continue + district = _normalize_congressional_district_geoid( + _fact_text(fact, "geography", "id") + ) + value = _fact_numeric_value(fact) + if not district or value is None: + continue + row = rows.setdefault( + district, + { + "congressional_district": district, + "acs_snap_households": None, + "acs_snap_households_moe": None, + }, + ) + if _is_acs_margin_of_error_fact(fact): + row["acs_snap_households_moe"] = value + else: + row["acs_snap_households"] = value + + records = [ + row for row in rows.values() if row.get("acs_snap_households") is not None + ] + if not records: + return None + return pd.DataFrame(records) + + +def _write_snap_local_proxy( + *, + release_dir: Path, + frame: Frame, + result, + registry: TargetRegistry, + ledger_facts: Iterable[object], + release_id: str, +) -> _SnapLocalProxyWriteResult: + missing_inputs = _missing_snap_local_proxy_inputs(frame) + if missing_inputs: + return _SnapLocalProxyWriteResult( + skipped_reason="missing required input columns", + missing_inputs=missing_inputs, + ) + household_frame = _snap_local_proxy_household_frame( + frame, + result=result, + registry=registry, + ) + if household_frame is None: + return _SnapLocalProxyWriteResult(skipped_reason="no household SNAP proxy rows") + payload = snap_local_proxy_diagnostics( + household_frame, + district_column="congressional_district", + weight_column="household_weight", + snap_receipt_column="raw_spm_snap_receipt", + snap_amount_column="raw_spm_snap_amount", + state_column="state_fips", + state_snap_relative_error_column="state_snap_relative_error", + acs_reference=_acs_s2201_snap_reference(ledger_facts), + acs_snap_households_moe_column="acs_snap_households_moe", + validation_source={ + "package_alias": SNAP_LOCAL_PROXY_PACKAGE_ALIAS, + "source_family": "snap_local_proxy", + "measure": "ACS S2201 congressional-district SNAP household estimates", + }, + ) + payload["release_id"] = release_id + return _SnapLocalProxyWriteResult( + path=write_snap_local_proxy_diagnostics( + payload, release_dir / SNAP_LOCAL_PROXY_FILENAME + ) + ) + + def _build_manifests( *, release_id: str, @@ -4306,6 +4649,18 @@ def _build_manifests( if (release_dir / "demographics.json").exists() else {} ), + **( + { + "snap_local_proxy": _artifact_entry( + SNAP_LOCAL_PROXY_FILENAME, + _sha256(release_dir / SNAP_LOCAL_PROXY_FILENAME), + kind="diagnostics", + revision=release_id, + ) + } + if (release_dir / SNAP_LOCAL_PROXY_FILENAME).exists() + else {} + ), }, } (release_dir / "release_manifest.json").write_text( @@ -4443,8 +4798,9 @@ def main() -> None: _assert_cd_vintage_support_matches( base_h5, congressional_district_vintage_crosswalk_metadata ) + ledger_facts = _load_ledger_facts(args.ledger_facts) target_registry = compile_us_fiscal_target_registry( - _load_ledger_facts(args.ledger_facts), + ledger_facts, target_period=PERIOD, include_congressional_district_targets=( args.include_congressional_district_targets @@ -4978,6 +5334,31 @@ def main() -> None: if telemetry is not None: telemetry.attach_artifact("demographics", release_dir / "demographics.json") + if not args.skip_snap_local_proxy: + if telemetry is not None: + telemetry.stage( + "snap_local_proxy", + message="Writing SNAP local proxy diagnostics.", + ) + snap_local_proxy_result = _write_snap_local_proxy( + release_dir=release_dir, + frame=export_frame, + result=result, + registry=registry, + ledger_facts=ledger_facts, + release_id=release_id, + ) + if telemetry is not None and snap_local_proxy_result.path is not None: + telemetry.attach_artifact("snap_local_proxy", snap_local_proxy_result.path) + elif telemetry is not None: + telemetry.stage( + "snap_local_proxy", + message="Skipped SNAP local proxy diagnostics.", + skipped=True, + skipped_reason=snap_local_proxy_result.skipped_reason, + missing_inputs=snap_local_proxy_result.missing_inputs, + ) + if telemetry is not None: telemetry.stage( "source_coverage", message="Writing source coverage diagnostics."