impl/python.py

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

from typing import Any, Dict, List, Sequence, Tuple

from .dates_uk_tax_year import uk_tax_year  ← from dates.uk-tax-year ^1.0.0 · built alongside by fune
from .invest_pension_annual_allowance_data import ANNUAL_ALLOWANCE_RULES, ANNUAL_ALLOWANCE_RULES_HISTORY, ANNUAL_ALLOWANCE_RULES_HORIZON  ← this capability’s own data, compiled from data/annual-allowance.json into the same file by fune build
from .invest_pension_annual_allowance_types import AnnualAllowanceResult, CarryForwardYear, PensionInputYear
from .money_amount import Money  ← from money.amount ^1.0.0 · built alongside by fune


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"]),
    )