Carry SIPP/SCF assets on the declared household head

changelog.d/us-wealth-carrier-household-head.fixed.md

@@ -0,0 +1 @@
+The SIPP/SCF liquid-asset stage carries each household's draw on the declared `is_household_head` (the CPS reference person, `P_SEQ == 1`), as `us/source_stages.json` declares and the housing stages use, instead of the lowest `A_LINENO` (part of #1215). A household whose flag does not mark exactly one person, or a frame without the flag, keeps the lowest-line rule. The two rules disagree in 243 of the certified frame's 57,240 households; on the 2024 income year, where `A_EXPRRP` is observed, the flag is the reference person in every household and the line rule misses 73. Both halves of the blend share the selector. Future builds only.

docs/evidence/spec-engine/us-f0-coverage.json

@@ -1642,13 +1642,13 @@
           "compiler_ir.node_slices"
         ],
         "expected": {
-          "map_sha256": "fa48af060da522bb1edcb80580098f258763757139292602759cbccbfff76b37",
-          "protocol_sha256": "5840ac6cefd5ce3d2172aca9e3bfc08d2344e97a8e9b42a723a4edcaa9f8ff53"
+          "map_sha256": "5f8b7c620abee639d31f3dd50ed61e9c9a0daad89b8b4886d485b71d5cb882bd",
+          "protocol_sha256": "3e432297573840bbae497370efe3e7cafed5c2c571db029208d10722035a7470"
         },
         "failures": [],
         "observed": {
-          "map_sha256": "fa48af060da522bb1edcb80580098f258763757139292602759cbccbfff76b37",
-          "protocol_sha256": "5840ac6cefd5ce3d2172aca9e3bfc08d2344e97a8e9b42a723a4edcaa9f8ff53"
+          "map_sha256": "5f8b7c620abee639d31f3dd50ed61e9c9a0daad89b8b4886d485b71d5cb882bd",
+          "protocol_sha256": "3e432297573840bbae497370efe3e7cafed5c2c571db029208d10722035a7470"
         },
         "status": "covered"
       },
@@ -1663,7 +1663,7 @@
           "compiler_ir.seed_stream_map"
         ],
         "expected": {
-          "implementation_sha256": "5840ac6cefd5ce3d2172aca9e3bfc08d2344e97a8e9b42a723a4edcaa9f8ff53",
+          "implementation_sha256": "3e432297573840bbae497370efe3e7cafed5c2c571db029208d10722035a7470",
           "protocol": "legacy-v1",
           "streams": [
             "build_model",
@@ -1684,7 +1684,7 @@
         },
         "failures": [],
         "observed": {
-          "implementation_sha256": "5840ac6cefd5ce3d2172aca9e3bfc08d2344e97a8e9b42a723a4edcaa9f8ff53",
+          "implementation_sha256": "3e432297573840bbae497370efe3e7cafed5c2c571db029208d10722035a7470",
           "protocol": "legacy-v1",
           "streams": [
             "build_model",
@@ -2585,7 +2585,7 @@
       "country": "us",
       "schema_id": "country_spec",
       "schema_version": 1,
-      "spec_sha256": "4ccbd01720871e59ad8e3c00e1c1296231facef91e83c561076ee499e0a138b2"
+      "spec_sha256": "915d29b0a5419d8e9ff4c9945fb5a92fb18576440dbbbcef2a7680546cfc4441"
     }
   },
   "report_schema_version": 3,
@@ -2595,7 +2595,7 @@
     "country": "us",
     "schema_id": "country_spec",
     "schema_version": 1,
-    "spec_sha256": "4ccbd01720871e59ad8e3c00e1c1296231facef91e83c561076ee499e0a138b2"
+    "spec_sha256": "915d29b0a5419d8e9ff4c9945fb5a92fb18576440dbbbcef2a7680546cfc4441"
   },
   "status": "pass"
 }

packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py

