diff --git a/changelog.d/576.added.md b/changelog.d/576.added.md new file mode 100644 index 00000000..f0b6313f --- /dev/null +++ b/changelog.d/576.added.md @@ -0,0 +1 @@ +Added the UK Budget reform contract (`policyengine.tax_benefit_models.uk.budget_reform_contract`), version 1.0.0. `run_budget_reform_contract()` checks that dated reforms (an April rate change, a finite change, an extended freeze, an April uprating schedule, a mid-year blended rate, a CPI-driven indexation change, a scenario or policy combining a parameter-setting modifier with dated changes, an unknown path, and each route's documented key readings) give independently derived results through the policyengine-uk and policyengine.py routes, leave the baseline and later unreformed runs unchanged, and survive bounded property sampling. It returns a JSON report for release qualification. diff --git a/src/policyengine/tax_benefit_models/uk/budget_reform_contract.py b/src/policyengine/tax_benefit_models/uk/budget_reform_contract.py new file mode 100644 index 00000000..4ca13fac --- /dev/null +++ b/src/policyengine/tax_benefit_models/uk/budget_reform_contract.py @@ -0,0 +1,2021 @@ +"""The UK Budget reform contract. + +A Budget dates its measures: a rate from 6 April, a freeze extended to a +later tax year, a change that ends, an April uprating schedule, a rate that +changes part-way through a year. This contract checks that each such reform +gives the right answer through every supported route, against expected +values derived independently of the model (statute arithmetic, or a +day-weighted fiscal-year annualisation written here), and that applying a +reform never changes the baseline. + +Qualification imports :func:`run_budget_reform_contract` and gates a +candidate model and data pair on its report. The contract lives here rather +than in policyengine-uk so a candidate model release never grades itself. + +Routes +------ + +``country.reform_dict`` + ``policyengine_uk.Simulation(reform=dict)`` with ``"start.stop"`` keys. +``country.scenario_parameter_changes`` + ``policyengine_uk.Simulation(scenario=Scenario(parameter_changes=dict))``. +``country.reform_from_dict`` + ``policyengine_uk.Simulation(reform=Reform.from_dict(dict))``. +``wrapper.household`` + ``policyengine.uk.calculate_household(reform=dict)``, with a lone date + for a change in force from that date and ``"start.stop"`` for a bounded + one. +``wrapper.policy`` + ``policyengine.core.Simulation(policy=Policy(...))`` over an in-memory + dataset holding the case's household (and, with ``dataset_path``, over a + deterministic subsample of the certified dataset). +``api.sim_api`` + Not run in-process: policyengine-sim-api's deploy gate imports + :data:`CASES` and routes them through its runtime. + +Readings +-------- + +Every case is written once as :class:`Change` objects on the statutory +timeline: a value in force from ``start`` to ``stop`` inclusive, or from +``start`` on (held flat) when ``stop`` is None. Each route receives the +representation it documents. Documented differences between readings (a +bare-year policyengine-uk dict key is that fiscal year only, while +``Reform.from_dict`` reads it as onwards) are covered by the country-only +reading cases. + +Prior law after a bounded change +-------------------------------- + +Contract 1.0.0 expects prior law's projected path to resume after a bounded +change (``PRIOR_LAW_OPTION = "projected"``). Whether indexation should +instead resume from a changed statutory amount (Income Tax Act 2007 s57 +indexes "the amount ... for the previous tax year") is an open methodology +call (cos d1344). Checks that depend on it carry ``pending`` and never gate. + +Report +------ + +A JSON-serialisable dict: ``contract_id``, ``contract_version``, +``prior_law_option``, installed ``versions``, the dataset's sha256 when one +is given, ``cases`` (one row per case, route and check, with ``status`` of +``pass``, ``fail``, ``error``, ``unsupported`` or ``pending``), +``properties``, ``population`` and ``summary`` counts. ``cases`` also +holds two run-level rows: ``module_parameters_unchanged`` (policyengine-uk's +shared parameter state, including the cached tree new simulations copy, +still equals its values before the run) and, when no case, property or +population check passed, an ``error`` row with case id ``run``. Any ``fail`` or +``error`` means the contract does not hold. + +Isolation +--------- + +Expected values read a copy of current law's parameters taken before any +reform runs. Every route's unreformed reference is built before its reform, +and after each reform a fresh unreformed simulation must reproduce it +(``unreformed_after_reform`` rows); the country routes also check each +reformed simulation's own baseline (``baseline_isolation`` rows). +""" + +from __future__ import annotations + +import datetime +import gc +import hashlib +import importlib.metadata +import json +import math +import random +import re +import sys +import traceback +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass, field +from decimal import ROUND_CEILING, Decimal +from typing import Any, Optional, Union + +CONTRACT_ID = "policyengine-uk/budget-reform-contract" +CONTRACT_VERSION = "1.0.0" +PRIOR_LAW_OPTION = "projected" +PRIOR_LAW_DECISION = "d1344" +FLAT_STOP = datetime.date(2100, 12, 31) + +COUNTRY_ROUTES = ( + "country.reform_dict", + "country.scenario_parameter_changes", + "country.reform_from_dict", +) +WRAPPER_ROUTES = ("wrapper.household", "wrapper.policy") +API_ROUTE = "api.sim_api" +ROUTES = COUNTRY_ROUTES + WRAPPER_ROUTES + (API_ROUTE,) +UNSUPPORTED_ROUTES = { + API_ROUTE: ( + "Runs in policyengine-sim-api's deploy gate, which imports CASES and " + "routes them through its runtime; it is not callable in-process." + ), +} + +PASS, FAIL, ERROR, UNSUPPORTED, PENDING = ( + "pass", + "fail", + "error", + "unsupported", + "pending", +) + +BASIC_RATE = "gov.hmrc.income_tax.rates.uk[0].rate" +DIVIDEND_BASIC_RATE = "gov.hmrc.income_tax.rates.dividends[0].rate" +DIVIDEND_ALLOWANCE = "gov.hmrc.income_tax.allowances.dividend_allowance" +PERSONAL_ALLOWANCE = "gov.hmrc.income_tax.allowances.personal_allowance.amount" +BASIC_RATE_LIMIT = "gov.hmrc.income_tax.rates.uk[1].threshold" +UC_SINGLE_OLD = "gov.dwp.universal_credit.standard_allowance.amount.SINGLE_OLD" +NEW_STATE_PENSION = "gov.dwp.state_pension.new_state_pension.amount" +CGT_BASIC_RATE = "gov.hmrc.cgt.basic_rate" +CGT_ANNUAL_EXEMPT_AMOUNT = "gov.hmrc.cgt.annual_exempt_amount" +VAT_STANDARD_RATE = "gov.hmrc.vat.standard_rate" +VAT_COVERAGE = "gov.simulation.microdata_vat_coverage" +PROTECTED_LCWRA = "gov.dwp.universal_credit.rebalancing.protected_health_element" +CPI_SEPTEMBER = "gov.economic_assumptions.statutory_uprating_inputs.cpi_september" +MA_INCOME_CONDITION = ( + "gov.contrib.cps.marriage_tax_reforms.expanded_ma.remove_income_condition" +) +SEPTEMBER_CPI_RISE = "gov.economic_assumptions.yoy_growth.september_cpi_uprating" +NICS = "gov.hmrc.national_insurance" +PRIMARY_THRESHOLD = f"{NICS}.class_1.thresholds.primary_threshold" +UPPER_EARNINGS_LIMIT = f"{NICS}.class_1.thresholds.upper_earnings_limit" +LOWER_PROFITS_LIMIT = f"{NICS}.class_4.thresholds.lower_profits_limit" +UPPER_PROFITS_LIMIT = f"{NICS}.class_4.thresholds.upper_profits_limit" + + +# Case definitions. + + +@dataclass(frozen=True) +class Change: + """A value in force on the statutory timeline. + + From ``start`` to ``stop`` inclusive, or from ``start`` on, held flat, + when ``stop`` is None. + """ + + path: str + start: datetime.date + value: Any + stop: Optional[datetime.date] = None + + @property + def interval_key(self) -> str: + return f"{self.start}.{self.stop or FLAT_STOP}" + + +@dataclass(frozen=True) +class Household: + """One benefit unit in one household; values hold in every year.""" + + people: tuple[Mapping[str, Any], ...] + benunit: Mapping[str, Any] = field(default_factory=dict) + household: Mapping[str, Any] = field(default_factory=dict) + + +Expected = Union[float, Callable[["Context"], float]] + + +@dataclass(frozen=True) +class Check: + """An expected value. + + ``target``: + - ``"parameter"``: the annual value of parameter ``name`` in fiscal year + ``year`` (read at 1 January of ``year``, as annual formulas read it); + - ``"variable"``: the household's total of variable ``name`` in ``year``; + - ``"change"``: that total under the reform minus without it, through + the same route. + + ``expected`` is a number or a function of the :class:`Context`. ``oracle`` + says how it was derived. ``pending`` names an open decision: the check is + reported but does not gate. + """ + + target: str + name: str + year: int + expected: Expected + oracle: str + tolerance: float = 1e-6 + pending: Optional[str] = None + + +@dataclass(frozen=True) +class Case: + """A dated reform and what it must produce. + + ``representations`` overrides the representation a route receives, for + the reading cases; ``routes`` limits the routes a case runs on. + ``expect_error`` is a regular expression the error must match for a + reform every route must reject. ``modifier`` holds changes a simulation + modifier writes onto the annualised parameters, combined with the + route's reform in one scenario or policy: modifier first when + ``modifier_first``, else reform first. + """ + + id: str + title: str + changes: tuple[Change, ...] + household: Household + checks: tuple[Check, ...] = () + expect_error: Optional[str] = None + routes: Optional[tuple[str, ...]] = None + representations: Mapping[str, Any] = field(default_factory=dict) + modifier: tuple[Change, ...] = () + modifier_first: bool = True + + +class Context: + """What an expected value may read: current law's annual parameters.""" + + def __init__(self, baseline_parameter: Callable[[str, int], float]): + self._baseline_parameter = baseline_parameter + + def baseline(self, path: str, year: int) -> float: + return self._baseline_parameter(path, year) + + +def _date(text: str) -> datetime.date: + return datetime.date.fromisoformat(text) + + +def adult(income: float = 30_000, **extra) -> Household: + return Household(people=({"age": 40, "employment_income": income, **extra},)) + + +def s57_personal_allowance(previous: float, increase: float) -> float: + """Income Tax Act 2007 s57(3): add the rise, rounded up to £10.""" + if increase <= 0: + return previous + rise = Decimal(repr(float(previous))) * Decimal(repr(float(increase))) + return float( + Decimal(repr(float(previous))) + + (rise / 10).to_integral_value(ROUND_CEILING) * 10 + ) + + +def s21_basic_rate_limit(previous: float, increase: float) -> float: + """Income Tax Act 2007 s21(3): scale, rounded up to £100.""" + if increase <= 0: + return previous + scaled = Decimal(repr(float(previous))) * (1 + Decimal(repr(float(increase)))) + return float((scaled / 100).to_integral_value(ROUND_CEILING) * 100) + + +def weekly(annual: float) -> float: + """The weekly NICs convention: the annual amount over 52, to the penny.""" + return float((Decimal(repr(float(annual))) / 52).quantize(Decimal("0.01"))) + + +def fiscal_year_average( + pieces: Sequence[tuple[datetime.date, float]], year: int +) -> float: + """Day-weighted average over 6 April to 5 April of a step function. + + ``pieces`` lists (date the value takes effect, value), in any order. + """ + start = datetime.date(year, 4, 6) + end = datetime.date(year + 1, 4, 6) + ordered = sorted(pieces) + total = 0.0 + day = start + while day < end: + value = None + for effective, piece_value in ordered: + if effective <= day: + value = piece_value + total += value + day += datetime.timedelta(days=1) + return total / (end - start).days + + +def _uplift( + year: int, rate: float, income: float = 30_000 +) -> Callable[[Context], float]: + def expected(context: Context) -> float: + allowance = context.baseline(PERSONAL_ALLOWANCE, year) + return (income - allowance) * (rate - context.baseline(BASIC_RATE, year)) + + return expected + + +def _freeze_change(year: int) -> Callable[[Context], float]: + def expected(context: Context) -> float: + frozen_out = context.baseline(PERSONAL_ALLOWANCE, year) - 12_570 + return context.baseline(BASIC_RATE, year) * frozen_out + + return expected + + +def _statutory_resumption(path: str) -> Callable[[Context], float]: + def expected(context: Context) -> float: + increase = context.baseline(SEPTEMBER_CPI_RISE, 2034) + if path == PERSONAL_ALLOWANCE: + return s57_personal_allowance(12_570, increase) + return s21_basic_rate_limit(37_700, increase) + + return expected + + +def _baseline_value(path: str, year: int) -> Callable[[Context], float]: + return lambda context: context.baseline(path, year) + + +GAINS = 10_000 + + +def _gains_uplift(year: int, share_of_year: float) -> Callable[[Context], float]: + """Extra CGT on gains taxed wholly at the basic rate, at 20% for part of a year.""" + + def expected(context: Context) -> float: + taxable = GAINS - context.baseline(CGT_ANNUAL_EXEMPT_AMOUNT, year) + rise = 0.20 - context.baseline(CGT_BASIC_RATE, year) + return taxable * rise * share_of_year + + return expected + + +GAINS_UPLIFT_ORACLE = ( + "(£10,000 of gains - the annual exempt amount) x (20% - current rate) x " + "the share of the tax year at 20%; the gains sit wholly in the basic rate " + "band, and the exempt amount and current rate are read from current law" +) + +DIVIDENDS = 10_000 + + +def _dividends_and_earnings( + year: int, rate: Optional[float] +) -> Callable[[Context], float]: + """The change in tax when dividends are taxed at 8.75%, and earnings at ``rate``. + + The dividends sit wholly in the basic rate band. + """ + + def expected(context: Context) -> float: + taxed = DIVIDENDS - context.baseline(DIVIDEND_ALLOWANCE, year) + change = taxed * (0.0875 - context.baseline(DIVIDEND_BASIC_RATE, year)) + if rate is not None: + change += _uplift(year, rate)(context) + return change + + return expected + + +MODIFIER_ORACLE = ( + "(£10,000 of dividends - the dividend allowance) x (8.75% - current " + "dividend rate), all within the basic rate band" +) + + +def _modifier_case(case_id: str, modifier_first: bool) -> Case: + # The modifier and the reform change different parameters, so either + # order must give the same values: what the two cases check is that + # neither side is lost however the scenario or policy is assembled. + order = "then" if modifier_first else "after" + return Case( + id=case_id, + title=( + "One scenario or policy with a modifier setting the dividend " + f"rate to 8.75% {order} a basic rate of 21% from 6 April 2027" + ), + changes=(Change(BASIC_RATE, _date("2027-04-06"), 0.21),), + modifier=(Change(DIVIDEND_BASIC_RATE, _date("2026-01-01"), 0.0875),), + modifier_first=modifier_first, + household=adult(dividend_income=DIVIDENDS), + checks=( + Check( + "parameter", DIVIDEND_BASIC_RATE, 2026, 0.0875, "set by the modifier" + ), + Check( + "parameter", DIVIDEND_BASIC_RATE, 2027, 0.0875, "set by the modifier" + ), + Check("parameter", BASIC_RATE, 2026, 0.20, "before the change"), + Check("parameter", BASIC_RATE, 2027, 0.21, "in force on 30 April 2027"), + Check( + "change", + "income_tax", + 2026, + _dividends_and_earnings(2026, None), + MODIFIER_ORACLE, + 0.005, + ), + Check( + "change", + "income_tax", + 2027, + _dividends_and_earnings(2027, 0.21), + MODIFIER_ORACLE + ", plus " + INCOME_UPLIFT_ORACLE, + 0.005, + ), + ), + # policyengine.uk.calculate_household takes a reform dict only. + routes=COUNTRY_ROUTES + ("wrapper.policy",), + ) + + +FULL_RATE_CONSUMPTION = 10_000 + + +def _vat_uplift(year: int, rate: float) -> Callable[[Context], float]: + def expected(context: Context) -> float: + rise = rate - context.baseline(VAT_STANDARD_RATE, year) + return FULL_RATE_CONSUMPTION * rise / context.baseline(VAT_COVERAGE, year) + + return expected + + +INCOME_UPLIFT_ORACLE = ( + "(30,000 - personal allowance) x (new rate - current rate); the allowance " + "and current rate are read from current law" +) + +CASES: tuple[Case, ...] = ( + Case( + id="april_rate_change", + title="Basic rate 21% from 6 April 2027", + changes=(Change(BASIC_RATE, _date("2027-04-06"), 0.21),), + household=adult(), + checks=( + Check("parameter", BASIC_RATE, 2026, 0.20, "unchanged before 2027-28"), + Check("parameter", BASIC_RATE, 2027, 0.21, "in force on 30 April 2027"), + Check("parameter", BASIC_RATE, 2028, 0.21, "held from 6 April 2027"), + Check("change", "income_tax", 2026, 0.0, "before the change", 0.005), + Check( + "change", + "income_tax", + 2027, + _uplift(2027, 0.21), + INCOME_UPLIFT_ORACLE, + 0.005, + ), + Check( + "change", + "income_tax", + 2028, + _uplift(2028, 0.21), + INCOME_UPLIFT_ORACLE, + 0.005, + ), + ), + ), + Case( + id="finite_rate_change", + title="Basic rate 21% for 2027-28 only", + changes=(Change(BASIC_RATE, _date("2027-04-06"), 0.21, _date("2028-04-05")),), + household=adult(), + checks=( + Check("parameter", BASIC_RATE, 2027, 0.21, "in force on 30 April 2027"), + Check("parameter", BASIC_RATE, 2028, 0.20, "prior law after 5 April 2028"), + Check( + "change", + "income_tax", + 2027, + _uplift(2027, 0.21), + INCOME_UPLIFT_ORACLE, + 0.005, + ), + Check("change", "income_tax", 2028, 0.0, "the change has ended", 0.005), + ), + ), + Case( + id="extended_freeze", + title="Personal allowance and basic rate limit frozen to 2033-34", + changes=( + Change( + PERSONAL_ALLOWANCE, _date("2031-04-06"), 12_570, _date("2034-04-05") + ), + Change(BASIC_RATE_LIMIT, _date("2031-04-06"), 37_700, _date("2034-04-05")), + ), + household=adult(), + checks=tuple( + check + for year in (2031, 2032, 2033) + for check in ( + Check("parameter", PERSONAL_ALLOWANCE, year, 12_570, "frozen"), + Check("parameter", BASIC_RATE_LIMIT, year, 37_700, "frozen"), + Check( + "parameter", + LOWER_PROFITS_LIMIT, + year, + 12_570, + "aligned with the allowance", + ), + Check( + "parameter", + UPPER_PROFITS_LIMIT, + year, + 50_270, + "allowance + basic rate limit", + ), + Check( + "parameter", + PRIMARY_THRESHOLD, + year, + weekly(12_570), + "12,570 / 52, to the penny", + ), + Check( + "parameter", + UPPER_EARNINGS_LIMIT, + year, + weekly(50_270), + "50,270 / 52, to the penny", + ), + Check( + "change", + "income_tax", + year, + _freeze_change(year), + "basic rate x (current law's allowance - 12,570)", + 0.005, + ), + ) + ) + + ( + Check( + "parameter", + PERSONAL_ALLOWANCE, + 2034, + _baseline_value(PERSONAL_ALLOWANCE, 2034), + "contract 1.0.0: prior law's projected path resumes", + ), + Check( + "parameter", + PERSONAL_ALLOWANCE, + 2034, + _statutory_resumption(PERSONAL_ALLOWANCE), + "ITA 2007 s57 from the frozen £12,570 with 2034's September CPI rise", + pending=PRIOR_LAW_DECISION, + ), + Check( + "parameter", + BASIC_RATE_LIMIT, + 2034, + _statutory_resumption(BASIC_RATE_LIMIT), + "ITA 2007 s21 from the frozen £37,700 with 2034's September CPI rise", + pending=PRIOR_LAW_DECISION, + ), + ), + ), + Case( + id="april_benefit_uprating", + title="UC single 25+ standard allowance £500 from April 2027, £520 from April 2028", + changes=( + Change(UC_SINGLE_OLD, _date("2027-04-06"), 500, _date("2028-04-05")), + Change(UC_SINGLE_OLD, _date("2028-04-06"), 520), + ), + household=adult(), + checks=( + Check("variable", "uc_standard_allowance", 2027, 6_000.0, "£500 x 12"), + Check("variable", "uc_standard_allowance", 2028, 6_240.0, "£520 x 12"), + Check( + "variable", + "uc_standard_allowance", + 2029, + 6_240.0, + "held from April 2028", + ), + Check("change", "uc_standard_allowance", 2026, 0.0, "before the change"), + ), + ), + Case( + id="bounded_change_on_uprated_benefit", + title="UC single 25+ standard allowance £500 for 2027-28 only", + changes=(Change(UC_SINGLE_OLD, _date("2027-04-06"), 500, _date("2028-04-05")),), + household=adult(), + checks=( + Check("variable", "uc_standard_allowance", 2027, 6_000.0, "£500 x 12"), + Check( + "change", + "uc_standard_allowance", + 2028, + 0.0, + "prior law's uprated path resumes", + ), + Check( + "change", + "uc_standard_allowance", + 2030, + 0.0, + "prior law's uprated path resumes", + ), + ), + ), + Case( + id="uc_award_follows_the_standard_allowance", + title="A claimant with no income is awarded the reformed standard allowance", + changes=(Change(UC_SINGLE_OLD, _date("2027-04-06"), 500, _date("2028-04-05")),), + household=adult(0), + checks=( + Check( + "variable", + "universal_credit", + 2027, + 6_000.0, + "with no income or other elements, the award is the standard " + "allowance: £500 x 12", + 0.01, + ), + Check("change", "universal_credit", 2026, 0.0, "before the change", 0.01), + Check( + "change", + "universal_credit", + 2028, + 0.0, + "prior law's uprated path resumes", + 0.01, + ), + ), + ), + Case( + id="new_state_pension_rate", + title="New State Pension £280 a week for 2027-28", + changes=( + Change(NEW_STATE_PENSION, _date("2027-04-06"), 280.0, _date("2028-04-05")), + ), + household=Household( + people=( + { + "age": 68, + "state_pension_type": "NEW", + # Above the full rate, so the person gets the full rate. + "state_pension_reported": 20_000, + }, + ) + ), + checks=( + Check( + "variable", + "new_state_pension", + 2027, + 14_560.0, + "the full weekly rate x 52: £280 x 52", + 0.01, + ), + Check("change", "new_state_pension", 2026, 0.0, "before the change", 0.01), + ), + ), + Case( + id="mid_year_blended_rate", + title="CGT basic rate 20% from 30 October 2027", + changes=(Change(CGT_BASIC_RATE, _date("2027-10-30"), 0.20),), + # £20,000 of earnings and £10,000 of gains: the taxable gains sit + # wholly in the basic rate band. + household=adult(20_000, capital_gains=GAINS), + checks=( + Check( + "parameter", + CGT_BASIC_RATE, + 2027, + fiscal_year_average( + [(_date("2024-10-30"), 0.18), (_date("2027-10-30"), 0.20)], 2027 + ), + "day-weighted: 207 days at 18% and 159 at 20% of 366", + 1e-12, + ), + Check("parameter", CGT_BASIC_RATE, 2028, 0.20, "in force all year"), + Check("parameter", CGT_BASIC_RATE, 2026, 0.18, "before the change"), + Check("change", "capital_gains_tax", 2026, 0.0, "before the change", 0.005), + Check( + "change", + "capital_gains_tax", + 2027, + _gains_uplift(2027, 159 / 366), + GAINS_UPLIFT_ORACLE + "; 159 of 2027-28's 366 days", + 0.005, + ), + Check( + "change", + "capital_gains_tax", + 2028, + _gains_uplift(2028, 1.0), + GAINS_UPLIFT_ORACLE + "; all of 2028-29", + 0.005, + ), + ), + ), + Case( + id="september_cpi_moves_indexed_thresholds", + title="A 5% September 2030 CPI rise indexes the 2031-32 thresholds", + changes=( + Change(CPI_SEPTEMBER, _date("2030-09-01"), 0.05, _date("2031-08-31")), + ), + household=adult(), + checks=( + Check( + "parameter", + PERSONAL_ALLOWANCE, + 2031, + s57_personal_allowance(12_570, 0.05), + "ITA 2007 s57(3)", + ), + Check( + "parameter", + BASIC_RATE_LIMIT, + 2031, + s21_basic_rate_limit(37_700, 0.05), + "ITA 2007 s21(3)", + ), + Check( + "parameter", + PRIMARY_THRESHOLD, + 2031, + weekly(s57_personal_allowance(12_570, 0.05)), + "aligned with the allowance, weekly", + ), + ), + routes=COUNTRY_ROUTES, + ), + Case( + id="uc_rebalancing_reform_after_data_load", + title="Protected LCWRA amount £500 a month for 2027-28", + changes=( + Change(PROTECTED_LCWRA, _date("2027-04-06"), 500.0, _date("2028-04-05")), + ), + household=adult(0, uc_limited_capability_for_WRA=True), + checks=(Check("variable", "uc_LCWRA_element", 2027, 6_000.0, "£500 x 12"),), + ), + Case( + id="structural_switch", + title="Marriage allowance without its income condition", + changes=( + Change(MA_INCOME_CONDITION, _date("2025-01-01"), True, _date("2030-12-31")), + ), + household=Household( + people=( + {"age": 40, "employment_income": 80_000}, + {"age": 40, "employment_income": 0}, + ), + benunit={"is_married": True}, + ), + checks=( + Check( + "variable", + "marriage_allowance", + 2026, + 1_260.0, + "10% of the £12,570 allowance, rounded up to £10 (ITA 2007 s55B)", + ), + ), + ), + Case( + id="baseline_vat_is_current_law", + title="VAT standard rate 22% from 2026-27 (#2188)", + changes=(Change(VAT_STANDARD_RATE, _date("2026-01-01"), 0.22),), + household=Household( + people=({"age": 40},), + household={ + "full_rate_vat_consumption": FULL_RATE_CONSUMPTION, + "reduced_rate_vat_consumption": 0, + }, + ), + checks=( + Check( + "baseline_equals", + "baseline_vat", + 2026, + "vat", + "baseline_vat is the unreformed vat", + ), + Check( + "change", + "vat", + 2026, + _vat_uplift(2026, 0.22), + "£10,000 of full-rate consumption x (22% - current rate), over " + "the share of consumption the microdata covers; the current " + "rate and coverage are read from current law", + 0.005, + ), + ), + ), + # A scenario's parameter changes used to rebuild the parameters after its + # modifier had run, so whatever the modifier set reverted. + _modifier_case("modifier_then_parameter_changes", True), + _modifier_case("parameter_changes_then_modifier", False), + Case( + id="unknown_path", + title="A parameter no certified model defines", + changes=( + Change("gov.budget_2026.new_credit.amount", _date("2027-04-06"), 500), + ), + household=adult(), + expect_error="new_credit", + ), + # Reading cases: each route's documented key reading. + Case( + id="reading_bare_year_dict", + title="A bare-year policyengine-uk dict key is that fiscal year only", + changes=(), + household=adult(), + routes=("country.reform_dict",), + representations={"country.reform_dict": {BASIC_RATE: {"2027": 0.21}}}, + checks=( + Check("parameter", BASIC_RATE, 2027, 0.21, "fiscal year 2027"), + Check("parameter", BASIC_RATE, 2028, 0.20, "that year only"), + ), + ), + Case( + id="reading_bare_year_reform_from_dict", + title="Reform.from_dict reads a bare year as from 1 January on", + changes=(), + household=adult(), + routes=("country.reform_from_dict",), + representations={"country.reform_from_dict": {BASIC_RATE: {"2027": 0.21}}}, + checks=( + Check("parameter", BASIC_RATE, 2026, 0.20, "before 1 January 2027"), + Check("parameter", BASIC_RATE, 2027, 0.21, "from 1 January 2027"), + Check("parameter", BASIC_RATE, 2028, 0.21, "onwards"), + ), + ), + Case( + id="reading_single_day_dict", + title="A single-day policyengine-uk dict key overrides one day", + changes=(), + household=adult(), + routes=("country.reform_dict",), + representations={"country.reform_dict": {BASIC_RATE: {"2027-04-06": 0.21}}}, + checks=( + Check("parameter", BASIC_RATE, 2027, 0.20, "one day is not 30 April"), + Check("parameter", BASIC_RATE, 2028, 0.20, "one day only"), + ), + ), + Case( + id="reading_effective_dates_reverse_order", + title="Effective dates given latest first", + changes=(), + household=adult(), + routes=WRAPPER_ROUTES, + representations={ + "wrapper.household": { + UC_SINGLE_OLD: {"2028-04-06": 520, "2027-04-06": 500} + }, + "wrapper.policy": {UC_SINGLE_OLD: {"2028-04-06": 520, "2027-04-06": 500}}, + }, + checks=( + Check("variable", "uc_standard_allowance", 2027, 6_000.0, "£500 x 12"), + Check("variable", "uc_standard_allowance", 2028, 6_240.0, "£520 x 12"), + ), + ), +) + + +# Representations. + + +def country_reform_dict(changes: Sequence[Change]) -> dict: + reform: dict = {} + for change in changes: + reform.setdefault(change.path, {})[change.interval_key] = change.value + return reform + + +def wrapper_reform_dict(changes: Sequence[Change]) -> dict: + """Lone dates for changes held from a date, intervals for bounded ones.""" + reform: dict = {} + for change in changes: + key = str(change.start) if change.stop is None else change.interval_key + reform.setdefault(change.path, {})[key] = change.value + return reform + + +def represent(case: Case, route: str) -> Any: + if route in case.representations: + return case.representations[route] + if route in COUNTRY_ROUTES: + return country_reform_dict(case.changes) + return wrapper_reform_dict(case.changes) + + +def reform_digest(representation: Any, route: str, case: Optional[Case] = None) -> str: + content: dict = {"route": route, "reform": representation} + if case is not None and case.modifier: + content["modifier"] = [ + [change.path, str(change.start), str(change.stop), change.value] + for change in case.modifier + ] + content["modifier_first"] = case.modifier_first + payload = json.dumps(content, sort_keys=True, default=str) + return hashlib.sha256(payload.encode()).hexdigest() + + +def parameter_modifier(changes: Sequence[Change]) -> Callable: + """A simulation modifier writing ``changes`` onto the annualised parameters.""" + + def modify(simulation): + from policyengine_core.periods import instant + + parameters = simulation.tax_benefit_system.parameters + for change in changes: + parameter = parameters.get_child(change.path) + if change.stop is None: + parameter.update(start=instant(str(change.start)), value=change.value) + else: + parameter.update( + start=instant(str(change.start)), + stop=instant(str(change.stop)), + value=change.value, + ) + return simulation + + return modify + + +# Routes. + + +def _years(case: Case) -> list[int]: + years = {check.year for check in case.checks} + return sorted(years) or [2027] + + +def _situation(household: Household, years: Sequence[int]) -> dict: + def periodised(values: Mapping[str, Any]) -> dict: + return {name: {year: value for year in years} for name, value in values.items()} + + people = { + f"person_{i}": periodised(person) for i, person in enumerate(household.people) + } + return { + "people": people, + "benunits": { + "benunit": {"members": list(people), **periodised(household.benunit)} + }, + "households": { + "household": {"members": list(people), **periodised(household.household)} + }, + } + + +class RouteRun: + """Values a route produced for a case. + + ``unreformed`` values come from simulations built before the reform. + ``isolation`` pairs the reformed simulation's own baseline with them and + ``after`` pairs a fresh unreformed simulation built after the reform with + them, as (name, year, value, unreformed value). + """ + + def __init__(self): + self.reformed: dict[tuple[str, str, int], float] = {} + self.unreformed: dict[tuple[str, str, int], float] = {} + self.isolation: list[tuple[str, int, float, float]] = [] + self.after: list[tuple[str, int, float, float]] = [] + + +def _total(values) -> float: + return float(sum(float(value) for value in values)) + + +class _CountryRoute: + def __init__(self, route: str): + self.route = route + + def simulation(self, representation, situation, case: Optional[Case] = None): + from policyengine_core.reforms import Reform + from policyengine_uk import Simulation + from policyengine_uk.utils.scenario import Scenario + + if case is not None and case.modifier: + # One scenario carrying the route's reform and the modifier. + if self.route == "country.reform_dict": + reform = Scenario.from_reform(representation) + elif self.route == "country.scenario_parameter_changes": + reform = Scenario(parameter_changes=representation) + else: + reform = Scenario.from_reform( + Reform.from_dict(representation, country_id="uk") + ) + modifier = Scenario(simulation_modifier=parameter_modifier(case.modifier)) + scenario = modifier + reform if case.modifier_first else reform + modifier + return Simulation(situation=situation, scenario=scenario) + if self.route == "country.reform_dict": + return Simulation(situation=situation, reform=representation) + if self.route == "country.scenario_parameter_changes": + return Simulation( + situation=situation, + scenario=Scenario(parameter_changes=representation), + ) + return Simulation( + situation=situation, + reform=Reform.from_dict(representation, country_id="uk"), + ) + + @staticmethod + def _value(simulation, request: tuple[str, str, int]) -> float: + kind, name, year = request + if kind == "parameter": + parameter = simulation.tax_benefit_system.parameters.get_child(name) + return float(parameter(f"{year}-01-01")) + return _total(simulation.calculate(name, year)) + + def run(self, case: Case, representation, cache: dict) -> RouteRun: + from policyengine_uk import Simulation + + years = _years(case) + situation = _situation(case.household, years) + situation_key = json.dumps(situation, sort_keys=True, default=str) + requests: list[tuple[str, str, int]] = [] + for check in case.checks: + kind = "parameter" if check.target == "parameter" else "variable" + requests.append((kind, check.name, check.year)) + if check.target == "baseline_equals": + requests.append(("variable", check.expected, check.year)) + # Two checks can read the same value; request it once. + requests = list(dict.fromkeys(requests)) + # Unreformed values are cached, not simulations, so each simulation + # can be released as soon as its case is done. + missing = [ + request + for request in requests + if (self.route, situation_key, request) not in cache + ] + if missing: + unreformed = Simulation(situation=situation) + for request in missing: + cache[(self.route, situation_key, request)] = self._value( + unreformed, request + ) + del unreformed + reformed = self.simulation(representation, situation, case) + baseline = getattr(reformed, "baseline", None) + run = RouteRun() + for request in requests: + run.reformed[request] = self._value(reformed, request) + run.unreformed[request] = cache[(self.route, situation_key, request)] + if request[0] == "variable" and baseline is not None: + run.isolation.append( + ( + request[1], + request[2], + self._value(baseline, request), + run.unreformed[request], + ) + ) + del reformed, baseline + after = Simulation(situation=situation) + for request in requests: + run.after.append( + ( + request[1], + request[2], + self._value(after, request), + run.unreformed[request], + ) + ) + return run + + +class _WrapperHouseholdRoute: + route = "wrapper.household" + + def _call(self, case: Case, year: int, reform, variables: Sequence[str]): + import policyengine as pe + + household = case.household + return pe.uk.calculate_household( + people=[dict(person) for person in household.people], + benunit=dict(household.benunit), + household=dict(household.household), + year=year, + reform=reform, + extra_variables=list(variables), + ) + + def run(self, case: Case, representation, cache: dict) -> RouteRun: + run = RouteRun() + by_year: dict[int, set[str]] = {} + for check in case.checks: + if check.target == "parameter": + continue + by_year.setdefault(check.year, set()).add(check.name) + if check.target == "baseline_equals": + by_year[check.year].add(check.expected) + if case.modifier: + raise ValueError( + "policyengine.uk.calculate_household takes a reform dict, not a " + "simulation modifier" + ) + for year, names in by_year.items(): + names = sorted(names) + key = (self.route, _household_key(case.household), year, tuple(names)) + # The unreformed reference is built before the reform runs. + if key not in cache: + cache[key] = self._call(case, year, None, names) + unreformed = cache[key] + reformed = self._call(case, year, representation, names) + after = self._call(case, year, None, names) + for name in names: + request = ("variable", name, year) + run.reformed[request] = _result_total(reformed, name) + run.unreformed[request] = _result_total(unreformed, name) + run.after.append( + (name, year, _result_total(after, name), run.unreformed[request]) + ) + return run + + +def _household_key(household: Household) -> str: + return json.dumps( + { + "people": list(household.people), + "benunit": dict(household.benunit), + "household": dict(household.household), + }, + sort_keys=True, + default=str, + ) + + +def _result_total(result, name: str) -> float: + """The household's total of a variable, read from the variable's entity. + + The household result can also carry a variable aggregated to another + entity (a benefit-unit award under the household too), so reading every + entity would count it twice. + """ + from policyengine_uk.system import system + + entity = system.variables[name].entity.key + rows = result[entity] if entity in result else [] + if isinstance(rows, Mapping): + rows = [rows] + values = [float(row[name]) for row in rows if name in row] + if not values: + raise KeyError(f"{name} is not in the household result's {entity} rows") + return sum(values) + + +class _WrapperPolicyRoute: + route = "wrapper.policy" + + def _dataset(self, household: Household, year: int): + import pandas as pd + from microdf import MicroDataFrame + + from policyengine.tax_benefit_models.uk.datasets import ( + PolicyEngineUKDataset, + UKYearData, + ) + + people = [dict(person) for person in household.people] + person = pd.DataFrame(people) + person["person_id"] = range(1, len(people) + 1) + person["person_benunit_id"] = 1 + person["person_household_id"] = 1 + person["person_weight"] = 1.0 + benunit = pd.DataFrame( + [{"benunit_id": 1, "benunit_weight": 1.0, **household.benunit}] + ) + # A UK dataset's household table carries a region (the run fails + # without one) and housing costs; these are neutral defaults. + household_frame = pd.DataFrame( + [ + { + "household_id": 1, + "household_weight": 1.0, + "region": "LONDON", + "tenure_type": "OWNED_OUTRIGHT", + "rent": 0.0, + "council_tax": 0.0, + **household.household, + } + ] + ) + return PolicyEngineUKDataset( + id=f"budget-reform-contract-{year}", + name="Budget reform contract household", + description="One household for the Budget reform contract", + filepath=None, + year=year, + data=UKYearData( + person=MicroDataFrame(person, weights="person_weight"), + benunit=MicroDataFrame(benunit, weights="benunit_weight"), + household=MicroDataFrame(household_frame, weights="household_weight"), + ), + ) + + def _run(self, dataset, policy, variables: Mapping[str, Sequence[str]]): + from policyengine.core import Simulation + from policyengine.tax_benefit_models.uk import uk_latest + + simulation = Simulation( + dataset=dataset, + tax_benefit_model_version=uk_latest, + policy=policy, + extra_variables={ + entity: list(names) for entity, names in variables.items() + }, + ) + simulation.run() + return simulation.output_dataset.data + + def run(self, case: Case, representation, cache: dict) -> RouteRun: + from policyengine.tax_benefit_models.common.reform import ( + compile_reform_to_policy, + ) + from policyengine.tax_benefit_models.uk import uk_latest + + run = RouteRun() + by_year: dict[int, set[str]] = {} + for check in case.checks: + if check.target in ("parameter", "baseline_equals"): + continue + by_year.setdefault(check.year, set()).add(check.name) + for year, names in by_year.items(): + variables = _by_entity(sorted(names)) + key = ( + self.route, + _household_key(case.household), + year, + tuple(sorted(names)), + ) + # The unreformed reference is built before the reform runs. + if key not in cache: + cache[key] = self._run( + self._dataset(case.household, year), None, variables + ) + unreformed = cache[key] + policy = compile_reform_to_policy( + representation, year=year, model_version=uk_latest + ) + if case.modifier: + from policyengine.core import Policy + + modifier = Policy( + name="modifier", + simulation_modifier=parameter_modifier(case.modifier), + ) + policy = modifier + policy if case.modifier_first else policy + modifier + reformed = self._run(self._dataset(case.household, year), policy, variables) + after = self._run(self._dataset(case.household, year), None, variables) + for entity, entity_names in variables.items(): + for name in entity_names: + request = ("variable", name, year) + run.reformed[request] = _total(getattr(reformed, entity)[name]) + run.unreformed[request] = _total(getattr(unreformed, entity)[name]) + run.after.append( + ( + name, + year, + _total(getattr(after, entity)[name]), + run.unreformed[request], + ) + ) + return run + + +def _by_entity(names: Sequence[str]) -> dict[str, list[str]]: + from policyengine_uk.system import system + + grouped: dict[str, list[str]] = {} + for name in names: + grouped.setdefault(system.variables[name].entity.key, []).append(name) + return grouped + + +def _route(route: str): + if route in COUNTRY_ROUTES: + return _CountryRoute(route) + if route == "wrapper.household": + return _WrapperHouseholdRoute() + if route == "wrapper.policy": + return _WrapperPolicyRoute() + raise ValueError(f"Unknown route {route!r}; expected one of {ROUTES}.") + + +# Memory. + +# policyengine-core registers each tax-benefit system's variable files in +# sys.modules as "__" and never removes them +# (PolicyEngine/policyengine-core#520), about 10 MB per simulation. The +# contract builds hundreds of simulations, so it drops the modules of systems +# that no longer exist, as policyengine-uk's test suite does. +_VARIABLE_MODULE_NAME = re.compile(r"(\d+)_-?\d+_") + + +def release_dead_variable_modules(max_owners: int = 8) -> int: + owners: dict[str, list[str]] = {} + for name in list(sys.modules): + match = _VARIABLE_MODULE_NAME.match(name) + if match: + owners.setdefault(match.group(1), []).append(name) + if len(owners) <= max_owners: + return 0 + from policyengine_core.taxbenefitsystems import TaxBenefitSystem + + gc.collect() + live = { + str(id(obj)) for obj in gc.get_objects() if isinstance(obj, TaxBenefitSystem) + } + released = 0 + for owner, names in owners.items(): + if owner not in live: + for name in names: + if sys.modules.pop(name, None) is not None: + released += 1 + return released + + +# Evaluation. + + +def reference_parameters(): + """A copy of current law's parameters, taken before any reform runs. + + Expected values read this copy, so a reform that leaks into the shared + system cannot move the reference it is checked against. + """ + from policyengine_uk.system import system + + return system.parameters.clone() + + +def parameter_reader(parameters) -> Callable[[str, int], float]: + def read(path: str, year: int) -> float: + return float(parameters.get_child(path)(f"{year}-01-01")) + + return read + + +def _baseline_parameter_reader() -> Callable[[str, int], float]: + return parameter_reader(reference_parameters()) + + +def _parameter_values(parameters) -> dict[str, list[tuple]]: + """Each parameter's dated values and metadata, by name. + + Every entry of its timeline (date and value) and its metadata, so a + change on any date, or to how it is uprated, shows. A name can occur + more than once (``parameters.baseline`` repeats the tree), so every + occurrence is kept, in tree order. + """ + from policyengine_core.parameters import Parameter + + values: dict[str, list[tuple]] = {} + for parameter in parameters.get_descendants(): + if isinstance(parameter, Parameter): + values.setdefault(parameter.name, []).append( + ( + tuple( + (entry.instant_str, entry.value) + for entry in parameter.values_list + ), + json.dumps(parameter.metadata, sort_keys=True, default=str), + ) + ) + return values + + +def _shared_trees() -> dict: + """policyengine-uk's shared parameter state that a reform could leak into. + + The module system's parameters, and (where the installed version has + them) the cached processed tree every new simulation copies, the cached + statutory tree reforms are written onto, and the cached projected + timelines. The caches are module globals, read without building them. + + Not watched: policyengine-uk's memo of rebuilt reform trees, which the + run itself fills, so there is no state from before the run to compare + it with. A leak into it would show in the reformed values the cases + check against their oracles. + """ + import importlib + + from policyengine_uk.system import system + + module = importlib.import_module("policyengine_uk.tax_benefit_system") + trees = {"system": system.parameters} + for name, attribute in ( + ("processed", "_processed_parameters_cache"), + ("statutory", "_statutory_parameters_cache"), + ("projected", "_projected_timelines_cache"), + ): + value = getattr(module, attribute, None) + if value is not None: + trees[name] = value + return trees + + +def _state_values(tree) -> dict: + if isinstance(tree, Mapping): + # Timelines by name: each entry's date and value. + return { + name: [(entry.instant_str, entry.value) for entry in timeline or []] + for name, timeline in tree.items() + } + return _parameter_values(tree) + + +def shared_state() -> dict: + """The values of policyengine-uk's shared parameter state, by tree.""" + return {name: _state_values(tree) for name, tree in _shared_trees().items()} + + +def module_parameters_unchanged(before: dict) -> dict: + """A report row: the shared parameter state still equals ``before``. + + ``before`` is :func:`shared_state` taken before any reform ran. + """ + now = shared_state() + changed = [] + for tree in sorted(before.keys() | now.keys()): + if tree not in before: + # Built during the run, so there is nothing to compare it with. + continue + old, new = before[tree], now.get(tree, {}) + changed += [ + f"{tree}:{path}" + for path in sorted(old.keys() | new.keys()) + if old.get(path) != new.get(path) + ] + return { + "case_id": "module_parameters_unchanged", + "route": "all", + "reform_digest": None, + "check": ( + "every parameter's dated values and metadata in " + f"{', '.join(sorted(before))}" + ), + "status": FAIL if changed else PASS, + "expected": "policyengine-uk's shared parameters as they were before the run", + "actual": f"{len(changed)} changed: {changed[:10]}" if changed else "unchanged", + "oracle": "the shared parameter state's values, read before any reform ran", + } + + +def _row(case: Case, route: str, digest: Optional[str], **fields) -> dict: + row = {"case_id": case.id, "route": route, "reform_digest": digest} + row.update(fields) + return row + + +def _close(actual: float, expected: float, tolerance: float) -> bool: + return math.isfinite(actual) and abs(actual - expected) <= tolerance + + +def evaluate_case( + case: Case, + route: str, + cache: Optional[dict] = None, + reference: Optional[Callable[[str, int], float]] = None, +) -> list[dict]: + """Run one case through one route; one report row per check. + + ``reference`` reads current law's annual parameters for expected values; + by default, a copy taken now. + """ + cache = {} if cache is None else cache + if route in UNSUPPORTED_ROUTES: + return [ + _row( + case, + route, + None, + check=None, + status=UNSUPPORTED, + reason=UNSUPPORTED_ROUTES[route], + ) + ] + representation = represent(case, route) + digest = reform_digest(representation, route, case) + context = Context(reference or _baseline_parameter_reader()) + if case.expect_error is not None: + try: + _route(route).run( + Case( + case.id, + case.title, + case.changes, + case.household, + (Check("change", "income_tax", 2027, 0.0, ""),), + ), + representation, + cache, + ) + except Exception as error: # noqa: BLE001 - any rejection is reported + matched = re.search(case.expect_error, str(error)) is not None + return [ + _row( + case, + route, + digest, + check="rejected", + status=PASS if matched else FAIL, + expected=f"an error matching {case.expect_error!r}", + actual=f"{type(error).__name__}: {error}"[:500], + ) + ] + return [ + _row( + case, + route, + digest, + check="rejected", + status=FAIL, + expected=f"an error matching {case.expect_error!r}", + actual="accepted", + ) + ] + try: + run = _route(route).run(case, representation, cache) + except Exception as error: # noqa: BLE001 - a crash is a contract error + return [ + _row( + case, + route, + digest, + check=None, + status=ERROR, + detail="".join( + traceback.format_exception_only(type(error), error) + ).strip()[:1000], + ) + ] + rows = [] + for check in case.checks: + label = f"{check.target}:{check.name}:{check.year}" + if check.target == "parameter" and route not in COUNTRY_ROUTES: + rows.append( + _row( + case, + route, + digest, + check=label, + status=UNSUPPORTED, + reason="this route exposes outputs, not parameter values", + ) + ) + continue + if check.target == "baseline_equals" and route == "wrapper.policy": + rows.append( + _row( + case, + route, + digest, + check=label, + status=UNSUPPORTED, + reason="the policy route's output dataset holds the reformed run only", + ) + ) + continue + try: + if check.target == "baseline_equals": + actual = run.reformed[("variable", check.name, check.year)] + expected = run.unreformed[("variable", check.expected, check.year)] + elif check.target == "change": + key = ("variable", check.name, check.year) + actual = run.reformed[key] - run.unreformed[key] + expected = ( + check.expected(context) + if callable(check.expected) + else check.expected + ) + else: + kind = "parameter" if check.target == "parameter" else "variable" + actual = run.reformed[(kind, check.name, check.year)] + expected = ( + check.expected(context) + if callable(check.expected) + else check.expected + ) + except Exception as error: # noqa: BLE001 + rows.append( + _row( + case, + route, + digest, + check=label, + status=PENDING if check.pending is not None else ERROR, + decision_id=check.pending, + detail=repr(error)[:500], + ) + ) + continue + passed = _close(actual, float(expected), check.tolerance) + status = PASS if passed else FAIL + if check.pending is not None: + status = PENDING + rows.append( + _row( + case, + route, + digest, + check=label, + status=status, + expected=float(expected), + actual=actual, + oracle=check.oracle, + decision_id=check.pending, + holds=passed if check.pending is not None else None, + ) + ) + for name, year, from_baseline, unreformed in run.isolation: + rows.append( + _row( + case, + route, + digest, + check=f"baseline_isolation:{name}:{year}", + status=PASS if from_baseline == unreformed else FAIL, + expected=unreformed, + actual=from_baseline, + oracle="the reformed simulation's baseline equals an unreformed simulation", + ) + ) + for name, year, after, unreformed in run.after: + rows.append( + _row( + case, + route, + digest, + check=f"unreformed_after_reform:{name}:{year}", + status=PASS if after == unreformed else FAIL, + expected=unreformed, + actual=after, + oracle=( + "an unreformed simulation built after the reform equals one " + "built before it" + ), + ) + ) + return rows + + +# Properties: bounded sampling. + +SAMPLED_PARAMETERS = { + BASIC_RATE: (False, (0.15, 0.19, 0.21, 0.25)), + UC_SINGLE_OLD: (False, (350.0, 480.0, 512.5, 600.0)), + PERSONAL_ALLOWANCE: (False, (11_000, 12_570, 15_000)), + CGT_BASIC_RATE: (True, (0.1, 0.2, 0.24)), +} +SAMPLED_YEARS = range(2025, 2036) + + +def _sampled_change(rng: random.Random) -> Change: + path = rng.choice(sorted(SAMPLED_PARAMETERS)) + _, values = SAMPLED_PARAMETERS[path] + start = datetime.date(2025, 6, 1) + datetime.timedelta( + days=rng.randrange(0, 8 * 365) + ) + stop = ( + None + if rng.random() < 0.5 + else start + datetime.timedelta(days=rng.randrange(0, 1500)) + ) + return Change(path, start, rng.choice(values), stop) + + +def expected_annual( + change: Change, year: int, baseline: Callable[[str, int], float] +) -> float: + """The annual value a single change gives, under contract 1.0.0. + + Prior law is current law's annual value outside the change. Blended + parameters are sampled only from 2025 on, where current law holds one + value through each fiscal year, so each day of the year takes the + change's value inside the change and prior law's outside it. + """ + blend, _ = SAMPLED_PARAMETERS[change.path] + stop = change.stop or FLAT_STOP + prior = baseline(change.path, year) + if not blend: + sample = datetime.date(year, 4, 30) + return change.value if change.start <= sample <= stop else prior + day = datetime.date(year, 4, 6) + end = datetime.date(year + 1, 4, 6) + total = 0.0 + days = 0 + while day < end: + total += change.value if change.start <= day <= stop else prior + days += 1 + day += datetime.timedelta(days=1) + return total / days + + +def _property_annual_values( + change: Change, + routes: Sequence[str], + cache: dict, + reference: Callable[[str, int], float], +) -> None: + representation = country_reform_dict([change]) + for route in routes: + simulation = _CountryRoute(route).simulation( + representation, _situation(adult(), [2027]) + ) + for year in SAMPLED_YEARS: + actual = float( + simulation.tax_benefit_system.parameters.get_child(change.path)( + f"{year}-01-01" + ) + ) + expected = expected_annual(change, year, reference) + assert abs(actual - expected) <= 1e-9, ( + f"{route}: {change.path} in {year} is {actual}, expected {expected}" + ) + + +SAMPLED_OUTPUTS = ( + "income_tax", + "capital_gains_tax", + "uc_standard_allowance", + "universal_credit", +) +SAMPLED_GAINS = 10_000 + + +def _property_routes_agree( + change: Change, + income: int, + routes: Sequence[str], + cache: dict, + reference: Callable[[str, int], float], +) -> None: + household = adult(income, capital_gains=SAMPLED_GAINS) + case = Case( + "sampled", + "sampled", + (change,), + household, + tuple( + Check("change", name, year, 0.0, "") + for name in SAMPLED_OUTPUTS + for year in (2026, 2028, 2031) + ), + ) + results = {} + for route in routes: + run = _route(route).run(case, represent(case, route), cache) + results[route] = dict(run.reformed) + for name, year, value, unreformed in run.isolation + run.after: + assert value == unreformed, ( + f"{route}: unreformed {name} in {year} is {value} after the " + f"reform, {unreformed} before it" + ) + first_route = next(iter(results)) + first = results[first_route] + for route, values in results.items(): + for key, value in values.items(): + assert abs(value - first[key]) <= 1e-6, ( + f"{key[1]} in {key[2]} is {value} through {route} and " + f"{first[key]} through {first_route}" + ) + + +PROPERTIES = { + "annual_values_match_the_oracle": "Random single changes on the country routes give the annual values of an independent fiscal-year annualisation.", + "routes_agree_and_baselines_hold": "Random changes give the same household income tax, capital gains tax, Universal Credit standard allowance and award on every in-process route, and leave each route's baseline and later unreformed simulations unchanged.", +} + + +def _counterexample(change: Optional[Change], income: Optional[int], detail) -> dict: + return { + "change": repr(change), + "income": income, + "detail": str(detail)[:2000], + } + + +def run_property( + property_id: str, + routes: Sequence[str], + seed: int, + max_examples: int, + cache: dict, + reference: Optional[Callable[[str, int], float]] = None, +) -> dict: + """Sample ``max_examples`` changes and check one property on each. + + A failed check is ``fail`` and any other exception ``error``, under + Hypothesis or, when it is not installed, a seeded random sampler; both + report the change and income that broke the property. + """ + if max_examples <= 0: + return { + "property_id": property_id, + "status": UNSUPPORTED, + "examples_run": 0, + "reason": "max_examples is 0, so no examples were requested", + } + reference = reference or _baseline_parameter_reader() + country = [route for route in routes if route in COUNTRY_ROUTES] + in_process = [route for route in routes if route not in UNSUPPORTED_ROUTES] + if property_id == "annual_values_match_the_oracle": + if not country: + return { + "property_id": property_id, + "status": UNSUPPORTED, + "examples_run": 0, + "reason": "needs a country route", + } + + def check(change, income): + _property_annual_values(change, country, cache, reference) + + else: + if len(in_process) < 2: + return { + "property_id": property_id, + "status": UNSUPPORTED, + "examples_run": 0, + "reason": "needs two in-process routes", + } + + def check(change, income): + _property_routes_agree(change, income, in_process, cache, reference) + + examples = 0 + current: list = [None, None] + + def example(change, income): + nonlocal examples + examples += 1 + current[:] = [change, income] + check(change, income) + release_dead_variable_modules() + + try: + from hypothesis import HealthCheck, Phase, given, settings + from hypothesis import seed as hypothesis_seed + from hypothesis import strategies as st + except ImportError: + engine = "random" + else: + engine = "hypothesis" + try: + if engine == "hypothesis": + + @hypothesis_seed(seed) + @settings( + max_examples=max_examples, + deadline=None, + database=None, + suppress_health_check=list(HealthCheck), + # Shrinking reruns a failing example many times; each example + # builds several simulations, so it is left out to bound + # runtime. The first failure is reported as found. + phases=[Phase.explicit, Phase.generate], + report_multiple_bugs=False, + ) + @given( + change=st.randoms(use_true_random=False).map(_sampled_change), + income=st.integers(0, 150_000), + ) + def sampled(change, income): + example(change, income) + + sampled() + else: + rng = random.Random(seed) + for _ in range(max_examples): + example(_sampled_change(rng), rng.randrange(0, 150_001)) + except AssertionError as failure: + return { + "property_id": property_id, + "engine": engine, + "examples_run": examples, + "status": FAIL, + "counterexample": _counterexample(*current, failure), + } + except Exception as error: # noqa: BLE001 - a crash is a contract error + return { + "property_id": property_id, + "engine": engine, + "examples_run": examples, + "status": ERROR, + "counterexample": _counterexample( + *current, + "".join(traceback.format_exception_only(type(error), error)).strip(), + ), + } + return { + "property_id": property_id, + "engine": engine, + "examples_run": examples, + "status": PASS, + } + + +# Entry point. + + +def _versions() -> dict: + versions = {} + for package in ("policyengine", "policyengine-uk", "policyengine-core"): + try: + versions[package] = importlib.metadata.version(package) + except importlib.metadata.PackageNotFoundError: + versions[package] = None + return versions + + +def _sha256(path: str) -> str: + digest = hashlib.sha256() + with open(path, "rb") as handle: + for block in iter(lambda: handle.read(1 << 20), b""): + digest.update(block) + return digest.hexdigest() + + +def run_budget_reform_contract( + *, + routes: Optional[Sequence[str]] = None, + dataset_path: Optional[str] = None, + seed: int = 0, + max_examples: int = 20, + cases: Optional[Sequence[str]] = None, +) -> dict: + """Run the UK Budget reform contract and return its JSON report. + + Args: + routes: Route ids to run (see :data:`ROUTES`). None runs every route; + ``api.sim_api`` reports ``unsupported`` with its reason. + dataset_path: A certified UK dataset file. When given, the population + checks also run on a deterministic subsample of it. + seed: Seed for the sampled properties. + max_examples: Examples per sampled property. + cases: Case ids to run; None runs every case. + """ + selected_routes = list(routes) if routes is not None else list(ROUTES) + if not selected_routes: + raise ValueError(f"No routes selected; expected some of {ROUTES}.") + unknown = [route for route in selected_routes if route not in ROUTES] + if unknown: + raise ValueError(f"Unknown routes {unknown}; expected some of {ROUTES}.") + case_ids = [case.id for case in CASES] + if cases is not None: + if not cases: + raise ValueError("No cases selected; pass None to run every case.") + unknown = [case_id for case_id in cases if case_id not in case_ids] + if unknown: + raise ValueError(f"Unknown cases {unknown}; expected some of {case_ids}.") + selected_cases = [case for case in CASES if cases is None or case.id in cases] + # Expected values read current law as it is before any reform runs. + reference = parameter_reader(reference_parameters()) + before = shared_state() + cache: dict = {} + rows: list[dict] = [] + for case in selected_cases: + for route in selected_routes: + if case.routes is not None and route not in case.routes: + continue + rows.extend(evaluate_case(case, route, cache, reference)) + release_dead_variable_modules() + properties = [ + run_property(property_id, selected_routes, seed, max_examples, cache, reference) + for property_id in PROPERTIES + ] + population: list[dict] = [] + if dataset_path is not None: + from policyengine.tax_benefit_models.uk.budget_reform_population import ( + run_population_checks, + ) + + population = run_population_checks(dataset_path, selected_routes, seed) + # Whether any check passed, before the run-level row (which passes + # whenever nothing leaked, including when nothing ran). + passed = any(row["status"] == PASS for row in rows + properties + population) + if any(route not in UNSUPPORTED_ROUTES for route in selected_routes): + rows.append(module_parameters_unchanged(before)) + if not passed: + # A run in which nothing passed certifies nothing. + rows.append( + { + "case_id": "run", + "route": "all", + "reform_digest": None, + "check": "at least one check passed", + "status": ERROR, + "detail": "no case, property or population check passed", + } + ) + summary = {status: 0 for status in (PASS, FAIL, ERROR, UNSUPPORTED, PENDING)} + for row in rows + properties + population: + summary[row["status"]] = summary.get(row["status"], 0) + 1 + return { + "contract_id": CONTRACT_ID, + "contract_version": CONTRACT_VERSION, + "prior_law_option": PRIOR_LAW_OPTION, + "versions": _versions(), + "dataset": ( + {"path": dataset_path, "sha256": _sha256(dataset_path)} + if dataset_path is not None + else None + ), + "routes": selected_routes, + "seed": seed, + "max_examples": max_examples, + "cases": rows, + "properties": properties, + "population": population, + "summary": summary, + "holds": summary[FAIL] == 0 and summary[ERROR] == 0, + } diff --git a/src/policyengine/tax_benefit_models/uk/budget_reform_population.py b/src/policyengine/tax_benefit_models/uk/budget_reform_population.py new file mode 100644 index 00000000..9ed54b55 --- /dev/null +++ b/src/policyengine/tax_benefit_models/uk/budget_reform_population.py @@ -0,0 +1,353 @@ +"""Population checks for the UK Budget reform contract. + +Run by :func:`~policyengine.tax_benefit_models.uk.budget_reform_contract.run_budget_reform_contract` +when it is given a certified dataset. They use a deterministic subsample of +the dataset's households, so a data-dependent failure (a reform reaching +the baseline through a cached value, or a route that diverges only for some +households) shows up without the cost of the full population. + +Checks: + +``population_baseline_isolation`` + An unreformed microsimulation, the ``baseline`` of a reformed one and an + unreformed one built after the reform give identical arrays. +``population_query_order`` + Calculating the outputs in reverse order gives identical arrays. +``population_routes_agree`` + A basic rate of 21% from 6 April 2027 gives identical income tax arrays + through the reform dict, ``parameter_changes`` and ``Reform.from_dict``. +``population_april_rate_change`` + That reform leaves 2026 income tax unchanged for every person and raises + total 2027 income tax. (It can lower one person's tax: in Scotland a + higher UK basic rate grosses up Gift Aid by more, restoring more of a + tapered personal allowance.) +``population_finite_change_expires`` + The same rate for 2027-28 only leaves 2028 identical to the baseline. +``population_uc_standard_allowance_plus_10`` + Every Universal Credit standard allowance 10% higher in 2027-28 raises + each benefit unit's standard allowance by exactly 10%, lowers no award + and raises total Universal Credit. +``population_state_pension_plus_10`` + The basic and new State Pension rates 10% higher in 2027-28 raise each + person's basic and new State Pension by exactly 10% and raise total + State Pension. + +The last two are responsiveness checks: a reform that reaches the parameter +but not the award (as when an award is fixed before the reform applies) +fails them. +""" + +from __future__ import annotations + +import os +import tempfile +import traceback +from collections.abc import Sequence +from typing import Any + +import numpy as np + +SUBSAMPLE_HOUSEHOLDS = 1_000 +OUTPUTS = ("income_tax", "universal_credit", "household_net_income") +YEARS = (2026, 2027, 2028) +BASIC_RATE = "gov.hmrc.income_tax.rates.uk[0].rate" +APRIL = {BASIC_RATE: {"2027-04-06.2100-12-31": 0.21}} +FINITE = {BASIC_RATE: {"2027-04-06.2028-04-05": 0.21}} +UC_STANDARD_ALLOWANCES = tuple( + f"gov.dwp.universal_credit.standard_allowance.amount.{claimant_type}" + for claimant_type in ("SINGLE_YOUNG", "SINGLE_OLD", "COUPLE_YOUNG", "COUPLE_OLD") +) +STATE_PENSION_RATES = ( + "gov.dwp.state_pension.basic_state_pension.amount", + "gov.dwp.state_pension.new_state_pension.amount", +) +RESPONSE_YEAR = 2027 + + +def plus_ten_percent(paths: Sequence[str]) -> dict: + """Each parameter 10% above current law for the response year.""" + from policyengine_uk.system import system + + return { + path: { + f"{RESPONSE_YEAR}-04-06.{RESPONSE_YEAR + 1}-04-05": 1.1 + * float(system.parameters.get_child(path)(f"{RESPONSE_YEAR}-01-01")) + } + for path in paths + } + + +def _filter_year(single_year, households: np.ndarray): + from policyengine_uk.data import UKSingleYearDataset + + person = single_year.person + benunit = single_year.benunit + household = single_year.household + keep_people = person["person_household_id"].isin(households) + kept_benunits = person.loc[keep_people, "person_benunit_id"].unique() + return UKSingleYearDataset( + person=person[keep_people].reset_index(drop=True), + benunit=benunit[benunit["benunit_id"].isin(kept_benunits)].reset_index( + drop=True + ), + household=household[household["household_id"].isin(households)].reset_index( + drop=True + ), + fiscal_year=int(str(single_year.time_period)[:4]), + ) + + +def _h5_path(dataset_path: str, directory: str) -> str: + """The file under a name ending in .h5, which policyengine-uk requires. + + A content-addressed cache stores the certified file without an + extension, so link it under one. + """ + if dataset_path.endswith(".h5"): + return dataset_path + linked = os.path.join(directory, "dataset.h5") + os.symlink(os.path.abspath(dataset_path), linked) + return linked + + +def subsample(dataset_path: str, households: int, seed: int): + """A deterministic subsample of a policyengine-uk dataset file.""" + with tempfile.TemporaryDirectory() as directory: + return _subsample(_h5_path(dataset_path, directory), households, seed) + + +def _subsample(dataset_path: str, households: int, seed: int): + from policyengine_uk.data import UKMultiYearDataset, UKSingleYearDataset + + if UKMultiYearDataset.validate_file_path(dataset_path, False): + dataset = UKMultiYearDataset(dataset_path) + first = dataset[dataset.years[0]] + else: + dataset = UKSingleYearDataset(dataset_path) + first = dataset + ids = np.sort(first.household["household_id"].to_numpy()) + rng = np.random.default_rng(seed) + chosen = np.sort(rng.choice(ids, size=min(households, len(ids)), replace=False)) + if isinstance(dataset, UKMultiYearDataset): + return UKMultiYearDataset( + datasets=[_filter_year(dataset[year], chosen) for year in dataset.years] + ) + return _filter_year(dataset, chosen) + + +def _arrays(simulation, order: int = 1) -> dict[str, np.ndarray]: + pairs = [(variable, year) for variable in OUTPUTS for year in YEARS][::order] + return { + f"{variable}:{year}": np.asarray( + simulation.calculate(variable, year), dtype=float + ) + for variable, year in pairs + } + + +def _same(a: dict[str, np.ndarray], b: dict[str, np.ndarray]) -> list[str]: + return [key for key in a if not np.array_equal(a[key], b[key], equal_nan=True)] + + +def _row(check_id: str, route: str, status: str, **fields: Any) -> dict: + return {"check_id": check_id, "route": route, "status": status, **fields} + + +def run_population_checks( + dataset_path: str, routes: Sequence[str], seed: int = 0 +) -> list[dict]: + """Run the population checks; one report row per check.""" + from policyengine_core.reforms import Reform + from policyengine_uk import Microsimulation + from policyengine_uk.utils.scenario import Scenario + + country = [route for route in routes if route.startswith("country.")] + if not country: + return [ + _row( + "population", + "-", + "unsupported", + reason="population checks need a country route", + ) + ] + rows: list[dict] = [] + try: + data = subsample(dataset_path, SUBSAMPLE_HOUSEHOLDS, seed) + + def build(route: str | None, reform: dict | None): + if reform is None: + return Microsimulation(dataset=data.copy()) + if route == "country.reform_dict": + return Microsimulation(dataset=data.copy(), reform=reform) + if route == "country.scenario_parameter_changes": + return Microsimulation( + dataset=data.copy(), scenario=Scenario(parameter_changes=reform) + ) + return Microsimulation( + dataset=data.copy(), + reform=Reform.from_dict(reform, country_id="uk"), + ) + + before = _arrays(build(None, None)) + reformed = {route: build(route, APRIL) for route in country} + for route, simulation in reformed.items(): + different = _same(before, _arrays(simulation.baseline)) + rows.append( + _row( + "population_baseline_isolation", + route, + "fail" if different else "pass", + detail=different or None, + ) + ) + after = _arrays(build(None, None)) + different = _same(before, after) + rows.append( + _row( + "population_baseline_isolation", + "unreformed", + "fail" if different else "pass", + detail=different or None, + ) + ) + different = _same(before, _arrays(build(None, None), order=-1)) + rows.append( + _row( + "population_query_order", + "unreformed", + "fail" if different else "pass", + detail=different or None, + ) + ) + tax = { + route: { + year: np.asarray(simulation.calculate("income_tax", year), dtype=float) + for year in (2026, 2027) + } + for route, simulation in reformed.items() + } + first = country[0] + for route in country[1:]: + same = all( + np.array_equal(tax[route][year], tax[first][year]) + for year in (2026, 2027) + ) + rows.append( + _row( + "population_routes_agree", + f"{first} = {route}", + "pass" if same else "fail", + ) + ) + for route in country: + unchanged_2026 = np.array_equal(tax[route][2026], before["income_tax:2026"]) + change_2027 = tax[route][2027] - before["income_tax:2027"] + # Some people's tax can fall: in Scotland a higher UK basic rate + # grosses up Gift Aid by more, which restores more of a tapered + # personal allowance while Scottish rates stay put. So the check + # is on the total, and the counts are reported. + total_rises = bool(np.sum(change_2027) > 1e-6) + ok = unchanged_2026 and total_rises + rows.append( + _row( + "population_april_rate_change", + route, + "pass" if ok else "fail", + detail={ + "2026_unchanged": unchanged_2026, + "2027_total_change": float(np.sum(change_2027)), + "2027_people_with_lower_tax": int(np.sum(change_2027 < -1e-6)), + "2027_people_with_higher_tax": int(np.sum(change_2027 > 1e-6)), + }, + ) + ) + for route in country: + simulation = build(route, FINITE) + same = np.array_equal( + np.asarray(simulation.calculate("income_tax", 2028), dtype=float), + before["income_tax:2028"], + ) + rows.append( + _row( + "population_finite_change_expires", + route, + "pass" if same else "fail", + ) + ) + + def values(simulation, variable): + return np.asarray( + simulation.calculate(variable, RESPONSE_YEAR), dtype=float + ) + + unreformed = build(None, None) + base = { + variable: values(unreformed, variable) + for variable in ( + "uc_standard_allowance", + "universal_credit", + "basic_state_pension", + "new_state_pension", + "state_pension", + ) + } + for route in country: + simulation = build(route, plus_ten_percent(UC_STANDARD_ALLOWANCES)) + allowance = values(simulation, "uc_standard_allowance") + award = values(simulation, "universal_credit") + scaled = bool( + np.allclose(allowance, 1.1 * base["uc_standard_allowance"], rtol=1e-6) + ) + lower = int(np.sum(award < base["universal_credit"] - 1e-3)) + higher = int(np.sum(award > base["universal_credit"] + 1e-3)) + rows.append( + _row( + "population_uc_standard_allowance_plus_10", + route, + "pass" if scaled and lower == 0 and higher > 0 else "fail", + detail={ + "standard_allowance_scaled_10_percent": scaled, + "awards_lower": lower, + "awards_higher": higher, + }, + ) + ) + for route in country: + simulation = build(route, plus_ten_percent(STATE_PENSION_RATES)) + scaled = all( + np.allclose( + values(simulation, variable), 1.1 * base[variable], rtol=1e-6 + ) + for variable in ("basic_state_pension", "new_state_pension") + ) + pension = values(simulation, "state_pension") + lower = int(np.sum(pension < base["state_pension"] - 1e-3)) + higher = int(np.sum(pension > base["state_pension"] + 1e-3)) + rows.append( + _row( + "population_state_pension_plus_10", + route, + "pass" if scaled and lower == 0 and higher > 0 else "fail", + detail={ + "basic_and_new_scaled_10_percent": bool(scaled), + "pensions_lower": lower, + "pensions_higher": higher, + }, + ) + ) + except Exception as error: # noqa: BLE001 - reported, never swallowed + rows.append( + _row( + "population", + "-", + "error", + detail="".join( + traceback.format_exception_only(type(error), error) + ).strip()[:1000], + ) + ) + for row in rows: + row.setdefault("subsample_households", SUBSAMPLE_HOUSEHOLDS) + row.setdefault("seed", seed) + return rows diff --git a/tests/test_uk_budget_reform_contract.py b/tests/test_uk_budget_reform_contract.py new file mode 100644 index 00000000..d07e91c9 --- /dev/null +++ b/tests/test_uk_budget_reform_contract.py @@ -0,0 +1,570 @@ +"""The UK Budget reform contract runner. + +The runner's oracles, representations and status logic are tested here +without depending on which policyengine-uk is installed. Whether an +installed model satisfies the contract is what qualification checks; the +integration test at the end runs one case when the installed +policyengine-uk has the statutory-timeline reform routes. +""" + +import datetime +import json + +import pytest + +pytest.importorskip("policyengine_uk") + +from policyengine.tax_benefit_models.uk import budget_reform_contract as contract +from policyengine.tax_benefit_models.uk.budget_reform_contract import ( + CASES, + Case, + Change, + Check, + RouteRun, + adult, + country_reform_dict, + evaluate_case, + expected_annual, + fiscal_year_average, + run_budget_reform_contract, + s21_basic_rate_limit, + s57_personal_allowance, + weekly, + wrapper_reform_dict, +) + +RATE = "gov.hmrc.income_tax.rates.uk[0].rate" + + +# Oracles. + + +def test_statutory_indexation_arithmetic(): + # ITA 2007 s57(3): 12,570 x 5% = 628.50, rounded up to 630. + assert s57_personal_allowance(12_570, 0.05) == 13_200 + # s21(3): 37,700 x 1.05 = 39,585, rounded up to 39,600. + assert s21_basic_rate_limit(37_700, 0.05) == 39_600 + # No rise without a CPI increase (s57(2), s21(1)). + assert s57_personal_allowance(12_570, 0) == 12_570 + assert s21_basic_rate_limit(37_700, -0.01) == 37_700 + # A rise that is already a multiple of £10 is not rounded further. + assert s57_personal_allowance(10_000, 0.02) == 10_200 + + +def test_weekly_thresholds_round_to_the_penny(): + assert weekly(12_570) == 241.73 + assert weekly(50_270) == 966.73 + + +def test_fiscal_year_average_is_day_weighted(): + pieces = [ + (datetime.date(2024, 10, 30), 0.18), + (datetime.date(2027, 10, 30), 0.20), + ] + # 6 April to 29 October 2027 is 207 days; 2027-28 has 366. + assert fiscal_year_average(pieces, 2027) == pytest.approx( + (0.18 * 207 + 0.20 * 159) / 366, abs=1e-12 + ) + assert fiscal_year_average(pieces, 2028) == pytest.approx(0.20) + + +def test_expected_annual_for_a_single_change(): + def baseline(path, year): + return 0.20 + + april = Change(RATE, datetime.date(2027, 4, 6), 0.21) + finite = Change(RATE, datetime.date(2027, 4, 6), 0.21, datetime.date(2028, 4, 5)) + may = Change(RATE, datetime.date(2027, 5, 1), 0.21) + assert [expected_annual(april, y, baseline) for y in (2026, 2027, 2028)] == [ + 0.20, + 0.21, + 0.21, + ] + assert [expected_annual(finite, y, baseline) for y in (2027, 2028)] == [0.21, 0.20] + # A change after 30 April misses that fiscal year's sample. + assert expected_annual(may, 2027, baseline) == 0.20 + + +# Representations. + + +def test_representations(): + changes = ( + Change(RATE, datetime.date(2027, 4, 6), 0.21, datetime.date(2028, 4, 5)), + Change(RATE, datetime.date(2028, 4, 6), 0.22), + ) + assert country_reform_dict(changes) == { + RATE: {"2027-04-06.2028-04-05": 0.21, "2028-04-06.2100-12-31": 0.22} + } + assert wrapper_reform_dict(changes) == { + RATE: {"2027-04-06.2028-04-05": 0.21, "2028-04-06": 0.22} + } + + +def test_case_ids_are_unique_and_every_check_has_an_oracle(): + ids = [case.id for case in CASES] + assert len(ids) == len(set(ids)) + for case in CASES: + for check in case.checks: + assert check.oracle, (case.id, check) + assert check.target in { + "parameter", + "variable", + "change", + "baseline_equals", + } + assert case.changes or case.representations, case.id + + +# Status logic, through a fake route. + + +class FakeRoute: + def __init__(self, reformed, unreformed, isolation=(), error=None, after=()): + self.reformed, self.unreformed = reformed, unreformed + self.isolation, self.error = list(isolation), error + self.after = list(after) + + def run(self, case, representation, cache): + if self.error is not None: + raise self.error + run = RouteRun() + run.reformed.update(self.reformed) + run.unreformed.update(self.unreformed) + run.isolation.extend(self.isolation) + run.after.extend(self.after) + return run + + +def fake(monkeypatch, route): + monkeypatch.setattr(contract, "_route", lambda name: route) + monkeypatch.setattr( + contract, "_baseline_parameter_reader", lambda: lambda p, y: 0.20 + ) + + +CASE = Case( + id="fake", + title="fake", + changes=(Change(RATE, datetime.date(2027, 4, 6), 0.21),), + household=adult(), + checks=( + Check("parameter", RATE, 2027, 0.21, "oracle"), + Check("change", "income_tax", 2027, 174.30, "oracle", 0.005), + Check("variable", "income_tax", 2027, 0.0, "oracle", pending="d1344"), + ), +) + + +def test_passing_failing_pending_and_isolation_rows(monkeypatch): + fake( + monkeypatch, + FakeRoute( + reformed={ + ("parameter", RATE, 2027): 0.21, + ("variable", "income_tax", 2027): 3_660.30, + }, + unreformed={("variable", "income_tax", 2027): 3_486.00}, + isolation=[("income_tax", 2027, 3_486.0, 3_486.0)], + ), + ) + rows = {row["check"]: row for row in evaluate_case(CASE, "country.reform_dict")} + assert rows[f"parameter:{RATE}:2027"]["status"] == "pass" + assert rows["change:income_tax:2027"]["status"] == "pass" + assert rows["variable:income_tax:2027"]["status"] == "pending" + assert rows["variable:income_tax:2027"]["decision_id"] == "d1344" + assert rows["variable:income_tax:2027"]["holds"] is False + assert rows["baseline_isolation:income_tax:2027"]["status"] == "pass" + + +def test_a_wrong_value_fails_and_a_leaking_baseline_fails(monkeypatch): + fake( + monkeypatch, + FakeRoute( + reformed={ + ("parameter", RATE, 2027): 0.20, + ("variable", "income_tax", 2027): 3_486.00, + }, + unreformed={("variable", "income_tax", 2027): 3_486.00}, + isolation=[("income_tax", 2027, 3_660.3, 3_486.0)], + ), + ) + rows = {row["check"]: row for row in evaluate_case(CASE, "country.reform_dict")} + assert rows[f"parameter:{RATE}:2027"]["status"] == "fail" + assert rows["change:income_tax:2027"]["status"] == "fail" + assert rows["baseline_isolation:income_tax:2027"]["status"] == "fail" + + +def test_a_crash_is_an_error_and_parameters_are_unsupported_on_wrapper_routes( + monkeypatch, +): + fake(monkeypatch, FakeRoute({}, {}, error=RuntimeError("boom"))) + (row,) = evaluate_case(CASE, "country.reform_dict") + assert row["status"] == "error" and "boom" in row["detail"] + fake( + monkeypatch, + FakeRoute( + reformed={("variable", "income_tax", 2027): 3_660.30}, + unreformed={("variable", "income_tax", 2027): 3_486.00}, + ), + ) + rows = {row["check"]: row for row in evaluate_case(CASE, "wrapper.household")} + assert rows[f"parameter:{RATE}:2027"]["status"] == "unsupported" + assert rows["change:income_tax:2027"]["status"] == "pass" + + +def test_rejection_cases(monkeypatch): + case = next(case for case in CASES if case.id == "unknown_path") + fake(monkeypatch, FakeRoute({}, {}, error=ValueError("no new_credit here"))) + (row,) = evaluate_case(case, "country.reform_dict") + assert row["status"] == "pass" + fake(monkeypatch, FakeRoute({}, {}, error=ValueError("something else"))) + (row,) = evaluate_case(case, "country.reform_dict") + assert row["status"] == "fail" + fake(monkeypatch, FakeRoute({}, {})) + (row,) = evaluate_case(case, "country.reform_dict") + assert row["status"] == "fail" and row["actual"] == "accepted" + + +def test_report_shape_for_an_unsupported_route(): + report = run_budget_reform_contract(routes=["api.sim_api"], max_examples=0) + json.dumps(report) + assert report["contract_id"] == contract.CONTRACT_ID + assert report["contract_version"] == contract.CONTRACT_VERSION + assert report["prior_law_option"] == "projected" + unsupported = [row for row in report["cases"] if row["case_id"] != "run"] + assert {row["status"] for row in unsupported} == {"unsupported"} + assert all(row["reason"] for row in unsupported) + # Nothing passed, so the run certifies nothing. + (run,) = [row for row in report["cases"] if row["case_id"] == "run"] + assert run["status"] == "error" + assert report["summary"]["error"] == 1 + assert report["holds"] is False + + +@pytest.mark.parametrize( + "selection, message", + [ + ({"routes": ["country.typo"]}, "Unknown routes"), + ({"routes": []}, "No routes"), + ({"cases": ["april_rate_chagne"]}, "Unknown cases"), + ({"cases": []}, "No cases"), + ], +) +def test_a_selection_that_would_certify_nothing_is_rejected(selection, message): + with pytest.raises(ValueError, match=message): + run_budget_reform_contract(max_examples=0, **selection) + + +def test_an_unreformed_run_that_moves_after_the_reform_fails(monkeypatch): + fake( + monkeypatch, + FakeRoute( + reformed={("variable", "income_tax", 2027): 3_660.30}, + unreformed={("variable", "income_tax", 2027): 3_486.00}, + after=[("income_tax", 2027, 3_660.30, 3_486.00)], + ), + ) + rows = {row["check"]: row for row in evaluate_case(CASE, "wrapper.household")} + assert rows["change:income_tax:2027"]["status"] == "pass" + assert rows["unreformed_after_reform:income_tax:2027"]["status"] == "fail" + + +def test_expected_values_read_the_reference_they_are_given(monkeypatch): + fake( + monkeypatch, + FakeRoute( + reformed={("variable", "income_tax", 2027): 3_660.30}, + unreformed={("variable", "income_tax", 2027): 3_486.00}, + ), + ) + case = next(case for case in CASES if case.id == "april_rate_change") + # A reference that a leak had moved to 21% would expect no change. + rows = evaluate_case(case, "wrapper.household", reference=lambda path, year: 0.21) + (row,) = [row for row in rows if row["check"] == "change:income_tax:2027"] + assert row["status"] == "fail" and row["expected"] == 0.0 + + +def test_module_parameters_unchanged_compares_the_shared_state(): + before = contract.shared_state() + assert "system" in before + assert contract.module_parameters_unchanged(before)["status"] == "pass" + # Changing the snapshot (never the shared state) stands in for a leak. + tree = "processed" if "processed" in before else "system" + entries, metadata = before[tree][RATE][0] + # A change on one date, 6 April 2027, that no 1 January reading sees. + before[tree][RATE][0] = (entries + (("2027-04-06", 0.5),), metadata) + row = contract.module_parameters_unchanged(before) + assert row["status"] == "fail" and f"{tree}:{RATE}" in row["actual"] + + +def test_module_parameters_unchanged_compares_projected_timelines(): + before = contract.shared_state() + if "projected" not in before: + pytest.skip("this policyengine-uk keeps no projected timelines") + name = next(iter(before["projected"])) + before["projected"][name] = before["projected"][name][1:] + row = contract.module_parameters_unchanged(before) + assert row["status"] == "fail" and f"projected:{name}" in row["actual"] + + +def test_a_run_in_which_nothing_passed_errors_with_an_in_process_route(): + # A country-only case selected with a wrapper route runs nothing; the + # run-level parameters row passes, but must not count as a pass. + report = run_budget_reform_contract( + routes=["wrapper.household"], + cases=["september_cpi_moves_indexed_thresholds"], + max_examples=0, + ) + statuses = {row["case_id"]: row["status"] for row in report["cases"]} + assert statuses == {"module_parameters_unchanged": "pass", "run": "error"} + assert report["holds"] is False + + +def _failing_property(monkeypatch, error): + def check(change, income, routes, cache, reference): + raise error + + monkeypatch.setattr(contract, "_property_routes_agree", check) + return contract.run_property( + "routes_agree_and_baselines_hold", + ["country.reform_dict", "wrapper.household"], + seed=0, + max_examples=3, + cache={}, + reference=lambda path, year: 0.20, + ) + + +def _without_hypothesis(monkeypatch): + import builtins + + real_import = builtins.__import__ + + def guarded(name, *args, **kwargs): + if name == "hypothesis" or name.startswith("hypothesis."): + raise ImportError(name) + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", guarded) + + +@pytest.mark.parametrize("hypothesis_installed", [True, False]) +@pytest.mark.parametrize( + "error, status", + [(AssertionError("routes disagree"), "fail"), (RuntimeError("crashed"), "error")], +) +def test_both_property_engines_report_failures_with_their_inputs( + monkeypatch, hypothesis_installed, error, status +): + if hypothesis_installed: + pytest.importorskip("hypothesis") + else: + _without_hypothesis(monkeypatch) + result = _failing_property(monkeypatch, error) + assert result["status"] == status + assert result["engine"] == ("hypothesis" if hypothesis_installed else "random") + counterexample = result["counterexample"] + assert counterexample["change"].startswith("Change(") + assert isinstance(counterexample["income"], int) + assert str(error) in counterexample["detail"] + + +# Integration: one case on the installed model. + + +def test_the_april_rate_case_holds_on_the_country_routes(): + pytest.importorskip("policyengine_uk.utils.dated_reforms") + report = run_budget_reform_contract( + routes=list(contract.COUNTRY_ROUTES), + cases=["april_rate_change", "finite_rate_change"], + max_examples=0, + ) + failures = [row for row in report["cases"] if row["status"] in ("fail", "error")] + assert failures == [] + checks = [row["check"] for row in report["cases"] if row["route"] != "all"] + assert "unreformed_after_reform:income_tax:2027" in checks + # Each value is checked once per case and route. + labelled = [(row["case_id"], row["route"], row["check"]) for row in report["cases"]] + assert len(labelled) == len(set(labelled)) + assert any( + row["case_id"] == "module_parameters_unchanged" for row in report["cases"] + ) + + +def test_a_parameter_modifier_writes_the_annualised_parameters(): + from policyengine_uk import Simulation + + simulation = Simulation(situation=contract._situation(adult(), [2027])) + contract.parameter_modifier( + [ + Change(RATE, datetime.date(2027, 1, 1), 0.25, datetime.date(2027, 12, 31)), + Change( + "gov.hmrc.income_tax.rates.dividends[0].rate", + datetime.date(2026, 1, 1), + 0.0875, + ), + ] + )(simulation) + parameters = simulation.tax_benefit_system.parameters + assert parameters.get_child(RATE)("2027-01-01") == 0.25 + assert parameters.get_child(RATE)("2028-01-01") == 0.20 + dividends = parameters.get_child("gov.hmrc.income_tax.rates.dividends[0].rate") + assert dividends("2030-01-01") == 0.0875 + + +def test_modifier_cases_differ_only_in_order_and_skip_the_household_route(): + first, second = ( + next(case for case in CASES if case.id == case_id) + for case_id in ( + "modifier_then_parameter_changes", + "parameter_changes_then_modifier", + ) + ) + assert first.modifier and first.modifier == second.modifier + assert first.changes == second.changes + assert first.modifier_first and not second.modifier_first + assert "wrapper.household" not in first.routes + assert contract.reform_digest({}, "wrapper.policy", first) != ( + contract.reform_digest({}, "wrapper.policy", second) + ) + # Cases without a modifier keep their digests. + plain = next(case for case in CASES if case.id == "april_rate_change") + assert contract.reform_digest({}, "r", plain) == contract.reform_digest({}, "r") + + +# Population checks. + + +def _tiny_dataset_file(tmp_path, households=5, recipients=False): + """Earners on £10,000 to £50,000, one household each. + + With ``recipients``, two more households: a pensioner above the full + State Pension and a working-age adult with no income, who claims + Universal Credit, so the responsiveness checks have awards to move. + """ + import pandas as pd + from policyengine_uk.data import UKSingleYearDataset + + ages = [40] * households + earnings = [10_000.0 * i for i in range(1, households + 1)] + pensions = [0.0] * households + if recipients: + ages += [70, 30] + earnings += [0.0, 0.0] + pensions += [20_000.0, 0.0] + households += 2 + ids = list(range(1, households + 1)) + person = pd.DataFrame( + { + "person_id": ids, + "person_benunit_id": ids, + "person_household_id": ids, + "age": ages, + "employment_income": earnings, + "state_pension_reported": pensions, + "person_weight": [1.0] * households, + } + ) + benunit = pd.DataFrame({"benunit_id": ids, "benunit_weight": [1.0] * households}) + household = pd.DataFrame( + { + "household_id": ids, + "household_weight": [1.0] * households, + "region": ["LONDON"] * households, + "tenure_type": ["OWNED_OUTRIGHT"] * households, + "council_tax": [0.0] * households, + "rent": [0.0] * households, + } + ) + path = tmp_path / "tiny.h5" + UKSingleYearDataset( + person=person, benunit=benunit, household=household, fiscal_year=2025 + ).save(str(path)) + return str(path) + + +def test_the_subsample_is_deterministic_and_keeps_whole_households(tmp_path): + from policyengine.tax_benefit_models.uk.budget_reform_population import subsample + + path = _tiny_dataset_file(tmp_path, households=5) + first = subsample(path, 3, seed=7) + again = subsample(path, 3, seed=7) + kept = sorted(first.household["household_id"]) + assert kept == sorted(again.household["household_id"]) + assert len(kept) == 3 + assert set(first.person["person_household_id"]) == set(kept) + assert set(first.benunit["benunit_id"]) == set(first.person["person_benunit_id"]) + everything = subsample(path, 50, seed=7) + assert len(everything.household) == 5 + + +def test_population_checks_hold_on_a_tiny_dataset(tmp_path): + pytest.importorskip("policyengine_uk.utils.dated_reforms") + from policyengine.tax_benefit_models.uk.budget_reform_population import ( + run_population_checks, + ) + + rows = run_population_checks( + _tiny_dataset_file(tmp_path, recipients=True), + list(contract.COUNTRY_ROUTES), + seed=0, + ) + assert rows and all(row["status"] == "pass" for row in rows), rows + + +def test_responsiveness_checks_fail_with_no_award_to_move(tmp_path): + pytest.importorskip("policyengine_uk.utils.dated_reforms") + from policyengine.tax_benefit_models.uk.budget_reform_population import ( + run_population_checks, + ) + + rows = run_population_checks( + _tiny_dataset_file(tmp_path), ["country.reform_dict"], seed=0 + ) + status = {row["check_id"]: row["status"] for row in rows} + # Earners only: no award to show a reform reaching it. + assert status["population_state_pension_plus_10"] == "fail" + assert status["population_april_rate_change"] == "pass" + + +def test_expected_annual_for_a_blended_change_that_started_earlier(): + def baseline(path, year): + return 0.18 + + cgt = "gov.hmrc.cgt.basic_rate" + earlier = Change(cgt, datetime.date(2025, 6, 1), 0.20) + assert expected_annual(earlier, 2026, baseline) == pytest.approx(0.20) + ends = Change(cgt, datetime.date(2025, 6, 1), 0.20, datetime.date(2026, 10, 29)) + # 6 April to 29 October 2026 is 207 days of 365. + assert expected_annual(ends, 2026, baseline) == pytest.approx( + (0.20 * 207 + 0.18 * 158) / 365 + ) + + +def test_the_subsample_reads_a_file_without_an_h5_extension(tmp_path): + import shutil + + from policyengine.tax_benefit_models.uk.budget_reform_population import subsample + + blob = tmp_path / "6a842565bc10" + shutil.copy(_tiny_dataset_file(tmp_path), blob) + assert len(subsample(str(blob), 3, seed=7).household) == 3 + + +def test_household_result_totals_read_each_variable_from_its_own_entity(): + from policyengine.tax_benefit_models.uk.budget_reform_contract import ( + _result_total, + ) + + # universal_credit is a benefit-unit variable; the household result can + # also carry it aggregated to the household. + result = { + "person": [{"income_tax": 100.0}, {"income_tax": 50.0}], + "benunit": {"universal_credit": 6_000.0}, + "household": {"universal_credit": 6_000.0}, + } + assert _result_total(result, "universal_credit") == 6_000.0 + assert _result_total(result, "income_tax") == 150.0 + with pytest.raises(KeyError, match="benunit"): + _result_total({"household": {"universal_credit": 1.0}}, "universal_credit")