Map FRS household codes from explicit data-dictionary tables

changelog.d/frs-household-codes.fixed.md

@@ -0,0 +1 @@
+Map the FRS council tax band (CTBAND) from an explicit table of the UKDS data dictionary codes. Code 10, which the 2024-25 dictionary labels "Information not found" and earlier releases "Household not valued separately", a blank band and the FRS missing-value codes now all take band D, the policyengine-uk default; a blank band used to become band A. A CTBAND, GVTREGNO, PTENTYP2 or TYPEACC code outside the data dictionary now stops the build with an error naming the FRS variable and code, instead of passing on silently (CTBAND) or as a missing value that policyengine-uk rejects without naming the variable.

policyengine_uk_data/datasets/frs.py

@@ -68,6 +68,26 @@
     10: EmploymentStatus.SHORT_TERM_DISABLED.name,  # Temporarily sick/injured
     11: EmploymentStatus.OTHER_INACTIVE.name,  # Other inactive
 }
+# Household-table CTBAND ("Council Tax band") value labels in the UKDS FRS
+# 2022-23, 2023-24 and 2024-25 data dictionaries. From 2024-25 DWP takes the
+# band from Valuation Office Agency and Scottish Assessors Association data
+# rather than asking the respondent, and labels band I "Wales only".
+FRS_CTBAND_COUNCIL_TAX_BAND = {
+    1: "A",
+    2: "B",
+    3: "C",
+    4: "D",
+    5: "E",
+    6: "F",
+    7: "G",
+    8: "H",
+    9: "I",
+}
+# CTBAND 10 gives no band: "Household not valued separately" to 2023-24 and
+# "Information not found" from 2024-25. Northern Ireland pays domestic rates
+# and leaves CTBAND blank. Both take policyengine-uk's default band.
+FRS_CTBAND_NO_BAND = 10
+COUNCIL_TAX_PLACEHOLDER_BAND = "D"
 ESA_HEALTH_EMPLOYMENT_STATUSES = (
     EmploymentStatus.LONG_TERM_DISABLED.name,
     EmploymentStatus.SHORT_TERM_DISABLED.name,
@@ -175,6 +195,55 @@ def derive_employment_status_from_frs(empstati, is_adult_record) -> np.ndarray:
     return np.where(is_adult_record, adult_status, EmploymentStatus.CHILD.name)
 
 
+def _list_codes(codes: pd.Series) -> str:
+    listed = [f"{code:g}" for code in sorted(codes.dropna().unique())]
+    listed += ["blank"] if codes.isna().any() else []
+    return ", ".join(listed)
+
+
+def map_frs_codes(values, mapping, variable: str, blank) -> np.ndarray:
+    """Map an FRS categorical variable's value codes through ``mapping``.
+
+    Blank values take ``mapping[blank]``. Any other code missing from
+    ``mapping`` fails the build naming the FRS variable, rather than becoming a
+    missing value that policyengine-uk later rejects as an unknown enum value
+    without saying which survey variable it came from.
+    """
+
+    codes = pd.Series(np.asarray(values, dtype=float))
+    mapped = codes.fillna(blank).map(mapping)
+    unknown = mapped.isna()
+    if unknown.any():
+        # Build logs are public: name the codes, never a count.
+        raise ValueError(
+            f"FRS {variable} has codes with no mapping: "
+            f"{_list_codes(codes[unknown])}. Map them from the release's data "
+            "dictionary."
+        )
+    return mapped.to_numpy()
+
+
+def derive_council_tax_band_from_frs(ctband) -> np.ndarray:
+    """Map FRS CTBAND codes to ``council_tax_band``.
+
+    Codes 1 to 9 are bands A to I. Code 10 (no band), blanks and the FRS
+    missing-value codes (zero or below) take ``COUNCIL_TAX_PLACEHOLDER_BAND``.
+    Any other code is a new data-dictionary value and fails the build.
+    """
+
+    codes = pd.Series(np.asarray(ctband, dtype=float))
+    band = codes.map(FRS_CTBAND_COUNCIL_TAX_BAND)
+    no_band = codes.isna() | (codes <= 0) | (codes == FRS_CTBAND_NO_BAND)
+    unknown = band.isna() & ~no_band
+    if unknown.any():
+        raise ValueError(
+            "FRS CTBAND has codes missing from FRS_CTBAND_COUNCIL_TAX_BAND: "
+            f"{_list_codes(codes[unknown])}. Map them from the release's data "
+            "dictionary."
+        )
+    return band.where(~no_band, COUNCIL_TAX_PLACEHOLDER_BAND).to_numpy()
+
+
 def derive_legacy_jobseeker_proxy(
     age,
     employment_status,
@@ -1174,9 +1243,12 @@ def determine_education_level(fted_val, typeed2_val, age_val):
         "NORTHERN_IRELAND",
         "UNKNOWN",
     ]
-    pe_household["region"] = categorical(
-        household.gvtregno, 14, [1, 2] + list(range(4, 15)), REGIONS
-    ).values
+    pe_household["region"] = map_frs_codes(
+        household.gvtregno,
+        dict(zip([1, 2] + list(range(4, 15)), REGIONS)),
+        "GVTREGNO",
+        blank=14,
+    )
     TENURES = [
         "RENT_FROM_COUNCIL",
         "RENT_FROM_HA",
@@ -1185,9 +1257,9 @@ def determine_education_level(fted_val, typeed2_val, age_val):
         "OWNED_OUTRIGHT",
         "OWNED_WITH_MORTGAGE",
     ]
-    pe_household["tenure_type"] = categorical(
-        household.ptentyp2, 3, range(1, 7), TENURES
-    ).values
+    pe_household["tenure_type"] = map_frs_codes(
+        household.ptentyp2, dict(zip(range(1, 7), TENURES)), "PTENTYP2", blank=3
+    )
     frs["num_bedrooms"] = household.bedroom6
     ACCOMMODATIONS = [
         "HOUSE_DETACHED",
@@ -1198,9 +1270,9 @@ def determine_education_level(fted_val, typeed2_val, age_val):
         "MOBILE",
         "OTHER",
     ]
-    pe_household["accommodation_type"] = categorical(
-        household.typeacc, 1, range(1, 8), ACCOMMODATIONS
-    ).values
+    pe_household["accommodation_type"] = map_frs_codes(
+        household.typeacc, dict(zip(range(1, 8), ACCOMMODATIONS)), "TYPEACC", blank=1
+    )
 
     # Impute Council Tax
 
@@ -1264,10 +1336,8 @@ def determine_education_level(fted_val, typeed2_val, age_val):
         )
     )
     pe_household["council_tax"] = council_tax.fillna(0)