@@ -379,9 +379,11 @@
     # gains the passive partnership/S-corp leaf, so the PUF QRF targets change.
     # microcosm#1259's recipient-share bound in puf_support moves both again.
     # microcosm#1261 moves both again: the PUF donor carries partnership and
-    # S-corp line-32 shares as separate leaves.
-    "seed_map": "fa48af060da522bb1edcb80580098f258763757139292602759cbccbfff76b37",
-    "seed_protocol": "5840ac6cefd5ce3d2172aca9e3bfc08d2344e97a8e9b42a723a4edcaa9f8ff53",
+    # S-corp line-32 shares as separate leaves. microcosm#1215 moves both again:
+    # scf_wealth and sipp_financial_assets carry the assets on the declared
+    # is_household_head carrier.
+    "seed_map": "5f8b7c620abee639d31f3dd50ed61e9c9a0daad89b8b4886d485b71d5cb882bd",
+    "seed_protocol": "3e432297573840bbae497370efe3e7cafed5c2c571db029208d10722035a7470",
     "source_manifest": "dc3879886d2a69f20fb982a675d7fb322c0e4017afef6471be54b0ee3cbd2af6",
     "take_up": "9522ce40f7dea569312dd7a7beb474e5a5afadd2f1e10cb5876534c3ef623d35",
     "tail": "565f0ceeadab69d3e96a886a00dd0305f7425fa3acde24d12ba48db2b37d5fc0",

packages/microcosm-build/src/microcosm/build/us_runtime/scf_wealth.py

@@ -115,6 +115,7 @@
     SIPP_FINANCIAL_ASSET_SOURCE_COLUMNS,
     SIPP_FINANCIAL_ASSET_TARGET_ALLOCATION_COLUMNS,
     SIPP_FINANCIAL_ASSET_TARGET_SOURCE_COLUMNS,
+    household_head_carrier_mask,
     impute_us_sipp_financial_assets,
 )
 from microcosm.frame import Frame
@@ -673,25 +674,13 @@ def _sum_present(person: pd.DataFrame, columns: tuple[str, ...]) -> np.ndarray:
 def _household_head_mask(person: pd.DataFrame) -> np.ndarray:
     """One reference person (household head) per household.
 
-    The head is the lowest-line-number person within each household (the CPS
-    householder is line 1); ties/absent line numbers fall back to the first
-    row. This mirrors the retired pipeline's reference-person rule (the first
-    person of each household).
+    The declared ``is_household_head`` (``P_SEQ == 1``) where it marks exactly
+    one person in the household, else the lowest-line-number person, then the
+    first row. The SIPP side shares this selector so both halves of the blend
+    carry the same person (microcosm#1215).
     """
 
-    household = pd.to_numeric(person[_HOUSEHOLD_ID_COLUMN], errors="coerce").to_numpy()
-    lineno = pd.to_numeric(person["A_LINENO"], errors="coerce").fillna(9_999).to_numpy()
-    position = np.arange(len(person))
-    # Sort by household, then line number, then original position; the first
-    # row of each household group in that order is its head.
-    order = np.lexsort((position, lineno, household))
-    ordered_household = household[order]
-    is_first_in_group = np.empty(len(person), dtype=bool)
-    is_first_in_group[0] = True
-    is_first_in_group[1:] = ordered_household[1:] != ordered_household[:-1]
-    mask = np.zeros(len(person), dtype=bool)
-    mask[order[is_first_in_group]] = True
-    return mask
+    return household_head_carrier_mask(person)
 
 
 def _household_head_predictor_table(person: pd.DataFrame) -> pd.DataFrame:

packages/microcosm-build/src/microcosm/build/us_runtime/sipp_financial_assets.py

@@ -558,7 +558,21 @@ def _recipient_is_married(person: pd.DataFrame) -> np.ndarray:
     return np.zeros(len(person), dtype=np.float64)
 
 
