Target SSI fiscal-year outlays in calibration

changelog.d/1109.changed

@@ -0,0 +1 @@
+Targeted single-year-compatible SSI federal fiscal-year outlays in calibration.

policyengine_us_data/calibration/sanity_checks.py

@@ -36,6 +36,10 @@
     "income_tax_before_credits",
 ]
 
+COMPUTED_KEY_MONETARY_VARS = [
+    "ssi_federal_fiscal_year_outlays",
+]
+
 TAKEUP_VARS = [
     "takes_up_snap_if_eligible",
     "takes_up_ssi_if_eligible",
@@ -375,6 +379,27 @@ def _get_person_weights(f, period, person_count, household_weights):
         except KeyError:
             return None
 
+    def _append_finite_check(var: str, vals) -> None:
+        vals = np.asarray(vals)
+        n_nan = int(np.isnan(vals).sum())
+        n_inf = int(np.isinf(vals).sum())
+        if n_nan > 0 or n_inf > 0:
+            results.append(
+                {
+                    "check": f"no_nan_inf_{var}",
+                    "status": "FAIL",
+                    "detail": f"{n_nan} NaN, {n_inf} Inf",
+                }
+            )
+        else:
+            results.append(
+                {
+                    "check": f"no_nan_inf_{var}",
+                    "status": "PASS",
+                    "detail": "",
+                }
+            )
+
     with h5py.File(h5_path, "r") as f:
         # 1. Weight non-negativity
         w_key = f"household_weight/{period}"
@@ -440,24 +465,7 @@ def _get_person_weights(f, period, person_count, household_weights):
             vals = _get(f, f"{var}/{period}")
             if vals is None:
                 continue
-            n_nan = int(np.isnan(vals).sum())
-            n_inf = int(np.isinf(vals).sum())
-            if n_nan > 0 or n_inf > 0:
-                results.append(
-                    {
-                        "check": f"no_nan_inf_{var}",
-                        "status": "FAIL",
-                        "detail": f"{n_nan} NaN, {n_inf} Inf",
-                    }
-                )
-            else:
-                results.append(
-                    {
-                        "check": f"no_nan_inf_{var}",
-                        "status": "PASS",
-                        "detail": "",
-                    }
-                )
+            _append_finite_check(var, vals)
 
         # 4. Person-to-household mapping
         person_hh_arr = _get(f, f"person_household_id/{period}")
@@ -650,9 +658,35 @@ def _get_person_weights(f, period, person_count, household_weights):
                 )
             )
 
+    for var, vals in _computed_key_monetary_values(h5_path, period).items():
+        _append_finite_check(var, vals)
+
     return results
 
 
+def _computed_key_monetary_values(h5_path: str, period: int) -> dict[str, np.ndarray]:
+    try:
+        from policyengine_us import Microsimulation
+
+        sim = Microsimulation(dataset=h5_path)
+    except Exception as error:
+        logger.info("Skipping computed monetary sanity checks: %s", error)
+        return {}
+
+    values = {}
+    for var in COMPUTED_KEY_MONETARY_VARS:
+        try:
+            result = sim.calculate(var, period)
+            values[var] = np.asarray(
+                result.values if hasattr(result, "values") else result
+            )
+        except Exception as error:
+            logger.info(
+                "Skipping computed monetary sanity check for %s: %s", var, error
+            )
+    return values
+
+
 def main():
     import argparse
 

policyengine_us_data/calibration/target_config.yaml

@@ -205,7 +205,7 @@ include:
     geo_level: national
   - variable: social_security_survivors
     geo_level: national
-  - variable: ssi
+  - variable: ssi_federal_fiscal_year_outlays
     geo_level: national
   - variable: person_count
     geo_level: national

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,
+    scale_ssi_fiscal_year_target_for_single_year_data,
 )
 from policyengine_us_data.utils.target_variables import (
     target_variable_components,
@@ -150,6 +153,33 @@ def _register_target_variable(session: Session, variable: str) -> None:
         session.flush()
 
 
+def _deactivate_replaced_national_target(
+    session: Session,
+    *,
+    stratum_id: int,
+    old_variable: str,
+    new_variable: str,
+    period: int,
+) -> None:
+    old_targets = session.exec(
+        select(Target).where(
+            Target.stratum_id == stratum_id,
+            Target.variable == old_variable,
+            Target.period == period,
+            Target.reform_id == 0,
+            Target.active,
+        )
+    ).all()
+    for target in old_targets:
+        target.active = False
+        replacement_note = (
+            f"Deactivated because {new_variable} replaced this target concept."
+        )
+        target.notes = (
+            f"{target.notes} | {replacement_note}" if target.notes else replacement_note
+        )
+
+
 WIC_NATIONAL_ANNUAL_SUMMARY_SOURCE = (
     "https://www.fns.usda.gov/sites/default/files/resource-files/wisummary-4.xlsx"
 )
@@ -751,13 +781,14 @@ def extract_national_targets(year: int = DEFAULT_YEAR):
         "income_tax_positive",
         "snap",
         "social_security",
-        "ssi",
+        "ssi_federal_fiscal_year_outlays",
         "unemployment_compensation",
     ]
 
     # Mapping from target variable to CBO parameter name (when different)
     cbo_param_name_map = {
         "income_tax_positive": "income_tax",  # CBO param is income_tax
+        "ssi_federal_fiscal_year_outlays": "ssi",
     }
 
     cbo_targets = []
@@ -767,12 +798,20 @@ 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_federal_fiscal_year_outlays":
+                value = scale_ssi_fiscal_year_target_for_single_year_data(
+                    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,
                 }
             )
