Normalize SSI calibration targets for payment timing

Author: MaxGhenis

changelog.d/1109.changed

@@ -0,0 +1 @@
+Normalized SSI amount calibration targets for federal fiscal-year payment timing.

policyengine_us_data/db/etl_national_targets.py

@@ -27,10 +27,13 @@
     get_geographic_strata,
 )
 from policyengine_us_data.utils.ssi_targets import (
+    SSI_PAYMENT_TARGET_SOURCE,
     SSI_RECIPIENT_TARGET_NOTES,
     SSI_RECIPIENT_TARGET_SOURCE,
     SSI_RECIPIENT_TARGET_YEAR,
     SSI_RECIPIENT_TARGETS_2024,
+    get_ssi_payment_target_notes,
+    normalize_ssi_payment_target,
 )
 from policyengine_us_data.utils.target_variables import (
     target_variable_components,
@@ -767,12 +770,18 @@ def extract_national_targets(year: int = DEFAULT_YEAR):
             value = tax_benefit_system.parameters(
                 time_period
             ).calibration.gov.cbo._children[param_name]
+            source = "CBO Budget Projections"
+            notes = f"CBO projection for {variable_name}"
+            if variable_name == "ssi":
+                value = normalize_ssi_payment_target(value, time_period)
+                source = SSI_PAYMENT_TARGET_SOURCE
+                notes = get_ssi_payment_target_notes(time_period)
             cbo_targets.append(
                 {
                     "variable": variable_name,
                     "value": float(value),
-                    "source": "CBO Budget Projections",
-                    "notes": f"CBO projection for {variable_name}",
+                    "source": source,
+                    "notes": notes,
                     "year": time_period,
                 }
             )

policyengine_us_data/utils/loss.py

@@ -26,7 +26,10 @@
 )
 from policyengine_core.reforms import Reform
 from policyengine_us_data.utils.soi import pe_to_soi, get_soi, get_tracked_soi_row
-from policyengine_us_data.utils.ssi_targets import SSI_RECIPIENT_TARGETS_2024
+from policyengine_us_data.utils.ssi_targets import (
+    SSI_RECIPIENT_TARGETS_2024,
+    normalize_ssi_payment_target,
+)
 from policyengine_us_data.utils.target_variables import (
     target_variable_components,
 )
@@ -93,6 +96,18 @@
     ),
 ]
 
