Functional Weave
Code in Rust

payroll.national-insurance@1.0.2

impl/python.py

6,609 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.

import re
from typing import NoReturn, Optional, Tuple

from .math_round_div import round_div  ← from math.round-div ^1.0.0 · built alongside by fune
from .money_amount import Money, money  ← from money.amount ^1.0.0 · built alongside by fune
from .payroll_national_insurance_data import NI_PRIMARY_RATES, NI_PRIMARY_RATES_HISTORY, NI_PRIMARY_RATES_HORIZON, NI_SECONDARY_RATES, NI_SECONDARY_RATES_HISTORY, NI_SECONDARY_RATES_HORIZON, NI_THRESHOLDS, NI_THRESHOLDS_HISTORY, NI_THRESHOLDS_HORIZON  ← this capability’s own data, compiled from data/ni-thresholds.json into the same file by fune build
from .payroll_national_insurance_types import DirectorNi, NationalInsurance
from .payroll_tax_period_types import PayFrequency

ISO_DATE = re.compile(r"[0-9]{4}-[0-9]{2}-[0-9]{2}")


def _in_force(valid_from: str, valid_to: Optional[str], on_date: str) -> bool:
    return on_date >= valid_from and (valid_to is None or on_date <= valid_to)


def _missing(what: str, on_date: str, history: str, horizon: Optional[str]) -> NoReturn:
    # A pruned build must refuse a date it no longer carries rules for rather
    # than answer it with a later year's rates.
    if history != "full" and horizon is not None and on_date < horizon:
        raise ValueError(
            "%s on %s: this build was installed with history=%s, so it only carries rules from %s. "
            "Reinstall with history=full for earlier tax years." % (what, on_date, history, horizon)
        )
    raise ValueError("%s on %s" % (what, on_date))


def _slice(earnings: int, lower: int, upper: Optional[int]) -> int:
    """Earnings falling in (lower, upper]; upper None means no ceiling."""
    top = earnings if upper is None else min(earnings, upper)
    return max(0, top - lower)


def _limits(t, frequency: str) -> Tuple[int, int, int, int, int, int]:
    """(lel, pt, st, fust, ust, uel) for the earnings period."""
    if frequency == "annual":
        return (t.lel_annual, t.pt_annual, t.st_annual, t.fust_annual, t.ust_annual, t.uel_annual)
    if frequency == "monthly":
        return (t.lel_monthly, t.pt_monthly, t.st_monthly, t.fust_monthly, t.ust_monthly, t.uel_monthly)
    # HMRC's CA38: for pay in multiples of a week, work on the weekly figures
    # and multiply by the number of weeks.
    k = {"weekly": 1, "fortnightly": 2, "four-weekly": 4}[frequency]
    return (t.lel_weekly * k, t.pt_weekly * k, t.st_weekly * k, t.fust_weekly * k, t.ust_weekly * k, t.uel_weekly * k)


def _contributions(earnings: int, limits, p, s) -> Tuple[int, int]:
    # Regulation 12(1) SSCR 2001: primary and secondary are worked out
    # separately and each total is rounded to the nearest penny, a half penny
    # going up.
    lel, pt, st, fust, ust, uel = limits
    primary = _slice(earnings, pt, uel) * p.pt_to_uel_basis_points + _slice(earnings, uel, None) * p.above_uel_basis_points
    secondary = (
        _slice(earnings, st, fust) * s.st_to_fust_basis_points
        + _slice(earnings, max(st, fust), ust) * s.fust_to_ust_basis_points
        + _slice(earnings, max(st, ust), None) * s.above_ust_basis_points
    )
    return round_div(primary, 10000, "half-up"), round_div(secondary, 10000, "half-up")


def national_insurance(
    earnings: Money, category: str, frequency: PayFrequency, pay_date: str, director: Optional[DirectorNi]
) -> NationalInsurance:
    """Class 1 National Insurance on one payment, by the exact percentage method.

    With ``director`` None this is the ordinary earnings-period calculation.
    With it, the director's annual earnings period: contributions on everything
    paid so far this tax year at the annual thresholds, less what has already
    been paid, so the amount can go down (or negative) as well as up.
    """
    if earnings.currency != "GBP":
        raise ValueError("National Insurance must be in GBP, received %s" % (earnings.currency,))
    if earnings.minor < 0:
        raise ValueError("earnings must not be negative, received %s" % (earnings.minor,))
    if frequency not in ("weekly", "fortnightly", "four-weekly", "monthly"):
        raise ValueError('unknown pay frequency "%s"' % (frequency,))
    if not isinstance(pay_date, str) or not ISO_DATE.fullmatch(pay_date):
        raise ValueError('payDate must be an ISO date (YYYY-MM-DD), received "%s"' % (pay_date,))

    thresholds = next((t for t in NI_THRESHOLDS if _in_force(t.valid_from, t.valid_to, pay_date)), None)
    if thresholds is None:
        _missing("no National Insurance thresholds", pay_date, NI_THRESHOLDS_HISTORY, NI_THRESHOLDS_HORIZON)

    def primary_for(basis: str):
        return next(
            (r for r in NI_PRIMARY_RATES if r.category == category and r.basis == basis and _in_force(r.valid_from, r.valid_to, pay_date)),
            None,
        )

    # Directors only have rates of their own in a year the main rate changed
    # mid-year (2023-24); otherwise they pay the ordinary rates.
    primary = primary_for("director") if director is not None else None
    if primary is None:
        primary = primary_for("standard")
    if primary is None:
        _missing("no National Insurance rates for category %s" % (category,), pay_date, NI_PRIMARY_RATES_HISTORY, NI_PRIMARY_RATES_HORIZON)
    secondary = next(
        (r for r in NI_SECONDARY_RATES if r.category == category and _in_force(r.valid_from, r.valid_to, pay_date)), None
    )
    if secondary is None:
        _missing("no National Insurance rates for category %s" % (category,), pay_date, NI_SECONDARY_RATES_HISTORY, NI_SECONDARY_RATES_HORIZON)

    if director is None:
        limits = _limits(thresholds, frequency)
        employee, employer = _contributions(earnings.minor, limits, primary, secondary)
        return NationalInsurance(
            employee=money(employee, "GBP"),
            employer=money(employer, "GBP"),
            lower_earnings_limit_reached=earnings.minor >= limits[0],
        )

    for name, value in (
        ("previousEarnings", director.previous_earnings),
        ("previousEmployee", director.previous_employee),
        ("previousEmployer", director.previous_employer),
    ):
        if value.currency != "GBP":
            raise ValueError("National Insurance must be in GBP, received %s for %s" % (value.currency, name))
    cumulative = director.previous_earnings.minor + earnings.minor
    if cumulative < 0:
        raise ValueError("earnings must not be negative, received %s to date" % (cumulative,))
    limits = _limits(thresholds, "annual")
    employee, employer = _contributions(cumulative, limits, primary, secondary)
    return NationalInsurance(
        employee=money(employee - director.previous_employee.minor, "GBP"),
        employer=money(employer - director.previous_employer.minor, "GBP"),
        lower_earnings_limit_reached=cumulative >= limits[0],
    )