Bind SSA SSI age-band recipient counts as ordinary indicator targets (#470)

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

@@ -2381,6 +2381,8 @@ def _population_age_reference_from_fact(
 #: are its dollar sum — the same variable the national ``ssi_total`` payment
 #: target already materializes.
 _SSA_SSI_BASE_VARIABLE = "ssi"
+_SSA_SSI_BY_AGE_RECORD_SET_TOKEN = ".ssi_federal_payment_recipients.by_age"
+SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE = "ssa_ssi_age_band_recipients"
 SSA_SSI_RECIPIENTS_TARGET_ROLE = "ssi_recipients"
 SSA_SSI_STATE_PAYMENTS_TARGET_ROLE = "ssi_state_payments"
 
@@ -2429,6 +2431,44 @@ def _ssa_ssi_reference_from_fact(
     if _SSA_OASDI_SSI_PAYMENTS_RECORD_SET_TOKEN in record_set_id:
         return _direct_reference_from_fact(fact, target_period=target_period)
 
+    if _SSA_SSI_BY_AGE_RECORD_SET_TOKEN in record_set_id:
+        # SSA SSI Monthly Statistics Table 1: federal-payment recipients by
+        # age group (populace#470). The age-band rows bind as national
+        # indicator counts of engine ``ssi`` receipt sliced by the fact's
+        # first-class age constraints — the ordinary-target replacement for
+        # the retired take-up assignment goals (#469/#473). The all-ages row
+        # never binds: it duplicates the by-area ``all_areas_total`` national
+        # count already compiled under role ``ssi_recipients``.
+        if _measure_id(fact) != "recipient_count":
+            return None
+        if _geography_level(fact) != "country":
+            return None
+        lower, upper = _age_bounds(fact)
+        if lower == "-inf" and upper == "inf":
+            return None
+        source_record_id = _source_record_id(fact)
+        if not source_record_id:
+            return None
+        return LedgerTargetReference(
+            name=source_record_id,
+            ledger_source_record_id=source_record_id,
+            entity="household",
+            measure=source_record_id,
+            period=target_period,
+            family="ssa",
+            metadata={
+                "materializer": "policyengine_variable",
+                "measure_mode": "indicator_sum",
+                "base_variable": _SSA_SSI_BASE_VARIABLE,
+                "target_role": SSA_SSI_AGE_BAND_RECIPIENTS_TARGET_ROLE,
+                "source_measure_id": "recipient_count",
+                "source_period": str(_period_value(fact)),
+                "target_period": str(target_period),
+                "age_lower_bound": lower,
+                "age_upper_bound": upper,
+            },
+        )
+
     measure_id = _measure_id(fact)
     if measure_id == "recipient_count":
         target_role = SSA_SSI_RECIPIENTS_TARGET_ROLE

packages/populace-build/tests/test_us_fiscal_targets.py

@@ -3008,6 +3008,77 @@ def test_age_targets_chains_w2_tips_through_soi_wages_bridge() -> None:
     assert abs(spec.value - 26_786_522_000 * expected_factor) < 1.0
 
 
+def test_ssa_ssi_age_band_counts_bind_as_person_age_indicator_targets() -> None:
+    # populace#470: the SSA SSI Monthly age-band recipient counts bind as
+    # national indicator counts of engine ssi receipt sliced by the fact's
+    # first-class age constraints; the all-ages row never binds (it would
+    # duplicate the by-area all_areas_total national count).
+    band_id = (
+        "ssa_ssi_monthly.month2024_12.ssi_federal_payment_recipients."
+        "by_age.under_18.recipient_count"
+    )
+    all_ages_id = (
+        "ssa_ssi_monthly.month2024_12.ssi_federal_payment_recipients."
+        "by_age.all_ages.recipient_count"
+    )
+    registry = compile_us_fiscal_target_registry(
+        [
+            *packaged_reference_facts(),
+            _dynamic_ledger_fact(
+                source_record_id=band_id,
+                source_name="ssa",
+                measure_id="recipient_count",
+                value=1_001_922,
+                period_value=2024,
+                layout_record_set_id=(
+                    "ssa_ssi_monthly.month2024_12.ssi_federal_payment_recipients.by_age"
+                ),
+                groupby_value_id="under_18",
+                universe_constraints=[
+                    {
+                        "operator": ">=",
+                        "role": "filter",
+                        "unit": "years",
+                        "value": 0,
+                        "variable": "age",
+                    },
+                    {
+                        "operator": "<",
+                        "role": "filter",
+                        "unit": "years",
+                        "value": 18,
+                        "variable": "age",
+                    },
+                ],
+            ),
+            _dynamic_ledger_fact(
+                source_record_id=all_ages_id,
+                source_name="ssa",
+                measure_id="recipient_count",
+                value=7_289_843,
+                period_value=2024,
+                layout_record_set_id=(
+                    "ssa_ssi_monthly.month2024_12.ssi_federal_payment_recipients.by_age"
+                ),
+                groupby_value_id="all_ages",
+            ),
+        ],
+        allow_unaged_dollar_targets=True,
+    )
+
+    specs = {spec.name: spec for spec in registry.specs}
+    spec = specs[band_id]
+    assert spec.family == "ssa"
+    assert spec.metadata["materializer"] == "policyengine_variable"
+    assert spec.metadata["measure_mode"] == "indicator_sum"
+    assert spec.metadata["base_variable"] == "ssi"
+    assert spec.metadata["target_role"] == "ssa_ssi_age_band_recipients"
+    assert spec.metadata["age_lower_bound"] == "0"
+    assert spec.metadata["age_upper_bound"] == "18"
+    assert spec.value == 1_001_922
+    assert all_ages_id not in specs
+
+
 def test_soi_itemized_deduction_targets_require_itemizing() -> None:
     medical_source_record_id = (
         "irs_soi.ty2022.historic_table_2.us.all.medical_dental_expense_returns"

tools/build_us_fiscal_refresh_release.py

@@ -4003,9 +4003,48 @@ def _materialize_target_frame(
     direct_value_cache: dict[
         tuple[tuple[str, ...], str, str, str, str], np.ndarray
     ] = {}
+    person_age_for_bands: np.ndarray | None = None
     for spec in direct_target_specs:
         base_variables = _base_variables_from_metadata(spec.metadata)
         mode = spec.metadata.get("measure_mode", "sum")
+        age_lower = spec.metadata.get("age_lower_bound")
+        age_upper = spec.metadata.get("age_upper_bound")
+        if age_lower is not None or age_upper is not None:
+            # Age-banded person-variable targets (populace#470, the SSA SSI
+            # recipients-by-age counts): mask the person-entity base variable
+            # by the fact's age constraints BEFORE the household collapse —
+            # the state/CD masks below act on household values and cannot
+            # express person-age slices.
+            if any(variable not in system.variables for variable in base_variables):
+                continue
+            for variable in base_variables:
+                if _variable_entity(system, variable) != "person":
+                    raise ValueError(
+                        "Age-banded target "
+                        f"{spec.name!r} requires person-entity base "
+                        f"variables; {variable!r} is "
+                        f"{_variable_entity(system, variable)!r}."
+                    )
+            if person_age_for_bands is None:
+                person_age_for_bands = np.asarray(
+                    _calculate_array(simulation, "age"), dtype=np.float64
+                )
+            person_values = np.zeros_like(person_age_for_bands)
+            for variable in base_variables:
+                person_values = person_values + np.asarray(
+                    _calculate_array(simulation, variable), dtype=np.float64
+                )
+            if mode == "indicator_sum":
+                person_values = (person_values > 0.0).astype(np.float64)
+            band_lower = _as_bound(str(age_lower if age_lower is not None else "-inf"))
+            band_upper = _as_bound(str(age_upper if age_upper is not None else "inf"))
+            band_mask = (person_age_for_bands >= band_lower) & (
+                person_age_for_bands < band_upper
+            )
+            hh[spec.measure] = _collapse_person(
+                base_frame, person_values * band_mask.astype(np.float64)
+            )
+            continue
         map_to = spec.metadata.get("indicator_map_to")
         filter_variable = spec.metadata.get("indicator_filter_variable")
         less_than = _less_than_from_metadata(spec.metadata)