Move legacy-benefit reporters onto Universal Credit at Move to UC claim rates

changelog.d/uc-managed-migration-takeup.fixed.md

@@ -0,0 +1 @@
+Draw `would_claim_uc_at_legacy_closure` for benefit units that report a legacy means-tested benefit but not Universal Credit, at DWP's Move to Universal Credit claim rates for their combination of legacy benefits (Stat-Xplore, notices to end of March 2026). policyengine-uk reads it once one of those benefits closes, so a legacy reporter whose `would_claim_uc` draw was false now moves to Universal Credit at the observed rate instead of losing its legacy award with nothing in its place. Units reporting Universal Credit, and units with no legacy benefit, are true. The draw has its own seeded generator, so every other take-up flag is unchanged, and SPI-synthetic rows are redrawn after stage-2 imputation rewrites their benefit receipt (#492).

policyengine_uk_data/datasets/frs.py

@@ -34,8 +34,19 @@
     fill_with_mean,
     STORAGE_FOLDER,
 )
-from policyengine_uk_data.parameters import load_take_up_rate, load_parameter
-from policyengine_uk_data.utils.takeup import assign_takeup_with_reported_anchors
+from policyengine_uk_data.parameters import (
+    load_parameter,
+    load_take_up_rate,
+    load_uc_managed_migration_claim_rates,
+)
+from policyengine_uk_data.utils.takeup import (
+    LEGACY_BENEFITS,
+    UC_MANAGED_MIGRATION_SEED,
+    UC_MANAGED_MIGRATION_YEARS,
+    assign_takeup_with_reported_anchors,
+    assign_uc_claim_at_legacy_closure,
+    reported_benunit_mask,
+)
 from policyengine_uk_data.datasets.childcare.assumptions import (
     EXTENDED_HOURS_MEAN,
     EXTENDED_HOURS_SD,
@@ -274,16 +285,6 @@ def derive_esa_support_group_proxy(
     )
 
 
-def reported_benunit_mask(
-    person: pd.DataFrame, benunit: pd.DataFrame, person_column: str
-) -> np.ndarray:
-    """Benefit units with any member reporting a positive ``person_column``."""
-    reporter_benunits = set(
-        person.loc[person[person_column] > 0, "person_benunit_id"].values
-    )
-    return benunit["benunit_id"].isin(reporter_benunits).values
-
-
 def assign_reported_takeup(
     person: pd.DataFrame,
     benunit: pd.DataFrame,
@@ -576,6 +577,111 @@ def derive_all_claimants_over_state_pension_age(
     return ((counts.claimants > 0) & (counts.over == counts.claimants)).to_numpy()
 
 
+def derive_moves_to_uc_at_legacy_closure(
+    reports: dict,
+    pension_age_route,
+    mixed_age_couple,
+) -> np.ndarray:
+    """Identify the benefit units policyengine-uk moves to Universal Credit
+    when their legacy benefits close, the units that read
+    ``would_claim_uc_at_legacy_closure``.
+
+    ``reports`` maps each legacy benefit in ``LEGACY_BENEFITS`` and
+    ``pension_credit`` to whether the claimant or partner reports it; the
+    other arguments are policyengine-uk's ``meets_pension_credit_age_conditions``
+    and ``is_mixed_age_couple`` for one model year. The rule is
+    ``legacy_benefits_closed`` with every closure in force. Tax credit units
+    move unless DWP sends them to Pension Credit: those on it, and those
+    meeting its age conditions without Working Tax Credit (SI 2019/167 art.
+    3A(1); SI 2014/1230 reg 60A). Income Support, income-based JSA and
+    income-related ESA units move unless on Pension Credit or wholly over the
+    qualifying age (reg 8(2B)). Housing Benefit units move unless they meet
+    the Pension Credit age conditions (SI 2025/1148 art. 7(4)(a), reg
+    6A(4)-(5)); the specified or temporary accommodation saving has no input
+    in the data. policyengine-uk reads the flag for income-related ESA and
+    Housing Benefit units only once it ends those awards (#1914, PE-UK
+    #1910); a unit on neither until then never reads it, so drawing it is
+    harmless.
+    """
+
+    def reported(benefit):
+        return np.asarray(reports[benefit], dtype=bool)
+
+    pension_age_route = np.asarray(pension_age_route, dtype=bool)
+    on_pension_credit = reported("pension_credit")
+    on_wtc = reported("working_tax_credit")
+    pension_credit_route = on_pension_credit | (pension_age_route & ~on_wtc)
+    tax_credits = reported("child_tax_credit") | on_wtc
+    working_age_benefits = (
+        reported("income_support") | reported("jsa_income") | reported("esa_income")
+    )
+    wholly_pension_age = pension_age_route & ~np.asarray(mixed_age_couple, dtype=bool)
+    return (
+        (tax_credits & ~pension_credit_route)
+        | (working_age_benefits & ~on_pension_credit & ~wholly_pension_age)
+        | (reported("housing_benefit") & ~pension_age_route)
+    )
+
+
+def first_year_moved_to_uc(sim, benunit, year, years) -> np.ndarray:
+    """The first of ``years`` in which policyengine-uk moves each unit in
+    ``benunit`` to Universal Credit at the closure of its legacy benefits, or
+    0 if it moves in none.
+
+    ``sim`` is the policyengine-uk simulation on the survey-year (``year``)
+    data. The claimant's and partner's reports are read in the survey year;
+    the age conditions in each of ``years``. The model holds ages at their
+    survey values, so inferred dates of birth move with the model year: a
+    unit can leave the Pension Credit route in a later year, for example a
+    mixed-age couple whose older member is then born too late for the SI
+    2019/37 art. 4 saving, and move from then.
+    """
+    benunit_ids = benunit.benunit_id
+    person_benunit = sim.calculate("person_benunit_id", year).values
+    owners = np.asarray(
+        sim.calculate(
+            claimant_or_partner_variable(sim.tax_benefit_system.variables), year
+        ).values,
+        dtype=bool,
+    )
+
+    def owners_report(variable):
+        reported = np.asarray(sim.calculate(variable, year).values) > 0
+        return (
+            pd.Series(reported & owners, index=person_benunit)
+            .groupby(level=0)
+            .any()
+            .reindex(benunit_ids, fill_value=False)
+            .to_numpy()
+        )
+
+    reports = {
+        benefit: owners_report(f"{benefit}_reported")
+        for benefit in (*LEGACY_BENEFITS, "pension_credit")
+    }
+    sim_benunit_ids = sim.calculate("benunit_id", year).values
+    first_year = np.zeros(len(benunit_ids), dtype=int)
+    for model_year in sorted(years, reverse=True):
+
+        def benunit_values(variable):
+            return (
+                pd.Series(
+                    np.asarray(sim.calculate(variable, model_year).values, dtype=bool),
+                    index=sim_benunit_ids,
+                )
+                .reindex(benunit_ids, fill_value=False)
+                .to_numpy()
+            )
+
+        moves = derive_moves_to_uc_at_legacy_closure(
+            reports,
+            pension_age_route=benunit_values("meets_pension_credit_age_conditions"),
+            mixed_age_couple=benunit_values("is_mixed_age_couple"),
+        )
+        first_year = np.where(moves, model_year, first_year)
+    return first_year
+
+
 def derive_is_claimant_or_partner_from_frs_microdata(
     person_ids,
     person_benunit_ids,
@@ -1781,6 +1887,27 @@ def determine_education_level(fted_val, typeed2_val, age_val):
         is_over_state_pension_age=sim.calculate("is_SP_age", year).values,
         benunit_ids=pe_benunit.benunit_id,
     )
+    # Whether a legacy-benefit family claims Universal Credit once its legacy
+    # benefits close, at DWP's Move to Universal Credit claim rates. Its own
+    # generator keeps every other draw on the seed=100 sequence. Checked
+    # because the dataset loader drops columns the model does not define.
+    require_variable(
+        "would_claim_uc_at_legacy_closure",
+        "Move to Universal Credit claim flag",
+    )
+    # Only units the model moves to Universal Credit read the flag, so only
+    # they are drawn and counted against DWP's rates, by the year they move.
+    first_move_year = first_year_moved_to_uc(
+        sim, pe_benunit, year, UC_MANAGED_MIGRATION_YEARS
+    )
+    pe_benunit["would_claim_uc_at_legacy_closure"] = assign_uc_claim_at_legacy_closure(
+        pe_person,
+        pe_benunit,
+        load_uc_managed_migration_claim_rates(),
+        seed=UC_MANAGED_MIGRATION_SEED,
+        would_claim_uc=pe_benunit["would_claim_uc"].values,
+        first_move_year=first_move_year,
+    )
     pe_benunit["would_claim_tfc"] = generator.random(len(pe_benunit)) < tfc_rate
 
     pe_benunit["would_claim_extended_childcare"] = (

policyengine_uk_data/datasets/imputations/income.py

@@ -532,6 +532,27 @@ def impute_income(dataset: UKSingleYearDataset) -> UKSingleYearDataset:
         train_dataset=dataset,
         target_dataset=zero_weight_copy,
     )
+    # Stage 2 rewrote these rows' legacy benefit and Universal Credit
+    # receipt, so redraw the Move to Universal Credit claim flag that
+    # create_frs drew from their donors' receipt.
+    if "would_claim_uc_at_legacy_closure" in zero_weight_copy.benunit.columns:
+        from policyengine_uk_data.parameters import (
+            load_uc_managed_migration_claim_rates,
+        )
+        from policyengine_uk_data.utils.takeup import (
+            UC_MANAGED_MIGRATION_SPI_SEED,
+            assign_uc_claim_at_legacy_closure,
+        )
+
+        zero_weight_copy.benunit["would_claim_uc_at_legacy_closure"] = (
+            assign_uc_claim_at_legacy_closure(
+                zero_weight_copy.person,
+                zero_weight_copy.benunit,
+                load_uc_managed_migration_claim_rates(),
+                seed=UC_MANAGED_MIGRATION_SPI_SEED,
+                would_claim_uc=zero_weight_copy.benunit.get("would_claim_uc"),
+            )
+        )
 
     # The copy keeps its FRS donor's employment status but now has SPI
     # incomes, so derive its gainful self-employment flag again from the

policyengine_uk_data/parameters/__init__.py

@@ -62,3 +62,19 @@ def load_take_up_rate(variable_name: str, year: int = 2015) -> float:
         Take-up rate as a float between 0 and 1
     """
     return load_parameter("take_up", variable_name, year)
+
+
+def load_uc_managed_migration_claim_rates() -> dict[str, float]:
+    """Load Move to Universal Credit claim rates by legacy benefit combination.
+
+    Returns:
+        Claim rate for each combination in ``take_up/uc_managed_migration.yaml``
+        (benefit names joined by "+", plus "all"): households that claimed
+        Universal Credit over those that claimed or did not claim.
+    """
+    with open(PARAMETERS_DIR / "take_up" / "uc_managed_migration.yaml") as f:
+        households = yaml.safe_load(f)["households"]
+    return {
+        combination: counts["claimed"] / (counts["claimed"] + counts["did_not_claim"])
+        for combination, counts in households.items()
+    }

policyengine_uk_data/parameters/take_up/uc_managed_migration.yaml

@@ -0,0 +1,42 @@
+description: >-
+  Households sent a Move to Universal Credit migration notice in Great Britain
+  (July 2022 to end of March 2026), by the legacy benefits they received and
+  whether they claimed Universal Credit. The claim rate for a combination is
+  claimed / (claimed + did_not_claim): the 538 households still in progress
+  are left out. Combinations DWP does not report use the all-households row.
+  "esa_income" is Stat-Xplore's "Employment Support Allowance"; notices go only
+  to claimants of an existing benefit, which is income-related ESA.
+metadata:
+  unit: /1
+  label: Move to Universal Credit claim rate by legacy benefit combination
+  reference:
+    - title: >-
+        DWP Stat-Xplore, "Households invited to Move to Universal Credit",
+        table "MtUC Households 3 - Migration notices by legacy benefit and
+        move status" (retrieved 2026-10-03)
+      href: https://stat-xplore.dwp.gov.uk
+    - title: "DWP: Move to Universal Credit, July 2022 to end March 2026 (12 May 2026)"
+      href: https://www.gov.uk/government/statistics/move-to-universal-credit-july-2022-to-end-march-2026
+# Benefit names follow the order child_tax_credit, working_tax_credit,
+# housing_benefit, esa_income, income_support, jsa_income, joined by "+".
+households:
+  child_tax_credit: {claimed: 36_723, did_not_claim: 17_733}
+  working_tax_credit: {claimed: 53_993, did_not_claim: 51_805}
+  child_tax_credit+working_tax_credit: {claimed: 363_984, did_not_claim: 126_397}
+  housing_benefit: {claimed: 30_900, did_not_claim: 10_121}
+  esa_income: {claimed: 321_075, did_not_claim: 13_944}
+  income_support: {claimed: 17_719, did_not_claim: 1_351}
+  jsa_income: {claimed: 5_867, did_not_claim: 329}
+  child_tax_credit+housing_benefit: {claimed: 15_646, did_not_claim: 1_042}
+  child_tax_credit+esa_income: {claimed: 11_171, did_not_claim: 381}
+  child_tax_credit+income_support: {claimed: 8_254, did_not_claim: 276}
+  child_tax_credit+jsa_income: {claimed: 381, did_not_claim: 8}
+  working_tax_credit+housing_benefit: {claimed: 7_820, did_not_claim: 1_042}
+  housing_benefit+esa_income: {claimed: 448_253, did_not_claim: 9_117}
+  housing_benefit+income_support: {claimed: 31_087, did_not_claim: 750}
+  housing_benefit+jsa_income: {claimed: 11_019, did_not_claim: 333}
+  child_tax_credit+working_tax_credit+housing_benefit: {claimed: 81_962, did_not_claim: 5_352}
+  child_tax_credit+housing_benefit+esa_income: {claimed: 73_859, did_not_claim: 949}
+  child_tax_credit+housing_benefit+income_support: {claimed: 58_764, did_not_claim: 691}
+  child_tax_credit+housing_benefit+jsa_income: {claimed: 2_286, did_not_claim: 30}
+  all: {claimed: 1_580_761, did_not_claim: 241_662}

policyengine_uk_data/tests/test_legacy_benefit_proxies.py

@@ -422,6 +422,14 @@ def calculate(self, variable, year=None):
             return pd.Series(self.dataset.person.age.values >= 18)
         if variable == "is_SP_age":
             return pd.Series(self.dataset.person.age.values >= 66)
+        if variable == "benunit_id":
+            return pd.Series(self.dataset.benunit.benunit_id.values)
+        if variable.endswith("_reported"):
+            if variable in self.dataset.person.columns:
+                return pd.Series(self.dataset.person[variable].values)
+            return pd.Series(np.zeros(len(self.dataset.person)))
+        if variable in ("meets_pension_credit_age_conditions", "is_mixed_age_couple"):
+            return pd.Series(np.zeros(len(self.dataset.benunit), dtype=bool))
         if variable in (
             "childcare_grant",
             "parents_learning_allowance",