Keep FRS-reported dividends and key them on person_id

.gitignore

@@ -26,3 +26,6 @@ policyengine_uk_data/storage/local_geography_weights.csv.gz
 
 # Hypothesis example database (property-based tests)
 .hypothesis/
+
+# Workspace-local hub job notes and tooling
+.job/

changelog.d/frs-reported-dividends.fixed.md

@@ -0,0 +1,2 @@
+- FRS investment-account dividends are now summed to the person who holds the account. They were keyed on each person's row position, so almost none matched and the base FRS carried about £40m of dividends instead of about £8bn in 2024-25.
+- The enhanced FRS keeps those reported dividends instead of replacing them with draws from the SPI income model, which predicts from age, gender and region within earnings groups and gave Universal Credit claimants six-figure dividends (policyengine-uk#1948). The SPI-donor half still carries the SPI income distribution.

policyengine_uk_data/datasets/frs.py

@@ -48,6 +48,8 @@
 # loaders (e.g. LCFS/ETB in `datasets/imputations/consumption.py`) can import
 # the same value rather than re-defining `* 52` locally and drifting.
 WEEKS_IN_YEAR = 365.25 / 7
+# Grosses up an amount taxed at source at the basic rate (1 / 0.8).
+INVERTED_BASIC_RATE = 1.25
 
 LEGACY_JOBSEEKER_MIN_AGE = 18
 HOURS_WORKED_WEEKS_PER_YEAR = 52
@@ -790,6 +792,28 @@ def validate_frs_survey_year(raw_frs_folder, year: int) -> None:
         )
 
 
+def frs_dividend_income(account: pd.DataFrame, person_ids) -> np.ndarray:
+    """Annual dividends each person reports on FRS investment accounts.
+
+    Gilt-edged stock taxed at source (account type 6), unit and investment
+    trusts (7) and stocks and shares (8). Amounts taxed at source are grossed
+    up at the basic rate. Summed to ``person_ids``, which must be the same
+    ``household_id * 1e3 + person`` keys as ``account.person_id``: keying on
+    the positional index instead matched almost no accounts, leaving the FRS
+    with about £40m of dividends rather than about £8bn in 2024-25.
+    """
+    dividends = (
+        account.accint * np.where(account.invtax == 1, INVERTED_BASIC_RATE, 1)
+    ) * (
+        ((account.account == 6) & (account.invtax == 1))  # GGES
+        | account.account.isin((7, 8))  # Stocks/shares/UITs
+    )
+    return np.maximum(
+        0,
+        sum_to_entity(dividends, account.person_id, person_ids) * WEEKS_IN_YEAR,
+    )
+
+
 def derive_pension_credit_reported_capital(benunit: pd.DataFrame) -> np.ndarray:
     """Each benefit unit's capital as the FRS records it, for Pension Credit.
 
@@ -1267,8 +1291,6 @@ def determine_education_level(fted_val, typeed2_val, age_val):
         * 52
     ).astype(float)
 
-    WEEKS_IN_YEAR = 365.25 / 7
-
     pe_person["employment_income"] = np.maximum(0, person.inearns) * WEEKS_IN_YEAR
 
     pension_payment = sum_to_entity(
@@ -1303,8 +1325,6 @@ def determine_education_level(fted_val, typeed2_val, age_val):
         )
     )
 
-    INVERTED_BASIC_RATE = 1.25
-
     pe_person["tax_free_savings_income"] = np.maximum(
         0,
         sum_to_entity(
@@ -1327,19 +1347,7 @@ def determine_education_level(fted_val, typeed2_val, age_val):
         0,
         taxable_savings_interest + pe_person["tax_free_savings_income"].values,
     )
-    pe_person["dividend_income"] = np.maximum(
-        0,
-        sum_to_entity(
-            (account.accint * np.where(account.invtax == 1, INVERTED_BASIC_RATE, 1))
-            * (
-                ((account.account == 6) & (account.invtax == 1))  # GGES
-                | account.account.isin((7, 8))  # Stocks/shares/UITs
-            ),
-            account.person_id,
-            person.index,
-        )
-        * 52,
-    )
+    pe_person["dividend_income"] = frs_dividend_income(account, person.person_id)
     pe_person["property_income"] = frs_property_income(person, household)
     maintenance_to_self = np.maximum(
         pd.Series(np.where(person.mntus1 == 2, person.mntusam1, person.mntamt1)).fillna(

policyengine_uk_data/datasets/imputations/income.py

@@ -450,8 +450,9 @@ def impute_over_incomes(
 
     # Housing costs (rent, mortgage interest, mortgage capital) used to be
     # rescaled here by new_income_total / original_income_total across
-    # INCOME_COMPONENTS. Because FRS dividend_income is near-zero and the
-    # SPI-trained QRF predicts materially larger dividends, the ratio
+    # INCOME_COMPONENTS. Because FRS dividend_income was then near-zero (a
+    # keying error in frs.py, since fixed) and the SPI-trained QRF predicts
+    # materially larger dividends, the ratio
     # inflated rent/mortgage by ~2.5× uniformly in the built enhanced FRS
     # — pushing AHC poverty rates 10–18 pp above HBAI for non-pensioners
     # (see issue #367). Housing costs now pass through unchanged; their
@@ -488,13 +489,11 @@ def impute_income(dataset: UKSingleYearDataset) -> UKSingleYearDataset:
     Returns:
         Combined dataset with original data plus synthetic high-income individuals.
     """