-    BANDS = ["A", "B", "C", "D", "E", "F", "G", "H", "I"]
-    # Band 1 is the most common
-    pe_household["council_tax_band"] = (
-        categorical(household.ctband, 1, range(1, 10), BANDS).fillna("D").values
+    pe_household["council_tax_band"] = derive_council_tax_band_from_frs(
+        household.ctband
     )
     # Domestic rates variables are all weeklyised, unlike Council Tax variables (despite the variable name suggesting otherwise)
     if year < 2021:

policyengine_uk_data/tests/test_frs_categorical_codes.py

@@ -0,0 +1,104 @@
+"""FRS household categorical codes against the UKDS data dictionaries.
+
+Value labels are from the UKDS FRS variable listings for 2022-23 (SN 9252),
+2023-24 (SN 9367) and 2024-25 (SN 9563), household table.
+"""
+
+import numpy as np
+import pytest
+from hypothesis import given, strategies as st
+from policyengine_uk.variables.input.council_tax_band import CouncilTaxBand
+
+from policyengine_uk_data.datasets.frs import (
+    COUNCIL_TAX_PLACEHOLDER_BAND,
+    derive_council_tax_band_from_frs,
+    map_frs_codes,
+)
+
+# CTBAND labels, identical in all three releases except code 9 ("Band I" to
+# 2023-24, "Band I - Wales only" in 2024-25) and code 10.
+CTBAND_BANDS = {
+    1: "A",
+    2: "B",
+    3: "C",
+    4: "D",
+    5: "E",
+    6: "F",
+    7: "G",
+    8: "H",
+    9: "I",
+}
+# Code 10: "Household not valued separately" (2022-23, 2023-24),
+# "Information not found" (2024-25).
+CTBAND_NO_BAND = 10
+
+
+@pytest.mark.parametrize("code, band", CTBAND_BANDS.items())
+def test_ctband_codes_1_to_9_are_bands_a_to_i(code, band):
+    assert derive_council_tax_band_from_frs([code]).tolist() == [band]
+
+
+def test_ctband_10_takes_the_policyengine_uk_default_band():
+    assert derive_council_tax_band_from_frs([CTBAND_NO_BAND]).tolist() == ["D"]
+    assert COUNCIL_TAX_PLACEHOLDER_BAND == CouncilTaxBand.D.name
+
+
+def test_blank_and_missing_value_ctband_take_the_same_band_as_code_10():
+    # A blank band used to become A while code 10 became D.
+    assert derive_council_tax_band_from_frs([np.nan, -1, 0, 10]).tolist() == [
+        "D",
+        "D",
+        "D",
+        "D",
+    ]
+
+
+@pytest.mark.parametrize("code", [11, 12, 99, 4.5])
+def test_a_ctband_code_outside_the_dictionary_fails_naming_ctband(code):
+    with pytest.raises(ValueError, match="CTBAND has codes missing .*" + f"{code:g}"):
+        derive_council_tax_band_from_frs([1, code])
+
+
+@given(
+    st.lists(
+        st.one_of(
+            st.sampled_from(list(CTBAND_BANDS) + [CTBAND_NO_BAND, 0, -1]),
+            st.just(np.nan),
+        ),
+        max_size=50,
+    )
+)
+def test_every_ctband_value_is_a_policyengine_uk_band(codes):
+    bands = derive_council_tax_band_from_frs(codes)
+    # policyengine-uk raises on any string that is not a band name.
+    CouncilTaxBand.encode(np.asarray(bands, dtype=str))
+    assert len(bands) == len(codes)
+
+
+TENURES = dict(
+    zip(
+        range(1, 7),
+        [
+            "RENT_FROM_COUNCIL",
+            "RENT_FROM_HA",
+            "RENT_PRIVATELY",
+            "RENT_PRIVATELY",
+            "OWNED_OUTRIGHT",
+            "OWNED_WITH_MORTGAGE",
+        ],
+    )
+)
+
+
+def test_map_frs_codes_maps_codes_and_blanks():
+    assert map_frs_codes([1, np.nan, 6], TENURES, "PTENTYP2", blank=3).tolist() == [
+        "RENT_FROM_COUNCIL",
+        "RENT_PRIVATELY",
+        "OWNED_WITH_MORTGAGE",
+    ]
+
+
+@pytest.mark.parametrize("code", [0, -1, 7])
+def test_map_frs_codes_names_the_variable_and_unknown_code(code):
+    with pytest.raises(ValueError, match=f"FRS PTENTYP2 .*: {code:g}\\."):
+        map_frs_codes([1, code], TENURES, "PTENTYP2", blank=3)

policyengine_uk_data/tests/test_legacy_benefit_proxies.py

@@ -442,6 +442,7 @@ def create_single_adult_frs(
     with_child=False,
     child_overrides=None,
     adult_overrides=None,
+    household_overrides=None,
 ):
     monkeypatch.setattr(policyengine_uk, "Microsimulation", FakeMicrosimulation)
     monkeypatch.setattr(frs_module, "load_take_up_rate", lambda *args, **kwargs: 0.0)
@@ -553,15 +554,18 @@ def create_single_adult_frs(
                 "gvtregno": 1,
                 "hhrent": 0,
                 "mortint": 0,
-                "ptentyp2": 0,
+                # Data-dictionary codes: rented privately unfurnished, and a
+                # detached house. 0 is not a PTENTYP2 or TYPEACC code.
+                "ptentyp2": 3,
                 "rt2rebam": 0,
                 "struins": 0,
                 "subrent": 0,
                 "tentyp2": 0,
-                "typeacc": 0,
+                "typeacc": 1,
                 "watsewrt": 0,
                 "niratlia": 0,
                 **{f"chrgamt{i}": 0 for i in range(1, 10)},
+                **(household_overrides or {}),
             }
         ]
     )
