From 38be34d68a1331656bb9c7591b34919e51ceba27 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 12 Aug 2026 17:07:48 -0400 Subject: [PATCH 1/2] Impute capital gains amounts from the published joint distribution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rebuilt as one commit on the renamed tree (microcosm.* namespace landed on main mid-flight, after the Frame carrier swap had already done the same), carrying the full reviewed history of the stage: the original rank-preserving amounts-only redraw from HMRC table 3; the Frame carrier adaptation; and two adversarial review rounds — repair of infeasible rounded means (four 2023-24 cells imply a mean outside their own band), suppressed-count cells taking the count their published gains imply at the band mean with every income column rescaled onto its published total, the bottom band's support starting one pound past the annual exempt amount, and the real published surface embedded as always-run test literals. The stage owns amounts, not incidence and not weights. Taxable income is an arithmetic proxy over persisted total_income components less a tapered Personal Allowance read from the policyengine-uk parameter tree through the uk extra's deferred import. Draws are seeded on the build period. Whether the band facts become calibration targets stays behind the reviewed fence adjudicated on #552. Co-Authored-By: Claude Fable 5 --- .../build/uk_runtime/cgt_imputation.py | 598 ++++++++++++++++++ .../tests/test_uk_cgt_imputation.py | 570 +++++++++++++++++ 2 files changed, 1168 insertions(+) create mode 100644 packages/microcosm-build/src/microcosm/build/uk_runtime/cgt_imputation.py create mode 100644 packages/microcosm-build/tests/test_uk_cgt_imputation.py diff --git a/packages/microcosm-build/src/microcosm/build/uk_runtime/cgt_imputation.py b/packages/microcosm-build/src/microcosm/build/uk_runtime/cgt_imputation.py new file mode 100644 index 00000000..04cc6faf --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/uk_runtime/cgt_imputation.py @@ -0,0 +1,598 @@ +"""Impute capital gains amounts from HMRC's published joint distribution. + +The certified candidate carries a ``capital_gains`` column whose amounts were +drawn from a percentile table that stops near £1m, so the population holds +almost none of the gains above £2m that carry most of the tax. This stage +redraws the amounts of existing gainers from HMRC table 3 — the published +joint distribution of size of gain by taxable income — which reaches the top +bands. See PolicyEngine/microcosm#552 for the measured gap. + +The stage owns amounts, not incidence and not weights. Who has gains comes +from the candidate; how much they have comes from the published distribution; +household weights pass through untouched, so the weight kind and mass log +carry through unchanged. Whether the published band facts also become calibration targets is a +separate adjudication (see the fence discussion on the issue), and nothing +here assumes it. + +Three documented approximations, in order of consequence: + +1. **Taxable income is an arithmetic proxy.** HMRC's rows condition on + taxable income after reliefs and the Personal Allowance. The population + carries raw inputs only, so the proxy sums the persisted components of the + model's ``total_income`` concept (ITA 2007 s.23) and subtracts a tapered + Personal Allowance, with the allowance parameters read from the + policyengine-uk parameter tree rather than maintained here. + ``state_pension_reported`` stands in for ``social_security_income``, whose + other taxable benefits are not persisted; reliefs such as pension + contributions and Gift Aid are not deducted. Both push the proxy up or + down at the margins, which can move a person one income band. +2. **Allocation is rank-preserving within income band.** Within each income + band, gainers are ranked by their existing gains and the top of the + ranking absorbs the published taxpayer mass, highest gain band first. A + person is not split across bands, so band mass is matched to the + granularity of one household weight. +3. **Rounded published values are repaired, not trusted raw.** Counts round + to the nearest thousand and amounts to the nearest million, and four + cells of the 2023-24 table imply a mean outside their own band. Implied + means are clamped just inside the violated boundary, keeping the signal + that the cell's mass sits near that edge; suppressed-count cells are + allocated the count their own published gains imply at the band mean, + and every income column is rescaled onto its published total, so the + allocation reconciles to published numbers by construction. + The bottom band's support is floored at the annual exempt amount, since + every allocated person is a taxpayer with a liability. +4. **Gainers beyond the published taxpayer mass keep their existing amounts, + capped at the annual exempt amount.** Table 3 covers only individuals + with a CGT liability, so the candidate's remaining gainers are treated as + sub-AEA gainers rather than being invented into the liability + distribution or deleted. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +import numpy as np +import pandas as pd + +from microcosm.build.uk_runtime.hmrc_capital_gains import ( + HMRC_CGT_GAIN_BAND_LOWER_BOUNDS, + HMRC_CGT_INCOME_BAND_LOWER_BOUNDS, + HMRC_CGT_SOURCE_VINTAGE, + HMRCCapitalGainsJointDistribution, + materialize_hmrc_capital_gains_joint_distribution, +) +from microcosm.build.uk_runtime.national_build import UKNationalStage +from microcosm.build.uk_runtime.national_frame import ( + uk_household_weight_kind, + uk_national_frame, + uk_time_period, + validate_uk_national_frame, +) +from microcosm.frame import Frame + +__all__ = [ + "UK_CGT_IMPUTATION_SEED", + "UK_CGT_IMPUTATION_STAGE_NAME", + "UK_CGT_TAXABLE_INCOME_PROXY_COMPONENTS", + "UKCGTImputationSummary", + "UKCGTPolicyParameters", + "impute_uk_capital_gains", + "summarize_uk_cgt_imputation", + "uk_capital_gains_imputation_stage", + "uk_cgt_policy_parameters", + "uk_cgt_taxable_income_proxy", +] + +UK_CGT_IMPUTATION_STAGE_NAME = "hmrc_cgt_gains" + +#: Base seed for the stage's draws. Combined with the build period so two +#: periods draw differently while each build is reproducible. +UK_CGT_IMPUTATION_SEED = 552 + +#: Persisted components of the model's ``total_income`` concept (ITA 2007 +#: s.23: taxable income after tax reliefs and before allowances). +#: ``state_pension_reported`` stands in for ``social_security_income``; the +#: other taxable benefits inside that concept are not persisted inputs. +UK_CGT_TAXABLE_INCOME_PROXY_COMPONENTS: tuple[str, ...] = ( + "employment_income", + "self_employment_income", + "state_pension_reported", + "private_pension_income", + "property_income", + "savings_interest_income", + "dividend_income", + "miscellaneous_income", +) + +#: A cell mean must sit strictly inside its band for a within-band +#: distribution to match it. Published values round to the nearest thousand +#: people and million pounds, and a rounded count of 1 against a rounded +#: amount can imply a mean outside the band — four cells of the 2023-24 +#: table do exactly that — so implied means are repaired into the band by +#: this margin (a fraction of band width) rather than trusted raw. +_MEAN_MARGIN = 0.02 + +#: Below this many people of published mass, a repaired cell is allocated +#: nothing rather than a distribution being fitted to noise. +_MINIMUM_ALLOCATION_PEOPLE = 1.0 + +#: The draw families require a mean strictly inside the support. Plans are +#: repaired before they reach the solvers, so a violation here signals a +#: caller passing an unrepaired mean, not published data. +_MEAN_POSITION_TOLERANCE = 1e-9 + + +@dataclass(frozen=True) +class UKCGTPolicyParameters: + """Policy amounts the proxy and the sub-AEA cap depend on. + + Read from the policyengine-uk parameter tree at a stated instant by + :func:`uk_cgt_policy_parameters`, or constructed directly in tests. + """ + + personal_allowance: float + personal_allowance_taper_threshold: float + personal_allowance_taper_rate: float + annual_exempt_amount: float + instant: str + source: str + + +def uk_cgt_policy_parameters(build_period: int | str) -> UKCGTPolicyParameters: + """Read the allowance parameters from the policyengine-uk tree. + + Values are read from the raw dated parameter files at 1 June of the tax + year starting in ``build_period``, so they are the statutory values with + their legislative references, not the model's fiscal-year snapshots. + + Requires the ``uk`` extra; the import is deferred so the base package + does not import policyengine-uk at import time. + """ + try: + import policyengine_uk + from policyengine_core.parameters import ParameterNode + except ImportError as exc: + raise ImportError( + "uk_cgt_policy_parameters requires the microcosm-build 'uk' extra " + "(policyengine-uk)." + ) from exc + + parameters_dir = Path(policyengine_uk.__file__).parent / "parameters" + parameters = ParameterNode(directory_path=str(parameters_dir)) + instant = f"{int(build_period)}-06-01" + allowances = parameters.gov.hmrc.income_tax.allowances + return UKCGTPolicyParameters( + personal_allowance=float(allowances.personal_allowance.amount(instant)), + personal_allowance_taper_threshold=float( + allowances.personal_allowance.maximum_ANI(instant) + ), + personal_allowance_taper_rate=float( + allowances.personal_allowance.reduction_rate(instant) + ), + annual_exempt_amount=float( + parameters.gov.hmrc.cgt.annual_exempt_amount(instant) + ), + instant=instant, + source="policyengine-uk parameters " + f"{getattr(policyengine_uk, '__version__', 'unknown')}", + ) + + +def uk_cgt_taxable_income_proxy( + person: pd.DataFrame, parameters: UKCGTPolicyParameters +) -> np.ndarray: + """Approximate taxable income after reliefs and the Personal Allowance.""" + missing = [ + column + for column in UK_CGT_TAXABLE_INCOME_PROXY_COMPONENTS + if column not in person.columns + ] + if missing: + raise ValueError( + f"Taxable income proxy components missing from person table: {missing}." + ) + total = np.zeros(len(person)) + for column in UK_CGT_TAXABLE_INCOME_PROXY_COMPONENTS: + total += pd.to_numeric(person[column], errors="raise").to_numpy(dtype=float) + + taper = parameters.personal_allowance_taper_rate * np.maximum( + 0.0, total - parameters.personal_allowance_taper_threshold + ) + allowance = np.maximum(0.0, parameters.personal_allowance - taper) + return np.maximum(0.0, total - allowance) + + +def _truncated_exponential_quantile( + quantiles: np.ndarray, lower: float, upper: float, mean: float +) -> np.ndarray: + """Map uniform quantiles onto [lower, upper) with the given mean. + + A truncated exponential is the one-parameter family on a bounded band + whose mean can sit anywhere inside it; the rate is solved by bisection, + and a mean at the midpoint degenerates to the uniform distribution. + """ + width = upper - lower + target = (mean - lower) / width + if not _MEAN_POSITION_TOLERANCE < target < 1 - _MEAN_POSITION_TOLERANCE: + raise ValueError(f"Band mean {mean} does not sit inside ({lower}, {upper}).") + if abs(target - 0.5) < 1e-9: + return lower + quantiles * width + + def normalized_mean(rate: float) -> float: + # Mean of x ~ TruncExp(rate) on [0, 1]: 1/rate - 1/(exp(rate) - 1), + # with a removable singularity at zero where the mean is one half. + if abs(rate) < 1e-9: + return 0.5 + return 1.0 / rate - 1.0 / np.expm1(rate) + + # normalized_mean is decreasing: rate -> +inf piles mass at 0 and + # rate -> -inf piles it at 1. + low_rate, high_rate = -700.0, 700.0 + for _ in range(200): + mid_rate = (low_rate + high_rate) / 2.0 + if normalized_mean(mid_rate) > target: + low_rate = mid_rate + else: + high_rate = mid_rate + rate = (low_rate + high_rate) / 2.0 + # Inverse CDF of the truncated exponential on [0, 1]. + positions = -np.log1p(quantiles * (np.exp(-rate) - 1.0)) / rate + return lower + np.clip(positions, 0.0, 1.0) * width + + +def _pareto_quantile(quantiles: np.ndarray, lower: float, mean: float) -> np.ndarray: + """Map uniform quantiles onto [lower, inf) with the given mean. + + The open top band takes a Pareto whose shape is pinned by the published + mean: alpha = mean / (mean - lower). + """ + if mean <= lower: + raise ValueError( + f"Open-band mean {mean} must exceed the band lower bound {lower}." + ) + alpha = mean / (mean - lower) + return lower * np.power(1.0 - quantiles, -1.0 / alpha) + + +@dataclass(frozen=True) +class _CellPlan: + """Allocation and draw parameters for one gain band within an income band.""" + + gain_lower_bound: int + gain_upper_bound: float + effective_lower_bound: float + allocation_people: float + mean: float + mean_is_band_fallback: bool + mean_repaired: bool + + +def _repair_mean( + candidate: float, *, effective_lower: float, upper: float +) -> tuple[float, bool]: + """Force an implied mean strictly inside its band. + + Rounded published values can imply a mean outside the band (a rounded + count of 1,000 against a rounded £463m gives £463,000 for the £500k-£1m + band). The repair clamps toward the violated boundary, keeping the + signal that the cell's mass sits near that edge. + """ + if np.isinf(upper): + floor = effective_lower * (1.0 + _MEAN_MARGIN) + if candidate < floor: + return floor, True + return candidate, False + width = upper - effective_lower + low = effective_lower + _MEAN_MARGIN * width + high = upper - _MEAN_MARGIN * width + if candidate < low: + return low, True + if candidate > high: + return high, True + return candidate, False + + +def _band_plans( + distribution: HMRCCapitalGainsJointDistribution, + income_lower_bound: int, + *, + annual_exempt_amount: float, +) -> tuple[_CellPlan, ...]: + """Build the per-gain-band plan for one income band, top band first. + + Suppressed-count cells split the residual between the published income + column total and its unsuppressed cells, so the allocation reconciles to + published totals instead of assuming a count. The bottom band's support + is floored at the annual exempt amount where that fits inside the band, + since every allocated person is a taxpayer with a liability. + """ + bounds = HMRC_CGT_GAIN_BAND_LOWER_BOUNDS + uppers = dict(zip(bounds, (*bounds[1:], np.inf), strict=True)) + + cells = { + gain_lower_bound: distribution.cell( + gain_lower_bound=gain_lower_bound, income_lower_bound=income_lower_bound + ) + for gain_lower_bound in bounds + } + + def band_mean(gain_lower_bound: int) -> float: + band_total = distribution.band_total(gain_lower_bound) + if band_total.individuals is None: + raise ValueError( + "Band total count is suppressed for gains from " + f"{gain_lower_bound}, leaving no defined mean." + ) + return band_total.gains / band_total.individuals + + # Raw allocation: published counts where published; for suppressed cells, + # the count the cell's own published gains imply at the band mean — a + # cell holding £1.3bn of gains at a £11.4m band mean holds people, and + # a bare residual can zero it when rounding makes the column's + # unsuppressed counts alone reach the column total. + raw_allocation: dict[int, float] = {} + for gain_lower_bound in bounds: + cell = cells[gain_lower_bound] + if cell.individuals is not None: + raw_allocation[gain_lower_bound] = cell.individuals + elif cell.gains is not None: + raw_allocation[gain_lower_bound] = cell.gains / band_mean(gain_lower_bound) + else: + raw_allocation[gain_lower_bound] = 0.0 + + # Published counts round to the nearest thousand, so a column's cells can + # sum past its own published total. Rescale the column onto the published + # total, which makes the allocation reconcile exactly by construction. + column = distribution.income_total(income_lower_bound) + if column.individuals is None: + raise ValueError( + "Income column total is suppressed for incomes from " + f"{income_lower_bound}, leaving the allocation nothing to " + "reconcile against." + ) + raw_total = sum(raw_allocation.values()) + scale = column.individuals / raw_total if raw_total > 0 else 0.0 + + plans = [] + for gain_lower_bound in reversed(bounds): + cell = cells[gain_lower_bound] + upper = uppers[gain_lower_bound] + effective_lower = float(max(gain_lower_bound, 1.0)) + # Every allocated person is a taxpayer with a liability, and + # liability requires gains strictly above the annual exempt amount, + # so the bottom band's support starts one pound past it. + liability_floor = annual_exempt_amount + 1.0 + if liability_floor > effective_lower and ( + np.isinf(upper) or liability_floor < 0.9 * upper + ): + effective_lower = float(liability_floor) + + if cell.individuals is not None and cell.gains is not None: + candidate = cell.gains / cell.individuals + else: + candidate = band_mean(gain_lower_bound) + mean, repaired = _repair_mean( + candidate, effective_lower=effective_lower, upper=upper + ) + + allocation = raw_allocation[gain_lower_bound] * scale + if allocation < _MINIMUM_ALLOCATION_PEOPLE: + allocation = 0.0 + plans.append( + _CellPlan( + gain_lower_bound=gain_lower_bound, + gain_upper_bound=upper, + effective_lower_bound=effective_lower, + allocation_people=allocation, + mean=mean, + mean_is_band_fallback=False, + mean_repaired=repaired, + ) + ) + return tuple(plans) + + +def _draw_amounts(plan: _CellPlan, quantiles: np.ndarray) -> np.ndarray: + lower = plan.effective_lower_bound + if np.isinf(plan.gain_upper_bound): + return _pareto_quantile(quantiles, lower, plan.mean) + return _truncated_exponential_quantile( + quantiles, lower, float(plan.gain_upper_bound), plan.mean + ) + + +@dataclass(frozen=True) +class UKCGTImputationSummary: + """Achieved allocation against the published surface, for reporting.""" + + rows: pd.DataFrame + taxpayer_mass: float + published_taxpayer_mass: float + remainder_mass: float + + +def impute_uk_capital_gains( + frame: Frame, + distribution: HMRCCapitalGainsJointDistribution, + parameters: UKCGTPolicyParameters, + *, + seed: int = UK_CGT_IMPUTATION_SEED, +) -> Frame: + """Redraw gainers' amounts from the published joint distribution.""" + validate_uk_national_frame(frame) + time_period = uk_time_period(frame) + person = frame.table("person").reset_index(drop=True) + if "capital_gains" not in person.columns: + raise ValueError("Person table has no capital_gains column to redraw.") + + weights_by_household = frame.table("household").set_index("household_id")[ + "household_weight" + ] + missing_households = set(person["person_household_id"]) - set( + weights_by_household.index + ) + if missing_households: + raise ValueError( + "Person rows reference households with no weight: " + f"{sorted(missing_households)[:5]}." + ) + person_weight = ( + person["person_household_id"].map(weights_by_household).to_numpy(dtype=float) + ) + + existing = pd.to_numeric(person["capital_gains"], errors="raise").to_numpy( + dtype=float + ) + taxable_income = uk_cgt_taxable_income_proxy(person, parameters) + income_band = np.asarray(HMRC_CGT_INCOME_BAND_LOWER_BOUNDS)[ + np.digitize(taxable_income, HMRC_CGT_INCOME_BAND_LOWER_BOUNDS[1:]) + ] + + rng = np.random.default_rng((seed, int(time_period))) + new_gains = existing.copy() + is_gainer = existing > 0 + + for income_lower_bound in HMRC_CGT_INCOME_BAND_LOWER_BOUNDS: + in_band = is_gainer & (income_band == income_lower_bound) + if not in_band.any(): + continue + indices = np.flatnonzero(in_band) + # Rank by existing gains, largest first; person_id breaks ties so + # the ordering, and with it every draw, is deterministic. + order = np.lexsort( + (person["person_id"].to_numpy()[indices], -existing[indices]) + ) + ranked = indices[order] + band_weights = person_weight[ranked] + cumulative = np.cumsum(band_weights) + + plans = _band_plans( + distribution, + income_lower_bound, + annual_exempt_amount=parameters.annual_exempt_amount, + ) + published_mass = sum(plan.allocation_people for plan in plans) + available_mass = float(cumulative[-1]) + if published_mass <= 0.0: + # No published taxpayer mass in this income band: every gainer + # here is a sub-AEA remainder. + new_gains[ranked] = np.minimum( + existing[ranked], parameters.annual_exempt_amount + ) + continue + # When the population holds less gainer mass than HMRC's taxpayers, + # allocate what exists in the published proportions rather than + # exhausting the top bands and emptying the bottom ones. + scale = min(1.0, available_mass / published_mass) + + boundary = 0.0 + assigned = np.zeros(len(ranked), dtype=bool) + for plan in plans: + boundary += plan.allocation_people * scale + in_cell = ~assigned & (cumulative - band_weights / 2.0 <= boundary) + count = int(in_cell.sum()) + if count == 0: + continue + assigned |= in_cell + quantiles = rng.random(count) + new_gains[ranked[in_cell]] = _draw_amounts(plan, quantiles) + + # Below the published taxpayer mass: sub-AEA gainers keep their + # existing amounts, capped at the annual exempt amount. + remainder = ranked[~assigned] + new_gains[remainder] = np.minimum( + existing[remainder], parameters.annual_exempt_amount + ) + + if not np.isfinite(new_gains).all(): + raise ValueError("Imputed capital gains contain non-finite values.") + if (new_gains < 0).any(): + raise ValueError("Imputed capital gains contain negative values.") + + new_person = person.copy() + new_person["capital_gains"] = new_gains + # Person-only replacement: mass is untouched, so the kind and mass log + # carry through unchanged; Frame construction re-runs linkage validation. + result_frame = uk_national_frame( + person=new_person, + benunit=frame.table("benunit"), + household=frame.table("household"), + time_period=time_period, + weight_kind=uk_household_weight_kind(frame), + mass_log=frame.mass_log, + ) + validate_uk_national_frame(result_frame) + return result_frame + + +def summarize_uk_cgt_imputation( + before: Frame, + after: Frame, + distribution: HMRCCapitalGainsJointDistribution, + parameters: UKCGTPolicyParameters, +) -> UKCGTImputationSummary: + """Compare achieved band totals with the published surface. + + Reporting, not a gate: where the population holds less gainer mass than + HMRC's taxpayers the achieved totals sit below the published ones by + construction, and holding levels to the published surface is the + calibration adjudication's question. + """ + person = after.table("person").reset_index(drop=True) + weights_by_household = after.table("household").set_index("household_id")[ + "household_weight" + ] + weight = ( + person["person_household_id"].map(weights_by_household).to_numpy(dtype=float) + ) + gains = pd.to_numeric(person["capital_gains"], errors="raise").to_numpy(dtype=float) + liable = gains > parameters.annual_exempt_amount + + bounds = HMRC_CGT_GAIN_BAND_LOWER_BOUNDS + uppers = (*bounds[1:], np.inf) + rows = [] + for lower, upper in zip(bounds, uppers, strict=True): + in_band = liable & (gains >= lower) & (gains < upper) + band_total = distribution.band_total(lower) + rows.append( + { + "gain_lower_bound": lower, + "achieved_people": float(weight[in_band].sum()), + "published_people": band_total.individuals, + "achieved_gains": float((gains[in_band] * weight[in_band]).sum()), + "published_gains": band_total.gains, + } + ) + return UKCGTImputationSummary( + rows=pd.DataFrame(rows), + taxpayer_mass=float(weight[liable].sum()), + published_taxpayer_mass=float(distribution.total_individuals), + remainder_mass=float(weight[(gains > 0) & ~liable].sum()), + ) + + +def uk_capital_gains_imputation_stage( + ods_path: str | Path, + *, + tax_year: str = HMRC_CGT_SOURCE_VINTAGE, + parameters: UKCGTPolicyParameters | None = None, + seed: int = UK_CGT_IMPUTATION_SEED, +) -> UKNationalStage: + """Build the national stage that redraws capital gains amounts. + + The published artifact is verified against its pinned fingerprint before + it is read. Parameters default to the policyengine-uk tree at the + dataset's build period, resolved when the stage runs. + """ + artifact_path = Path(ods_path) + + def transform(frame: Frame) -> Frame: + distribution = materialize_hmrc_capital_gains_joint_distribution( + artifact_path, tax_year=tax_year + ) + resolved = parameters or uk_cgt_policy_parameters(uk_time_period(frame)) + return impute_uk_capital_gains(frame, distribution, resolved, seed=seed) + + return UKNationalStage(name=UK_CGT_IMPUTATION_STAGE_NAME, transform=transform) diff --git a/packages/microcosm-build/tests/test_uk_cgt_imputation.py b/packages/microcosm-build/tests/test_uk_cgt_imputation.py new file mode 100644 index 00000000..67f0c0ae --- /dev/null +++ b/packages/microcosm-build/tests/test_uk_cgt_imputation.py @@ -0,0 +1,570 @@ +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.uk_runtime.cgt_imputation import ( + UK_CGT_IMPUTATION_STAGE_NAME, + UK_CGT_TAXABLE_INCOME_PROXY_COMPONENTS, + UKCGTPolicyParameters, + _band_plans, + _pareto_quantile, + _truncated_exponential_quantile, + impute_uk_capital_gains, + summarize_uk_cgt_imputation, + uk_capital_gains_imputation_stage, + uk_cgt_taxable_income_proxy, +) +from microcosm.build.uk_runtime.hmrc_capital_gains import ( + HMRC_CGT_GAIN_BAND_LOWER_BOUNDS, + HMRC_CGT_INCOME_BAND_LOWER_BOUNDS, + HMRCCapitalGainsBandTotal, + HMRCCapitalGainsCell, + HMRCCapitalGainsIncomeTotal, + HMRCCapitalGainsJointDistribution, + HMRCCapitalGainsSourceProvenance, +) +from microcosm.build.uk_runtime.national_frame import uk_national_frame +from microcosm.frame import Frame + +PARAMETERS = UKCGTPolicyParameters( + personal_allowance=12_570.0, + personal_allowance_taper_threshold=100_000.0, + personal_allowance_taper_rate=0.5, + annual_exempt_amount=6_000.0, + instant="2023-06-01", + source="test", +) + + +def _distribution( + *, cell_people: float = 1_000.0, suppress_top_low_income: bool = False +) -> HMRCCapitalGainsJointDistribution: + """A synthetic joint distribution with in-band cell means.""" + bounds = HMRC_CGT_GAIN_BAND_LOWER_BOUNDS + uppers = (*bounds[1:], None) + cells = [] + band_totals = [] + column_people = dict.fromkeys(HMRC_CGT_INCOME_BAND_LOWER_BOUNDS, 0.0) + column_gains = dict.fromkeys(HMRC_CGT_INCOME_BAND_LOWER_BOUNDS, 0.0) + for lower, upper in zip(bounds, uppers, strict=True): + if upper is None: + mean = lower * 2.0 + else: + mean = lower + (upper - lower) * 0.4 + band_people = 0.0 + band_gains = 0.0 + for income_lower in HMRC_CGT_INCOME_BAND_LOWER_BOUNDS: + suppressed = ( + suppress_top_low_income + and upper is None + and income_lower == HMRC_CGT_INCOME_BAND_LOWER_BOUNDS[0] + ) + # Real suppression withholds the count and keeps the amount in + # all but one published cell, so the fixture matches that shape. + people = None if suppressed else cell_people + gains = cell_people * mean + cells.append( + HMRCCapitalGainsCell( + gain_lower_bound=lower, + income_lower_bound=income_lower, + individuals=people, + gains=gains, + ) + ) + band_people += cell_people + band_gains += cell_people * mean + column_people[income_lower] += cell_people + column_gains[income_lower] += cell_people * mean + band_totals.append( + HMRCCapitalGainsBandTotal( + gain_lower_bound=lower, + individuals=band_people, + gains=band_gains, + ) + ) + income_totals = tuple( + HMRCCapitalGainsIncomeTotal( + income_lower_bound=income_lower, + individuals=column_people[income_lower], + gains=column_gains[income_lower], + ) + for income_lower in HMRC_CGT_INCOME_BAND_LOWER_BOUNDS + ) + return HMRCCapitalGainsJointDistribution( + cells=tuple(cells), + band_totals=tuple(band_totals), + income_totals=income_totals, + source=HMRCCapitalGainsSourceProvenance( + local_path=None, + sha256="synthetic", + size_bytes=0, + sheet_name="synthetic", + source_vintage="2023-24", + build_period="2023", + ), + total_individuals=sum(t.individuals for t in band_totals), + total_gains=sum(t.gains for t in band_totals), + ) + + +def _frame(person_rows: int, *, gains, incomes) -> Frame: + person = pd.DataFrame( + { + "person_id": np.arange(person_rows, dtype="int64"), + "person_household_id": np.arange(person_rows, dtype="int64"), + "person_benunit_id": np.arange(person_rows, dtype="int64"), + "capital_gains": np.asarray(gains, dtype=float), + "employment_income": np.asarray(incomes, dtype=float), + } + ) + for column in UK_CGT_TAXABLE_INCOME_PROXY_COMPONENTS: + if column not in person.columns: + person[column] = 0.0 + household = pd.DataFrame( + { + "household_id": np.arange(person_rows, dtype="int64"), + "household_weight": np.full(person_rows, 100.0), + } + ) + benunit = pd.DataFrame({"benunit_id": np.arange(person_rows, dtype="int64")}) + return uk_national_frame( + person=person, + benunit=benunit, + household=household, + time_period="2023", + ) + + +class TestTaxableIncomeProxy: + def test_subtracts_the_personal_allowance(self) -> None: + person = pd.DataFrame( + {column: [0.0] for column in UK_CGT_TAXABLE_INCOME_PROXY_COMPONENTS} + ) + person["employment_income"] = [30_000.0] + + proxy = uk_cgt_taxable_income_proxy(person, PARAMETERS) + + assert proxy[0] == pytest.approx(30_000.0 - 12_570.0) + + def test_tapers_the_allowance_above_the_threshold(self) -> None: + person = pd.DataFrame( + {column: [0.0, 0.0] for column in UK_CGT_TAXABLE_INCOME_PROXY_COMPONENTS} + ) + # £120,000 tapers the allowance to £2,570; £130,000 removes it. + person["employment_income"] = [120_000.0, 130_000.0] + + proxy = uk_cgt_taxable_income_proxy(person, PARAMETERS) + + assert proxy[0] == pytest.approx(120_000.0 - 2_570.0) + assert proxy[1] == pytest.approx(130_000.0) + + def test_rejects_a_missing_component(self) -> None: + person = pd.DataFrame({"employment_income": [1.0]}) + + with pytest.raises(ValueError, match="components missing"): + uk_cgt_taxable_income_proxy(person, PARAMETERS) + + +class TestWithinBandDraws: + def test_truncated_exponential_matches_the_target_mean(self) -> None: + rng = np.random.default_rng(0) + quantiles = rng.random(200_000) + + for target in (30_000.0, 50_000.0, 70_000.0): + draws = _truncated_exponential_quantile( + quantiles, 25_000.0, 100_000.0, target + ) + assert draws.min() >= 25_000.0 + assert draws.max() <= 100_000.0 + assert draws.mean() == pytest.approx(target, rel=5e-3) + + def test_midpoint_mean_degenerates_to_uniform(self) -> None: + quantiles = np.asarray([0.0, 0.25, 0.5, 1.0]) + + draws = _truncated_exponential_quantile(quantiles, 0.0, 100.0, 50.0) + + assert draws == pytest.approx([0.0, 25.0, 50.0, 100.0]) + + def test_pareto_matches_the_published_mean(self) -> None: + rng = np.random.default_rng(0) + quantiles = rng.random(500_000) + + draws = _pareto_quantile(quantiles, 5_000_000.0, 11_357_000.0) + + assert draws.min() >= 5_000_000.0 + # alpha ~ 1.79 has a heavy tail; the sample mean converges slowly, + # so hold it loosely and the median (analytic) tightly. + assert draws.mean() == pytest.approx(11_357_000.0, rel=0.15) + alpha = 11_357_000.0 / (11_357_000.0 - 5_000_000.0) + assert np.median(draws) == pytest.approx( + 5_000_000.0 * 2.0 ** (1.0 / alpha), rel=5e-3 + ) + + def test_rejects_a_mean_outside_the_band(self) -> None: + with pytest.raises(ValueError, match="does not sit inside"): + _truncated_exponential_quantile(np.asarray([0.5]), 0.0, 10.0, 20.0) + with pytest.raises(ValueError, match="must exceed"): + _pareto_quantile(np.asarray([0.5]), 5_000_000.0, 4_000_000.0) + + +class TestImputation: + def test_reaches_the_top_band_and_is_deterministic(self) -> None: + distribution = _distribution() + rows = 2_000 + rng = np.random.default_rng(1) + gains = np.where(rng.random(rows) < 0.5, rng.lognormal(10, 1, rows), 0.0) + incomes = rng.lognormal(10.5, 0.8, rows) + frame = _frame(rows, gains=gains, incomes=incomes) + + first = impute_uk_capital_gains(frame, distribution, PARAMETERS) + second = impute_uk_capital_gains(frame, distribution, PARAMETERS) + + drawn = first.table("person")["capital_gains"].to_numpy() + assert (drawn == second.table("person")["capital_gains"].to_numpy()).all() + assert drawn.max() >= 5_000_000.0, "no draw reached the open top band" + + def test_non_gainers_stay_at_zero_and_weights_pass_through(self) -> None: + distribution = _distribution() + frame = _frame( + 6, + gains=[0.0, 0.0, 50_000.0, 20_000.0, 10_000.0, 5_000.0], + incomes=[10_000.0] * 6, + ) + + result = impute_uk_capital_gains(frame, distribution, PARAMETERS) + + drawn = result.table("person")["capital_gains"].to_numpy() + assert (drawn[:2] == 0.0).all() + assert (drawn[2:] > 0.0).all() + pd.testing.assert_frame_equal( + result.table("household"), frame.table("household") + ) + assert result.mass_log == frame.mass_log + + def test_remainder_keeps_existing_amounts_capped_at_the_aea(self) -> None: + # One income band holds far more gainer mass than the published + # taxpayers, so the ranking's tail lands in the sub-AEA remainder. + distribution = _distribution(cell_people=10.0) + rows = 3_000 + rng = np.random.default_rng(2) + gains = rng.lognormal(9, 1.5, rows) + frame = _frame(rows, gains=gains, incomes=np.full(rows, 20_000.0)) + + result = impute_uk_capital_gains(frame, distribution, PARAMETERS) + + drawn = result.table("person")["capital_gains"].to_numpy() + remainder = drawn <= PARAMETERS.annual_exempt_amount + assert remainder.any() + expected = np.minimum(gains, PARAMETERS.annual_exempt_amount) + assert drawn[remainder] == pytest.approx(expected[remainder]) + + def test_allocation_preserves_the_existing_ranking(self) -> None: + # Everyone sits in one income band, whose cells hold 100 people each + # against person weights of 100 — so the £5m+ band holds exactly one + # person of mass, and it must be the largest existing gainer. + distribution = _distribution(cell_people=100.0) + rows = 4_000 + rng = np.random.default_rng(3) + gains = rng.lognormal(9, 2, rows) + frame = _frame(rows, gains=gains, incomes=np.full(rows, 20_000.0)) + + result = impute_uk_capital_gains(frame, distribution, PARAMETERS) + drawn = result.table("person")["capital_gains"].to_numpy() + + ranked = np.argsort(-gains) + assert drawn[ranked[0]] >= 5_000_000.0 + # Walking down the existing ranking, assigned gain bands never rise. + taxpayer_mass = int( + sum(100.0 for _ in HMRC_CGT_GAIN_BAND_LOWER_BOUNDS) + ) # people of mass in the band, at weight 100 each -> 10 persons + assigned = drawn[ranked[: taxpayer_mass // 100]] + bands = np.digitize(assigned, HMRC_CGT_GAIN_BAND_LOWER_BOUNDS) + assert (np.diff(bands) <= 0).all() + + def test_suppressed_cell_falls_back_to_the_band_mean(self) -> None: + distribution = _distribution(suppress_top_low_income=True) + rows = 4_000 + rng = np.random.default_rng(4) + gains = rng.lognormal(9, 2, rows) + frame = _frame(rows, gains=gains, incomes=np.full(rows, 20_000.0)) + + result = impute_uk_capital_gains(frame, distribution, PARAMETERS) + + drawn = result.table("person")["capital_gains"].to_numpy() + assert drawn.max() >= 5_000_000.0 + + def test_summary_reports_achieved_against_published(self) -> None: + distribution = _distribution() + rows = 2_000 + rng = np.random.default_rng(5) + gains = np.where(rng.random(rows) < 0.5, rng.lognormal(10, 1, rows), 0.0) + frame = _frame(rows, gains=gains, incomes=np.full(rows, 20_000.0)) + + result = impute_uk_capital_gains(frame, distribution, PARAMETERS) + summary = summarize_uk_cgt_imputation(frame, result, distribution, PARAMETERS) + + assert len(summary.rows) == len(HMRC_CGT_GAIN_BAND_LOWER_BOUNDS) + assert summary.published_taxpayer_mass == distribution.total_individuals + assert summary.taxpayer_mass > 0 + assert (summary.rows["published_gains"] > 0).all() + + +class TestStage: + def test_stage_runs_end_to_end_on_the_pinned_artifact(self) -> None: + """The factory's own transform path, not just its failure branch. + + Regression test for the transform keeping a retired carrier type in + its signature: with postponed annotation evaluation, only running the + stage exercises the closure. + """ + from pathlib import Path + + from microcosm.build.uk_runtime.hmrc_capital_gains import ( + HMRC_CGT_JOINT_ODS_FILENAME, + ) + + repo_root = Path(__file__).resolve().parents[3] + pinned_ods = repo_root / "inputs" / "hmrc" / HMRC_CGT_JOINT_ODS_FILENAME + if not pinned_ods.is_file(): + pytest.skip("reviewed HMRC capital gains ODS is an optional local input") + stage = uk_capital_gains_imputation_stage(pinned_ods, parameters=PARAMETERS) + incomes = [20_000.0, 45_000.0, 80_000.0, 120_000.0, 180_000.0, 400_000.0] + frame = _frame( + 60, + gains=[float(5_000 * (i + 1)) for i in range(60)], + incomes=[incomes[i % 6] for i in range(60)], + ) + + result = stage.run(frame) + + drawn = result.table("person")["capital_gains"].to_numpy() + assert (drawn >= 0).all() + assert drawn.max() > 0 + + def test_stage_carries_the_reviewed_name(self) -> None: + stage = uk_capital_gains_imputation_stage("unused.ods", parameters=PARAMETERS) + + assert stage.name == UK_CGT_IMPUTATION_STAGE_NAME + + def test_stage_verifies_the_artifact_before_reading(self, tmp_path) -> None: + wrong = tmp_path / "wrong.ods" + wrong.write_bytes(b"not the pinned artifact") + stage = uk_capital_gains_imputation_stage(wrong, parameters=PARAMETERS) + frame = _frame(1, gains=[10_000.0], incomes=[20_000.0]) + + with pytest.raises(ValueError, match="bytes, not the pinned"): + stage.run(frame) + + +REAL_2023_24 = { + # (gain band lower, income band lower): (thousands of people, £m of gains), + # None for a suppressed count. HMRC Capital Gains Tax statistics + # (24 July 2025), table 3.1 — public published values, embedded so CI + # exercises the real surface without the artifact. Four cells imply a + # mean outside their band through rounding: (500000, 37700), + # (2000000, 50000), (1000000, 100000), (250000, 150000). + (0, 0): (33, 247), + (0, 37700): (6, 42), + (0, 50000): (11, 79), + (0, 100000): (4, 32), + (0, 150000): (2, 18), + (0, 200000): (6, 44), + (10000, 0): (62, 1027), + (10000, 37700): (8, 136), + (10000, 50000): (15, 244), + (10000, 100000): (5, 85), + (10000, 150000): (3, 47), + (10000, 200000): (8, 131), + (25000, 0): (48, 1706), + (25000, 37700): (6, 199), + (25000, 50000): (10, 349), + (25000, 100000): (3, 126), + (25000, 150000): (2, 63), + (25000, 200000): (6, 202), + (50000, 0): (32, 2246), + (50000, 37700): (4, 288), + (50000, 50000): (8, 556), + (50000, 100000): (3, 208), + (50000, 150000): (2, 110), + (50000, 200000): (5, 323), + (100000, 0): (19, 2837), + (100000, 37700): (3, 463), + (100000, 50000): (6, 961), + (100000, 100000): (3, 407), + (100000, 150000): (1, 227), + (100000, 200000): (5, 755), + (250000, 0): (5, 1867), + (250000, 37700): (1, 394), + (250000, 50000): (3, 907), + (250000, 100000): (1, 423), + (250000, 150000): (1, 237), + (250000, 200000): (3, 938), + (500000, 0): (3, 1829), + (500000, 37700): (1, 463), + (500000, 50000): (2, 1105), + (500000, 100000): (1, 569), + (500000, 150000): (None, 334), + (500000, 200000): (2, 1405), + (1000000, 0): (1, 1494), + (1000000, 37700): (None, 436), + (1000000, 50000): (1, 1340), + (1000000, 100000): (1, 709), + (1000000, 150000): (None, 481), + (1000000, 200000): (1, 1929), + (2000000, 0): (None, 1424), + (2000000, 37700): (None, 471), + (2000000, 50000): (1, 1582), + (2000000, 100000): (None, 1149), + (2000000, 150000): (None, 815), + (2000000, 200000): (1, 3748), + (5000000, 0): (None, 1298), + (5000000, 37700): (None, 420), + (5000000, 50000): (None, 1373), + (5000000, 100000): (None, 1513), + (5000000, 150000): (None, 1480), + (5000000, 200000): (1, 16631), +} +REAL_BAND_TOTALS = { + 0: (63, 462), + 10000: (101, 1669), + 25000: (74, 2645), + 50000: (53, 3731), + 100000: (37, 5649), + 250000: (14, 4766), + 500000: (8, 5705), + 1000000: (5, 6390), + 2000000: (3, 9189), + 5000000: (2, 22714), +} +REAL_INCOME_TOTALS = { + 0: (203, 15975), + 37700: (29, 3311), + 50000: (55, 8496), + 100000: (22, 5221), + 150000: (12, 3812), + 200000: (37, 26106), +} + + +def _real_distribution() -> HMRCCapitalGainsJointDistribution: + cells = tuple( + HMRCCapitalGainsCell( + gain_lower_bound=gain, + income_lower_bound=income, + individuals=None if people is None else people * 1_000.0, + gains=amount * 1_000_000.0, + ) + for (gain, income), (people, amount) in REAL_2023_24.items() + ) + band_totals = tuple( + HMRCCapitalGainsBandTotal( + gain_lower_bound=gain, + individuals=people * 1_000.0, + gains=amount * 1_000_000.0, + ) + for gain, (people, amount) in REAL_BAND_TOTALS.items() + ) + income_totals = tuple( + HMRCCapitalGainsIncomeTotal( + income_lower_bound=income, + individuals=people * 1_000.0, + gains=amount * 1_000_000.0, + ) + for income, (people, amount) in REAL_INCOME_TOTALS.items() + ) + return HMRCCapitalGainsJointDistribution( + cells=cells, + band_totals=band_totals, + income_totals=income_totals, + source=HMRCCapitalGainsSourceProvenance( + local_path=None, + sha256="embedded-2023-24", + size_bytes=0, + sheet_name="3_1_2023-24", + source_vintage="2023-24", + build_period="2023", + ), + total_individuals=359_000.0, + total_gains=62_921_000_000.0, + ) + + +class TestRealPublishedSurface: + """The published 2023-24 values, embedded so CI cannot mask them. + + A synthetic distribution with feasible means hid that four real cells + imply a mean outside their band through rounding, crashing the stage on + any population reaching those income bands. + """ + + def test_every_income_band_plans_without_raising(self) -> None: + distribution = _real_distribution() + + for income in HMRC_CGT_INCOME_BAND_LOWER_BOUNDS: + plans = _band_plans( + distribution, + income, + annual_exempt_amount=PARAMETERS.annual_exempt_amount, + ) + for plan in plans: + if np.isinf(plan.gain_upper_bound): + assert plan.mean > plan.effective_lower_bound + else: + assert ( + plan.effective_lower_bound < plan.mean < plan.gain_upper_bound + ) + + def test_infeasible_rounded_means_are_repaired_toward_the_boundary(self) -> None: + distribution = _real_distribution() + + plans = _band_plans( + distribution, 37700, annual_exempt_amount=PARAMETERS.annual_exempt_amount + ) + by_band = {plan.gain_lower_bound: plan for plan in plans} + + # Rounded mean £463,000 for the £500k-£1m band clamps just inside. + repaired = by_band[500_000] + assert repaired.mean_repaired + assert 500_000 < repaired.mean < 520_000 + + def test_imputation_runs_across_every_income_band(self) -> None: + distribution = _real_distribution() + rows = 3_000 + rng = np.random.default_rng(7) + gains = np.where(rng.random(rows) < 0.6, rng.lognormal(10, 1.5, rows), 0.0) + incomes = rng.choice( + [20_000.0, 45_000.0, 80_000.0, 120_000.0, 180_000.0, 400_000.0], rows + ) + frame = _frame(rows, gains=gains, incomes=incomes) + + result = impute_uk_capital_gains(frame, distribution, PARAMETERS) + + drawn = result.table("person")["capital_gains"].to_numpy() + assert np.isfinite(drawn).all() + assert (drawn >= 0).all() + redrawn = drawn[gains > 0] + liable = redrawn[redrawn > PARAMETERS.annual_exempt_amount] + # Every draw allocated to the liability distribution clears the AEA. + assert liable.size > 0 + below = redrawn[(redrawn > 0) & (redrawn <= PARAMETERS.annual_exempt_amount)] + # The remainder keeps capped existing amounts, never band draws. + assert below.size == 0 or below.max() <= PARAMETERS.annual_exempt_amount + assert drawn.max() >= 5_000_000.0 + + +class TestPolicyParameters: + def test_reads_the_2023_values_from_the_parameter_tree(self) -> None: + pytest.importorskip( + "policyengine_uk", reason="requires the microcosm-build 'uk' extra" + ) + from microcosm.build.uk_runtime.cgt_imputation import uk_cgt_policy_parameters + + parameters = uk_cgt_policy_parameters(2023) + + assert parameters.personal_allowance == 12_570.0 + assert parameters.personal_allowance_taper_threshold == 100_000.0 + assert parameters.personal_allowance_taper_rate == 0.5 + assert parameters.annual_exempt_amount == 6_000.0 + assert parameters.instant == "2023-06-01" From 399a816dcb027b1f79d9e9cef66e75f10af8f889 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 13 Aug 2026 01:15:41 -0400 Subject: [PATCH 2/2] =?UTF-8?q?Reach=20the=20=C2=A337,700=20band=20in=20th?= =?UTF-8?q?e=20end-to-end=20fixtures,=20and=20prove=20it?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous round claimed the end-to-end incomes covered all six income bands; they did not — £45,000 maps to £32,430 after the Personal Allowance, below the £37,700 bound, so the band went unexercised through the imputation path in both fixtures. £55,000 reaches it, and the test now asserts the proxy visits every published band, so the fixture cannot silently regress the way it silently overclaimed. Co-Authored-By: Claude Fable 5 --- .../tests/test_uk_cgt_imputation.py | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/packages/microcosm-build/tests/test_uk_cgt_imputation.py b/packages/microcosm-build/tests/test_uk_cgt_imputation.py index 67f0c0ae..195cf808 100644 --- a/packages/microcosm-build/tests/test_uk_cgt_imputation.py +++ b/packages/microcosm-build/tests/test_uk_cgt_imputation.py @@ -330,7 +330,7 @@ def test_stage_runs_end_to_end_on_the_pinned_artifact(self) -> None: if not pinned_ods.is_file(): pytest.skip("reviewed HMRC capital gains ODS is an optional local input") stage = uk_capital_gains_imputation_stage(pinned_ods, parameters=PARAMETERS) - incomes = [20_000.0, 45_000.0, 80_000.0, 120_000.0, 180_000.0, 400_000.0] + incomes = [20_000.0, 55_000.0, 80_000.0, 120_000.0, 180_000.0, 400_000.0] frame = _frame( 60, gains=[float(5_000 * (i + 1)) for i in range(60)], @@ -535,10 +535,21 @@ def test_imputation_runs_across_every_income_band(self) -> None: rng = np.random.default_rng(7) gains = np.where(rng.random(rows) < 0.6, rng.lognormal(10, 1.5, rows), 0.0) incomes = rng.choice( - [20_000.0, 45_000.0, 80_000.0, 120_000.0, 180_000.0, 400_000.0], rows + [20_000.0, 55_000.0, 80_000.0, 120_000.0, 180_000.0, 400_000.0], rows ) frame = _frame(rows, gains=gains, incomes=incomes) + proxy = uk_cgt_taxable_income_proxy(frame.table("person"), PARAMETERS) + visited = set( + np.asarray(HMRC_CGT_INCOME_BAND_LOWER_BOUNDS)[ + np.digitize(proxy, HMRC_CGT_INCOME_BAND_LOWER_BOUNDS[1:]) + ] + ) + assert visited == set(HMRC_CGT_INCOME_BAND_LOWER_BOUNDS), ( + "fixture incomes must reach every published income band after " + f"the Personal Allowance; missing {set(HMRC_CGT_INCOME_BAND_LOWER_BOUNDS) - visited}" + ) + result = impute_uk_capital_gains(frame, distribution, PARAMETERS) drawn = result.table("person")["capital_gains"].to_numpy()