-    # Impute wealth, assuming same time period as trained data
     dataset = dataset.copy()
     # gift_aid and charitable_investment_gifts are in IMPUTATIONS but are not
     # columns on the raw FRS build, so initialise them to zero everywhere
-    # before imputation. Without this, the full-FRS half stays NaN for these
-    # columns (they're never touched by the dividend-only impute_over_incomes
-    # call below), and the eventual stacked dataset fails validate().
+    # before imputation. The FRS half keeps its reported incomes, so these
+    # columns must already be populated when the stacked dataset is validated.
     for column in ("gift_aid", "charitable_investment_gifts"):
         if column not in dataset.person.columns:
             dataset.person[column] = 0.0
@@ -507,7 +506,7 @@ def impute_income(dataset: UKSingleYearDataset) -> UKSingleYearDataset:
 
     model = create_income_model()
 
-    # Impute just dividends on the original, full variable set on the copy
+    # Impute the full income set on the SPI-donor copy only.
 
     zero_weight_copy = impute_over_incomes(
         zero_weight_copy,
@@ -550,11 +549,14 @@ def impute_income(dataset: UKSingleYearDataset) -> UKSingleYearDataset:
             )
         )
 
-    dataset = impute_over_incomes(
-        dataset,
-        model,
-        ["dividend_income"],
-    )
+    # The FRS half keeps its reported dividends. Replacing them with a draw
+    # from the SPI model, which predicts from age, gender and region within
+    # earnings groups (whether the person has pay or a trade), gave
+    # dividends to FRS respondents without regard to their investments or
+    # benefits, so Universal Credit claimants received them
+    # (policyengine-uk#1948). The SPI-donor half carries the SPI dividend
+    # distribution, and calibration to the HMRC dividend targets reweights
+    # between the two.
 
     zero_weight_copy.validate()
     dataset.validate()

policyengine_uk_data/tests/test_frs_dividend_income.py

@@ -0,0 +1,82 @@
+"""FRS dividends land on the person who holds the account (policyengine-uk#1948)."""
+
+import numpy as np
+import pandas as pd
+import pytest
+from hypothesis import given, settings, strategies as st
+
+from policyengine_uk_data.datasets.frs import WEEKS_IN_YEAR, frs_dividend_income
+
+
+def test_dividends_are_keyed_on_person_id_not_row_position():
+    # Person ids are household_id * 1000 + person number, so they never
+    # coincide with the positional index the rows happen to sit at.
+    person_ids = pd.Series([1_001, 1_002, 2_001, 2_002])
+    account = pd.DataFrame(
+        {
+            "person_id": [1_002, 2_001, 2_001, 2_002, 1_001, 2_002],
+            # 8: stocks and shares; 7: unit/investment trusts; 6: gilts; 1: bank account
+            "account": [8, 7, 6, 6, 1, 8],
+            "accint": [10.0, 4.0, 2.0, 3.0, 50.0, 0.0],
+            "invtax": [2, 1, 1, 2, 2, 2],
+        }
+    )
+    dividends = frs_dividend_income(account, person_ids)
+    weekly = np.array(
+        [
+            0.0,  # 1,001 has only a bank account
+            10.0,  # 1,002: shares, not taxed at source
+            4.0 * 1.25 + 2.0 * 1.25,  # 2,001: trusts and gilts taxed at source
+            0.0,  # 2,002: gilts not taxed at source are excluded; shares pay 0
+        ]
+    )
+    assert dividends == pytest.approx(weekly * WEEKS_IN_YEAR)
+
+
+def test_dividends_are_never_negative():
+    person_ids = pd.Series([1_001])
+    account = pd.DataFrame(
+        {"person_id": [1_001], "account": [8], "accint": [-5.0], "invtax": [2]}
+    )
+    assert frs_dividend_income(account, person_ids).tolist() == [0.0]
+
+
+@st.composite
+def dividend_account_cases(draw):
+    household_ids = draw(
+        st.lists(st.integers(1, 50), min_size=1, max_size=8, unique=True)
+    )
+    person_ids = [household_id * 1_000 + 1 for household_id in household_ids]
+    rows = draw(
+        st.lists(
+            st.tuples(
+                st.sampled_from(person_ids),
+                st.sampled_from([1, 6, 7, 8, 9]),
+                st.integers(0, 1_000_000),
+                st.sampled_from([1, 2]),
+            ),
+            max_size=40,
+        )
+    )
+    account = pd.DataFrame(rows, columns=["person_id", "account", "pence", "invtax"])
+    account["accint"] = account.pop("pence") / 100
+    # Neither frame's index can be mistaken for an FRS person identifier.
+    account.index = np.arange(len(account)) * 7 + 100
+    person_ids = pd.Series(person_ids[::-1], index=np.arange(len(person_ids)) * 3 + 10)
+    return account, person_ids
+
+
+@given(dividend_account_cases())
+@settings(max_examples=100, deadline=None)
+def test_eligible_gross_dividends_are_conserved_when_joining_to_people(case):
+    account, person_ids = case
+    expected_by_person = {person_id: 0.0 for person_id in person_ids}
+    for row in account.itertuples():
+        if row.account in (7, 8) or (row.account == 6 and row.invtax == 1):
+            gross_weekly_amount = row.accint / (0.8 if row.invtax == 1 else 1)
+            expected_by_person[row.person_id] += gross_weekly_amount * WEEKS_IN_YEAR
+
+    dividends = frs_dividend_income(account, person_ids)
+
+    assert dividends.sum() == pytest.approx(sum(expected_by_person.values()))
+    assert dict(zip(person_ids, dividends)) == pytest.approx(expected_by_person)