Populate reported TANF, SNAP, and WIC enrollment from CPS admin-linkage fields (#591)

changelog.d/591-reported-enrollment.added.md

@@ -0,0 +1 @@
+Populate monthly `is_tanf_enrolled` and `receives_snap` inputs from positive annual CPS ASEC reports at SPM-unit grain, preserve the flags across support clones, and transfer them to ACS units through the model-required boolean plan. Gate `is_tanf_enrolled` on the reported program type (`PAW_TYP` 1 or 3), restoring `PAW_TYP` from the pinned official ASEC archives into the raw-stage checkpoint (schema v3), because `PAW_VAL` alone conflates TANF with other cash welfare — in the 2023 ASEC, 271 of 682 PAW-positive SPM units (42.6% SPM-weighted) report no TANF type. Carry annual reported WIC receipt on its adult-female reporter for SPM-unit-only consumption, with a CI guard against person-grain engine consumers.

packages/populace-build/src/populace/build/us_runtime/__init__.py

@@ -159,7 +159,13 @@
     CPS_CARRIED_FORMULA_OWNED_COLUMNS,
     CPS_CARRIED_PERSON_INPUTS,
     CPS_CARRIED_SPM_UNIT_INPUTS,
+    CPS_REPORTED_TANF_AMOUNT_RAW_COLUMN,
+    CPS_REPORTED_TANF_TYPE_RAW_COLUMN,
+    CPS_REPORTED_WIC_RAW_COLUMN,
+    WIC_CARRIER_ADJUDICATION_URL,
     derive_us_cps_carried_inputs,
+    reported_tanf_enrollment_by_spm_unit,
+    reported_wic_receipt_carrier,
 )
 from populace.build.us_runtime.demographics import (
     AGE_BANDS,
@@ -514,6 +520,14 @@
     us_prior_year_income_summary,
     with_us_prior_year_income_inputs,
 )
+from populace.build.us_runtime.public_assistance_type_source import (
+    ASEC_PUBLIC_ASSISTANCE_TYPE_AUDIT_PINS,
+    ASEC_PUBLIC_ASSISTANCE_TYPE_INCOME_YEARS,
+    PAW_TYPE_TANF_CODES,
+    PAW_TYPE_VALID_CODES,
+    fill_asec_public_assistance_type_source,
+    load_asec_public_assistance_type_sources,
+)
 from populace.build.us_runtime.puf_capital_gains_tail import (
     PUF_CAPITAL_GAINS_TAIL_APPLIED_COLUMN,
     PUF_CAPITAL_GAINS_TAIL_DONOR_AGI_BAND_COLUMN,
@@ -1073,6 +1087,10 @@
     "CPS_CARRIED_FORMULA_OWNED_COLUMNS",
     "CPS_CARRIED_PERSON_INPUTS",
     "CPS_CARRIED_SPM_UNIT_INPUTS",
+    "CPS_REPORTED_TANF_AMOUNT_RAW_COLUMN",
+    "CPS_REPORTED_TANF_TYPE_RAW_COLUMN",
+    "CPS_REPORTED_WIC_RAW_COLUMN",
+    "WIC_CARRIER_ADJUDICATION_URL",
     "SimpleTaxExpenditureReform",
     "ReformValidationSpec",
     "REFORM_VALIDATION_SCHEMA_VERSION",
@@ -1435,6 +1453,12 @@
     "fetch_asec_education_assistance_source",
     "fill_asec_education_assistance_source",
     "load_asec_education_assistance_sources",
+    "ASEC_PUBLIC_ASSISTANCE_TYPE_AUDIT_PINS",
+    "ASEC_PUBLIC_ASSISTANCE_TYPE_INCOME_YEARS",
+    "PAW_TYPE_TANF_CODES",
+    "PAW_TYPE_VALID_CODES",
+    "fill_asec_public_assistance_type_source",
+    "load_asec_public_assistance_type_sources",
     "US_EDUCATION_INPUTS_REQUIRED_SOURCE_COLUMNS",
     "US_EDUCATION_INPUTS_STAGE_NAME",
     "derive_us_education_inputs_from_manifest",
@@ -1819,6 +1843,8 @@
     "congressional_district_assignment_summary",
     "congressional_district_distribution_from_ledger_facts",
     "derive_us_cps_carried_inputs",
+    "reported_tanf_enrollment_by_spm_unit",
+    "reported_wic_receipt_carrier",
     "disaggregate_us_puf_aggregate_records_from_manifest",
     "finalize_us_puf_tax_detail_predictions",
     "hard_target_package_aliases",

packages/populace-build/src/populace/build/us_runtime/acs_transfer.py

@@ -284,6 +284,10 @@
             "is_disabled",
             "is_full_time_college_student",
             "is_pregnant",
+            # WICYN's adult-female reporter is only a physical carrier for an
+            # SPM-unit receipt fact. Engine consumers are separately guarded
+            # to aggregate receives_wic at SPM-unit grain (populace#591).
+            "receives_wic",
         ),
         "model_required_discrete": ("own_children_in_household",),
     },
@@ -307,6 +311,9 @@
         # Housing-assistance receipt is source-observed in the raw pool. Other
         # takes_up_* leaves are runtime-owned draws and are not donor targets.
         "benefit_participation": ("takes_up_housing_assistance_if_eligible",),
+        # Reported TANF/SNAP receipt is produced from annual ASEC amounts before
+        # cloning and transferred like the person-level required booleans.
+        "model_required_boolean": ("is_tanf_enrolled", "receives_snap"),
         "model_required_numeric": ("spm_unit_pre_subsidy_childcare_expenses",),
     },
 }