@@ -912,6 +951,14 @@ def load_national_targets(
         for _, target_data in direct_targets_df.iterrows():
             target_year = target_data["year"]
             _register_target_variable(session, target_data["variable"])
+            if target_data["variable"] == "ssi_federal_fiscal_year_outlays":
+                _deactivate_replaced_national_target(
+                    session,
+                    stratum_id=us_stratum.stratum_id,
+                    old_variable="ssi",
+                    new_variable="ssi_federal_fiscal_year_outlays",
+                    period=target_year,
+                )
             # Check if target already exists
             existing_target = session.exec(
                 select(Target).where(

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,
+    scale_ssi_fiscal_year_target_for_single_year_data,
+)
 from policyengine_us_data.utils.target_variables import (
     target_variable_components,
 )
@@ -93,6 +96,19 @@
     ),
 ]
 
+CBO_PROGRAMS = [
+    "income_tax_positive",
+    "snap",
+    "social_security",
+    "ssi_federal_fiscal_year_outlays",
+    "unemployment_compensation",
+]
+
+CBO_PARAM_NAME_MAP = {
+    "income_tax_positive": "income_tax",
+    "ssi_federal_fiscal_year_outlays": "ssi",
+}
+
 HARD_CODED_TOTALS = {
     MEDICARE_PART_B_PREMIUM_VARIABLE: (
         get_beneficiary_paid_medicare_part_b_premiums_target(2024)
@@ -233,6 +249,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_federal_fiscal_year_outlays":
+        return scale_ssi_fiscal_year_target_for_single_year_data(value, time_period)
+    return value
+
+
 ACA_SPENDING_TARGETS = {
     2024: 98e9,
 }
@@ -1316,30 +1342,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,99 @@
 """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/oact/ssir/SSI24/IV_C_Payments.html"
+SSI_PAYMENT_RULE_SOURCE = "https://www.ssa.gov/OP_Home/cfr20/416/416-0502.htm"
+SSI_PAYMENT_TARGET_SOURCE = (
+    f"{SSI_CBO_TARGET_SOURCE}; {SSI_PAYMENT_TIMING_SOURCE}; {SSI_PAYMENT_RULE_SOURCE}"
+)
+
+
+def _as_fiscal_year(year) -> int:
+    return int(str(year)[:4])
+
+
+def _is_new_years_day_observed(day: date) -> bool:
+    new_years_day = date(day.year, 1, 1)
+    next_new_years_day = date(day.year + 1, 1, 1)
+    return (
+        day == new_years_day
+        or (new_years_day.weekday() == 6 and day == date(day.year, 1, 2))
+        or (next_new_years_day.weekday() == 5 and day == date(day.year, 12, 31))
+    )
+
+
+def _is_labor_day(day: date) -> bool:
+    return day.month == 9 and day.weekday() == 0 and day.day <= 7
+
+
+def _is_federal_holiday_affecting_ssi_payment(day: date) -> bool:
+    return _is_new_years_day_observed(day) or _is_labor_day(day)
+
+
+def _ssi_payment_date(year: int, month: int) -> date:
+    payment_date = date(year, month, 1)
+    while payment_date.weekday() >= 5 or _is_federal_holiday_affecting_ssi_payment(
+        payment_date
+    ):
+        payment_date -= timedelta(days=1)
+    return payment_date
+
+
+def _ssi_fiscal_year_benefit_months(year) -> list[date]:
+    fiscal_year = _as_fiscal_year(year)
+    fiscal_year_start = date(fiscal_year - 1, 10, 1)
+    fiscal_year_end = date(fiscal_year, 9, 30)
+
+    benefit_months = []
+    for calendar_year in (fiscal_year - 1, fiscal_year):
+        for month in range(1, 13):
+            payment_day = _ssi_payment_date(calendar_year, month)
+            if fiscal_year_start <= payment_day <= fiscal_year_end:
+                benefit_months.append(date(calendar_year, month, 1))
+    return benefit_months
+
+
+def get_ssi_fiscal_year_payment_count(year) -> int:
+    """Return SSI benefit months with payment dates in the federal fiscal year."""
+    return len(_ssi_fiscal_year_benefit_months(year))
+
+
+def get_ssi_single_year_available_payment_count(year) -> int:
+    """Return fiscal-year SSI benefit months available from a single-year H5."""
+    fiscal_year = _as_fiscal_year(year)
+    return sum(
+        benefit_month.year == fiscal_year
+        for benefit_month in _ssi_fiscal_year_benefit_months(year)
+    )
+
+
+def scale_ssi_fiscal_year_target_for_single_year_data(value, year) -> float:
+    """Scale full fiscal-year SSI outlays to months computable from one H5 year."""
+    return (
+        float(value)
+        * get_ssi_single_year_available_payment_count(year)
+        / get_ssi_fiscal_year_payment_count(year)
+    )
+
+
+def get_ssi_payment_target_notes(year) -> str:
+    fiscal_year = _as_fiscal_year(year)
+    available_count = get_ssi_single_year_available_payment_count(year)
+    payment_count = get_ssi_fiscal_year_payment_count(year)
+    return (
+        "CBO SSI federal fiscal-year outlays scaled to the benefit months "
+        "computable from a single-year PolicyEngine-US-data H5 using "
+        "policyengine-us ssi_federal_fiscal_year_outlays; "
+        f"FY{fiscal_year} has {payment_count} SSI benefit months paid in the "
+        f"federal fiscal year, of which {available_count} are benefit months "
+        f"in calendar year {fiscal_year}"
+    )
+
+
 SSI_RECIPIENT_TARGET_YEAR = 2024
 SSI_RECIPIENT_TARGET_SOURCE = (
     "https://www.ssa.gov/policy/docs/statcomps/ssi_monthly/2024-12/table01.html"