dodil/erp-brick-payroll

DODIL ERP brick: payroll

src/brick_payroll/domain/statutory.py

"""Statutory sick pay and statutory maternity pay: PURE functions of the facts time-off recorded and the dated rows in force
(no rate is written here - module.yaml in_code_not_rows). time-off computes no money; it serves the facts (qualifying
days, the period of incapacity, the linked earlier spell, the expected week of childbirth) and this module decides pay.

SSP (SSCBA 1992 ss. 151-153): a period of incapacity for work (4 or more consecutive days) pays on qualifying days; the first
`ssp_waiting_days` of a period are not paid (a dated row: 3 before 6 April 2026, 0 after - and none at all in a period linked to an
earlier one, whose waiting days were served there); the weekly rate is the flat rate, or the lower of it and `ssp_earnings_cap_pct`
of average weekly earnings where that row exists; before April 2026 average earnings had to reach the lower earnings limit
(`ssp_lel_test`); it stops after `ssp_max_weeks` weeks across the linked periods; a fit note is needed from the day after the
self-certification days. SMP (SSCBA 1992 ss. 164-166): 26 weeks' service by the qualifying week (the 15th before the expected week
of childbirth), average weekly earnings at least the lower earnings limit, the evidence in hand; `smp_higher_pct` of average weekly
earnings for `smp_higher_weeks`, then the lower of the flat rate and that percentage, up to `smp_total_weeks` in all; a day of SMP
is a seventh of its week's rate. Anything that needs a figure that is not a row is refused by name, never guessed."""
from dataclasses import dataclass, field
from datetime import date, timedelta
from fractions import Fraction
from typing import Optional

from brick_payroll.domain.errors import RuleBroken
from brick_payroll.domain.money import F, money, pct, rate4
from brick_payroll.domain.paye import PERIODS, Rates

WEEKLY = "weekly"


def average_weekly_earnings(paydays: list, frequency: str) -> Optional[Fraction]:
    """Earnings of the paydays in the relevant period, annualised by the schedule and spread over 52 weeks: two monthly paydays of
    2,816.67 -> 5,633.34 x 12 / 2 / 52 = 650.00. None when no payday falls in the period (the caller names its fallback)."""
    if not paydays:
        return None
    return sum((F(g) for _, g in paydays), Fraction(0)) * PERIODS[frequency] / len(paydays) / 52


def awe_penny(awe: Fraction):
    """Average weekly earnings to the penny, half up (rounding rule chosen here: module.yaml names none; AGENTS.md)."""
    return money(awe)


# ── SSP ─────────────────────────────────────────────────────────────────────────────────────

@dataclass
class SspFacts:
    absence_ref: str
    revision: int
    first_day: date
    last_day: date                     # the last day off (closed) or the last day known (open)
    piw: bool
    linked: bool                       # linked to an earlier period of incapacity (its waiting days were served there)
    group: str                         # the linked group: the first spell's ref
    evidence_ok: bool
    qualifying: list                   # the spell's qualifying days (hr's pattern: employed and contracted), sorted, first_day .. last_day
    per_week: int                      # qualifying days in a week of the spell
    paid_before: int = 0               # days of this group already paid by approved runs


@dataclass
class SspResult:
    deduct_days: list = field(default_factory=list)       # qualifying days of the spell inside the period: salary comes off at the day rate
    paid_days: list = field(default_factory=list)
    waiting_days: list = field(default_factory=list)
    weekly: Optional[Fraction] = None
    daily: Optional[Fraction] = None
    reason: str = ""                   # why no SSP (empty when paid)
    notes: dict = field(default_factory=dict)


def ssp(f: SspFacts, lo: date, hi: date, awe: Optional[Fraction], rates: Rates) -> SspResult:
    spell = [d for d in f.qualifying if f.first_day <= d <= f.last_day]
    res = SspResult(deduct_days=[d for d in spell if lo <= d <= hi])
    res.notes = {"absence_ref": f.absence_ref, "revision": f.revision, "piw": f.piw, "linked_to_earlier": f.linked, "group": f.group}
    if not res.deduct_days:
        return res
    if not f.piw and rates.maybe("ssp_requires_piw"):
        res.reason = "not a period of incapacity for work (fewer than 4 consecutive days): no SSP"
        return res
    if f.per_week <= 0:
        raise RuleBroken(f"{f.absence_ref}: hr's work pattern has no qualifying day in the week the sickness began, so SSP cannot be a daily rate",
                         gate="work_pattern")
    flat = rates.get("ssp_weekly_rate", WEEKLY)
    weekly = flat
    cap = rates.maybe("ssp_earnings_cap_pct")
    if cap is not None and awe is not None:
        weekly = min(flat, awe * pct(cap))
    if rates.maybe("ssp_lel_test") and awe is not None and awe < rates.get("ni_lower_earnings_limit", WEEKLY):
        res.reason = f"average weekly earnings {money(awe)} are below the lower earnings limit: no SSP"
        res.notes.update({"awe": str(money(awe))})
        return res
    waiting = 0 if f.linked else int(rates.get("ssp_waiting_days"))
    res.waiting_days = spell[:waiting]
    self_cert = int(rates.get("ssp_self_certification_days"))
    cap_days = int(rates.get("ssp_max_weeks")) * f.per_week
    remaining = max(cap_days - f.paid_before, 0)
    payable = [d for d in spell[waiting:] if lo <= d <= hi and (f.evidence_ok or (d - f.first_day).days < self_cert)]
    res.paid_days = payable[:remaining]
    res.weekly, res.daily = weekly, weekly / f.per_week
    res.notes.update({"awe": None if awe is None else str(money(awe)), "weekly_rate": str(money(weekly)), "qualifying_days_a_week": f.per_week,
                      "waiting_days": len(res.waiting_days), "paid_days": len(res.paid_days), "days_paid_before": f.paid_before,
                      "limit_days": cap_days, "evidence_ok": f.evidence_ok})
    if not res.paid_days:
        res.reason = "no payable day in this period" + (" (the 28 week limit is reached)" if remaining == 0 else "")
    elif len(payable) > remaining:
        res.notes["stopped_at_limit"] = True
    return res