+CBO_PROGRAMS = [
+    "income_tax_positive",
+    "snap",
+    "social_security",
+    "ssi",
+    "unemployment_compensation",
+]
+
+CBO_PARAM_NAME_MAP = {
+    "income_tax_positive": "income_tax",
+}
+
 HARD_CODED_TOTALS = {
     MEDICARE_PART_B_PREMIUM_VARIABLE: (
         get_beneficiary_paid_medicare_part_b_premiums_target(2024)
@@ -233,6 +248,16 @@ def _add_ssi_recipient_targets(loss_matrix, targets_array, sim, time_period):
     return targets_array, loss_matrix
 
 
+def _cbo_program_target_value(sim, variable_name: str, time_period):
+    param_name = CBO_PARAM_NAME_MAP.get(variable_name, variable_name)
+    value = sim.tax_benefit_system.parameters(
+        time_period
+    ).calibration.gov.cbo._children[param_name]
+    if variable_name == "ssi":
+        return normalize_ssi_payment_target(value, time_period)
+    return value
+
+
 ACA_SPENDING_TARGETS = {
     2024: 98e9,
 }
@@ -1316,30 +1341,12 @@ def build_loss_matrix(dataset: type, time_period):
     # refundable credit payments in excess of liability are classified as
     # outlays, not negative receipts. See: https://www.cbo.gov/publication/43767
 
-    CBO_PROGRAMS = [
-        "income_tax_positive",
-        "snap",
-        "social_security",
-        "ssi",
-        "unemployment_compensation",
-    ]
-
-    # Mapping from variable name to CBO parameter name (when different)
-    CBO_PARAM_NAME_MAP = {
-        "income_tax_positive": "income_tax",
-    }
-
     for variable_name in CBO_PROGRAMS:
         label = f"nation/cbo/{variable_name}"
         loss_matrix[label] = sim.calculate(variable_name, map_to="household").values
         if any(loss_matrix[label].isna()):
             raise ValueError(f"Missing values for {label}")
-        param_name = CBO_PARAM_NAME_MAP.get(variable_name, variable_name)
-        targets_array.append(
-            sim.tax_benefit_system.parameters(
-                time_period
-            ).calibration.gov.cbo._children[param_name]
-        )
+        targets_array.append(_cbo_program_target_value(sim, variable_name, time_period))
 
     targets_array, loss_matrix = _add_ssi_recipient_targets(
         loss_matrix,

policyengine_us_data/utils/ssi_targets.py

@@ -1,5 +1,66 @@
 """Shared SSI calibration targets."""
 
+from datetime import date, timedelta
+
+
+SSI_CBO_TARGET_SOURCE = (
+    "https://www.cbo.gov/system/files/2026-02/51313-2026-02-ssi.xlsx"
+)
+SSI_PAYMENT_TIMING_SOURCE = (
+    "https://www.ssa.gov/budget/assets/materials/2026/2026BST.pdf"
+)
+SSI_PAYMENT_TARGET_SOURCE = f"{SSI_CBO_TARGET_SOURCE}; {SSI_PAYMENT_TIMING_SOURCE}"
+
+
+def _as_fiscal_year(year) -> int:
+    return int(str(year)[:4])
+
+
+def _is_first_day_federal_holiday(day: date) -> bool:
+    is_new_years_day = day.month == 1 and day.day == 1
+    is_labor_day = day.month == 9 and day.weekday() == 0 and 1 <= day.day <= 7
+    return is_new_years_day or is_labor_day
+
+
+def _ssi_payment_date(year: int, month: int) -> date:
+    day = date(year, month, 1)
+    while day.weekday() >= 5 or _is_first_day_federal_holiday(day):
+        day -= timedelta(days=1)
+    return day
+
+
+def get_ssi_fiscal_year_payment_count(year) -> int:
+    """Return SSI monthly benefit payments counted in the federal fiscal year."""
+    fiscal_year = _as_fiscal_year(year)
+    start = date(fiscal_year - 1, 10, 1)
+    end = date(fiscal_year, 9, 30)
+    payment_count = 0
+
+    for calendar_year in (fiscal_year - 1, fiscal_year):
+        for month in range(1, 13):
+            payment_day = _ssi_payment_date(calendar_year, month)
+            if start <= payment_day <= end:
+                payment_count += 1
+
+    return payment_count
+
+
+def normalize_ssi_payment_target(value, year) -> float:
+    """Convert fiscal-year SSI outlays to a 12-payment-equivalent target."""
+    payment_count = get_ssi_fiscal_year_payment_count(year)
+    return float(value) * 12 / payment_count
+
+
+def get_ssi_payment_target_notes(year) -> str:
+    payment_count = get_ssi_fiscal_year_payment_count(year)
+    return (
+        "CBO SSI total outlays normalized to a 12-payment-equivalent "
+        "annual target for PolicyEngine's annual SSI variable; "
+        f"FY{_as_fiscal_year(year)} has {payment_count} monthly SSI "
+        "payments under federal budget timing"
+    )
+
+
 SSI_RECIPIENT_TARGET_YEAR = 2024
 SSI_RECIPIENT_TARGET_SOURCE = (
     "https://www.ssa.gov/policy/docs/statcomps/ssi_monthly/2024-12/table01.html"

tests/unit/calibration/test_loss_targets.py

@@ -28,6 +28,7 @@
     _add_real_estate_tax_targets,
     _add_ssi_recipient_targets,
     _add_transfer_balance_targets,
+    _cbo_program_target_value,
     _get_medicaid_national_targets,
     _get_aca_national_targets,
     _load_aca_spending_and_enrollment_targets,
@@ -39,7 +40,11 @@
     get_target_loss_weights,
 )
 from policyengine_us_data.db import etl_national_targets
-from policyengine_us_data.utils.ssi_targets import SSI_RECIPIENT_TARGETS_2024
+from policyengine_us_data.utils.ssi_targets import (
+    SSI_RECIPIENT_TARGETS_2024,
+    get_ssi_fiscal_year_payment_count,
+    normalize_ssi_payment_target,
+)
 
 
 def test_legacy_loss_targets_include_aggregate_qbi_deduction():
@@ -228,6 +233,27 @@ def map_result(self, values, source_entity, target_entity, how=None):
         return np.asarray(values, dtype=np.float32)
 
 
+class _FakeCBOProgramTargetSimulation:
+    def __init__(self):
+        self.tax_benefit_system = SimpleNamespace(
+            parameters=lambda period: SimpleNamespace(
+                calibration=SimpleNamespace(
+                    gov=SimpleNamespace(
+                        cbo=SimpleNamespace(
+                            _children={
+                                "income_tax": 2_000.0,
+                                "snap": 1_000.0,
+                                "social_security": 3_000.0,
+                                "ssi": 57_000_000_000.0,
+                                "unemployment_compensation": 4_000.0,
+                            }
+                        )
+                    )
+                )
+            )
+        )
+
+
 class _FakeCapitalGainsSimulation:
     def __init__(self):
         self.calculate_calls = []
@@ -354,6 +380,28 @@ def test_add_ssi_recipient_targets_adds_total_and_age_counts():
     )
 
 
+def test_ssi_payment_targets_normalize_fiscal_year_payment_timing():
+    assert get_ssi_fiscal_year_payment_count(2024) == 11
+    assert get_ssi_fiscal_year_payment_count(2025) == 12
+    assert get_ssi_fiscal_year_payment_count(2028) == 13
+
+    assert normalize_ssi_payment_target(57_000_000_000, 2024) == pytest.approx(
+        57_000_000_000 * 12 / 11
+    )
+    assert normalize_ssi_payment_target(75_400_000_000, 2028) == pytest.approx(
+        75_400_000_000 * 12 / 13
+    )
+
+
+def test_legacy_cbo_ssi_target_uses_12_payment_equivalent():
+    sim = _FakeCBOProgramTargetSimulation()
+
+    assert _cbo_program_target_value(sim, "ssi", 2024) == pytest.approx(
+        57_000_000_000 * 12 / 11
+    )
+    assert _cbo_program_target_value(sim, "snap", 2024) == 1_000.0
+
+
 def test_add_ctc_targets(monkeypatch):
     monkeypatch.setattr(
         "policyengine_us_data.utils.loss.get_national_geography_soi_target",

tests/unit/calibration/test_target_config.py

@@ -546,6 +546,7 @@ def test_training_config_includes_ssi_recipient_count_targets(self):
         )
 
         include_rules = config["include"]
+        assert {"variable": "ssi", "geo_level": "national"} in include_rules
         assert {
             "variable": "person_count",
             "geo_level": "national",

tests/unit/test_etl_national_targets.py

@@ -1,6 +1,7 @@
 import inspect
 
 import pandas as pd
+import pytest
 from sqlalchemy import text
 from sqlmodel import Session, select
 
@@ -19,7 +20,10 @@
     load_national_targets,
     load_state_acs_rent_targets,
 )
-from policyengine_us_data.utils.ssi_targets import SSI_RECIPIENT_TARGETS_2024
+from policyengine_us_data.utils.ssi_targets import (
+    SSI_PAYMENT_TARGET_SOURCE,
+    SSI_RECIPIENT_TARGETS_2024,
+)
 
 
 def test_national_targets_do_not_extract_treasury_eitc():
@@ -439,6 +443,57 @@ def test_extract_national_targets_includes_ssi_count_targets():
     }
 
 
+def test_extract_national_targets_normalizes_ssi_cbo_amount_target(monkeypatch):
+    class FakeIncomeBySource:
+        _children = {
+            target["parameter"]: 0
+            for target in etl_national_targets.CBO_INCOME_BY_SOURCE_TARGETS
+        }
+
+    class FakeCBO:
+        income_by_source = FakeIncomeBySource()
+        _children = {
+            "income_tax": 0,
+            "snap": 0,
+            "social_security": 0,
+            "ssi": 57_000_000_000,
+            "unemployment_compensation": 0,
+        }
+
+    class FakeSOI:
+        _children = {"long_term_capital_gains": 0}
+
+    class FakeGov:
+        cbo = FakeCBO()
+        irs = type("FakeIRS", (), {"soi": FakeSOI()})()
+
+    class FakeCalibration:
+        gov = FakeGov()
+
+    class FakeParameters:
+        def __call__(self, year):
+            return self
+
+        calibration = FakeCalibration()
+
+    class FakeTaxBenefitSystem:
+        parameters = FakeParameters()
+
+    monkeypatch.setattr(
+        "policyengine_us.CountryTaxBenefitSystem",
+        FakeTaxBenefitSystem,
+    )
+
+    raw_targets = extract_national_targets(year=2024)
+    ssi_target = next(
+        target for target in raw_targets["cbo_targets"] if target["variable"] == "ssi"
+    )
+
+    assert ssi_target["value"] == pytest.approx(57_000_000_000 * 12 / 11)
+    assert ssi_target["source"] == SSI_PAYMENT_TARGET_SOURCE
+    assert "12-payment-equivalent" in ssi_target["notes"]
+
+
 def test_load_national_targets_uses_medicaid_enrolled_for_enrollment_counts(
     tmp_path, monkeypatch
 ):