dodil/erp-brick-payroll

DODIL ERP brick: payroll

src/brick_payroll/domain/paye.py

"""HMRC's method: the tax year and its periods, tax codes, cumulative and week1/month1 PAYE, Class 1 NICs (category A) by
the exact percentage method, and pension contributions on qualifying earnings. PURE functions of their inputs and the
rate rows in force - no rate is written here (module.yaml in_code_not_rows).

Slice A covers the monthly schedule, rest-of-UK tax codes, NI category A and the net pay arrangement; anything else is
refused by name (UnsupportedHere), never computed approximately."""
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from fractions import Fraction
from typing import Optional

from brick_payroll.domain.errors import RuleBroken
from brick_payroll.domain.money import F, down_penny, down_pound, money, ni_penny, pct, up_penny, up_pound
from brick_payroll.domain.values import TAX_YEAR_START

# Pay periods in a tax year per pay frequency: the divisor of the free pay and the rate bands (a 53rd week is n = 53 over 52).
PERIODS = {"monthly": 12, "weekly": 52, "fortnightly": 26, "four_weekly": 13}
PERIOD_LETTER = {"monthly": "M", "weekly": "W", "fortnightly": "F", "four_weekly": "P"}
PERIOD_WORD = {"monthly": "month", "weekly": "week", "fortnightly": "fortnight", "four_weekly": "four-week period"}


class UnsupportedHere(RuleBroken):
    """A case this slice does not calculate. Named, so a person is never paid on an approximation."""


# ── the tax year ─────────────────────────────────────────────────────────────────────────────

def tax_year_of(d: date) -> int:
    """The year the tax year starts in: 2026-11-30 -> 2026 (6 April 2026 to 5 April 2027)."""
    m, day = TAX_YEAR_START
    return d.year if (d.month, d.day) >= (m, day) else d.year - 1


def tax_year_label(start_year: int) -> str:
    return f"{start_year}-{(start_year + 1) % 100:02d}"


def tax_month(d: date) -> int:
    """Tax month 1 runs 6 April to 5 May; a payment is in the tax month its PAY DATE falls in (2026-11-30 -> 8)."""
    m = d.month - 4 if d.day >= 6 else d.month - 5
    return m % 12 + 1


def tax_week(d: date) -> int:
    """Tax week 1 runs 6 to 12 April; the 53rd week is the one or two days left over at the end of the year."""
    m, day = TAX_YEAR_START
    return (d - date(tax_year_of(d), m, day)).days // 7 + 1


def tax_period_of(frequency: str, d: date) -> int:
    """The number of the pay period a payment date falls in: tax month (monthly), tax week, fortnight (1 and 2 are period 1) or four
    weeks. Week 53 is period 27 of a fortnightly and 14 of a four-weekly schedule (HMRC's tables run on)."""
    if frequency == "monthly":
        return tax_month(d)
    w = tax_week(d)
    if frequency == "weekly":
        return w
    if frequency == "fortnightly":
        return (w + 1) // 2
    if frequency == "four_weekly":
        return (w + 3) // 4
    raise UnsupportedHere(f"the {frequency} schedule is not known", gate="schedule")


# ── tax codes ────────────────────────────────────────────────────────────────────────────────

@dataclass(frozen=True)
class TaxCode:
    code: str                         # as HMRC writes it: 1257L, K475, BR, D0, D1, NT, 0T
    kind: str                         # allowance | k | flat | nt
    number: int = 0                   # the numeric part of an allowance or K code
    flat_rate: str = ""               # the rate row of a flat code: tax_basic_rate (BR), tax_higher_rate (D0), ...

    def annual_adjustment(self) -> int:
        """HMRC's pay adjustment for a year: code number x 10 + 9 (1257 -> 12,579); 0 for 0T."""
        return self.number * 10 + 9 if self.number else 0


FLAT = {"BR": "tax_basic_rate", "D0": "tax_higher_rate", "D1": "tax_additional_rate"}


