diff --git a/changelog.d/591-strike-benefits.fixed.md b/changelog.d/591-strike-benefits.fixed.md new file mode 100644 index 00000000..f0950a5f --- /dev/null +++ b/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. diff --git a/packages/populace-build/src/populace/build/us_runtime/__init__.py b/packages/populace-build/src/populace/build/us_runtime/__init__.py index cde117aa..7cbc75f7 100644 --- a/packages/populace-build/src/populace/build/us_runtime/__init__.py +++ b/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", diff --git a/packages/populace-build/src/populace/build/us_runtime/alimony.py b/packages/populace-build/src/populace/build/us_runtime/alimony.py index 12c1b514..6193169f 100644 --- a/packages/populace-build/src/populace/build/us_runtime/alimony.py +++ b/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, diff --git a/packages/populace-build/src/populace/build/us_runtime/cps_carried.py b/packages/populace-build/src/populace/build/us_runtime/cps_carried.py index 3c6d15c4..faf2b575 100644 --- a/packages/populace-build/src/populace/build/us_runtime/cps_carried.py +++ b/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, } ) diff --git a/packages/populace-build/tests/test_us_alimony.py b/packages/populace-build/tests/test_us_alimony.py index 37d7db11..0139130a 100644 --- a/packages/populace-build/tests/test_us_alimony.py +++ b/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" diff --git a/packages/populace-build/tests/test_us_multispine_pool.py b/packages/populace-build/tests/test_us_multispine_pool.py index 34af94bc..381ff0f7 100644 --- a/packages/populace-build/tests/test_us_multispine_pool.py +++ b/packages/populace-build/tests/test_us_multispine_pool.py @@ -660,15 +660,22 @@ def test_pool_transfer_plan_extends_legacy_except_receipted_asset_deferrals() -> "person", "source_operator_hours_worked", ) + assert owners["strike_benefits"] == ( + "person", + "source_operator_cps_carried", + ) + assert POOL_OPERATOR_CONTRACTS["derive_us_cps_carried_inputs"].phases == ( + "pre_clone", + ) assert "weeks_worked" not in owners assert "medicare_part_b_premiums_reported" not in owners assert "has_marketplace_health_coverage" not in owners target_names = sorted(owners) - assert len(target_names) == 114 + assert len(target_names) == 115 assert ( hashlib.sha256(("\n".join(target_names) + "\n").encode()).hexdigest() - == "4c106c69c9791b4a323088cb7f4894578b60a713bec308e807a956001fc9b965" + == "d33a6afdcc6e32f5f38d2de4e7bb0b617b557cadbe601fcdc6019d54bd6d83a7" ) @@ -676,13 +683,13 @@ def test_pool_input_surface_normalizes_all_four_source_registries() -> None: surface = pool_input_surface() by_name = {entry.variable: entry for entry in surface} - assert len(surface) == len(by_name) == 135 + assert len(surface) == len(by_name) == 136 assert [entry.variable for entry in surface] == sorted(by_name) assert Counter( provenance for entry in surface for provenance in entry.provenance ) == Counter( { - "pool_transfer_target_families": 114, + "pool_transfer_target_families": 115, "POOL_DEFERRED_TRANSFER_INPUTS": 3, "PRIMARY_QRF_TARGET_ORDER": 65, "load_take_up_contract": 13, @@ -700,6 +707,12 @@ def test_pool_input_surface_normalizes_all_four_source_registries() -> None: family="primary_puf_qrf_nontransfer", provenance=("PRIMARY_QRF_TARGET_ORDER",), ) + assert by_name["strike_benefits"] == PoolInputSurfaceEntry( + variable="strike_benefits", + entity="person", + family="source_operator_cps_carried", + provenance=("pool_transfer_target_families",), + ) assert by_name["first_home_mortgage_balance"] == PoolInputSurfaceEntry( variable="first_home_mortgage_balance", entity="tax_unit", @@ -1214,7 +1227,7 @@ def test_every_pool_transfer_target_is_an_installed_engine_input_leaf() -> None: for columns in families.values() for target in columns } - assert len(targets) == 114 + assert len(targets) == 115 acs_transfer_module.assert_acs_transfer_targets_are_input_leaves( targets, require_known=True, @@ -1239,7 +1252,7 @@ def test_every_pool_transfer_family_accepts_its_produced_physical_dtype( ) ) - assert len(targets) == 114 + assert len(targets) == 115 assert len(predictors) == 32 assert len(primary_predictor_sets) == 65 primary_targets = tuple( @@ -1256,7 +1269,7 @@ def test_every_pool_transfer_family_accepts_its_produced_physical_dtype( assert len(primary_predictor_sets[0][1]) == 8 assert len(primary_predictor_sets[-1][1]) == 72 assert len(POOL_DEFERRED_TRANSFER_INPUTS) == 3 - assert len(targets) + len(POOL_DEFERRED_TRANSFER_INPUTS) == 117 + assert len(targets) + len(POOL_DEFERRED_TRANSFER_INPUTS) == 118 assert set(POOL_SOURCE_OPERATOR_ORDER) <= set(calls) assert all(calls[name] > 0 for name in POOL_SOURCE_OPERATOR_ORDER) assert calls["with_us_prior_year_income_inputs"] == 2 diff --git a/packages/populace-build/tests/test_us_pool_input_consumers.py b/packages/populace-build/tests/test_us_pool_input_consumers.py index fbb9e3fd..d32d6bb3 100644 --- a/packages/populace-build/tests/test_us_pool_input_consumers.py +++ b/packages/populace-build/tests/test_us_pool_input_consumers.py @@ -348,6 +348,15 @@ def test_model_api_reference_helper_consumer_is_discoverable() -> None: ) +def test_strike_benefits_market_income_consumer_is_discoverable() -> None: + assert any( + receipt.consumer == "market_income" + and receipt.path.endswith("/market_income.py") + and receipt.line > 0 + for receipt in _ENGINE_INDEX.consumer_receipts("strike_benefits") + ) + + @pytest.mark.parametrize( ("variable", "consumer", "filename"), ( diff --git a/packages/populace-build/tests/test_us_puf_support.py b/packages/populace-build/tests/test_us_puf_support.py index 35a9dc63..b3d9e3a4 100644 --- a/packages/populace-build/tests/test_us_puf_support.py +++ b/packages/populace-build/tests/test_us_puf_support.py @@ -1059,6 +1059,7 @@ def test_cps_carried_derivations_create_leaf_inputs_not_aggregates() -> None: assert "farm_income" not in person assert person["unemployment_compensation"].tolist() == [0.0, 70.0, 0.0] assert person["alimony_income"].tolist() == [0.0, 0.0, 0.0] + assert person["strike_benefits"].tolist() == [0.0, 0.0, 0.0] assert person["miscellaneous_income"].tolist() == [3.0, 0.0, 0.0] assert person["health_insurance_premiums_without_medicare_part_b"].tolist() == [ 400.0,