Supply is_claimant_or_partner from the FRS adult table

changelog.d/frs-claimant-or-partner.added.md

@@ -0,0 +1 @@
+Supply `is_claimant_or_partner`, each benefit unit's single adult or couple, from FRS adult-table membership, so policyengine-uk reads the survey's roles instead of inferring them from ages. SPI records and the public transfer dataset (`enhanced_cps_2025.h5`, column added in place) carry it too, and stacking refuses to combine a person table that has it with one that does not.

policyengine_uk_data/datasets/enhanced_cps.py

@@ -559,6 +559,15 @@ def _build_base_dataset(
         adults = scenario["adults"]
         children = scenario["children"]
         people = adults + children
+        # Each source household is one tax unit with its head listed first and,
+        # for joint filers, the spouse second: the benefit unit's claimant and
+        # partner.
+        claimants_or_partners = 2 if row["filing_status"] == "joint" else 1
+        if len(adults) < claimants_or_partners:
+            raise ValueError(
+                f"Source household {source_household_id} files "
+                f"{row['filing_status']} but lists {len(adults)} adult(s)."
+            )
 
         has_mortgage = any(
             "deductible_mortgage_interest" in person.get("inputs", {})
@@ -663,6 +672,7 @@ def _build_base_dataset(
                     "person_id": person_id,
                     "person_household_id": household_id,
                     "person_benunit_id": benunit_id,
+                    "is_claimant_or_partner": person_index <= claimants_or_partners,
                     "age": int(person["age"]),
                     "gender": "MALE" if (household_id + person_index) % 2 else "FEMALE",
                     "employment_income_before_lsr": _gbp(

policyengine_uk_data/datasets/frs.py

@@ -392,6 +392,32 @@ def derive_is_parent_from_frs_microdata(
     return is_adult_record & has_dependent_children
 
 
+def derive_is_claimant_or_partner_from_frs_microdata(
+    person_ids,
+    person_benunit_ids,
+    adult_person_ids,
+) -> np.ndarray:
+    """Identify each FRS benefit unit's single adult or couple.
+
+    An FRS benefit unit is one adult or a couple plus any dependent children.
+    The adult table holds the head (UPERSON 1) and any partner (UPERSON 2);
+    the child table holds the dependent children. Any other adult in the
+    household, such as a grown-up son or daughter, forms and heads their own
+    benefit unit. So adult-table membership is policyengine-uk's
+    `is_claimant_or_partner`, which the model otherwise infers from ages.
+    """
+
+    is_adult_record = np.isin(np.asarray(person_ids), np.asarray(adult_person_ids))
+    per_benunit = pd.Series(is_adult_record).groupby(np.asarray(person_benunit_ids))
+    counts = per_benunit.sum()
+    if not counts.between(1, 2).all():
+        raise ValueError(
+            "Every FRS benefit unit needs one or two adult-table records; "
+            f"{int((~counts.between(1, 2)).sum())} benefit units do not."
+        )
+    return is_adult_record
+
+
 def _as_non_negative_array(values) -> np.ndarray:
     values = np.asarray(values, dtype=float)
     return np.maximum(np.nan_to_num(values, nan=0.0), 0.0)
@@ -728,6 +754,13 @@ def create_frs(
         benunit_ids=pe_benunit.benunit_id,
         dependent_children=dependent_children,
     )
+    pe_person["is_claimant_or_partner"] = (
+        derive_is_claimant_or_partner_from_frs_microdata(
+            person_ids=pe_person.person_id,
+            person_benunit_ids=pe_person.person_benunit_id,
+            adult_person_ids=frs["adult"].person_id,
+        )
+    )
     MARITAL = [
         "MARRIED",
         "SINGLE",

policyengine_uk_data/datasets/spi.py

@@ -112,6 +112,8 @@ def create_spi(
     person["person_id"] = df.SREF
     person["person_household_id"] = df.SREF
     person["person_benunit_id"] = df.SREF
+    # Each SPI taxpayer is their own one-person benefit unit, so its claimant.
+    person["is_claimant_or_partner"] = np.ones(len(df), dtype=bool)
     benunit["benunit_id"] = df.SREF
     household["household_id"] = df.SREF
 

policyengine_uk_data/tests/test_claimant_or_partner.py

@@ -0,0 +1,181 @@
+"""The claimant-or-partner role: FRS derivation, stacking, and built datasets."""
+
+import numpy as np
+import pandas as pd
+import pytest
+from hypothesis import given, settings
+from hypothesis import strategies as st
+from policyengine_uk.data import UKSingleYearDataset
+
+from policyengine_uk_data.datasets.frs import (
+    derive_is_claimant_or_partner_from_frs_microdata,
+    derive_is_parent_from_frs_microdata,
+)
+from policyengine_uk_data.utils.stack import stack_datasets
+
+
+def test_grown_up_child_heads_their_own_benefit_unit():
+    # Lone parent and dependent child in one benefit unit; grown-up son in his own.
+    result = derive_is_claimant_or_partner_from_frs_microdata(
+        person_ids=np.array([1_001, 1_002, 1_003]),
+        person_benunit_ids=np.array([101, 102, 101]),
+        adult_person_ids=np.array([1_001, 1_002]),
+    )
+    assert result.tolist() == [True, True, False]
+
+
+def test_couple_with_children():
+    result = derive_is_claimant_or_partner_from_frs_microdata(
+        person_ids=np.array([2_003, 2_001, 2_002]),
+        person_benunit_ids=np.array([201, 201, 201]),
+        adult_person_ids=np.array([2_002, 2_001]),
+    )
+    assert result.tolist() == [False, True, True]
+
+
+@pytest.mark.parametrize("adults", [[], [3_001, 3_002, 3_003]])
+def test_benefit_unit_without_one_or_two_adults_is_rejected(adults):
+    with pytest.raises(ValueError, match="one or two adult-table records"):
+        derive_is_claimant_or_partner_from_frs_microdata(
+            person_ids=np.array([3_001, 3_002, 3_003]),
+            person_benunit_ids=np.array([301, 301, 301]),
+            adult_person_ids=np.array(adults, dtype=int),
+        )
+
+
+@st.composite
+def frs_households(draw):
+    """FRS-shaped records: benefit units of one or two adults plus children."""
+    rows = []
+    for household in range(1, draw(st.integers(1, 6)) + 1):
+        person = 0
+        for benunit in range(1, draw(st.integers(1, 3)) + 1):
+            n_adults = draw(st.integers(1, 2))
+            n_children = draw(st.integers(0, 4))
+            for index in range(n_adults + n_children):
+                person += 1
+                rows.append(
+                    (
+                        household * 1_000 + person,
+                        household * 100 + benunit,
+                        index < n_adults,
+                    )
+                )
+    order = draw(st.permutations(range(len(rows))))
+    frame = pd.DataFrame(rows, columns=["person_id", "benunit_id", "adult"])
+    return frame.iloc[list(order)].reset_index(drop=True)
+
+
+def _derive(frame, adult_mask=None):
+    adult = frame.adult if adult_mask is None else adult_mask
+    return derive_is_claimant_or_partner_from_frs_microdata(
+        person_ids=frame.person_id,
+        person_benunit_ids=frame.benunit_id,
+        adult_person_ids=frame.person_id[adult],
+    )
+
+
+@settings(max_examples=300, deadline=None)
+@given(frs_households())
+def test_every_benefit_unit_has_a_claimant_and_at_most_one_partner(frame):
+    result = _derive(frame)
+    assert (result == frame.adult.to_numpy()).all()
+    counts = pd.Series(result).groupby(frame.benunit_id.to_numpy()).sum()
+    assert counts.between(1, 2).all()
+
+
+@settings(max_examples=200, deadline=None)
+@given(frs_households(), st.randoms(use_true_random=False))
+def test_record_order_does_not_change_anyones_role(frame, rng):
+    order = list(range(len(frame)))
+    rng.shuffle(order)
+    shuffled = frame.iloc[order].reset_index(drop=True)
+    by_person = dict(zip(frame.person_id, _derive(frame)))
+    assert [by_person[p] for p in shuffled.person_id] == _derive(shuffled).tolist()
+
+
+@settings(max_examples=200, deadline=None)
+@given(frs_households(), st.data())
+def test_parents_are_always_claimant_or_partner(frame, data):
+    benunits = np.sort(frame.benunit_id.unique())
+    children = data.draw(
+        st.lists(st.integers(0, 4), min_size=len(benunits), max_size=len(benunits))
+    )
+    is_parent = derive_is_parent_from_frs_microdata(
+        person_ids=frame.person_id,
+        person_benunit_ids=frame.benunit_id,
+        adult_person_ids=frame.person_id[frame.adult],
+        benunit_ids=benunits,
+        dependent_children=np.array(children),
+    )
+    assert not (is_parent & ~_derive(frame)).any()
+
+
+@settings(max_examples=200, deadline=None)
+@given(frs_households(), st.data())
+def test_a_benefit_unit_with_no_adult_or_a_third_adult_is_rejected(frame, data):
+    benunit = data.draw(st.sampled_from(sorted(frame.benunit_id.unique())))
+    members = frame.benunit_id == benunit
+    if data.draw(st.booleans()):
+        adult = frame.adult & ~members
+    else:
+        n_extra = 3 - int(frame.adult[members].sum())
+        start = frame.person_id.max() + 1
+        extra = pd.DataFrame(
+            {
+                "person_id": range(start, start + n_extra),
+                "benunit_id": benunit,
+                "adult": True,
+            }
+        )
+        frame = pd.concat([frame, extra], ignore_index=True)
+        adult = frame.adult
+    with pytest.raises(ValueError, match="one or two adult-table records"):
+        _derive(frame, adult)
+
+
+def _tiny_dataset(roles=True):
+    person = pd.DataFrame(
+        {
+            "person_id": [1, 2, 3],
+            "person_benunit_id": [1, 1, 1],
+            "person_household_id": [1, 1, 1],
+            "age": [40, 38, 10],
+        }
+    )
+    if roles:
+        person["is_claimant_or_partner"] = [True, True, False]
+    return UKSingleYearDataset(
+        person=person,
+        benunit=pd.DataFrame({"benunit_id": [1]}),
+        household=pd.DataFrame({"household_id": [1], "household_weight": [1.0]}),
+        fiscal_year=2024,
+    )
+
+
+def test_stacking_keeps_roles():
+    stacked = stack_datasets(_tiny_dataset(), _tiny_dataset())
+    stacked.validate()
+    assert stacked.person.is_claimant_or_partner.dtype == bool
+    assert stacked.person.is_claimant_or_partner.tolist() == [True, True, False] * 2
+
+
+@pytest.mark.parametrize("roles", [(True, False), (False, True)])
+def test_stacking_refuses_a_table_without_roles(roles):
+    with pytest.raises(ValueError, match="is_claimant_or_partner"):
+        stack_datasets(_tiny_dataset(roles[0]), _tiny_dataset(roles[1]))
+
+
+@pytest.mark.parametrize("fixture", ["frs", "enhanced_frs"])
+def test_built_dataset_roles(fixture, request):
+    person = request.getfixturevalue(fixture).person
+    role = person.is_claimant_or_partner
+    assert role.dtype == bool
+    counts = role.groupby(person.person_benunit_id).sum()
+    assert counts.between(1, 2).all()
+    assert not (person.is_benunit_head & ~role).any()
+    assert not (person.is_parent & ~role).any()
+    assert (person.age[role] >= 16).all()
+    # FRS dependent children are 19 at most, so everyone older is in the
+    # adult table: a claimant or partner.
+    assert role[person.age >= 20].all()

policyengine_uk_data/tests/test_clone_and_assign.py

@@ -256,6 +256,20 @@ def test_data_preserved_across_clones(self, toy_dataset, small_crosswalk):
             original_regions = toy_dataset.household["region"].values
             np.testing.assert_array_equal(regions, original_regions)
 
+    def test_claimant_or_partner_follows_each_clone(self, toy_dataset, small_crosswalk):
+        roles = np.tile([True, False], len(toy_dataset.household))
+        toy_dataset.person["is_claimant_or_partner"] = roles
+        result = clone_and_assign(
+            toy_dataset,
+            n_clones=3,
+            crosswalk_path=str(small_crosswalk),
+        )
+
+        assert result.person["is_claimant_or_partner"].dtype == bool
+        np.testing.assert_array_equal(
+            result.person["is_claimant_or_partner"].values, np.tile(roles, 3)
+        )
+
     def test_single_clone_is_near_identity(self, toy_dataset, small_crosswalk):
         """With n_clones=1, output should match input dimensions."""
         result = clone_and_assign(

policyengine_uk_data/tests/test_legacy_benefit_proxies.py

@@ -548,8 +548,10 @@ def fake_read_csv(path, *args, **kwargs):
         "age_started_or_accepted_current_education_or_training",
         "is_before_universal_credit_qualifying_young_person_terminal_date",
         "is_parent",
+        "is_claimant_or_partner",
     }.issubset(dataset.person.columns)
     assert not dataset.person["is_parent"].iloc[0]
+    assert dataset.person["is_claimant_or_partner"].iloc[0]
     assert not dataset.person["is_in_non_advanced_education"].iloc[0]
     assert not dataset.person["is_in_approved_training"].iloc[0]
     assert (

policyengine_uk_data/tests/test_policybench_transfer.py

@@ -26,6 +26,10 @@
     "pip_dl_category",
     "pip_m_category",
 }
+# Benefit-unit roles the source records, which policyengine-uk infers by
+# formula when a dataset leaves them out (is_claimant_or_partner from
+# policyengine-uk#1896). Releases that predate a role ignore its column.
+DATASET_SUPPLIED_ROLES = {"is_claimant_or_partner"}
 
 
 def _subset_source(tmp_path: Path, rows: int) -> Path:
@@ -81,11 +85,15 @@ def test_policybench_transfer_writes_only_valid_leaf_inputs(tmp_path: Path):
         invalid_columns = [
             column
             for column in frame.columns
-            if column not in system.variables
-            or system.variables[column].entity.key != entity
+            if (column not in system.variables and column not in DATASET_SUPPLIED_ROLES)
             or (
-                not system.variables[column].is_input_variable()
-                and column not in ALLOWED_COMPATIBILITY_INPUTS
+                column in system.variables
+                and system.variables[column].entity.key != entity
+            )
+            or (
+                column in system.variables
+                and not system.variables[column].is_input_variable()
+                and column not in ALLOWED_COMPATIBILITY_INPUTS | DATASET_SUPPLIED_ROLES
             )
         ]
         assert invalid_columns == []
@@ -228,6 +236,38 @@ def test_policybench_transfer_family_structure_matches_person_membership(
         assert observed == expected
 
 
+def test_policybench_transfer_marks_head_and_joint_spouse_as_claimant_or_partner(
+    tmp_path: Path,
+):
+    source_file_path = _subset_source(tmp_path, 200)
+    source = pd.read_csv(source_file_path)
+    dataset = create_enhanced_cps(source_file_path=source_file_path, calibrate=False)
+    person = dataset.person
+
+    counts = person.groupby("person_benunit_id").is_claimant_or_partner.sum()
+    expected = np.where(source.filing_status == "joint", 2, 1)
+    np.testing.assert_array_equal(counts.sort_index().to_numpy(), expected)
+    # The head and any spouse are listed before every other member.
+    first = person.groupby(
+        "person_benunit_id"
+    ).cumcount() < person.person_benunit_id.map(counts)
+    assert (person.is_claimant_or_partner == first).all()
+
+
+def test_checked_in_enhanced_cps_h5_roles_match_the_builder(tmp_path: Path):
+    checked_in = UKSingleYearDataset(file_path=str(ENHANCED_CPS_FILE)).person
+    built = create_enhanced_cps(
+        source_file_path=_subset_source(tmp_path, 500),
+        calibrate=False,
+    ).person
+    head = checked_in.iloc[: len(built)]
+
+    for column in ("person_id", "person_benunit_id", "age", "is_claimant_or_partner"):
+        np.testing.assert_array_equal(head[column].to_numpy(), built[column].to_numpy())
+    counts = checked_in.groupby("person_benunit_id").is_claimant_or_partner.sum()
+    assert counts.between(1, 2).all()
+
+
 def test_assign_council_tax_bands_handles_upper_percentile_edge():
     households = pd.DataFrame(
         {

policyengine_uk_data/tests/test_spi_build.py

@@ -467,3 +467,15 @@ def test_income_model_cache_accepts_current_spi_release(tmp_path, monkeypatch):
     )
 
     assert income_module.create_income_model().metadata == current_metadata
+
+
+def test_create_spi_marks_every_taxpayer_as_their_benefit_units_claimant(tmp_path):
+    from policyengine_uk_data.datasets.spi import create_spi
+
+    tab = tmp_path / "spi.tab"
+    _write_fake_spi(tab)
+
+    person = create_spi(tab, 2020, seed=0).person
+    assert person["is_claimant_or_partner"].dtype == bool
+    assert person["is_claimant_or_partner"].all()
+    assert person["person_benunit_id"].is_unique

policyengine_uk_data/utils/stack.py

@@ -1,10 +1,17 @@
 from policyengine_uk.data import UKSingleYearDataset
 import pandas as pd
 
+# Benefit-unit roles taken from the survey, which policyengine-uk otherwise
+# infers from ages. A stacked table without them would leave missing roles.
+SURVEY_ROLE_COLUMNS = ("is_claimant_or_partner",)
+
 
 def stack_datasets(
     data_1: UKSingleYearDataset, data_2: UKSingleYearDataset
 ) -> UKSingleYearDataset:
+    for column in SURVEY_ROLE_COLUMNS:
+        if (column in data_1.person.columns) != (column in data_2.person.columns):
+            raise ValueError(f"Only one of the stacked datasets has {column}.")
     person_id_offset = data_1.person.person_id.max() + 1
     benunit_id_offset = data_1.benunit.benunit_id.max() + 1
     household_id_offset = data_1.household.household_id.max() + 1