Restore strike_benefits from the OI_VAL decomposition (#591 — ECPS parity)

changelog.d/591-strike-benefits.fixed.md

@@ -0,0 +1,3 @@
+Restore the observed annual `strike_benefits` input from the ASEC other-income
+code-12 amount, conserve the `OI_VAL` decomposition, and carry the leaf onto
+ACS rows through the pre-clone CPS source family.

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

@@ -50,9 +50,11 @@
 from populace.build.us_runtime.alimony import (
     ALIMONY_ASEC_ARCHIVED_DERIVATION_URL,
     ALIMONY_PUF_ARCHIVED_DERIVATION_URL,
+    STRIKE_BENEFITS_ASEC_ARCHIVED_DERIVATION_URL,
     US_ALIMONY_NONCONSTANT_PERSON_COLUMNS,
     US_ALIMONY_OUTPUT_COLUMNS,
     US_ALIMONY_STAGE_NAME,
+    US_ASEC_OTHER_INCOME_OUTPUT_COLUMNS,
     derive_us_alimony_from_asec,
     derive_us_alimony_from_puf,
     us_alimony_signal_gate,
@@ -1215,6 +1217,8 @@
     "with_us_relationship_inputs",
     "ALIMONY_ASEC_ARCHIVED_DERIVATION_URL",
     "ALIMONY_PUF_ARCHIVED_DERIVATION_URL",
+    "STRIKE_BENEFITS_ASEC_ARCHIVED_DERIVATION_URL",
+    "US_ASEC_OTHER_INCOME_OUTPUT_COLUMNS",
     "US_ADULT_CARE_CHILD_QUALIFYING_AGE_LIMIT",
     "US_ADULT_CARE_EARNED_INCOME_SOURCES",
     "US_ADULT_CARE_OUTPUT_COLUMNS",

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

@@ -1,12 +1,12 @@
-"""ASEC/IRS PUF alimony inputs for the US build.
+"""ASEC other-income decomposition and IRS PUF alimony inputs.
 
 The retired eCPS pipeline measured recipient income from Census ASEC other-
 income records and tax-return income/expense from two IRS PUF fields.  The
-ASEC half keeps reported ``OI_VAL`` only when ``OI_OFF == 20``; the PUF support
-half is populated from the direct ``E00800`` / ``E03500`` mappings through the
-shared weighted PUF QRF.  ``miscellaneous_income`` must simultaneously exclude
-alimony (and the retired pipeline's strike-benefit code 12), otherwise the ASEC
-amount is counted twice in gross income.
+ASEC half splits ``OI_VAL`` into reported alimony when ``OI_OFF == 20``, strike
+benefits when ``OI_OFF == 12``, and miscellaneous income otherwise.  The PUF
+support half is populated from the direct ``E00800`` / ``E03500`` mappings
+through the shared weighted PUF QRF.  The three ASEC outputs are mutually
+exclusive so every reported amount is carried exactly once.
 
 This module owns only factual input leaves. PolicyEngine-US owns taxable-income
 and above-the-line-deduction formulas.
@@ -31,6 +31,8 @@
 __all__ = [
     "ALIMONY_ASEC_ARCHIVED_DERIVATION_URL",
     "ALIMONY_PUF_ARCHIVED_DERIVATION_URL",
+    "STRIKE_BENEFITS_ASEC_ARCHIVED_DERIVATION_URL",
+    "US_ASEC_OTHER_INCOME_OUTPUT_COLUMNS",
     "US_ALIMONY_NONCONSTANT_PERSON_COLUMNS",
     "US_ALIMONY_OUTPUT_COLUMNS",
     "US_ALIMONY_STAGE_NAME",
@@ -48,6 +50,7 @@
     f"{_ARCHIVED_DATA_REPOSITORY}/blob/{_ARCHIVED_COMMIT}/"
     "policyengine_" + "us_data/datasets/cps/cps.py#L1481-L1492"
 )
+STRIKE_BENEFITS_ASEC_ARCHIVED_DERIVATION_URL = ALIMONY_ASEC_ARCHIVED_DERIVATION_URL
 ALIMONY_PUF_ARCHIVED_DERIVATION_URL = (
     "https://github.com/PolicyEngine/"
     f"{_ARCHIVED_DATA_REPOSITORY}/blob/{_ARCHIVED_COMMIT}/"
@@ -60,9 +63,20 @@
     "alimony_expense",
 )
 US_ALIMONY_NONCONSTANT_PERSON_COLUMNS = US_ALIMONY_OUTPUT_COLUMNS
+US_ASEC_OTHER_INCOME_OUTPUT_COLUMNS: tuple[str, ...] = (
+    "alimony_income",
+    "strike_benefits",
+    "miscellaneous_income",
+)
 
 _ASEC_ALIMONY_OTHER_INCOME_CODE = 20
-_ASEC_SEPARATE_OTHER_INCOME_CODES = frozenset({12, 20})
+_ASEC_STRIKE_BENEFITS_OTHER_INCOME_CODE = 12
+_ASEC_SEPARATE_OTHER_INCOME_CODES = frozenset(
+    {
+        _ASEC_ALIMONY_OTHER_INCOME_CODE,
+        _ASEC_STRIKE_BENEFITS_OTHER_INCOME_CODE,
+    }
+)
 _NONZERO_SHARE_BAND = (0.00001, 0.02)
 
 
@@ -109,18 +123,18 @@ def _strict_numeric_source(
 
 
 def derive_us_alimony_from_asec(person: pd.DataFrame) -> pd.DataFrame:
-    """Carry reported ASEC alimony and remove it from miscellaneous income.
+    """Split reported ASEC other income into three exhaustive input leaves.
 
-    An already materialized pair is preserved, which makes the CPS-carried
-    transform idempotent on a staged base artifact. If either leaf is missing,
-    both raw ASEC fields are mandatory; a missing source must not be healed with
-    fabricated zeros.
+    An already materialized split is preserved, which makes the CPS-carried
+    transform idempotent on a staged base artifact. If any leaf is missing,
+    both raw ASEC fields are mandatory; a missing source must not be healed
+    with fabricated zeros.
     """
 
     raw_sources = {"OI_VAL", "OI_OFF"}
     present_sources = raw_sources.intersection(person.columns)
     if not present_sources:
-        materialized = {"alimony_income", "miscellaneous_income"}
+        materialized = set(US_ASEC_OTHER_INCOME_OUTPUT_COLUMNS)
         if materialized.issubset(person.columns):
             return person.copy(deep=True)
         missing = sorted(materialized - set(person.columns))
@@ -162,6 +176,11 @@ def derive_us_alimony_from_asec(person: pd.DataFrame) -> pd.DataFrame:
         amounts,
         0.0,
     )
+    result["strike_benefits"] = np.where(
+        integer_codes == _ASEC_STRIKE_BENEFITS_OTHER_INCOME_CODE,
+        amounts,
+        0.0,
+    )
     result["miscellaneous_income"] = np.where(
         np.isin(integer_codes, list(_ASEC_SEPARATE_OTHER_INCOME_CODES)),
         0.0,
@@ -256,11 +275,21 @@ def us_alimony_signal_gate(frame: Frame) -> GateResult:
                 f"band [{low}, {high}]."
             )
     raw_sources = {"OI_VAL", "OI_OFF"}
-    if raw_sources.issubset(person.columns) and "miscellaneous_income" not in person:
-        failures.append(
-            "ASEC raw OI_VAL/OI_OFF are present but miscellaneous_income is missing."
+    if raw_sources.issubset(person.columns):
+        missing_split = sorted(
+            set(US_ASEC_OTHER_INCOME_OUTPUT_COLUMNS) - set(person.columns)
         )
-    elif raw_sources.issubset(person.columns):
+        if missing_split:
+            failures.append(
+                "ASEC raw OI_VAL/OI_OFF are present but other-income split "
+                f"column(s) are missing: {missing_split}."
+            )
+            return GateResult(
+                name="alimony_inputs_signal",
+                passed=False,
+                failures=tuple(failures),
+                details=summary,
+            )
         source_mask = np.ones(len(person), dtype=bool)
         if has_support_role_metadata(person, entity="person"):
             source_mask = (
@@ -281,6 +310,11 @@ def us_alimony_signal_gate(frame: Frame) -> GateResult:
                 amounts,
                 0.0,
             )
+            expected_strike_benefits = np.where(
+                integer_codes == _ASEC_STRIKE_BENEFITS_OTHER_INCOME_CODE,
+                amounts,
+                0.0,
+            )
             expected_miscellaneous = np.where(
                 np.isin(integer_codes, list(_ASEC_SEPARATE_OTHER_INCOME_CODES)),
                 0.0,
@@ -289,24 +323,45 @@ def us_alimony_signal_gate(frame: Frame) -> GateResult:
             actual_alimony = pd.to_numeric(
                 person.loc[source_mask, "alimony_income"], errors="coerce"
             ).to_numpy(dtype=np.float64)
+            actual_strike_benefits = pd.to_numeric(
+                person.loc[source_mask, "strike_benefits"], errors="coerce"
+            ).to_numpy(dtype=np.float64)
             actual_miscellaneous = pd.to_numeric(
                 person.loc[source_mask, "miscellaneous_income"], errors="coerce"
             ).to_numpy(dtype=np.float64)
             alimony_mismatch = ~np.isclose(actual_alimony, expected_alimony)
+            strike_benefits_mismatch = ~np.isclose(
+                actual_strike_benefits,
+                expected_strike_benefits,
+            )
             miscellaneous_mismatch = ~np.isclose(
                 actual_miscellaneous,
                 expected_miscellaneous,
             )
+            conservation_mismatch = ~np.isclose(
+                actual_alimony + actual_strike_benefits + actual_miscellaneous,
+                amounts,
+            )
             if bool(alimony_mismatch.any()):
                 failures.append(
                     "ASEC alimony_income disagrees with OI_OFF == 20 / OI_VAL "
                     f"on {int(np.count_nonzero(alimony_mismatch))} row(s)."
                 )
+            if bool(strike_benefits_mismatch.any()):
+                failures.append(
+                    "ASEC strike_benefits disagrees with OI_OFF == 12 / OI_VAL "
+                    f"on {int(np.count_nonzero(strike_benefits_mismatch))} row(s)."
+                )
             if bool(miscellaneous_mismatch.any()):
                 failures.append(
                     "ASEC miscellaneous_income does not exclude alimony/strike "
                     f"codes on {int(np.count_nonzero(miscellaneous_mismatch))} row(s)."
                 )
+            if bool(conservation_mismatch.any()):
+                failures.append(
+                    "ASEC other-income split does not conserve OI_VAL on "
+                    f"{int(np.count_nonzero(conservation_mismatch))} row(s)."
+                )
     return GateResult(
         name="alimony_inputs_signal",
         passed=not failures,

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

@@ -13,7 +13,10 @@
 import numpy as np
 import pandas as pd
 
-from populace.build.us_runtime.alimony import derive_us_alimony_from_asec
+from populace.build.us_runtime.alimony import (
+    US_ASEC_OTHER_INCOME_OUTPUT_COLUMNS,
+    derive_us_alimony_from_asec,
+)
 from populace.frame import US_SCHEMA, Frame
 
 __all__ = [
@@ -75,8 +78,7 @@
         "has_va_health_coverage_at_interview",
         "is_female",
         "unemployment_compensation",
-        "alimony_income",
-        "miscellaneous_income",
+        *US_ASEC_OTHER_INCOME_OUTPUT_COLUMNS,
     }
 )
 

packages/populace-build/tests/test_us_alimony.py

@@ -12,6 +12,7 @@
 from populace.build.us_runtime.alimony import (
     ALIMONY_ASEC_ARCHIVED_DERIVATION_URL,
     ALIMONY_PUF_ARCHIVED_DERIVATION_URL,
+    STRIKE_BENEFITS_ASEC_ARCHIVED_DERIVATION_URL,
     derive_us_alimony_from_asec,
     derive_us_alimony_from_puf,
     us_alimony_signal_gate,
@@ -50,6 +51,7 @@ def test_archived_coordinates_are_commit_and_line_pinned() -> None:
     for url in (
         ALIMONY_ASEC_ARCHIVED_DERIVATION_URL,
         ALIMONY_PUF_ARCHIVED_DERIVATION_URL,
+        STRIKE_BENEFITS_ASEC_ARCHIVED_DERIVATION_URL,
     ):
         assert "42ed5d45c56df80d754fbe24cce21cfeb8d05cbe" in url
         assert "#L" in url
@@ -66,15 +68,56 @@ def test_asec_mapping_splits_alimony_and_miscellaneous_income_exactly() -> None:
     result = derive_us_alimony_from_asec(source)
 
     assert result["alimony_income"].tolist() == [5_000.0, 0.0, 0.0, 0.0]
+    assert result["strike_benefits"].tolist() == [0.0, 700.0, 0.0, 0.0]
     assert result["miscellaneous_income"].tolist() == [0.0, 0.0, 900.0, 0.0]
+    assert all(
+        result[column].dtype == np.dtype("float64")
+        for column in (
+            "alimony_income",
+            "strike_benefits",
+            "miscellaneous_income",
+        )
+    )
     assert "alimony_income" not in source
+    assert "strike_benefits" not in source
     assert "miscellaneous_income" not in source
 
 
-def test_asec_mapping_preserves_an_already_materialized_pair() -> None:
+def test_strike_benefits_matches_archived_ecps_mapping_semantics() -> None:
+    source = pd.DataFrame(
+        {
+            "OI_OFF": [12, 20, 19, 12, 0],
+            "OI_VAL": [700.0, 5_000.0, 900.0, 0.0, 0.0],
+        }
+    )
+
+    result = derive_us_alimony_from_asec(source)
+    archived_ecps = (source["OI_OFF"] == 12) * source["OI_VAL"]
+
+    np.testing.assert_array_equal(result["strike_benefits"], archived_ecps)
+
+
+def test_asec_other_income_split_conserves_oi_val_per_person() -> None:
+    source = pd.DataFrame(
+        {
+            "OI_OFF": [12, 20, 19, 0, 7],
+            "OI_VAL": [700.0, 5_000.0, 900.0, 0.0, 125.5],
+        }
+    )
+
+    result = derive_us_alimony_from_asec(source)
+    split_total = result.loc[
+        :, ["alimony_income", "strike_benefits", "miscellaneous_income"]
+    ].sum(axis="columns")
+
+    np.testing.assert_array_equal(split_total, source["OI_VAL"])
+
+
+def test_asec_mapping_preserves_an_already_materialized_split() -> None:
     source = pd.DataFrame(
         {
             "alimony_income": [10.0],
+            "strike_benefits": [30.0],
             "miscellaneous_income": [20.0],
         }
     )
@@ -282,6 +325,7 @@ def test_signal_gate_rejects_asec_alimony_left_in_miscellaneous_income() -> None
                 "OI_VAL": amounts,
                 "alimony_income": income,
                 "alimony_expense": expense,
+                "strike_benefits": np.zeros(n),
                 "miscellaneous_income": miscellaneous,
             }
         )
@@ -296,6 +340,40 @@ def test_signal_gate_rejects_asec_alimony_left_in_miscellaneous_income() -> None
     )
 
 
+def test_signal_gate_rejects_discarded_asec_strike_benefits() -> None:
+    n = 1_000
+    codes = np.zeros(n)
+    amounts = np.zeros(n)
+    income = np.zeros(n)
+    expense = np.zeros(n)
+    strike_benefits = np.zeros(n)
+    miscellaneous = np.zeros(n)
+    codes[10] = 12
+    amounts[10] = 2_000.0
+    expense[20] = 3_000.0
+    frame = _PersonFrame(
+        pd.DataFrame(
+            {
+                "OI_OFF": codes,
+                "OI_VAL": amounts,
+                "alimony_income": income,
+                "alimony_expense": expense,
+                "strike_benefits": strike_benefits,
+                "miscellaneous_income": miscellaneous,
+            }
+        )
+    )
+
+    result = us_alimony_signal_gate(frame)  # type: ignore[arg-type]
+
+    assert not result.passed
+    assert any(
+        "strike_benefits disagrees with OI_OFF == 12" in failure
+        for failure in result.failures
+    )
+    assert any("does not conserve OI_VAL" in failure for failure in result.failures)
+
+
 @pytest.mark.parametrize(
     "person",
     [
@@ -314,11 +392,11 @@ def test_signal_gate_rejects_missing_default_or_invalid_surface(
 
 
 @requires_us
-def test_policyengine_us_contract_is_two_person_year_input_leaves() -> None:
+def test_policyengine_us_contract_includes_strike_person_year_input_leaf() -> None:
     from policyengine_us import CountryTaxBenefitSystem
 
     variables = CountryTaxBenefitSystem().variables
-    for name in ("alimony_income", "alimony_expense"):
+    for name in ("alimony_income", "alimony_expense", "strike_benefits"):
         variable = variables[name]
         assert variable.is_input_variable()
         assert variable.entity.key == "person"