Functional Weave
Code in Rust

payroll.statutory-sick-pay@1.0.1

impl/python.py

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

from .dates_add_days import add_days, days_between, parse_iso_date  ← from dates.add-days ^1.0.0 · built alongside by fune
from .dates_add_months import add_months  ← from dates.add-months ^1.0.0 · built alongside by fune
from .dates_day_of_week import day_of_week  ← from dates.day-of-week ^1.0.0 · built alongside by fune
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_statutory_sick_pay_data import SSP_RULES, SSP_RULES_HISTORY, SspRule  ← this capability’s own data, compiled from data/ssp-rules.json into the same file by fune build
from .payroll_statutory_sick_pay_types import SicknessPeriod, SickPay, SickPayWeek

# Periods of incapacity separated by no more than 8 weeks are one period (SSCBA 1992 s.152(3)).
LINK_GAP_DAYS = 56
# 28 weeks' entitlement, counted in 1/420ths of a week so any 1-7 qualifying-day week divides exactly.
WEEK_UNITS = 420
MAX_UNITS = 28 * WEEK_UNITS


def _rule_for(date: str) -> int:
    for i, rule in enumerate(SSP_RULES):
        if date >= rule.valid_from and (rule.valid_to is None or date <= rule.valid_to):
            return i
    if SSP_RULES_HISTORY != "full":
        earliest = min((r.valid_from for r in SSP_RULES), default=None)
        if earliest is not None and date < earliest:
            raise ValueError(
                "no SSP rule for %s: this build was installed with history=%s, so it only carries rules from %s. "
                "Reinstall with history=full for older sickness." % (date, SSP_RULES_HISTORY, earliest)
            )
    raise ValueError("no SSP rule for %s" % (date,))


def statutory_sick_pay(periods: Sequence[SicknessPeriod], qualifying_weekdays: Sequence[int], normal_weekly_earnings: Money) -> SickPay:
    """Statutory Sick Pay due for a run of sickness spells.

    Every day is judged by the rules in force on that day, which is how the
    6 April 2026 reforms land: a spell straddling that date serves waiting days
    before it and is paid from the first qualifying day after it.
    """
    if len(qualifying_weekdays) == 0:
        raise ValueError("there must be at least one qualifying day a week")
    qualifying = set()
    for d in qualifying_weekdays:
        if isinstance(d, bool) or not isinstance(d, int) or d < 1 or d > 7:
            raise ValueError("qualifying weekdays must be 1 (Monday) to 7 (Sunday), received %s" % (d,))
        if d in qualifying:
            raise ValueError("qualifying weekday %s is listed twice" % (d,))
        qualifying.add(d)
    per_week = len(qualifying)
    if normal_weekly_earnings.currency != "GBP":
        raise ValueError("SSP is paid in GBP, received %s" % (normal_weekly_earnings.currency,))
    if normal_weekly_earnings.minor < 0:
        raise ValueError("normalWeeklyEarnings must not be negative")

    lines: List[dict] = []

    def line_for(week_start: str, rule: int) -> dict:
        for line in lines:
            if line["week_start"] == week_start and line["rule"] == rule:
                return line
        line = {"week_start": week_start, "rule": rule, "paid": 0, "waiting": 0}
        lines.append(line)
        return line

    previous_end: Optional[str] = None
    last_piw_end: Optional[str] = None
    series_end = ""
    waiting_served = 0
    units_used = 0
    exhausted = False

    for period in periods:
        parse_iso_date(period.start)
        parse_iso_date(period.end)
        if period.end < period.start:
            raise ValueError("sickness period %s to %s ends before it starts" % (period.start, period.end))
        if previous_end is not None and period.start <= previous_end:
            raise ValueError("sickness periods must be in date order and must not overlap")
        previous_end = period.end

        # Too short to be a period of incapacity under the rules of its first
        # day: it is not paid and does not link anything.
        length = days_between(period.start, period.end) + 1
        if length < SSP_RULES[_rule_for(period.start)].minimum_period_days:
            continue

        if last_piw_end is None or days_between(last_piw_end, period.start) - 1 > LINK_GAP_DAYS:
            # A new period of entitlement: waiting days and the 28 weeks start
            # again, and it ends 3 years on at the latest (SSP (General) Regs 1982 reg 3(3)).
            waiting_served = 0
            units_used = 0
            series_end = add_months(period.start, 36)
        last_piw_end = period.end

        day = period.start
        while day <= period.end:
            weekday = day_of_week(day)
            current = day
            day = add_days(day, 1)
            if weekday not in qualifying:
                continue
            index = _rule_for(current)
            rule = SSP_RULES[index]
            if rule.lower_earnings_limit is not None and normal_weekly_earnings.minor < rule.lower_earnings_limit:
                continue
            week_start = add_days(current, -(weekday % 7))
            if waiting_served < rule.waiting_days:
                waiting_served += 1
                line_for(week_start, index)["waiting"] += 1
                continue
            units = WEEK_UNITS // per_week
            if current >= series_end or units_used + units > MAX_UNITS:
                exhausted = True
                continue
            units_used += units
            line_for(week_start, index)["paid"] += 1

    lines.sort(key=lambda l: (l["week_start"], l["rule"]))
    weeks = [
        SickPayWeek(
            week_start=l["week_start"],
            paid_days=l["paid"],
            waiting_days=l["waiting"],
            amount=money(_week_amount(SSP_RULES[l["rule"]], normal_weekly_earnings.minor, l["paid"], per_week), "GBP"),
        )
        for l in lines
    ]
    return SickPay(
        total=money(sum(w.amount.minor for w in weeks), "GBP"),
        paid_days=sum(w.paid_days for w in weeks),
        waiting_days=sum(w.waiting_days for w in weeks),
        weeks=weeks,
        entitlement_exhausted=exhausted,
    )


def _week_amount(rule: SspRule, earnings: int, days: int, per_week: int) -> int:
    """SSP for some days of one week: the weekly rate times days over
    qualifying days in the week, rounded up to the penny, as HMRC's daily-rate
    tables do. The weekly rate is kept exact (in ten-thousandths of a penny) so
    that 80% of earnings is never rounded twice."""
    rate = rule.weekly_rate * 10000
    if rule.earnings_cap_basis_points is not None:
        rate = min(rate, earnings * rule.earnings_cap_basis_points)
    return round_div(rate * days, 10000 * per_week, "up")