# ── SMP ─────────────────────────────────────────────────────────────────────────────────────

def week_start(d: date) -> date:
    """The Sunday that starts the week containing d (SMP weeks run Sunday to Saturday)."""
    return d - timedelta(days=(d.weekday() + 1) % 7)


def qualifying_week(ewc: date) -> tuple:
    """(start, end) of the 15th week before the week the baby is due; the relevant period of 8 weeks ends with it."""
    start = week_start(ewc) - timedelta(weeks=15)
    return start, start + timedelta(days=6)


def relevant_period(ewc: date) -> tuple:
    _, end = qualifying_week(ewc)
    return end - timedelta(days=55), end


@dataclass
class SmpFacts:
    absence_ref: str
    revision: int
    ewc: date
    evidence_received_on: Optional[date]
    start: date                        # the first day of the leave: SMP weeks count from it
    last_day: date                     # the last day of leave inside the period (extended over a trailing weekend)
    employed_26_weeks: bool            # employed at the start of the 26 weeks ending with the qualifying week


@dataclass
class SmpResult:
    amount: Fraction = Fraction(0)     # exact: the caller rounds ONCE
    days: int = 0
    reason: str = ""
    eligible: bool = False
    notes: dict = field(default_factory=dict)
    weeks_in_period: list = field(default_factory=list)


def smp(f: SmpFacts, lo: date, hi: date, awe: Optional[Fraction], rates: Rates) -> SmpResult:
    res = SmpResult()
    q_start, q_end = qualifying_week(f.ewc)
    res.notes = {"absence_ref": f.absence_ref, "revision": f.revision, "ewc": f.ewc.isoformat(), "leave_start": f.start.isoformat(),
                 "qualifying_week": f"{q_start.isoformat()}..{q_end.isoformat()}", "relevant_period": "%s..%s" % tuple(d.isoformat() for d in relevant_period(f.ewc))}
    if awe is None:
        raise RuleBroken(f"{f.absence_ref}: no earnings to average for SMP", gate="average_earnings")
    lel = rates.get("ni_lower_earnings_limit", WEEKLY)
    awe_p = awe_penny(awe)
    res.notes.update({"awe": str(awe_p), "lower_earnings_limit": str(money(lel))})
    if f.evidence_received_on is None:
        res.reason = "no maternity evidence (MATB1) received: no SMP"
    elif not f.employed_26_weeks:
        res.reason = "not employed for 26 weeks by the qualifying week: no SMP (Maternity Allowance instead)"
    elif F(awe_p) < lel:
        res.reason = f"average weekly earnings {awe_p} are below the lower earnings limit {money(lel)}: no SMP"
    if res.reason:
        return res
    res.eligible = True
    higher = F(money(F(awe_p) * pct(rates.get("smp_higher_pct"))))
    flat = rates.get("smp_weekly_flat_rate", WEEKLY)
    later = min(flat, higher)
    h_weeks, total = int(rates.get("smp_higher_weeks")), int(rates.get("smp_total_weeks"))
    res.notes.update({"weekly_higher": str(money(higher)), "weekly_later": str(money(later)), "higher_weeks": h_weeks, "total_weeks": total,
                      "schedule": [{"weeks": f"1-{h_weeks}", "weekly": str(money(higher))}, {"weeks": f"{h_weeks + 1}-{total}", "weekly": str(money(later))}]})
    d, total_amount, counted = max(lo, f.start), Fraction(0), {}
    while d <= min(hi, f.last_day):
        week = (d - f.start).days // 7 + 1
        if week > total:
            break
        w = higher if week <= h_weeks else later
        total_amount += w / 7
        counted[week] = counted.get(week, 0) + 1
        res.days += 1
        d += timedelta(days=1)
    res.amount = total_amount
    res.weeks_in_period = [{"week": k, "days": v} for k, v in sorted(counted.items())]
    if not res.days:
        res.reason = "no day of SMP falls in this period"
    return res


def recovery(amount, small_employer: bool, rates: Rates):
    """What the employer recovers from HMRC: 92% of SMP, or 109% (2026/27) when small employers' relief is set (module.yaml PRL-09, PRL-19)."""
    key = "smp_small_employer_recovery_pct" if small_employer else "smp_recovery_pct"
    return money(F(amount) * pct(rates.get(key)))


def day_rate4(weekly: Fraction, per_week: int):
    return rate4(weekly / per_week)