@@ -660,6 +664,31 @@ def test_create_frs_child_rows_are_child(tmp_path, monkeypatch):
     }
 
 
+# CTBAND 10 is "Information not found" in the FRS 2024-25 data dictionary and
+# "Household not valued separately" before it; a blank is Northern Ireland.
+@pytest.mark.parametrize(
+    "ctband, band", [(1, "A"), (8, "H"), (9, "I"), (10, "D"), (np.nan, "D")]
+)
+def test_create_frs_maps_ctband(tmp_path, monkeypatch, ctband, band):
+    household = create_single_adult_frs(
+        tmp_path, monkeypatch, household_overrides={"ctband": ctband}
+    ).household
+
+    assert household["council_tax_band"].tolist() == [band]
+
+
+@pytest.mark.parametrize(
+    "column, code", [("ctband", 11), ("gvtregno", 3), ("ptentyp2", 7), ("typeacc", 8)]
+)
+def test_create_frs_names_an_unknown_household_code(
+    tmp_path, monkeypatch, column, code
+):
+    with pytest.raises(ValueError, match=f"FRS {column.upper()} has codes"):
+        create_single_adult_frs(
+            tmp_path, monkeypatch, household_overrides={column: code}
+        )
+
+
 def test_create_frs_rejects_unknown_adult_empstati(tmp_path, monkeypatch):
     with pytest.raises(ValueError, match="EMPSTATI"):
         create_single_adult_frs(tmp_path, monkeypatch, empstati=12)