Functional Weave
Code in Python

invest.capital-gains-tax@1.0.1

impl/python.py

6,123 bytes · the Python implementation · view raw

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 Dict, List, Optional, Sequence

from .dates_uk_tax_year import uk_tax_year  ← from dates.uk-tax-year ^1.0.0 · built alongside by fune
from .invest_capital_gains_tax_data import CGT_RATES, CGT_RATES_HISTORY, CGT_RATES_HORIZON, CGT_YEARS, CGT_YEARS_HISTORY, CGT_YEARS_HORIZON  ← this capability’s own data, compiled from data/cgt-years.json into the same file by fune build
from .invest_capital_gains_tax_types import CgtDisposal, CgtLine, CgtResult
from .money_amount import Money  ← from money.amount ^1.0.0 · built alongside by fune
from .money_apply_rate import apply_rate  ← from money.apply-rate ^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) -> None:
    if value.currency != "GBP":
        raise ValueError("%s must be in GBP, received %s" % (name, value.currency))
    if isinstance(value.minor, bool) or not isinstance(value.minor, int):
        raise ValueError("%s must be a whole number of pence, received %s" % (name, value.minor))


def _pruned(history: str, horizon: Optional[str], date: str) -> str:
    if horizon is not None and date < horizon:
        return (
            ": this build was installed with history=%s, so it only carries rules from %s. "
            "Reinstall with history=full for earlier years." % (history, horizon)
        )
    return ""


def _rate_for(disposal: CgtDisposal) -> List[int]:
    for rule in CGT_RATES:
        if rule.kind != disposal.kind or disposal.date < rule.valid_from:
            continue
        if rule.valid_to is not None and disposal.date > rule.valid_to:
            continue
        return [rule.lower_basis_points, rule.higher_basis_points, 0]
    raise ValueError(
        "no CGT rate for %s gains on %s" % (disposal.kind, disposal.date)
        + _pruned(CGT_RATES_HISTORY, CGT_RATES_HORIZON, disposal.date)
    )


def capital_gains_tax(
    tax_year_date: str,
    disposals: Sequence[CgtDisposal],
    taxable_income: Money,
    losses_brought_forward: Money,
) -> CgtResult:
    """Capital Gains Tax for an individual for one UK tax year.

    The year's losses come off its gains in full, even when that wastes the
    annual exempt amount; losses brought forward only bring the net gains down
    to the annual exempt amount. Losses and the exempt amount go against the
    gains charged at the highest rate first, and what taxable income leaves of
    the basic rate band goes to the gains whose two rates are furthest apart:
    that is the "most beneficial" order TCGA 1992 s1F and s1K allow.
    """
    year = uk_tax_year(tax_year_date)
    rules = next(
        (row for row in CGT_YEARS if row.valid_from <= year.start and (row.valid_to is None or year.start <= row.valid_to)),
        None,
    )
    if rules is None:
        raise ValueError(
            "no CGT rules for the %s tax year" % (year.label,)
            + _pruned(CGT_YEARS_HISTORY, CGT_YEARS_HORIZON, year.start)
        )
    _check_money(taxable_income, "taxableIncome")
    _check_money(losses_brought_forward, "lossesBroughtForward")
    if taxable_income.minor < 0:
        raise ValueError("taxableIncome must not be negative, received %d" % (taxable_income.minor,))
    if losses_brought_forward.minor < 0:
        raise ValueError("lossesBroughtForward must not be negative, received %d" % (losses_brought_forward.minor,))

    # [lower, higher, amount]: gains that share a pair of rates.
    buckets: List[List[int]] = []
    gains = 0
    losses = 0
    for disposal in disposals:
        _check_money(disposal.gain, "gain")
        if disposal.kind not in ("residential", "other"):
            raise ValueError('kind must be residential or other, received "%s"' % (disposal.kind,))
        if disposal.date < year.start or disposal.date > year.end:
            raise ValueError("the disposal on %s is outside the %s tax year" % (disposal.date, year.label))
        rate = _rate_for(disposal)
        if disposal.gain.minor <= 0:
            losses -= disposal.gain.minor
            continue
        gains += disposal.gain.minor
        bucket = next((b for b in buckets if b[0] == rate[0] and b[1] == rate[1]), None)
        if bucket is None:
            buckets.append([rate[0], rate[1], disposal.gain.minor])
        else:
            bucket[2] += disposal.gain.minor

    net = gains - losses
    annual_exempt_amount = rules.annual_exempt_amount
    brought_forward_used = 0
    exempt_used = 0
    taxable = 0
    carried = losses_brought_forward.minor
    if net > 0:
        brought_forward_used = min(losses_brought_forward.minor, max(0, net - annual_exempt_amount))
        exempt_used = min(annual_exempt_amount, net - brought_forward_used)
        taxable = net - brought_forward_used - exempt_used
        carried -= brought_forward_used
    else:
        carried += -net

    # Deductions against the gains taxed hardest first.
    deductions = gains - taxable
    for bucket in sorted(buckets, key=lambda b: (-b[1], -b[0])):
        used = min(bucket[2], deductions)
        bucket[2] -= used
        deductions -= used

    # The basic rate band where it saves the most: the widest gap between the two rates.
    band_available = max(0, rules.basic_rate_band - taxable_income.minor)
    band = band_available
    charged: Dict[int, int] = {}

    def charge(rate: int, amount: int) -> None:
        if amount > 0:
            charged[rate] = charged.get(rate, 0) + amount

    for bucket in sorted(buckets, key=lambda b: (-(b[1] - b[0]), -b[1])):
        lower = min(bucket[2], band)
        band -= lower
        charge(bucket[0], lower)
        charge(bucket[1], bucket[2] - lower)

    lines = [
        CgtLine(rate_basis_points=rate, gains=_gbp(amount), tax=apply_rate(_gbp(amount), rate, "down"))
        for rate, amount in sorted(charged.items())
    ]
    tax = sum(line.tax.minor for line in lines)

    return CgtResult(
        tax_year=year.label,
        gains=_gbp(gains),
        losses=_gbp(losses),
        losses_brought_forward_used=_gbp(brought_forward_used),
        annual_exempt_amount=_gbp(annual_exempt_amount),
        annual_exempt_amount_used=_gbp(exempt_used),
        taxable_gains=_gbp(taxable),
        basic_rate_band_available=_gbp(band_available),
        lines=lines,
        tax=_gbp(tax),
        losses_carried_forward=_gbp(carried),
    )