from typing import Any, Dict, List, Sequence, Tuple from .dates_uk_tax_year import uk_tax_year from .invest_pension_annual_allowance_data import ANNUAL_ALLOWANCE_RULES, ANNUAL_ALLOWANCE_RULES_HISTORY, ANNUAL_ALLOWANCE_RULES_HORIZON from .invest_pension_annual_allowance_types import AnnualAllowanceResult, CarryForwardYear, PensionInputYear from .money_amount import Money def _gbp(minor: int) -> Money: return Money(minor=minor, currency="GBP") def _check_money(value: Money, name: str, label: str) -> int: if value.currency != "GBP": raise ValueError("%s must be in GBP, received %s for %s" % (name, value.currency, label)) minor = value.minor if isinstance(minor, bool) or not isinstance(minor, int) or minor < 0: raise ValueError( "%s must be a whole number of pence, not negative, received %s for %s" % (name, minor, label) ) return minor def _rule_for(start_year: int) -> Tuple[Any, str]: start = "%04d-04-06" % (start_year,) label = uk_tax_year(start).label for rule in ANNUAL_ALLOWANCE_RULES: if rule.valid_from <= start and (rule.valid_to is None or start <= rule.valid_to): return rule, label why = "" horizon = ANNUAL_ALLOWANCE_RULES_HORIZON if horizon is not None and start < horizon: why = ( ": this build was installed with history=%s, so it only carries rules from %s. " "Reinstall with history=full for earlier years." % (ANNUAL_ALLOWANCE_RULES_HISTORY, horizon) ) raise ValueError("no annual allowance rules for the %s tax year%s" % (label, why)) def _test(start_year: int, year: PensionInputYear, carry_forward: int) -> Dict[str, Any]: """How one tax year comes out against its own allowance, given the carry-forward it can reach.""" rule, label = _rule_for(start_year) mp = _check_money(year.money_purchase_input, "moneyPurchaseInput", label) other = _check_money(year.other_input, "otherInput", label) threshold = _check_money(year.threshold_income, "thresholdIncome", label) adjusted = _check_money(year.adjusted_income, "adjustedIncome", label) total = mp + other if not year.member and total > 0: raise ValueError("pension inputs in %s need membership of a registered pension scheme in that year" % (label,)) # Taper: 1 pound off for every 2 of adjusted income over the limit, rounded # down to a whole pound, when threshold income is also over its limit. allowance = rule.annual_allowance tapered = threshold > rule.threshold_income and adjusted > rule.adjusted_income if tapered: reduction = (adjusted - rule.adjusted_income) // 200 * 100 allowance = max(rule.minimum_tapered_allowance, allowance - reduction) out: Dict[str, Any] = {"label": label, "standard": rule.annual_allowance, "allowance": allowance, "tapered": tapered, "total": total} default_used = min(carry_forward, max(0, total - allowance)) default_excess = max(0, total - allowance - default_used) if year.mpaa_applies and mp > rule.money_purchase_allowance: # Carry-forward can top up the alternative allowance, never the MPAA. alternative = max(0, allowance - rule.money_purchase_allowance) alt_used = min(carry_forward, max(0, other - alternative)) alt_excess = mp - rule.money_purchase_allowance + max(0, other - alternative - alt_used) use_alternative = alt_excess > default_excess out.update( mpaa=rule.money_purchase_allowance, alternative=alternative, basis="alternative" if use_alternative else "standard", carry_forward_used=alt_used if use_alternative else default_used, excess=alt_excess if use_alternative else default_excess, unused=max(0, alternative - other) if year.member else 0, tested_allowance=alternative, tested_input=other, ) return out out.update( mpaa=rule.money_purchase_allowance if year.mpaa_applies else None, alternative=None, basis="standard", carry_forward_used=default_used, excess=default_excess, unused=max(0, allowance - total) if year.member else 0, tested_allowance=allowance, tested_input=total, ) return out def _available(entry: Dict[str, Any]) -> int: return entry["unused"] - entry["used_by_later_years"] - entry["used_this_year"] def _consume(entries: List[Dict[str, Any]], amount: int, field: str) -> None: """Take `amount` from the entries' unused allowance, earliest year first.""" left = amount for entry in entries: take = min(left, _available(entry)) entry[field] += take left -= take def pension_annual_allowance( tax_year_date: str, current: PensionInputYear, previous_years: Sequence[PensionInputYear] ) -> AnnualAllowanceResult: """The annual allowance for a UK tax year, and how much of the year's pension input amounts is subject to the annual allowance charge. The earlier years are replayed oldest first, because an excess in one of them uses up unused allowance from the years before it (earliest first), and that is allowance the year being tested can no longer carry forward. A year's own allowance is used before any carried forward; carry-forward is then taken from the earliest of the three previous years first. """ year = uk_tax_year(tax_year_date) start_year = int(year.start[:4]) if len(previous_years) < 3: raise ValueError( "previousYears must hold at least the three tax years before this one, oldest first, received %d" % (len(previous_years),) ) ledger: List[Dict[str, Any]] = [] for i, previous in enumerate(previous_years): window = ledger[max(0, i - 3) : i] result = _test(start_year - len(previous_years) + i, previous, sum(_available(e) for e in window)) _consume(window, result["carry_forward_used"], "used_by_later_years") ledger.append( { "tax_year": result["label"], "allowance": result["tested_allowance"], "input": result["tested_input"], "unused": result["unused"], "used_by_later_years": 0, "used_this_year": 0, } ) window = ledger[-3:] carry_forward_available = sum(_available(e) for e in window) result = _test(start_year, current, carry_forward_available) _consume(window, result["carry_forward_used"], "used_this_year") carry_forward = [ CarryForwardYear( tax_year=e["tax_year"], allowance=_gbp(e["allowance"]), input=_gbp(e["input"]), unused=_gbp(e["unused"]), used_by_later_years=_gbp(e["used_by_later_years"]), used_this_year=_gbp(e["used_this_year"]), ) for e in window ] return AnnualAllowanceResult( tax_year=result["label"], standard_allowance=_gbp(result["standard"]), allowance=_gbp(result["allowance"]), tapered=result["tapered"], money_purchase_allowance=None if result["mpaa"] is None else _gbp(result["mpaa"]), alternative_allowance=None if result["alternative"] is None else _gbp(result["alternative"]), total_input=_gbp(result["total"]), basis=result["basis"], carry_forward=carry_forward, carry_forward_available=_gbp(carry_forward_available), carry_forward_used=_gbp(result["carry_forward_used"]), excess=_gbp(result["excess"]), )