diff --git a/changelog.d/468.fixed.md b/changelog.d/468.fixed.md new file mode 100644 index 000000000..3980abe1f --- /dev/null +++ b/changelog.d/468.fixed.md @@ -0,0 +1,10 @@ +Align local Universal Credit targets by area code, preserve constituencies' 2024 boundaries, and use each matched constituency's country for its children split. + +Restore the correct 2024 code order for the mapped income, age and housing targets of all 57 Scottish constituencies. This is a pure reorder; Scotland's totals are unchanged. Target moves from the previous positional assignment are: + +| Target | Rows moving >5% | Rows moving >25% | Maximum move | +|---|---:|---:|---:| +| Age population | 44 | 11 | 269.7% | +| Housing anchor | 44 | 11 | 269.7% | +| Employment income amount | 53 | 26 | 245% | +| Self-employment income amount | 53 | 37 | 799% | diff --git a/policyengine_uk_data/calibration/matrix_builder.py b/policyengine_uk_data/calibration/matrix_builder.py index 820e7e74e..0d213b7fd 100644 --- a/policyengine_uk_data/calibration/matrix_builder.py +++ b/policyengine_uk_data/calibration/matrix_builder.py @@ -189,14 +189,22 @@ def _load_area_targets( get_uk_total_population, ) - y = pd.DataFrame() time_period = dataset.time_period # ── Income targets ────────────────────────────────────────────── if area_type == "constituency": incomes = get_constituency_income_targets() + age_targets = get_constituency_age_targets() else: incomes = get_la_income_targets() + age_targets = get_la_age_targets() + + # SPI income and age inputs share this source order. Keep the codes + # through any boundary mapping before selecting the requested rows. + assert incomes["code"].equals(age_targets["code"]), ( + "Income and age code order differs" + ) + y = pd.DataFrame(index=pd.Index(age_targets["code"], name="code")) national_incomes = get_national_income_projections(int(time_period)) @@ -215,11 +223,6 @@ def _load_area_targets( ) # ── Age targets ───────────────────────────────────────────────── - if area_type == "constituency": - age_targets = get_constituency_age_targets() - else: - age_targets = get_la_age_targets() - uk_total_population = get_uk_total_population(int(time_period)) age_cols = [c for c in age_targets.columns if c.startswith("age/")] @@ -231,31 +234,40 @@ def _load_area_targets( for col in age_cols: y[col] *= uk_total_population / targets_total_pop * 0.9 + # ── Boundary mapping (constituency 2010 → 2024 only) ─────────── + if area_type == "constituency": + from policyengine_uk_data.datasets.local_areas.constituencies.boundary_changes.mapping_matrix import ( + mapping_matrix, + old_codes, + new_codes, + ) + + y = pd.DataFrame( + mapping_matrix @ y.loc[old_codes].to_numpy(), + index=pd.Index(new_codes, name="code"), + columns=y.columns, + ) + + # Both the OA assignment matrix and the LA extras use area_codes order. + y = y.loc[area_codes["code"]] + # ── UC targets ────────────────────────────────────────────────── + # Constituency UC counts are already on 2024 boundaries and must stay + # outside the mapping applied to the older income and age sources. if area_type == "constituency": from policyengine_uk_data.targets.sources.local_uc import ( get_constituency_uc_targets, get_constituency_uc_by_children_targets, ) - y["uc_households"] = get_constituency_uc_targets().values - uc_by_children = get_constituency_uc_by_children_targets() + y["uc_households"] = get_constituency_uc_targets().loc[y.index] + uc_by_children = get_constituency_uc_by_children_targets().loc[y.index] for col in uc_by_children.columns: - y[col] = uc_by_children[col].values + y[col] = uc_by_children[col] else: from policyengine_uk_data.targets.sources.local_uc import get_la_uc_targets - y["uc_households"] = get_la_uc_targets().values - - # ── Boundary mapping (constituency 2010 → 2024 only) ─────────── - if area_type == "constituency": - from policyengine_uk_data.datasets.local_areas.constituencies.boundary_changes.mapping_matrix import ( - mapping_matrix, - ) - - y_columns = list(y.columns) - y_values = mapping_matrix @ y.values - y = pd.DataFrame(y_values, columns=y_columns) + y["uc_households"] = get_la_uc_targets().loc[y.index] # ── LA extras: ONS income, tenure, rent ───────────────────────── if area_type == "la": @@ -264,8 +276,7 @@ def _load_area_targets( load_household_counts, load_tenure_data, load_private_rents, - UPRATING_NET_INCOME_BHC_2020_TO_2025, - UPRATING_HOUSING_COSTS_2020_TO_2025, + get_ons_income_uprating_factors, ) original_weights = sim.calculate("household_weight", 2025).values @@ -288,16 +299,19 @@ def _load_area_targets( "equiv_hbai_household_net_income_ahc" ).values housing_costs = hbai_net_income - hbai_net_income_ahc + income_bhc_uprating_factor, housing_costs_uprating_factor = ( + get_ons_income_uprating_factors(int(time_period)) + ) ons_merged["equiv_net_income_bhc_target"] = ( ons_merged["net_income_bhc"] * ons_merged["households"] - * UPRATING_NET_INCOME_BHC_2020_TO_2025 + * income_bhc_uprating_factor ) ons_merged["equiv_housing_costs_target"] = ( ons_merged["net_income_bhc"] * ons_merged["households"] - ons_merged["net_income_ahc"] * ons_merged["households"] - ) * UPRATING_HOUSING_COSTS_2020_TO_2025 + ) * housing_costs_uprating_factor ons_merged["equiv_net_income_ahc_target"] = ( ons_merged["equiv_net_income_bhc_target"] - ons_merged["equiv_housing_costs_target"] @@ -385,7 +399,7 @@ def _load_area_targets( national_rent * la_household_share, ) - return y + return y.reset_index(drop=True) def create_cloned_target_matrix( diff --git a/policyengine_uk_data/datasets/local_areas/constituencies/boundary_changes/mapping_matrix.py b/policyengine_uk_data/datasets/local_areas/constituencies/boundary_changes/mapping_matrix.py index 457133881..27bd965c9 100644 --- a/policyengine_uk_data/datasets/local_areas/constituencies/boundary_changes/mapping_matrix.py +++ b/policyengine_uk_data/datasets/local_areas/constituencies/boundary_changes/mapping_matrix.py @@ -5,6 +5,19 @@ # 1. Read the boundary change data into a DataFrame df = pd.read_csv(Path(__file__).parent / "boundary_changes.csv") +# Five unchanged Scottish constituencies retain their 2010 codes in the +# boundary-change source. Use their 2024 codes from constituencies_2024.csv +# so the mapped rows can be joined to the other 2024 targets by code. +df["code_2024"] = df["code_2024"].replace( + { + "S14000006": "S14000107", # Ayr, Carrick and Cumnock + "S14000008": "S14000108", # Berwickshire, Roxburgh and Selkirk + "S14000010": "S14000109", # Central Ayrshire + "S14000040": "S14000110", # Kilmarnock and Loudoun + "S14000058": "S14000111", # West Aberdeenshire and Kincardine + } +) + # 2. Get unique constituency codes for 2010 and 2024, sorted alphabetically old_codes = sorted(df["code_2010"].unique()) new_codes = sorted(df["code_2024"].unique()) diff --git a/policyengine_uk_data/datasets/local_areas/constituencies/loss.py b/policyengine_uk_data/datasets/local_areas/constituencies/loss.py index 1fb55f13a..745739766 100644 --- a/policyengine_uk_data/datasets/local_areas/constituencies/loss.py +++ b/policyengine_uk_data/datasets/local_areas/constituencies/loss.py @@ -18,6 +18,8 @@ from policyengine_uk_data.storage import STORAGE_FOLDER from policyengine_uk_data.datasets.local_areas.constituencies.boundary_changes.mapping_matrix import ( mapping_matrix, + old_codes, + new_codes, ) from policyengine_uk_data.targets.sources.local_age import ( get_constituency_age_targets, @@ -49,10 +51,14 @@ def create_constituency_target_matrix( sim.default_calculation_period = time_period matrix = pd.DataFrame() - y = pd.DataFrame() + age_targets = get_constituency_age_targets() + y = pd.DataFrame(index=pd.Index(age_targets["code"], name="code")) # ── Income targets ───────────────────────────────────────────── incomes = get_constituency_income_targets() + assert incomes["code"].equals(age_targets["code"]), ( + "Income and age code order differs" + ) national_incomes = get_national_income_projections(int(time_period)) for income_variable in INCOME_VARIABLES: @@ -78,7 +84,6 @@ def create_constituency_target_matrix( ) # ── Age targets ──────────────────────────────────────────────── - age_targets = get_constituency_age_targets() uk_total_population = get_uk_total_population(int(time_period)) age = sim.calculate("age").values @@ -97,7 +102,6 @@ def create_constituency_target_matrix( y[col] *= uk_total_population / targets_total_pop * 0.9 # ── UC targets ───────────────────────────────────────────────── - y["uc_households"] = get_constituency_uc_targets().values on_uc = (sim.calculate("universal_credit").values > 0).astype(int) matrix["uc_households"] = sim.map_result(on_uc, "benunit", "household") @@ -113,10 +117,6 @@ def create_constituency_target_matrix( matrix["uc_hh_2_children"] = (on_uc_hh & (children_per_hh == 2)).astype(float) matrix["uc_hh_3plus_children"] = (on_uc_hh & (children_per_hh >= 3)).astype(float) - uc_by_children = get_constituency_uc_by_children_targets() - for col in uc_by_children.columns: - y[col] = uc_by_children[col].values - # ── Wales/Scotland housing anchors ─────────────────────────────── add_private_rent_targets( matrix, @@ -130,15 +130,28 @@ def create_constituency_target_matrix( # ── Boundary mapping (2010 → 2024) ──────────────────────────── const_2024 = pd.read_csv(STORAGE_FOLDER / "constituencies_2024.csv") - y_columns = list(y.columns) - y_values = mapping_matrix @ y.values - y = pd.DataFrame(y_values, columns=y_columns) + # The mapping's columns and rows are sorted by 2010 and 2024 codes, + # respectively; neither order should be inferred from a target CSV. + y = pd.DataFrame( + mapping_matrix @ y.loc[old_codes].to_numpy(), + index=pd.Index(new_codes, name="code"), + columns=y.columns, + ) + + # UC is already on 2024 boundaries, so attach it only after mapping + # the older income, age and housing targets. + y = y.loc[const_2024["code"]] + y["uc_households"] = get_constituency_uc_targets().loc[y.index] + uc_by_children = get_constituency_uc_by_children_targets().loc[y.index] + for col in uc_by_children.columns: + y[col] = uc_by_children[col] + y = y.loc[:, matrix.columns] country_mask = create_country_mask( household_countries=sim.calculate("country").values, codes=const_2024.code, ) - return matrix, y, country_mask + return matrix, y.reset_index(drop=True), country_mask def create_country_mask( diff --git a/policyengine_uk_data/datasets/local_areas/local_authorities/loss.py b/policyengine_uk_data/datasets/local_areas/local_authorities/loss.py index d2455e22f..38abd4d01 100644 --- a/policyengine_uk_data/datasets/local_areas/local_authorities/loss.py +++ b/policyengine_uk_data/datasets/local_areas/local_authorities/loss.py @@ -58,15 +58,23 @@ def create_local_authority_target_matrix( time_period = dataset.time_period la_codes = pd.read_csv(STORAGE_FOLDER / "local_authorities_2021.csv") + age_targets = get_la_age_targets() + if not pd.Index(age_targets["code"]).equals(pd.Index(la_codes["code"])): + raise ValueError( + "LA age targets and local authority codes must have the same order" + ) sim = Microsimulation(dataset=dataset, reform=reform) sim.default_calculation_period = time_period matrix = pd.DataFrame() - y = pd.DataFrame() + y = pd.DataFrame(index=pd.Index(age_targets["code"], name="code")) # ── Income targets ───────────────────────────────────────────── incomes = get_la_income_targets() + assert incomes["code"].equals(age_targets["code"]), ( + "Income and age code order differs" + ) national_incomes = get_national_income_projections(int(time_period)) for income_variable in INCOME_VARIABLES: @@ -92,7 +100,6 @@ def create_local_authority_target_matrix( ) # ── Age targets ──────────────────────────────────────────────── - age_targets = get_la_age_targets() uk_total_population = get_uk_total_population(int(time_period)) age = sim.calculate("age").values @@ -110,7 +117,7 @@ def create_local_authority_target_matrix( y[col] *= uk_total_population / targets_total_pop * 0.9 # ── UC targets ───────────────────────────────────────────────── - y["uc_households"] = get_la_uc_targets().values + y["uc_households"] = get_la_uc_targets().loc[y.index] matrix["uc_households"] = sim.map_result( (sim.calculate("universal_credit").values > 0).astype(int), "benunit", @@ -305,7 +312,7 @@ def create_local_authority_target_matrix( codes=la_codes.code, ) - return matrix, y, country_mask + return matrix, y.reset_index(drop=True), country_mask def create_country_mask( diff --git a/policyengine_uk_data/db/etl.py b/policyengine_uk_data/db/etl.py index 84303a6ae..3b4ab578a 100644 --- a/policyengine_uk_data/db/etl.py +++ b/policyengine_uk_data/db/etl.py @@ -429,21 +429,19 @@ def _load_local_uc_targets(session: Session) -> int: from policyengine_uk_data.utils import uc_data - df = getattr(uc_data, loader_attr) + df = getattr(uc_data, loader_attr).set_index("code", verify_integrity=True) if level == "constituency": codes_df = pd.read_csv(STORAGE_FOLDER / "constituencies_2024.csv") else: codes_df = pd.read_csv(STORAGE_FOLDER / "local_authorities_2021.csv") - for i, hh_count in enumerate(df.household_count.values): - if i >= len(codes_df): - break - code = codes_df.iloc[i]["code"] - area_name = codes_df.iloc[i].get("name", "") + areas = codes_df.set_index("code", verify_integrity=True) + for code, hh_count in df["household_count"].items(): + area_name = areas.loc[code, "name"] value = float(hh_count) - if np.isnan(value): - continue + if not np.isfinite(value): + raise ValueError(f"Non-finite UC household target for {code}") target_name = f"dwp/{level}/{code}/uc_households" _insert_target( session, diff --git a/policyengine_uk_data/targets/sources/local_uc.py b/policyengine_uk_data/targets/sources/local_uc.py index a8420c964..a2b2f802f 100644 --- a/policyengine_uk_data/targets/sources/local_uc.py +++ b/policyengine_uk_data/targets/sources/local_uc.py @@ -71,14 +71,12 @@ def _scaled_uc_children_by_country(gb_total_households: float) -> dict[str, np.n def get_constituency_uc_targets() -> pd.Series: - """UC household counts for 650 constituencies (positional order). - - Returns Series of household_count values, aligned to the same - ordering as the constituency age.csv. - """ + """UC household counts indexed by 2024 constituency code.""" from policyengine_uk_data.utils.uc_data import uc_pc_households - return uc_pc_households.household_count + return uc_pc_households.set_index("code", verify_integrity=True)[ + "household_count" + ].sort_index() def get_constituency_uc_by_children_targets() -> pd.DataFrame: @@ -87,50 +85,43 @@ def get_constituency_uc_by_children_targets() -> pd.DataFrame: Applies country-level proportions from Stat-Xplore to each constituency's total UC count. Returns a DataFrame with columns ``uc_hh_0_children``, ``uc_hh_1_child``, ``uc_hh_2_children``, - ``uc_hh_3plus_children``, in the same positional order as - :func:`get_constituency_uc_targets`. + ``uc_hh_3plus_children``, indexed by 2024 constituency code. Integer + buckets sum exactly to each area's total, using its matched country. """ - from policyengine_uk_data.utils.uc_data import uc_pc_households - from policyengine_uk_data.storage import STORAGE_FOLDER - - codes = pd.read_csv(STORAGE_FOLDER / "constituencies_2024.csv")["code"] - totals = uc_pc_households.household_count.values.astype(float) + totals = get_constituency_uc_targets() - result = pd.DataFrame(index=range(len(totals))) cols = [ "uc_hh_0_children", "uc_hh_1_child", "uc_hh_2_children", "uc_hh_3plus_children", ] - for col in cols: - result[col] = 0.0 - - gb_total = totals[codes.str[0].isin(["E", "W", "S"])].sum() + gb_total = totals[totals.index.str[0].isin(["E", "W", "S"])].sum() country_buckets = _scaled_uc_children_by_country(gb_total) - for i, (total, code) in enumerate(zip(totals, codes)): - country_prefix = code[0] - proportions = country_buckets.get( - country_prefix, - country_buckets["N"], # fallback - ) + rows = [] + for code, total in totals.items(): + proportions = country_buckets[code[0]] shares = proportions / proportions.sum() - for j, col in enumerate(cols): - result.loc[i, col] = round(total * shares[j]) + expected = total * shares + counts = np.floor(expected).astype(int) + # Allocate remaining households to the largest fractional parts. + # Independent rounding can otherwise lose or add a household. + remainder = int(total - counts.sum()) + order = np.argsort(-(expected - counts), kind="stable") + counts[order[:remainder]] += 1 + rows.append(counts) - return result + return pd.DataFrame(rows, index=totals.index, columns=cols) def get_la_uc_targets() -> pd.Series: - """UC household counts for 360 local authorities (positional order). - - Returns Series of household_count values, aligned to the same - ordering as the LA age.csv. - """ + """UC household counts indexed by local authority code.""" from policyengine_uk_data.utils.uc_data import uc_la_households - return uc_la_households.household_count + return uc_la_households.set_index("code", verify_integrity=True)[ + "household_count" + ].sort_index() REFERENCE_URL = _REF diff --git a/policyengine_uk_data/tests/test_devolved_housing_targets.py b/policyengine_uk_data/tests/test_devolved_housing_targets.py index 0619bd938..4fc5c351d 100644 --- a/policyengine_uk_data/tests/test_devolved_housing_targets.py +++ b/policyengine_uk_data/tests/test_devolved_housing_targets.py @@ -119,8 +119,10 @@ def map_result(self, values, source_entity, target_entity): def test_constituency_target_matrix_includes_devolved_housing_targets(monkeypatch): age_targets = _age_targets().iloc[[0, 2]].reset_index(drop=True) + codes = age_targets["code"].tolist() income_targets = pd.DataFrame( { + "code": codes, "self_employment_income_amount": [1.0, 1.0], "self_employment_income_count": [1.0, 1.0], "employment_income_amount": [1.0, 1.0], @@ -141,7 +143,8 @@ def test_constituency_target_matrix_includes_devolved_housing_targets(monkeypatc "uc_hh_1_child": [0.0, 0.0], "uc_hh_2_children": [0.0, 0.0], "uc_hh_3plus_children": [0.0, 0.0], - } + }, + index=pd.Index(codes, name="code"), ) monkeypatch.setattr(constituency_loss, "Microsimulation", _FakeSim) @@ -160,7 +163,7 @@ def test_constituency_target_matrix_includes_devolved_housing_targets(monkeypatc monkeypatch.setattr( constituency_loss, "get_constituency_uc_targets", - lambda: pd.Series([1.0, 1.0]), + lambda: pd.Series([1.0, 1.0], index=pd.Index(codes, name="code")), ) monkeypatch.setattr( constituency_loss, @@ -168,6 +171,8 @@ def test_constituency_target_matrix_includes_devolved_housing_targets(monkeypatc lambda: uc_by_children, ) monkeypatch.setattr(constituency_loss, "mapping_matrix", np.eye(2)) + monkeypatch.setattr(constituency_loss, "old_codes", codes) + monkeypatch.setattr(constituency_loss, "new_codes", codes) monkeypatch.setattr( constituency_loss.pd, "read_csv", diff --git a/policyengine_uk_data/tests/test_la_loss_missing_sources.py b/policyengine_uk_data/tests/test_la_loss_missing_sources.py index 11f078fe2..f2c7d394b 100644 --- a/policyengine_uk_data/tests/test_la_loss_missing_sources.py +++ b/policyengine_uk_data/tests/test_la_loss_missing_sources.py @@ -50,6 +50,7 @@ def _patch_common_la_inputs(monkeypatch, tmp_path): "get_la_income_targets", lambda: pd.DataFrame( { + "code": _fake_la_codes()["code"], "employment_income_amount": [1.0, 1.0, 1.0, 1.0], "employment_income_count": [1.0, 1.0, 1.0, 1.0], } @@ -69,10 +70,19 @@ def _patch_common_la_inputs(monkeypatch, tmp_path): monkeypatch.setattr( loss, "get_la_age_targets", - lambda: pd.DataFrame({"age/0_100": [1.0, 1.0, 1.0, 1.0]}), + lambda: pd.DataFrame( + { + "code": _fake_la_codes()["code"], + "age/0_100": [1.0, 1.0, 1.0, 1.0], + } + ), ) monkeypatch.setattr(loss, "get_uk_total_population", lambda year: 4.0) - monkeypatch.setattr(loss, "get_la_uc_targets", lambda: pd.Series([0, 1, 0, 0])) + monkeypatch.setattr( + loss, + "get_la_uc_targets", + lambda: pd.Series([0, 1, 0, 0], index=_fake_la_codes()["code"]), + ) monkeypatch.setattr( loss, "get_ons_income_uprating_factors", diff --git a/policyengine_uk_data/tests/test_local_uc_alignment.py b/policyengine_uk_data/tests/test_local_uc_alignment.py new file mode 100644 index 000000000..bd3537eb0 --- /dev/null +++ b/policyengine_uk_data/tests/test_local_uc_alignment.py @@ -0,0 +1,459 @@ +"""Local UC targets retain their area identity through both calibration paths. + +The oracle reads only committed public Stat-Xplore/DfC workbooks and geography +metadata. A four-household simulation stub exercises the target builders without +opening survey data or building/recalibrating a dataset. +""" + +from functools import lru_cache +from pathlib import Path +import re +from types import SimpleNamespace +import unicodedata + +from hypothesis import given, settings, strategies as st +import numpy as np +import pandas as pd +import pytest + +from policyengine_uk_data.storage import STORAGE_FOLDER +from policyengine_uk_data.targets.sources import local_uc +from policyengine_uk_data.utils import uc_data + + +_READ_EXCEL = pd.read_excel +_AREA_FILES = { + "constituency": "constituencies_2024.csv", + "la": "local_authorities_2021.csv", +} +_CHILD_COLUMNS = [ + "uc_hh_0_children", + "uc_hh_1_child", + "uc_hh_2_children", + "uc_hh_3plus_children", +] +_COUNTRIES = { + "E": "ENGLAND", + "W": "WALES", + "S": "SCOTLAND", + "N": "NORTHERN_IRELAND", +} + + +@lru_cache +def _workbook(filename, sheet_name=0): + return _READ_EXCEL(STORAGE_FOLDER / filename, sheet_name=sheet_name, header=None) + + +def _name_key(name): + # Stat-Xplore's Welsh LA labels have the English name before the slash. + english = str(name).split(" / ")[0].casefold().replace("&", "and") + unaccented = "".join( + c + for c in unicodedata.normalize("NFKD", english) + if not unicodedata.combining(c) + ) + return re.sub(r"[^a-z0-9]", "", unaccented) + + +@lru_cache +def _area_metadata(area_type): + return pd.read_csv(STORAGE_FOLDER / _AREA_FILES[area_type]) + + +@lru_cache +def _national_payment_total(): + table = _workbook("uc_national_payment_dist.xlsx") + rows = table.iloc[9:] + paying = rows[~rows[1].isin(["No payment", "Total"])] + return sum( + pd.to_numeric(paying[column], errors="coerce").fillna(0).sum() + for column in range(3, 7) + ) + + +@lru_cache +def _workbook_oracle(area_type): + """Independent raw-name join and the existing uniform national rescale.""" + areas = _area_metadata(area_type) + codes_by_name = {_name_key(row["name"]): row["code"] for _, row in areas.iterrows()} + assert len(codes_by_name) == len(areas) + name_column = 1 if area_type == "constituency" else 2 + filename = ( + "uc_pc_households.xlsx" + if area_type == "constituency" + else "uc_la_households.xlsx" + ) + gb = _workbook(filename) + counts = {} + for _, row in gb.iloc[8:].iterrows(): + name, count = row[name_column], row[3] + if pd.isna(name) or pd.isna(count) or name in {"Unknown", "Total"}: + continue + code = codes_by_name[_name_key(name)] + assert code not in counts + counts[code] = int(count) + + ni = _workbook( + "dfc-ni-uc-stats-supp-tables-may-2025.ods", + "5b" if area_type == "constituency" else "5c", + ) + month = ni.loc[ni[0].eq("May 2025")].iloc[0] + stop = 19 if area_type == "constituency" else 12 + for column in range(1, stop): + name = ni.iloc[2, column] + if pd.isna(name) or (area_type == "la" and name == "Ards and North Down"): + continue + code = codes_by_name[_name_key(name)] + assert code not in counts + counts[code] = int(month[column]) + + raw = pd.Series(counts, name="household_count").rename_axis("code").sort_index() + assert set(raw.index) == set(areas["code"]) + scaled = (raw * (_national_payment_total() / raw.sum())).round().astype(int) + return raw, scaled + + +def _get_targets(area_type): + getter = ( + local_uc.get_constituency_uc_targets + if area_type == "constituency" + else local_uc.get_la_uc_targets + ) + return getter() + + +def _map_constituency_targets_by_name(source): + """Aggregate the public boundary table without the production matrix or aliases.""" + boundary = pd.read_csv( + STORAGE_FOLDER.parent + / "datasets/local_areas/constituencies/boundary_changes/boundary_changes.csv" + ) + old_areas = pd.read_csv(STORAGE_FOLDER / "constituencies_2010.csv") + current_areas = _area_metadata("constituency") + codes_by_name = { + _name_key(row["name"]): row["code"] for _, row in current_areas.iterrows() + } + old_names = old_areas.set_index("code")["name"] + # The boundary source retains 2010 codes for some unchanged Scottish seats. + # Resolve those codes through their names instead of copying the code aliases + # from mapping_matrix.py; every other destination is already a current code. + historical = ~boundary["code_2024"].isin(current_areas["code"]) + boundary.loc[historical, "code_2024"] = ( + boundary.loc[historical, "code_2024"] + .map(old_names) + .map(_name_key) + .map(codes_by_name) + ) + assert boundary["code_2024"].notna().all() + assert set(boundary["code_2024"]) == set(current_areas["code"]) + shares = boundary["old_population_present"] / boundary.groupby("code_2010")[ + "old_population_present" + ].transform("sum") + contributions = source.loc[boundary["code_2010"]].reset_index(drop=True) + return contributions.mul(shares, axis=0).groupby(boundary["code_2024"]).sum() + + +@pytest.mark.parametrize("area_type", ["constituency", "la"]) +def test_all_uc_rows_and_country_totals_match_public_workbooks(area_type): + """Keep the national payment rescale and per-area rounding unchanged.""" + raw, expected = _workbook_oracle(area_type) + actual = _get_targets(area_type).sort_index() + pd.testing.assert_series_equal(actual, expected, check_dtype=False) + for country in _COUNTRIES: + codes = expected.index[expected.index.str.startswith(country)] + assert actual.loc[codes].sum() == expected.loc[codes].sum() + assert actual.sum() == (raw * (_national_payment_total() / raw.sum())).round().sum() + # The national scale is deliberate; individual integer rounding can drift. + assert abs(actual.sum() - _national_payment_total()) <= len(actual) / 2 + + +@pytest.mark.parametrize("area_type", ["constituency", "la"]) +@settings(max_examples=8, deadline=None) +@given(seed=st.integers(min_value=0, max_value=2**32 - 1)) +def test_permuting_workbook_areas_preserves_code_indexed_targets(area_type, seed): + _, expected = _workbook_oracle(area_type) + filename = ( + "uc_pc_households.xlsx" + if area_type == "constituency" + else "uc_la_households.xlsx" + ) + name_column = 1 if area_type == "constituency" else 2 + rng = np.random.default_rng(seed) + shuffled = _workbook(filename).copy() + valid = shuffled.index[ + (shuffled.index >= 8) + & shuffled[name_column].notna() + & shuffled[3].notna() + & ~shuffled[name_column].isin(["Unknown", "Total"]) + ] + shuffled.loc[valid] = shuffled.loc[rng.permutation(valid)].to_numpy() + + def read_excel(path, *args, **kwargs): + if Path(path).name == filename: + return shuffled.copy() + if Path(path).suffix == ".ods": + table = _workbook(Path(path).name, kwargs["sheet_name"]).copy() + stop = 19 if kwargs["sheet_name"] == "5b" else 12 + columns = list(range(1, stop)) + table.iloc[:, columns] = table.iloc[:, rng.permutation(columns)].to_numpy() + return table + return _workbook(Path(path).name).copy() + + parser = ( + uc_data._parse_uc_pc_households + if area_type == "constituency" + else uc_data._parse_uc_la_households + ) + attribute = ( + "uc_pc_households" if area_type == "constituency" else "uc_la_households" + ) + with pytest.MonkeyPatch.context() as patch: + patch.setattr(uc_data.pd, "read_excel", read_excel) + patch.setattr(uc_data, attribute, parser()) + pd.testing.assert_series_equal( + _get_targets(area_type).sort_index(), expected, check_dtype=False + ) + + +def test_children_split_sums_exactly_and_uses_matched_country_shares(): + _, totals = _workbook_oracle("constituency") + actual = local_uc.get_constituency_uc_by_children_targets().sort_index() + assert list(actual.columns) == _CHILD_COLUMNS + pd.testing.assert_index_equal(actual.index, totals.index) + np.testing.assert_array_equal(actual.sum(axis=1), totals) + gb_total = totals.loc[totals.index.str[0].isin(["E", "W", "S"])].sum() + buckets = local_uc._scaled_uc_children_by_country(gb_total) + for code, total in totals.items(): + shares = buckets[code[0]] / buckets[code[0]].sum() + # Integer apportionment must stay within one household of each share. + assert (np.abs(actual.loc[code].to_numpy() - total * shares) < 1).all(), code + + +@pytest.mark.parametrize("problem", ["unmatched", "duplicate", "missing"]) +def test_area_matching_rejects_incomplete_or_ambiguous_names(problem): + areas = pd.DataFrame({"code": ["E01", "W01"], "name": ["A & B", "Cardiff"]}) + names = { + "unmatched": ["Unknown place", "Cardiff"], + "duplicate": ["A and B", "A & B"], + "missing": ["A and B"], + }[problem] + data = pd.DataFrame({"area_name": names, "household_count": [1] * len(names)}) + with pytest.raises(ValueError): + uc_data._match_uc_area_codes(data, areas, "area_name") + + +@pytest.mark.parametrize("bilingual", ["CARDIFF / CAERDYDD", "CAERDYDD / CARDIFF"]) +def test_area_matching_normalises_labels_and_prefers_explicit_codes(bilingual): + areas = pd.DataFrame( + { + "code": ["E01", "W01", "W02"], + "name": ["King's Lynn and West Norfolk", "Cardiff", "Ynys Môn"], + } + ) + names = pd.DataFrame( + { + "area_name": [bilingual, "kings lynn & west-norfolk", "YNYS MON"], + "household_count": [10, 20, 30], + } + ) + matched = uc_data._match_uc_area_codes(names, areas, "area_name") + assert matched["code"].tolist() == ["W01", "E01", "W02"] + coded = names.assign(code=["E01", "W01", "W02"], area_name="Unrecognised label") + assert uc_data._match_uc_area_codes(coded, areas, "area_name")["code"].tolist() == [ + "E01", + "W01", + "W02", + ] + + +@pytest.mark.parametrize("codes", [["E01", "E01"], ["E01", "X01"]]) +def test_area_matching_rejects_duplicate_or_unknown_explicit_codes(codes): + areas = pd.DataFrame({"code": ["E01", "W01"], "name": ["Area A", "Area B"]}) + data = pd.DataFrame({"code": codes, "area_name": ["Area A", "Area B"]}) + with pytest.raises(ValueError): + uc_data._match_uc_area_codes(data, areas, "area_name") + + +@pytest.mark.parametrize("problem", ["ambiguous_alias", "duplicate_code"]) +def test_area_matching_rejects_ambiguous_reference_geography(problem): + areas = pd.DataFrame( + { + "code": ["E01", "W01"] if problem == "ambiguous_alias" else ["E01", "E01"], + "name": ["A & B", "A and B"], + } + ) + data = pd.DataFrame({"area_name": ["A and B"], "household_count": [1]}) + with pytest.raises(ValueError): + uc_data._match_uc_area_codes(data, areas, "area_name") + + +class _FakeSim: + """Four artificial households, one in each UK country; no dataset IO.""" + + def __init__(self, *args, **kwargs): + self.default_calculation_period = 2025 + + def calculate(self, variable, *args, **kwargs): + if variable == "country": + values = list(_COUNTRIES.values()) + elif variable == "tenure_type": + values = ["RENT_PRIVATELY"] * 4 + elif variable == "council_tax_band": + values = ["A"] * 4 + elif variable == "age": + values = [20, 30, 40, 50] + elif variable == "is_child": + values = [False] * 4 + else: + values = [1.0] * 4 + return SimpleNamespace(values=np.asarray(values)) + + def map_result(self, values, source_entity, target_entity): + return np.asarray(values) + + +def _patch_unrelated_sources(monkeypatch, dense_loss, area_type): + from policyengine_uk_data.targets.sources import local_income, local_la_extras + + for module in [local_income, dense_loss]: + monkeypatch.setattr( + module, + "INCOME_VARIABLES", + ["employment_income"] if area_type == "constituency" else [], + ) + if area_type == "la": + monkeypatch.setattr( + module, "get_national_income_projections", lambda year: pd.DataFrame() + ) + monkeypatch.setattr(dense_loss, "Microsimulation", _FakeSim) + if hasattr(dense_loss, "add_private_rent_targets"): + monkeypatch.setattr( + dense_loss, "add_private_rent_targets", lambda *args, **kwargs: None + ) + + codes = _area_metadata("la")["code"].to_numpy() + loaders = { + "load_ons_la_income": pd.DataFrame( + {"la_code": codes, "net_income_bhc": 1.0, "net_income_ahc": 1.0} + ), + "load_household_counts": pd.DataFrame({"la_code": codes, "households": 1.0}), + "load_tenure_data": pd.DataFrame( + { + "la_code": codes, + **{ + name + "_pct": 25.0 + for name in [ + "owned_outright", + "owned_mortgage", + "private_rent", + "social_rent", + ] + }, + } + ), + "load_private_rents": pd.DataFrame( + {"area_code": codes, "median_annual_rent": 1.0} + ), + } + for name, frame in loaders.items(): + for module in [local_la_extras, dense_loss]: + if hasattr(module, name): + monkeypatch.setattr(module, name, lambda frame=frame: frame.copy()) + for module in [local_la_extras, dense_loss]: + if hasattr(module, "get_ons_income_uprating_factors"): + monkeypatch.setattr( + module, "get_ons_income_uprating_factors", lambda year: (1.0, 1.0) + ) + + +@pytest.mark.parametrize("area_type", ["constituency", "la"]) +@pytest.mark.parametrize("builder", ["dense", "oa"]) +def test_each_builder_uc_row_matches_its_area_and_country( + monkeypatch, area_type, builder +): + from policyengine_uk_data.calibration import matrix_builder + from policyengine_uk_data.datasets.local_areas.constituencies import ( + loss as constituency_loss, + ) + from policyengine_uk_data.datasets.local_areas.local_authorities import ( + loss as la_loss, + ) + from policyengine_uk_data.targets.sources import local_age, local_income + + loss = constituency_loss if area_type == "constituency" else la_loss + _patch_unrelated_sources(monkeypatch, loss, area_type) + dataset = SimpleNamespace(time_period=2025) + areas = _area_metadata(area_type).copy() + if builder == "dense": + create = ( + loss.create_constituency_target_matrix + if area_type == "constituency" + else loss.create_local_authority_target_matrix + ) + matrix, y, mask = create(dataset) + expected_mask = np.equal.outer( + areas["code"].str[0].map(_COUNTRIES).to_numpy(), list(_COUNTRIES.values()) + ) + np.testing.assert_array_equal(mask, expected_mask) + assert list(matrix.columns) == list(y.columns) + else: + # Assignment rows can arrive in any code order, including interleaved countries. + areas = areas.sample(frac=1, random_state=468).reset_index(drop=True) + y = matrix_builder._load_area_targets(area_type, areas, dataset, _FakeSim()) + source_age = ( + local_age.get_constituency_age_targets() + if area_type == "constituency" + else local_age.get_la_age_targets() + ) + age_columns = [column for column in source_age if column.startswith("age/")] + adjustment = ( + 0.9 + * local_age.get_uk_total_population(dataset.time_period) + / source_age[age_columns].to_numpy().sum() + ) + expected_country_age = ( + source_age[age_columns].groupby(source_age["code"].str[0]).sum() * adjustment + ) + actual_country_age = y[age_columns].groupby(areas["code"].str[0].to_numpy()).sum() + pd.testing.assert_frame_equal( + actual_country_age.sort_index(), + expected_country_age.sort_index(), + check_names=False, + rtol=1e-12, + atol=1e-7, + ) + if area_type == "la": + np.testing.assert_allclose( + y[age_columns].to_numpy(), + source_age.set_index("code").loc[areas["code"], age_columns].to_numpy() + * adjustment, + ) + if area_type == "constituency": + source = source_age.set_index("code")[age_columns] * adjustment + incomes = local_income.get_constituency_income_targets().set_index("code") + national = local_income.get_national_income_projections(dataset.time_period) + national_amount = national.loc[ + national["total_income_lower_bound"].eq(12_570) + & national["total_income_upper_bound"].eq(np.inf), + "employment_income_amount", + ].iloc[0] + income_adjustment = national_amount / incomes["employment_income_amount"].sum() + for metric in ["amount", "count"]: + source[f"hmrc/employment_income/{metric}"] = ( + incomes[f"employment_income_{metric}"] * income_adjustment + ) + expected_mapped = _map_constituency_targets_by_name(source) + # Per-code checks catch the Scottish permutation even though its country + # totals and all UC targets are unchanged. Check all seats in both paths. + np.testing.assert_allclose( + y[source.columns].to_numpy(), + expected_mapped.loc[areas["code"], source.columns].to_numpy(), + rtol=1e-12, + atol=1e-7, + ) + _, expected = _workbook_oracle(area_type) + np.testing.assert_array_equal(y["uc_households"], expected.loc[areas["code"]]) + if area_type == "constituency": + children = local_uc.get_constituency_uc_by_children_targets().loc[areas["code"]] + np.testing.assert_array_equal(y[_CHILD_COLUMNS], children) diff --git a/policyengine_uk_data/tests/test_local_uc_etl.py b/policyengine_uk_data/tests/test_local_uc_etl.py new file mode 100644 index 000000000..8a00bc4de --- /dev/null +++ b/policyengine_uk_data/tests/test_local_uc_etl.py @@ -0,0 +1,45 @@ +"""Local UC database targets keep the public workbook's matched area codes.""" + +from unittest.mock import Mock + +import pandas as pd + +from policyengine_uk_data.db import etl +from policyengine_uk_data.storage import STORAGE_FOLDER +from policyengine_uk_data.targets.sources.local_uc import ( + get_constituency_uc_targets, + get_la_uc_targets, +) + + +def test_local_uc_etl_inserts_all_targets_with_matched_codes_and_names(monkeypatch): + session = Mock() + inserted = [] + + def insert_target(actual_session, **target): + assert actual_session is session + inserted.append(target) + + monkeypatch.setattr(etl, "_insert_target", insert_target) + + count = etl._load_local_uc_targets(session) + + assert count == len(inserted) == 1_010 + session.commit.assert_called_once_with() + for level, getter, area_file in [ + ("constituency", get_constituency_uc_targets, "constituencies_2024.csv"), + ("local_authority", get_la_uc_targets, "local_authorities_2021.csv"), + ]: + expected = getter() + areas = pd.read_csv(STORAGE_FOLDER / area_file).set_index( + "code", verify_integrity=True + ) + local = [row for row in inserted if row["geographic_level"] == level] + codes = [row["geo_code"] for row in local] + assert len(codes) == len(set(codes)) == len(expected) + assert set(codes) == set(expected.index) + for target in local: + code = target["geo_code"] + assert target["name"] == f"dwp/{level}/{code}/uc_households" + assert target["geo_name"] == areas.loc[code, "name"] + assert target["values"] == {2025: float(expected.loc[code])} diff --git a/policyengine_uk_data/utils/uc_data.py b/policyengine_uk_data/utils/uc_data.py index ffe20896e..2c8d1116a 100644 --- a/policyengine_uk_data/utils/uc_data.py +++ b/policyengine_uk_data/utils/uc_data.py @@ -1,6 +1,73 @@ import numpy as np import pandas as pd from pathlib import Path +import re +import unicodedata + + +def _normalise_area_name(name: str) -> str: + """Ignore case, accents, punctuation and the spelling of ampersands.""" + name = unicodedata.normalize("NFKD", name.casefold().replace("&", "and")) + return "".join(character for character in name if character.isalnum()) + + +def _area_name_aliases(name: str) -> set[str]: + # Welsh and English names in Stat-Xplore labels may appear in either order. + # Match either label, without guessing a translation. + return {_normalise_area_name(part) for part in re.split(r"/", name)} - {""} + + +def _match_uc_area_codes( + data: pd.DataFrame, areas: pd.DataFrame, name_column: str +) -> pd.DataFrame: + """Attach canonical codes with a complete, one-to-one area match. + + Extracts with codes use those directly; the committed workbooks have + names only. Reject incomplete or ambiguous extracts before scaling so + the national rescale cannot hide a missing or duplicated area. + """ + if areas["code"].isna().any() or areas["code"].duplicated().any(): + raise ValueError("UC reference areas contain missing or duplicate codes") + + aliases = {} + for area in areas.itertuples(index=False): + for alias in _area_name_aliases(area.name): + if alias in aliases and aliases[alias] != area.code: + raise ValueError(f"Ambiguous UC reference area name: {area.name!r}") + aliases[alias] = area.code + + expected_codes = set(areas["code"]) + matched_codes = [] + for row in data.to_dict("records"): + code = row.get("code") + if pd.notna(code) and code != "": + if code not in expected_codes: + raise ValueError(f"Unmatched UC area code: {code!r}") + else: + name = row[name_column] + candidates = ( + { + aliases[alias] + for alias in _area_name_aliases(name) + if alias in aliases + } + if isinstance(name, str) + else set() + ) + if len(candidates) != 1: + raise ValueError(f"Unmatched or ambiguous UC area name: {name!r}") + code = candidates.pop() + matched_codes.append(code) + + result = data.copy() + result["code"] = matched_codes + duplicates = result.loc[result["code"].duplicated(keep=False), "code"].tolist() + if duplicates: + raise ValueError(f"Duplicate UC areas: {duplicates}") + missing = expected_codes - set(matched_codes) + if missing: + raise ValueError(f"Missing UC areas: {sorted(missing)}") + return result def parse_monthly_award_band(band: str) -> tuple[float, float]: @@ -166,6 +233,11 @@ def _parse_uc_pc_households(): # Combine GB and NI data result_df = pd.DataFrame(gb_data_rows + ni_data_rows) + result_df = _match_uc_area_codes( + result_df, + pd.read_csv(storage_path / "constituencies_2024.csv"), + "constituency_name", + ) # Scale constituency counts to match national total national_total = _parse_uc_national_payment_dist()["household_count"].sum() @@ -237,6 +309,11 @@ def _parse_uc_la_households(): # Combine GB and NI data result_df = pd.DataFrame(gb_data_rows + ni_data_rows) + result_df = _match_uc_area_codes( + result_df, + pd.read_csv(storage_path / "local_authorities_2021.csv"), + "la_name", + ) # Scale LA counts to match national total national_total = _parse_uc_national_payment_dist()["household_count"].sum()