def parse_tax_code(text: str) -> TaxCode:
    """A rest-of-UK code. A Scottish (S) or Welsh (C) prefix needs its own band rows, which slice A has not seeded."""
    t = (text or "").strip().upper().replace(" ", "")
    if not t or len(t) > 8:
        raise RuleBroken(f"{text!r} is not a tax code: use one HMRC issued, such as 1257L, BR, D0, NT or K475", gate="bounds")
    if t[0] in ("S", "C") and len(t) > 1 and (t[1].isdigit() or t[1:] in FLAT or t[1:] in ("NT", "K")):
        raise UnsupportedHere(f"tax code {t}: Scottish and Welsh codes need their own band rows, not set up yet", gate="tax_region")
    if t in FLAT:
        return TaxCode(t, "flat", flat_rate=FLAT[t])
    if t == "NT":
        return TaxCode(t, "nt")
    if t[0] == "K" and t[1:].isdigit() and 1 <= len(t[1:]) <= 4:
        return TaxCode(t, "k", int(t[1:]))
    if t[:-1].isdigit() and 1 <= len(t[:-1]) <= 4 and t[-1] in "LMNT":
        return TaxCode(t, "allowance", int(t[:-1]))
    raise RuleBroken(f"{text!r} is not a tax code: use one HMRC issued, such as 1257L, BR, D0, NT or K475", gate="bounds")


# ── rates in force ───────────────────────────────────────────────────────────────────────────

class Rates:
    """The statutory rows in force on a pay date: (rate_key, frequency) -> value, and which rows were used."""

    def __init__(self, rows: list):
        self.rows = {(r["rate_key"], r["frequency"]): r for r in rows}
        self.used: set = set()

    def maybe(self, key: str, frequency: str = "any") -> Optional[Fraction]:
        """A row that may legitimately not exist for a date (a rule that stopped or has not started): None, not a refusal."""
        return self.get(key, frequency) if (key, frequency) in self.rows else None

    def get(self, key: str, frequency: str = "any") -> Fraction:
        row = self.rows.get((key, frequency))
        if row is None:
            raise RuleBroken(f"no statutory rate {key} ({frequency}) is in force on the pay date: enter it in Settings", gate="rates_in_force")
        self.used.add(f"{key}/{frequency}@{row['valid_from'].isoformat()}")
        return F(row["value"])


# ── PAYE ─────────────────────────────────────────────────────────────────────────────────────

@dataclass
class TaxResult:
    tax: Decimal                      # this period (negative: a refund)
    detail: dict


def paye(code: TaxCode, basis: str, frequency: str, period_no: int, pay_this: Decimal, prev_pay: Decimal,
         prev_tax: Decimal, rates: Rates) -> TaxResult:
    """PAYE on `pay_this` (taxable pay of the period). Cumulative: on pay and tax to date (`prev_*` include a P45's
    figures, so a starter's code works from 6 April); week1/month1 and flat codes: on this period alone.
      free pay to date = code x 10 + 9, x n / periods, rounded UP to the penny (1257L month 8: 8,386.00)
      taxable to date  = pay to date - free pay (+ for a K code), rounded DOWN to the pound
      bands to date    = annual band x n / periods, rounded UP to the pound
      tax to date      = sum over bands, rounded DOWN to the penny; this period = to date - tax already deducted
      the overriding limit: tax in a period never exceeds 50% of that period's taxable pay (a K code)."""
    periods = PERIODS.get(frequency)
    if periods is None:
        raise UnsupportedHere(f"the {frequency} schedule is not calculated yet", gate="schedule")
    pay_this, prev_pay, prev_tax = F(pay_this), F(prev_pay), F(prev_tax)
    detail: dict = {"code": code.code, "basis": basis, "period": period_no}
    if code.kind == "nt":
        return TaxResult(money(0), {**detail, "rule": "NT: no tax"})
    if code.kind == "flat":
        tax = down_penny(max(pay_this, Fraction(0)) * pct(rates.get(code.flat_rate)))
        return TaxResult(tax, {**detail, "rule": f"{code.code}: flat rate on this period"})
    if basis == "week1_month1":
        n, pay_td, before = 1, pay_this, Fraction(0)
    elif basis == "cumulative":
        n, pay_td, before = period_no, prev_pay + pay_this, prev_tax
    else:
        raise RuleBroken(f"basis {basis!r} is cumulative or week1_month1", gate="bounds")
    free = up_penny(Fraction(code.annual_adjustment() * n, periods))
    taxable = pay_td + F(free) if code.kind == "k" else pay_td - F(free)
    taxable_td = max(down_pound(taxable), Decimal(0))
    b1 = up_pound(rates.get("tax_basic_limit", "annual") * n / periods)
    b2 = up_pound(rates.get("tax_higher_limit", "annual") * n / periods)
    t = F(taxable_td)
    raw = (min(t, F(b1)) * pct(rates.get("tax_basic_rate"))
           + max(min(t, F(b2)) - F(b1), Fraction(0)) * pct(rates.get("tax_higher_rate"))
           + max(t - F(b2), Fraction(0)) * pct(rates.get("tax_additional_rate")))
    tax_td = down_penny(raw)
    tax = F(tax_td) - before
    limit = down_penny(max(pay_this, Fraction(0)) * pct(rates.get("tax_regulatory_limit")))
    capped = tax > F(limit)
    if capped:
        tax = F(limit)
    detail.update({"pay_to_date": str(money(pay_td)), "free_pay_to_date": str(free), "taxable_to_date": str(taxable_td),
                   "basic_band_to_date": str(b1), "higher_band_to_date": str(b2), "tax_to_date_due": str(tax_td),
                   "tax_before": str(money(before)), "overriding_limit_applied": capped})
    return TaxResult(money(tax), detail)