-def _household_head_mask(person: pd.DataFrame) -> np.ndarray:
+def household_head_carrier_mask(person: pd.DataFrame) -> np.ndarray:
+    """One financial-asset carrier per household: the declared household head.
+
+    ``us/source_stages.json`` declares the wealth stage's carry as
+    ``head_carry`` with ``condition: is_household_head``, the CPS reference
+    person (``P_SEQ == 1``; ``relationship_inputs``), as the housing stages
+    use. A household whose flag does not mark exactly one person, or a frame
+    without the flag, falls back to its lowest ``A_LINENO``, then row order.
+    On the 2024 income year, where ``A_EXPRRP`` is observed, the flag is the
+    reference person in every household and the line rule misses 73
+    (microcosm#1215). Both the SCF and SIPP sides use this one selector.
+    """
+
+    if len(person) == 0:
+        return np.zeros(0, dtype=bool)
     household = pd.to_numeric(person[_HOUSEHOLD_ID_COLUMN], errors="coerce").to_numpy()
     if "A_LINENO" in person.columns:
         line = (
@@ -572,9 +586,23 @@ def _household_head_mask(person: pd.DataFrame) -> np.ndarray:
     first = np.empty(len(person), dtype=bool)
     first[0] = True
     first[1:] = ordered_household[1:] != ordered_household[:-1]
-    mask = np.zeros(len(person), dtype=bool)
-    mask[order[first]] = True
-    return mask
+    lowest_line = np.zeros(len(person), dtype=bool)
+    lowest_line[order[first]] = True
+    if "is_household_head" not in person.columns:
+        return lowest_line
+    raw_flag = person["is_household_head"]
+    if pd.api.types.is_bool_dtype(raw_flag.dtype):
+        flag = raw_flag.fillna(False).astype(bool).to_numpy()
+    else:
+        # Only true booleans and nonzero numbers flag a head; strings such as
+        # "False" or "" never do.
+        flag = pd.to_numeric(raw_flag, errors="coerce").fillna(0).ne(0).to_numpy()
+    # Count flags within the same groups the fallback sorts, so a missing
+    # household id is its own group in both rules.
+    group = np.empty(len(person), dtype=np.int64)
+    group[order] = np.cumsum(first) - 1
+    flags_in_household = np.bincount(group, weights=flag.astype(np.float64))[group]
+    return np.where(flags_in_household == 1, flag, lowest_line)
 
 
 def _recipient_sipp_asset_predictor_table(person: pd.DataFrame) -> pd.DataFrame:
@@ -660,7 +688,7 @@ def impute_us_sipp_financial_assets(
             f"SIPP financial-asset donor table missing column(s): {missing}."
         )
 
-    head_mask = _household_head_mask(person)
+    head_mask = household_head_carrier_mask(person)
     head_features = _recipient_sipp_asset_predictor_table(person).loc[head_mask]
     augmented = head_features.copy()
     draws = pd.DataFrame(index=head_features.index)

packages/microcosm-build/tests/engine_free/shared/test_spec_engine_loader.py

@@ -25,7 +25,7 @@ def test_semantic_hash_has_golden_vector_and_surface_separation(tmp_path) -> Non
     # ssi_disability_criteria's attested bytes.
     # The opt-in cross-fitted gate calibration moves the fit kernels' bytes.
     assert first.spec_sha256 == (
-        "5b39d83cafb6f312ce73e5550ab36da6676ce265779d615f49c7987f5cc5d36e"
+        "ddb925d9603cd269c358f46ff433d5c4adf67991aa17782331a23c0b55767a05"
     )
 
     second_root = _rich_minimal(tmp_path / "xy", note="second", store="local:b")

packages/microcosm-build/tests/engine_free/us/test_us_scf_wealth.py

@@ -1,6 +1,9 @@
 """Tests split from packages/microcosm-build/tests/test_us_scf_wealth.py."""
 
 # ruff: noqa: F403, F405
+from hypothesis import given, settings
+from hypothesis import strategies as st
+
 from test_support.microcosm_build.us_scf_wealth import *
 
 
@@ -119,6 +122,83 @@ def test_household_head_mask_is_one_per_household_at_lowest_line() -> None:
     assert mask.sum() == 2
 
 
+def test_household_head_mask_follows_the_declared_household_head() -> None:
+    """microcosm#1215: the carry is declared ``condition: is_household_head``.
+    A household whose flag marks exactly one person carries on that person,
+    even when another member has a lower line number; a household with no
+    flagged person or two falls back to its lowest line."""
+    person = pd.DataFrame(
+        {
+            "person_household_id": [5, 5, 5, 9, 9, 12, 12, 12],
+            "A_LINENO": [1, 2, 3, 1, 2, 2, 1, 3],
+            "is_household_head": [False, True, False, False, False, True, True, False],
+        }
+    )
+    mask = _household_head_mask(person)
+    # hh 5: the flagged line-2 person; hh 9: no flag, lowest line (index 3);
+    # hh 12: two flags, lowest line (index 6).
+    np.testing.assert_array_equal(
+        mask, [False, True, False, True, False, False, True, False]
+    )
+
+
+def test_both_blend_halves_share_one_carrier_selector() -> None:
+    from microcosm.build.us_runtime import sipp_financial_assets
+
+    person = pd.DataFrame(
+        {
+            "person_household_id": [1, 1, 2],
+            "A_LINENO": [1, 2, 1],
+            "is_household_head": pd.array([False, True, pd.NA], dtype="boolean"),
+        }
+    )
+    np.testing.assert_array_equal(
+        _household_head_mask(person),
+        sipp_financial_assets.household_head_carrier_mask(person),
+    )
+    np.testing.assert_array_equal(_household_head_mask(person), [False, True, True])
+    assert _household_head_mask(person.iloc[:0]).shape == (0,)
+
+
+@settings(max_examples=100, deadline=None)
+@given(
+    st.lists(
+        st.tuples(
+            st.integers(0, 6),  # household
+            st.integers(1, 9),  # line number
+            st.booleans(),  # is_household_head
+        ),
+        min_size=1,
+        max_size=40,
+    )
+)
+def test_household_head_mask_marks_exactly_one_carrier_per_household(rows) -> None:
+    # CPS line numbers are unique within a household.
+    person = (
+        pd.DataFrame(
+            rows, columns=["person_household_id", "A_LINENO", "is_household_head"]
+        )
+        .drop_duplicates(["person_household_id", "A_LINENO"])
+        .reset_index(drop=True)
+    )
+    mask = _household_head_mask(person)
+    per_household = pd.Series(mask).groupby(person["person_household_id"]).sum()
+    assert (per_household == 1).all()
+    flagged = person.groupby("person_household_id")["is_household_head"].transform(
+        "sum"
+    )
+    single = (flagged == 1).to_numpy()
+    # Where exactly one person is flagged, the carrier is that person.
+    np.testing.assert_array_equal(
+        mask[single], person["is_household_head"].to_numpy()[single]
+    )
+    # Permuting rows moves the carrier with its row.
+    order = np.random.default_rng(len(rows)).permutation(len(person))
+    np.testing.assert_array_equal(
+        _household_head_mask(person.iloc[order].reset_index(drop=True)), mask[order]
+    )
+
+
 def test_impute_head_carries_and_zeroes_non_heads() -> None:
     person = _person_rows(60)
     donor = _donor_table()
@@ -233,6 +313,96 @@ def test_blend_uses_one_complete_source_vector_and_reference_person_carry(
     )
 
 
+def test_blend_carries_on_a_declared_head_above_line_one(
+    sipp_scf_blend_case: dict[str, object],
+) -> None:
+    """microcosm#1215: where the declared head is the line-2 person, all three
+    leaves land on that person, every other member holds zero, and the gate's
+    audit check uses the same carrier."""
+    person = sipp_scf_blend_case["person"].copy()
+    assert isinstance(person, pd.DataFrame)
+    line_two_head = (person["person_household_id"] % 2 == 0).to_numpy()
+    person["is_household_head"] = np.where(
+        line_two_head, person["A_LINENO"] == 2, person["A_LINENO"] == 1
+    )
+    blended = impute_us_sipp_scf_financial_assets(
+        person,
+        _donor_table(),
+        _sipp_donor_table(),
+        seed=23,
+        time_period=TIME_PERIOD,
+        n_estimators=8,
+    )
+    head = person["is_household_head"].to_numpy()
+    np.testing.assert_array_equal(_household_head_mask(person), head)
+    leaves = list(US_SCF_FINANCIAL_ASSET_OUTPUT_COLUMNS)
+    assert blended.loc[~head, leaves].eq(0.0).all(axis=None)
+    assert blended.loc[head & line_two_head, "bank_account_assets"].sum() > 0
+    assert (
+        blended.loc[(person["A_LINENO"] == 1).to_numpy() & line_two_head, leaves]
+        .eq(0.0)
+        .all(axis=None)
+    )
+
+    base = _us_frame(person)
+    tables = {entity: base.table(entity).copy() for entity in base.entities}
+    for column in leaves:
+        tables["person"][column] = blended[column].to_numpy(dtype=np.float64)
+    tables["person"]["is_household_head"] = head
+    tables["person"].attrs[FINANCIAL_ASSET_BLEND_AUDIT_KEY] = dict(
+        blended.attrs[FINANCIAL_ASSET_BLEND_AUDIT_KEY]
+    )
+    net_worth = np.linspace(10_000.0, 250_000.0, len(tables["household"]))
+    net_worth[:24] *= -1.0
+    tables["household"]["net_worth"] = net_worth
+    frame = Frame(
+        tables,
+        base.schema,
+        {entity: base.weights_for(entity) for entity in base.weighted_entities},
+    )
+    gate = us_scf_wealth_signal_gate(frame, require_sipp_blend=True)
+    # The gate recomputes each household's source and the bank medians from the
+    # carrier mask; none of its blend-audit checks may disagree. (This small
+    # synthetic donor can trip unrelated distribution-plausibility checks.)
+    audit_failures = [
+        failure
+        for failure in gate.failures
+        if any(
+            token in failure.lower()
+            for token in ("blend", "median", "low-tail", "source share")
+        )
+    ]
+    assert audit_failures == []
+    assert gate.details["financial_asset_blend"]["sipp_household_count"] > 0
+
+
+def test_household_head_mask_ignores_string_flags_and_isolates_missing_ids() -> None:
+    from microcosm.build.us_runtime import sipp_financial_assets
+
+    strings = pd.DataFrame(
+        {
+            "person_household_id": [1, 1],
+            "A_LINENO": [2, 1],
+            "is_household_head": ["False", ""],
+        }
+    )
+    # No real flag: the lowest line carries.
+    np.testing.assert_array_equal(
+        sipp_financial_assets.household_head_carrier_mask(strings), [False, True]
+    )
+    missing_ids = pd.DataFrame(
+        {
+            "person_household_id": [np.nan, np.nan],
+            "A_LINENO": [1, 2],
+            "is_household_head": [False, True],
+        }
+    )
+    # Each missing id is its own household, as in the lowest-line rule.
+    np.testing.assert_array_equal(
+        sipp_financial_assets.household_head_carrier_mask(missing_ids), [True, True]
+    )
+
+
 def test_impute_missing_donor_column_raises() -> None:
     person = _person_rows(10)
     donor = _donor_table().drop(columns=["bond_assets"])

packages/microcosm-build/tests/engine_workflow/us/test_us_multispine_pool_tool.py

@@ -353,7 +353,7 @@ def capture_equality(expected: object, actual: object) -> None:
             # The opt-in cross-fitted gate calibration in microcosm.fit.qrf moves it.
             # microcosm#1073's scf_wealth education-debt declaration moves it
             # through the bundle's source stages.
-            "spec_sha256": "4ccbd01720871e59ad8e3c00e1c1296231facef91e83c561076ee499e0a138b2",
+            "spec_sha256": "915d29b0a5419d8e9ff4c9945fb5a92fb18576440dbbbcef2a7680546cfc4441",
         },
     }