dodil/erp-brick-payroll
DODIL ERP brick: payroll
src/brick_payroll/domain/pay_run.py
"""The pay run and its payslips: the aggregate, its one table of arrows, and gross to net.
A payslip is a PURE function of what hr, time-off and the dated rows answered (module.yaml decisions): calculating again
gives the same payslips. After approval nothing is recalculated - a correction is the next period's (cumulative tax
corrects itself) or an off-cycle run's - so no state is ever re-entered (workflow note)."""
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from decimal import Decimal
from fractions import Fraction
from typing import Optional
from brick_payroll.domain import paye
from brick_payroll.domain.errors import RuleBroken
from brick_payroll.domain.money import MONEY_MAX, ZERO, F, money
# The arrows of the signed workflow (module.yaml workflows: pay_run), and nothing else moves a run. tests/test_rules.py
# checks this table against process.yaml, so a transition cannot be added here without a signature.
MOVES = {
"open_pay_run": (("none",), "open"),
"open_off_cycle_run": (("none",), "open"),
"calculate_pay_run": (("open",), "calculated"),
"recalculate_pay_run": (("calculated",), "calculated"),
"approve_pay_run": (("calculated",), "approved"),
"approve_approver_payslip": (("calculated",), "calculated"),
"record_fps_accepted": (("approved",), "submitted"),
"confirm_pay_run_paid": (("submitted",), "paid"),
"close_pay_run": (("paid",), "closed"),
"cancel_pay_run": (("open", "calculated"), "cancelled"),
}
TERMINAL = ("closed", "cancelled")
FROZEN = ("approved", "submitted", "paid", "closed") # payslips of a run in these states never change
EMPLOYEE_MOVES = {
"take_starter": (("none",), "pending_details"),
"complete_payroll_details": (("pending_details",), "active"),
"issue_p45": (("active",), "left"),
}
@dataclass(frozen=True)
class Identity:
"""Both of a caller's names. The same person under either is the same person (email AND sub are both checked)."""
email: str
sub: str
def same_as(self, other: "Identity") -> bool:
return bool((self.email and other.email and self.email.lower() == other.email.lower())
or (self.sub and other.sub and self.sub == other.sub))
def move(state: str, transition: str, moves: dict = MOVES) -> str:
frm, to = moves[transition]
if state not in frm:
raise RuleBroken(f"{transition.replace('_', ' ')} is not possible from {state} (only from {', '.join(frm)}): "
"an approved run is never edited - a correction goes in the next period or an off-cycle run",
gate="state")
return to
# ── the period, the pay date, the id ───────────────────────────────────────────────────────────
def month_bounds(period_code: str) -> tuple:
try:
y, m = (int(x) for x in period_code.split("-"))
first = date(y, m, 1)
except (ValueError, TypeError):
raise RuleBroken(f"period {period_code!r} is a calendar month like 2026-11", gate="bounds") from None
if not 2000 <= y <= 2100:
raise RuleBroken(f"period {period_code} is outside 2000 .. 2100", gate="bounds")
last = (first.replace(day=28) + timedelta(days=4)).replace(day=1) - timedelta(days=1)
return first, last
def last_working_day(first: date, last: date) -> date:
"""The monthly schedule's default pay day: the last Monday-to-Friday of the month (bank holidays: see AGENTS.md)."""
d = last
while d.weekday() >= 5:
d -= timedelta(days=1)
return d
def week_bounds(period_code: str) -> tuple:
"""An ISO week like 2026-W46 (Monday to Sunday): the period of a weekly schedule, worked in arrears of its pay date."""
try:
y, w = period_code.split("-W")
first = date.fromisocalendar(int(y), int(w), 1)
except (ValueError, TypeError):
raise RuleBroken(f"period {period_code!r} is an ISO week like 2026-W46", gate="bounds") from None
if not 2000 <= first.year <= 2100:
raise RuleBroken(f"period {period_code} is outside 2000 .. 2100", gate="bounds")
return first, first + timedelta(days=6)
def period_dates(frequency: str, period_code: str) -> tuple:
"""(first day, last day, pay date) of a period: a calendar month paid on its last working day, or an ISO week paid the Friday
after it (a week in arrears). Bank holidays are not consulted (AGENTS.md)."""
if frequency == "monthly":
first, last = month_bounds(period_code)
return first, last, last_working_day(first, last)
if frequency == "weekly":
first, last = week_bounds(period_code)
return first, last, last + timedelta(days=5)
raise RuleBroken(f"the {frequency} schedule has no period in hr's payroll extract: monthly and weekly only", gate="schedule")
def period_key(frequency: str, tax_period: int) -> str:
"""M08, W33: the letter of the schedule's frequency and the period's number in the tax year (the period claim and the run id)."""
return f"{paye.PERIOD_LETTER[frequency]}{tax_period:02d}"
def pay_run_id(company_code: str, schedule_code: str, pay_date: date, attempt: int = 1, frequency: str = "monthly") -> str:
"""UK01-M-2026-27-M08, UK01-W-2026-27-W33: the company, the schedule, the tax year and the pay period of the pay date. A period
whose run was cancelled is opened again as -R2, -R3 (the cancelled run keeps its id and its history)."""
base = f"{company_code}-{schedule_code}-{paye.tax_year_label(paye.tax_year_of(pay_date))}-{period_key(frequency, paye.tax_period_of(frequency, pay_date))}"
return base if attempt == 1 else f"{base}-R{attempt}"
def working_days(days: list, lo: date, hi: date) -> int:
"""Days in [lo, hi] hr's work pattern contracts minutes for."""
return sum(1 for d in days if lo <= d["date"] <= hi and int(d.get("contracted_minutes") or 0) > 0)
def bacs_lead_met(today: date, pay_date: date, lead: int) -> bool:
"""Approved at least `lead` working days (Monday to Friday) before the pay date, counted from the business date."""
d, left = today, lead
while left:
d += timedelta(days=1)
if d.weekday() < 5:
left -= 1
return d <= pay_date
def settled(p: "Payslip") -> bool:
"""A payslip whose net pay is dealt with: paid or returned by the bank, or nothing to pay / nobody to pay it to (decided when the journal posted)."""
return p.payout_status in ("paid", "returned", "no_payee", "nothing_to_pay")
def advance(run: "PayRun", fps_accepted: bool, filings_done: bool) -> list:
"""The arrows a run takes by itself once the world has answered, each from its own state and never backwards: HMRC accepted the FPS
(approved -> submitted), every payout settled (submitted -> paid), the journal posted and the pension file sent or its manual task done
(paid -> closed). Returns the arrows taken, in order."""
taken = []
if run.status == "approved" and fps_accepted:
run.move("record_fps_accepted")
taken.append("record_fps_accepted")
if run.status == "submitted" and run.payslips and all(settled(p) for p in run.payslips):
run.move("confirm_pay_run_paid")
taken.append("confirm_pay_run_paid")
if run.status == "paid" and run.journal_status == "posted" and filings_done:
run.move("close_pay_run")
taken.append("close_pay_run")
return taken
# ── the aggregate ─────────────────────────────────────────────────────────────────────────────
@dataclass
class Line:
line_no: int
element_code: str
kind: str # earning | deduction | employer
quantity: Optional[Decimal]
rate_amount: Optional[Decimal]
amount: Decimal # signed: earnings +, a deduction from pay -, deductions -, employer +
source_kind: str # hr_extract | absence | statutory | manual
source_ref: str
rate_rows_used: str = ""
@dataclass
class Payslip:
pay_run_id: str
payroll_id: str
person_id: int
calc_no: int
gross: Decimal = ZERO
taxable: Decimal = ZERO
tax: Decimal = ZERO
ee_ni: Decimal = ZERO
er_ni: Decimal = ZERO
ee_pension: Decimal = ZERO
er_pension: Decimal = ZERO
student_loan: Decimal = ZERO
net: Decimal = ZERO
tax_code_used: str = ""
basis_used: str = ""
ni_category_used: str = ""
ytd: dict = field(default_factory=dict)
cost_centre: str = ""
bp_id: Optional[int] = None
inputs_sha: str = ""
stale: bool = False
stale_reason: str = ""
approved_by: str = ""
approved_sub: str = ""
payout_ref: str = ""
payout_status: str = ""
payout_reason: str = ""
paid_on: Optional[date] = None
payment_id: str = ""
document_key: str = ""
published_on: Optional[date] = None
leaving_date: Optional[date] = None
lines: list = field(default_factory=list)
@dataclass
class PayRun:
pay_run_id: str
company_code: str
schedule_code: str
frequency: str
kind: str # regular | off_cycle
tax_year: str # 2026-27
tax_period: int # 8
period_code: str # 2026-11
period_from: date
period_to: date
pay_date: date
paye_ref: str
status: str = "open"
calc_no: int = 0
opened_by: str = ""
opened_sub: str = ""
prepared_by: str = ""
prepared_sub: str = ""
calculated_at: Optional[datetime] = None
approved_by: str = ""
approved_sub: str = ""
approved_at: Optional[datetime] = None
approved_on: Optional[date] = None
journal_id: Optional[int] = None
journal_status: str = "" # pending | posted | refused (with journal_reason)
journal_reason: str = ""
hr_batch_id: str = ""
cancel_reason: str = ""
payslips: list = field(default_factory=list)
def preparer(self) -> Identity:
return Identity(self.prepared_by, self.prepared_sub)
def move(self, transition: str) -> None:
self.status = move(self.status, transition)
# ── gross to net ──────────────────────────────────────────────────────────────────────────────
@dataclass
class Earning:
element_code: str
quantity: Optional[Decimal]
rate_amount: Optional[Decimal]
amount: Decimal
source_kind: str
source_ref: str
taxable: bool = True
niable: bool = True
pensionable: bool = True
rate_rows: str = ""
@dataclass
class PersonInput:
payroll_id: str
person_id: int
earnings: list
tax_code: paye.TaxCode
basis: str
ni_category: str
scheme: Optional[paye.Scheme]
prev: dict # this employer's figures to date before this run: gross, taxable, tax, ee_ni, er_ni, ee_pension, er_pension
p45_pay: Decimal = ZERO
p45_tax: Decimal = ZERO
cost_centre: str = ""
bp_id: Optional[int] = None
leaving_date: Optional[date] = None
inputs_sha: str = ""
parts: dict = field(default_factory=dict) # the fingerprint of each input (extract, absences, payroll record, rates ...)
notes: dict = field(default_factory=dict) # how statutory pay was worked out (SSP, SMP): shown on the payslip, never a second source of truth
YTD_KEYS = ("gross", "taxable", "tax", "ee_ni", "er_ni", "ee_pension", "er_pension")
def gross_to_net(run: PayRun, inp: PersonInput, rates: paye.Rates) -> Payslip:
"""One payslip, every figure shown: the earnings, then pension (net pay arrangement: before tax), PAYE on taxable pay
to date, Class 1 NICs on gross, net pay. Each line names its source and the rate rows it used."""
rates.used = set()
lines = [Line(i + 1, e.element_code, "earning", e.quantity, e.rate_amount, e.amount, e.source_kind, e.source_ref, e.rate_rows)
for i, e in enumerate(inp.earnings)]
gross = money(sum((F(e.amount) for e in inp.earnings), Fraction(0)))
pensionable = money(sum((F(e.amount) for e in inp.earnings if e.pensionable), Fraction(0)))
niable = money(sum((F(e.amount) for e in inp.earnings if e.niable), Fraction(0)))
taxable_gross = money(sum((F(e.amount) for e in inp.earnings if e.taxable), Fraction(0)))
if gross < 0:
raise RuleBroken(f"{inp.payroll_id}: gross pay would be {gross} - a deduction larger than the pay; carry it to the next period",
gate="bounds")
pen = paye.pension(inp.scheme, run.frequency, pensionable, rates)
pen_rows = sorted(rates.used)
rates.used = set()
taxable = money(F(taxable_gross) - F(pen.employee)) # the net pay arrangement: relief at the top rate
prev_taxable = F(inp.prev.get("taxable", 0)) + F(inp.p45_pay)
prev_tax = F(inp.prev.get("tax", 0)) + F(inp.p45_tax)
t = paye.paye(inp.tax_code, inp.basis, run.frequency, run.tax_period, taxable, money(prev_taxable), money(prev_tax), rates)
tax_rows = sorted(rates.used)
rates.used = set()
ni = paye.class1_ni(inp.ni_category, run.frequency, niable, rates)
ni_rows = sorted(rates.used)
net = money(F(gross) - F(pen.employee) - F(t.tax) - F(ni.employee))
n = len(lines)
tail = [("EE_PENSION", "deduction", -pen.employee, "statutory", f"{inp.scheme.scheme_code} qualifying earnings {pen.pensionable}" if inp.scheme else "", pen_rows),
("TAX", "deduction", -t.tax, "statutory", f"{inp.tax_code.code} {inp.basis} {paye.PERIOD_WORD[run.frequency]} {run.tax_period}", tax_rows),
("EE_NI", "deduction", -ni.employee, "statutory", f"category {inp.ni_category}", ni_rows),
("ER_NI", "employer", ni.employer, "statutory", f"category {inp.ni_category}", ni_rows),
("ER_PENSION", "employer", pen.employer, "statutory", f"{inp.scheme.scheme_code} qualifying earnings {pen.pensionable}" if inp.scheme else "", pen_rows)]
for code, kind, amount, sk, ref, used in tail:
if amount == 0 and code in ("EE_PENSION", "ER_PENSION") and inp.scheme is None:
continue
n += 1
lines.append(Line(n, code, kind, None, None, money(amount), sk, ref, ";".join(used)))
this = {"gross": gross, "taxable": taxable, "tax": t.tax, "ee_ni": ni.employee, "er_ni": ni.employer, "ee_pension": pen.employee,
"er_pension": pen.employer}
ytd = {k: str(money(F(inp.prev.get(k, 0)) + F(this[k]))) for k in YTD_KEYS}
ytd["tax_detail"] = t.detail
ytd["ni_detail"] = ni.detail
ytd["inputs"] = dict(inp.parts)
if inp.scheme is not None:
ytd["pension"] = {"scheme_code": inp.scheme.scheme_code, "qualifying_earnings": str(pen.pensionable)}
if inp.notes:
ytd["statutory"] = {k: v for k, v in inp.notes.items() if k != "warnings"}
ytd["warnings"] = inp.notes.get("warnings", [])
for v in (gross, net, t.tax, ni.employee, ni.employer):
if abs(v) >= MONEY_MAX:
raise RuleBroken(f"{inp.payroll_id}: an amount of {v} is outside NUMERIC(18,2)", gate="bounds")
return Payslip(run.pay_run_id, inp.payroll_id, inp.person_id, run.calc_no, gross=gross, taxable=taxable, tax=t.tax, ee_ni=ni.employee,
er_ni=ni.employer, ee_pension=pen.employee, er_pension=pen.employer, net=net, tax_code_used=inp.tax_code.code,
basis_used=inp.basis, ni_category_used=inp.ni_category, ytd=ytd, cost_centre=inp.cost_centre, bp_id=inp.bp_id,
inputs_sha=inp.inputs_sha, leaving_date=inp.leaving_date, lines=lines)
def salary_line(rate_basis: str, rate_amount: Decimal, frequency: str, slice_days: Optional[int], period_days: Optional[int]) -> tuple:
"""(quantity, period rate, amount) of a salary slice: the period's pay (a year's / 12) x working days in the slice /
working days in the period, rounded ONCE (2,500.00 x 11 / 21 = 1,309.52). A whole period: the period's pay."""
periods = paye.PERIODS.get(frequency)
if periods is None:
raise paye.UnsupportedHere(f"the {frequency} schedule is not calculated", gate="schedule")
if rate_basis == "year":
per = F(rate_amount) / periods
elif rate_basis == "month":
per = F(rate_amount) * 12 / periods
elif rate_basis == "week" and frequency == "weekly":
per = F(rate_amount)
else:
raise paye.UnsupportedHere(f"pay on a rate per {rate_basis} needs approved hours from timesheets (PRL-22), which this brick does not read yet", gate="pay_basis")
if slice_days is None:
return Decimal("1.0000"), money(per), money(per)
if not period_days:
raise RuleBroken("hr's work pattern has no working day in the period, so a part period cannot be prorated", gate="work_pattern")
return Decimal(slice_days), money(per), money(per * Fraction(slice_days, period_days))
# ── the gates before approval ─────────────────────────────────────────────────────────────────
def approval_gates(run: PayRun, me: Identity, my_person_id: Optional[int], today: date, lead: int, stale: dict) -> list:
"""Every gate of approve_pay_run (workflow readiness), met or not, in the spec's order."""
slips = run.payslips
stale_named = sorted(stale)
negative = [p.payroll_id for p in slips if p.net < 0]
own = [p for p in slips if my_person_id is not None and p.person_id == my_person_id]
own_ok = all(p.approved_by or p.approved_sub for p in own)
return [
{"gate": "not_stale", "met": not stale_named, "reason": "; ".join(f"{k}: {v}" for k, v in sorted(stale.items())) if stale_named else "",
"fix": "recalculate"},
{"gate": "nmw_met", "met": True, "reason": "", "fix": "correct the rate in hr"}, # no hourly rate in a slice-A run (salaries only)
{"gate": "net_not_negative", "met": not negative, "reason": f"{', '.join(negative)}: net pay is below 0.00" if negative else "",
"fix": "carry the deduction to the next period"},
{"gate": "bacs_lead", "met": bacs_lead_met(today, run.pay_date, lead),
"reason": "" if bacs_lead_met(today, run.pay_date, lead) else f"the pay date {run.pay_date} is fewer than {lead} working days away",
"fix": "approve now and ask finance for a faster payment, or move the pay date"},
{"gate": "second_person", "met": not me.same_as(run.preparer()), "reason": "" if not me.same_as(run.preparer()) else "you calculated this run",
"fix": "another payroll manager approves it"},
{"gate": "own_payslip_by_peer", "met": own_ok, "reason": "" if own_ok else "your own payslip is in this run and no peer has approved it",
"fix": "another payroll manager approves your payslip first"},
]
# ── the journal and the payouts ───────────────────────────────────────────────────────────────
def journal_lines(slips: list) -> list:
"""ONE journal per run, summed per purpose and cost centre (never per person in the ledger): debit gross pay, employer NI
and employer pension on each cost centre; credit what is owed to HMRC (tax + both NICs), to the scheme (both pension
parts) and to the people (net). A negative sum (a run of refunds) goes on the other side."""
cost: dict = {}
for p in slips:
for purpose, amount in (("gross_pay", p.gross), ("employer_ni", p.er_ni), ("employer_pension", p.er_pension)):
cost[(purpose, p.cost_centre)] = cost.get((purpose, p.cost_centre), ZERO) + amount
owed = {"paye_ni_payable": sum((p.tax + p.ee_ni + p.er_ni for p in slips), ZERO),
"pension_payable": sum((p.ee_pension + p.er_pension for p in slips), ZERO),
"net_pay_payable": sum((p.net for p in slips), ZERO)}
out = []
order = {"gross_pay": 0, "employer_ni": 1, "employer_pension": 2}
for (purpose, cc), amount in sorted(cost.items(), key=lambda kv: (kv[0][1], order[kv[0][0]])):
if amount:
side = "debit" if amount > 0 else "credit"
out.append({"purpose": purpose, **({"cost_centre": cc} if cc else {}), side: f"{abs(amount):.2f}"})
for purpose, amount in owed.items():
if amount:
side = "credit" if amount > 0 else "debit"
out.append({"purpose": purpose, side: f"{abs(amount):.2f}"})
dr = sum((Decimal(x.get("debit", "0")) for x in out), ZERO)
cr = sum((Decimal(x.get("credit", "0")) for x in out), ZERO)
if dr != cr:
raise RuleBroken(f"the run's journal does not balance ({dr} against {cr})", gate="balanced")
return out
def payout_ref(run_id: str, person_id: int, n: int = 1) -> str:
return f"prl:{run_id}:{person_id}:net" + (f":{n}" if n > 1 else "")