# ── National Insurance ───────────────────────────────────────────────────────────────────────

@dataclass
class NiResult:
    employee: Decimal
    employer: Decimal
    detail: dict


def class1_ni(category: str, frequency: str, gross: Decimal, rates: Rates) -> NiResult:
    """Class 1 NICs on the period's NIable pay, thresholds per pay period (not prorated for a part period), by the exact
    percentage method; each band rounded to the nearest penny, an exact half penny down. Category A only in slice A."""
    if category != "A":
        raise UnsupportedHere(f"NI category {category} is not calculated yet (slice A: category A)", gate="ni_category")
    if frequency not in PERIODS:
        raise UnsupportedHere(f"the {frequency} schedule is not calculated yet", gate="schedule")
    g = F(gross)
    pt, uel, st = rates.get("ni_primary_threshold", frequency), rates.get("ni_upper_earnings_limit", frequency), rates.get("ni_secondary_threshold", frequency)
    main = max(min(g, uel) - pt, Fraction(0))
    above = max(g - uel, Fraction(0))
    ee = ni_penny(main * pct(rates.get("ni_a_employee_main"))) + ni_penny(above * pct(rates.get("ni_a_employee_additional")))
    er = ni_penny(max(g - st, Fraction(0)) * pct(rates.get("ni_a_employer")))
    return NiResult(ee, er, {"category": category, "pt_to_uel": str(money(main)), "above_uel": str(money(above)),
                             "above_st": str(money(max(g - st, Fraction(0))))})


# ── pension ──────────────────────────────────────────────────────────────────────────────────

@dataclass(frozen=True)
class Scheme:
    scheme_code: str
    basis: str                        # qualifying_earnings (slice A) | pensionable_pay | basic_pay
    employee_pct: Decimal
    employer_pct: Decimal
    tax_relief: str                   # net_pay (slice A) | relief_at_source


@dataclass
class PensionResult:
    employee: Decimal
    employer: Decimal
    pensionable: Decimal


def pension(scheme: Optional[Scheme], frequency: str, pensionable_pay: Decimal, rates: Rates) -> PensionResult:
    """Contributions on qualifying earnings: pay in the band (lower to upper, per pay period), x each percentage, half up
    to the penny. Not a member: nothing. Other bases and relief at source are refused by name."""
    if scheme is None:
        return PensionResult(money(0), money(0), money(0))
    if scheme.basis != "qualifying_earnings":
        raise UnsupportedHere(f"pension scheme {scheme.scheme_code}: basis {scheme.basis} is not calculated yet", gate="pension_basis")
    if scheme.tax_relief != "net_pay":
        raise UnsupportedHere(f"pension scheme {scheme.scheme_code}: relief at source is not calculated yet", gate="pension_basis")
    lo, hi = rates.get("ae_qualifying_lower", frequency), rates.get("ae_qualifying_upper", frequency)
    qe = min(max(F(pensionable_pay) - lo, Fraction(0)), hi - lo)
    return PensionResult(money(qe * pct(scheme.employee_pct)), money(qe * pct(scheme.employer_pct)), money(qe))