packages/populace-build/src/populace/build/us_runtime/asec_checkpoint.py

@@ -56,7 +56,10 @@
 ASEC_RAW_STAGE_ARTIFACT_KIND = "populace_us_asec_raw_stage"
 ASEC_RAW_STAGE_CHECKPOINT_FILENAME = "asec_raw_stage.checkpoint.h5"
 ASEC_RAW_STAGE_OPERATOR_STATUS = "operator_untouched"
-ASEC_RAW_STAGE_SCHEMA_VERSION = 2
+# Version 3 added the PAW_TYP restoration that gates TANF enrollment; older
+# artifacts lack the gate column and must fail loudly rather than let
+# PAW_VAL-only conflation back in (populace#591).
+ASEC_RAW_STAGE_SCHEMA_VERSION = 3
 ASEC_RAW_STAGE_STAGE = "raw_source_mapping"
 _RAW_STAGE_BINDING_KEYS = frozenset(
     {
@@ -71,9 +74,9 @@
         "stage",
     }
 )
-_RAW_SOURCE_MAPPING_COLUMNS = frozenset({"ED_VAL", "LKWEEKS"})
+_RAW_SOURCE_MAPPING_COLUMNS = frozenset({"ED_VAL", "LKWEEKS", "PAW_TYP"})
 _RAW_STAGE_REQUIRED_PERSON_COLUMNS = frozenset(
-    {"ED_VAL", "LKWEEKS", "PERIDNUM", "source_year"}
+    {"ED_VAL", "LKWEEKS", "PAW_TYP", "PERIDNUM", "source_year"}
 )
 _RAW_SOURCE_MAPPING_KEYS = frozenset(
     {
@@ -183,9 +186,7 @@ def load_asec_raw_stage_checkpoint(
             "source-construction structural identity it declares."
         )
     metadata["identity"] = stored_identity.to_payload()
-    metadata["source_construction_identity"] = (
-        source_construction_identity.to_payload()
-    )
+    metadata["source_construction_identity"] = source_construction_identity.to_payload()
     return loaded.frame, metadata
 
 
@@ -254,9 +255,8 @@ def _validate_raw_stage_binding(
             "ASEC artifact."
         )
     schema_version = metadata["schema_version"]
-    if (
-        schema_version != ASEC_RAW_STAGE_SCHEMA_VERSION
-        or isinstance(schema_version, bool)
+    if schema_version != ASEC_RAW_STAGE_SCHEMA_VERSION or isinstance(
+        schema_version, bool
     ):
         raise ValueError(
             f"ASEC raw-stage checkpoint {path} has an unsupported raw-stage "
@@ -331,8 +331,7 @@ def _validate_source_receipt(receipt: object, *, path: Path) -> None:
 def _validate_raw_source_mappings(mappings: object, *, path: Path) -> None:
     if not isinstance(mappings, Mapping):
         raise ValueError(
-            f"ASEC raw-stage checkpoint {path} raw_source_mappings must be "
-            "an object."
+            f"ASEC raw-stage checkpoint {path} raw_source_mappings must be an object."
         )
     if frozenset(mappings) != _RAW_SOURCE_MAPPING_COLUMNS:
         raise ValueError(
@@ -415,13 +414,10 @@ def _validate_raw_stage_source_columns(frame: Frame, *, path: Path) -> None:
     peridnum = person["PERIDNUM"]
     valid_peridnum = peridnum.notna()
     if pd.api.types.is_string_dtype(peridnum.dtype) or peridnum.dtype == object:
-        valid_peridnum &= (
-            peridnum.astype("string").str.strip().ne("").fillna(False)
-        )
+        valid_peridnum &= peridnum.astype("string").str.strip().ne("").fillna(False)
     if not valid_peridnum.all():
         raise ValueError(
-            f"ASEC raw-stage checkpoint {path} PERIDNUM must be complete and "
-            "nonempty."
+            f"ASEC raw-stage checkpoint {path} PERIDNUM must be complete and nonempty."
         )
 
     education = pd.to_numeric(person["ED_VAL"], errors="coerce").to_numpy(
@@ -433,9 +429,7 @@ def _validate_raw_stage_source_columns(frame: Frame, *, path: Path) -> None:
             "and nonnegative."
         )
 
-    weeks = pd.to_numeric(person["LKWEEKS"], errors="coerce").to_numpy(
-        dtype=np.float64
-    )
+    weeks = pd.to_numeric(person["LKWEEKS"], errors="coerce").to_numpy(dtype=np.float64)
     valid_weeks = np.isfinite(weeks) & np.equal(weeks, np.floor(weeks))
     valid_weeks &= (weeks == -1.0) | ((weeks >= 0.0) & (weeks <= 52.0))
     if not valid_weeks.all():
@@ -444,6 +438,16 @@ def _validate_raw_stage_source_columns(frame: Frame, *, path: Path) -> None:
             "in {-1, 0, ..., 52}."
         )
 
+    paw_type = pd.to_numeric(person["PAW_TYP"], errors="coerce").to_numpy(
+        dtype=np.float64
+    )
+    valid_paw_type = np.isfinite(paw_type) & np.isin(paw_type, (0.0, 1.0, 2.0, 3.0))
+    if not valid_paw_type.all():
+        raise ValueError(
+            f"ASEC raw-stage checkpoint {path} PAW_TYP must be complete integers "
+            "in {0, 1, 2, 3}."
+        )
+
 
 def _validate_asec_frame(
     frame: Frame,