dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/domain/document_tax.py
"""The DocumentTax aggregate: the tax on one selling document (an order, an invoice, a credit note), from determination to
freezing (TAX-06, 08, 09, 10, 11).
The document (sales-order, ar) owns its own copy of the answer; this lifecycle is the tax on it: computed -> frozen.
determine each line's code from the rule valid on the tax point, the rate valid on it, the rounding method
override a person changes a line's code with a reason; one that LOWERS tax waits for a second person
suggest the AI proposes a code for a line no rule covers; a person accepts or rejects (never applied by itself)
freeze at issue: never recomputed; a rate change afterwards changes nothing ("issued: credit it instead")
Pure: the caller (the use case) hands in what the rules, rates, registration and settings say on the tax point.
"""
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import ROUND_HALF_UP, Decimal
from typing import Optional
from brick_tax.domain import rounding, rules, schedule
from brick_tax.domain.errors import Gate, GateUnmet, RuleBroken, unmet
from brick_tax.domain.events import (
CodeSuggested,
LineOverridden,
OverrideApproved,
SuggestionDecided,
TaxDetermined,
TaxFrozen,
)
from brick_tax.domain.values import CodeState, DocState, Settings, Treatment
P2 = Decimal("0.01")
RULE, OVERRIDE, SUGGESTION, NOT_REGISTERED, PINNED = "rule", "override", "suggestion", "not_registered", "pinned"
NONE, WAITING, APPROVED = "none", "waiting", "approved"
OPEN, ACCEPTED, REJECTED = "open", "accepted", "rejected"
@dataclass(frozen=True)
class CodeFacts:
"""A code as it stands on a tax point: its meaning and the rate valid then (None: it has none yet)."""
tax_code: str
treatment: str
lifecycle: str
description: str = ""
wording: str = ""
reporting_box: str = ""
rate_pct: Optional[Decimal] = None
charged_pct: Optional[Decimal] = None
direction: str = "sale"
input_reporting_box: str = ""
value_box: str = ""
reclaimable_pct: Optional[Decimal] = None
attribution: str = ""
eu_goods: bool = False
@property
def usable(self) -> bool:
return self.lifecycle == CodeState.ACTIVE.value and self.rate_pct is not None
@dataclass
class Context:
"""What the rules, rates, registration and settings say on one tax point."""
tax_point: date
registered: Optional[bool] # None: nothing recorded
treatment_code: Optional[str]
rules: list
codes: dict # tax_code -> CodeFacts
settings: Settings
prices_include_tax: Optional[bool] = None
direction: str = "sale" # sale | purchase: the rules and codes read, and whose treatment `treatment_code` is
ship_to_country: str = "" # v4: where the goods go (ISO alpha-2): decides the place of supply of goods lines
pins: dict = field(default_factory=dict) # v4: line_no -> (CodeFacts of the earlier frozen line, "pinned to <invoice_id>")
@dataclass
class LineIn:
line_no: int
description: str = ""
category_code: str = ""
supply: str = "any"
amount: Decimal = Decimal(0)
quantity: Optional[Decimal] = None
unit_price: Optional[Decimal] = None
pinned_from: str = "" # v4: an earlier frozen document whose code and rate this line takes
pinned_code: str = ""
consumption: str = "" # v6 (TAX-51): eat_in | takeaway | "" (not said)
@dataclass
class DocLine:
line_no: int
description: str = ""
category_code: str = ""
supply: str = "any"
amount: Decimal = Decimal(0)
quantity: Optional[Decimal] = None
unit_price: Optional[Decimal] = None
tax_code: str = ""
treatment: str = ""
rate_pct: Optional[Decimal] = None
charged_pct: Optional[Decimal] = None
wording: str = ""
reporting_box: str = ""
source: str = "" # rule | override | suggestion | not_registered
rule_ref: str = "" # the rule that chose the code, in words
rule_code: str = "" # the code the rule chose (the baseline an override is measured against)
refusal: str = ""
override_reason: str = ""
override_by: str = ""
override_state: str = NONE
override_approved_by: str = ""
override_approval_reason: str = ""
lowers_tax: bool = False
suggestion_state: str = ""
override_by_sub: str = "" # the overrider's sub beside their email (override_by)
reclaimable_pct: Optional[Decimal] = None # a purchase line: its code's reclaimable share, as it stood
input_reporting_box: str = ""
value_box: str = ""
pinned_from: str = ""
attribution: str = ""
eu_goods: bool = False
consumption: str = "" # v6 (TAX-51): part of the frozen line
@property
def resolved(self) -> bool:
return bool(self.tax_code) and not self.refusal
@dataclass
class Suggestion:
"""One AI suggestion, kept with its inputs, the model, the output, the reason and who decided, and when."""
line_no: int
seq: int
tax_code: str
reason: str = ""
confidence: Optional[int] = None
model: str = ""
inputs: str = ""
state: str = OPEN
created_by: str = ""
decided_by: str = ""
decided_at: Optional[datetime] = None
decision_reason: str = ""
@dataclass
class DocumentTax:
document_ref: str
bp_id: Optional[int] = None
tax_point: Optional[date] = None
currency: str = "GBP"
treatment_code: str = ""
status: str = DocState.COMPUTED.value
rounding_method: str = "per_line_nearest"
prices_include_tax: bool = False
registered: bool = False
vat_number: str = ""
title: str = "Invoice"
fx_rate: Optional[Decimal] = None
vat_total_gbp: Optional[Decimal] = None
frozen_by: str = ""
frozen_at: Optional[datetime] = None
direction: str = "sale"
ship_to_country: str = ""
self_billed: bool = False
due_date: Optional[date] = None
sealed: Optional[rounding.Computed] = None # the tax AS ISSUED, stored at freeze: an issued document is never recomputed
lines: list = field(default_factory=list)
suggestions: list = field(default_factory=list)
events: list = field(default_factory=list, repr=False, compare=False)
# ── determination (TAX-10) ────────────────────────────────────────────────────────────────────────────────
@classmethod
def determine(cls, document_ref: str, lines_in: list, ctx: Context, *, bp_id: Optional[int], currency: str,
fx_rate: Optional[Decimal], vat_number: str, previous: Optional["DocumentTax"], by: str) -> "DocumentTax":
"""Determine every line's code and compute the document's tax. Run twice with the same inputs, the same answer.
A person's decisions (an override, a suggestion) on a line number survive a recompute; a frozen document does not
recompute at all."""
if previous is not None and previous.status == DocState.FROZEN.value:
raise RuleBroken(f"the tax on {document_ref} is frozen at issue and never recomputed: issued: credit it instead")
if not lines_in:
raise RuleBroken("a document has at least one line")
if len({ln.line_no for ln in lines_in}) != len(lines_in):
raise RuleBroken("two lines share a line number")
if ctx.registered is None:
raise unmet("registration_known", "nothing is recorded about whether the business is VAT registered on this tax point",
"Tax settings > Registration")
purchase = ctx.direction == "purchase"
if (ctx.registered or purchase) and not ctx.treatment_code:
raise unmet("supplier_treatment" if purchase else "customer_treatment",
"the supplier has no tax treatment" if purchase else "the customer has no tax treatment",
"Supplier tax" if purchase else "Customer tax")
inclusive = ctx.settings.prices_include_tax if ctx.prices_include_tax is None else ctx.prices_include_tax
doc = cls(document_ref, bp_id, ctx.tax_point, currency, ctx.treatment_code or "", DocState.COMPUTED.value,
ctx.settings.rounding_method, bool(inclusive), bool(ctx.registered),
vat_number if ctx.registered else "", "Bill" if purchase else ("VAT invoice" if ctx.registered else "Invoice"), fx_rate)
doc.direction, doc.ship_to_country = ctx.direction, ctx.ship_to_country
before = {ln.line_no: ln for ln in (previous.lines if previous else [])}
if previous:
doc.suggestions = [s for s in previous.suggestions if s.line_no in {ln.line_no for ln in lines_in}]
for li in sorted(lines_in, key=lambda x: x.line_no):
line = DocLine(li.line_no, li.description, li.category_code, li.supply, li.amount, li.quantity, li.unit_price)
line.pinned_from, line.consumption = li.pinned_from, li.consumption
doc._choose(line, ctx)
old = before.get(li.line_no)
if old is not None and ctx.registered:
doc._keep_decision(line, old, ctx)
doc.lines.append(line)
doc.vat_total_gbp = doc._gbp()
doc.events.append(TaxDetermined(document_ref, by))
return doc
def _facts_into(self, line: DocLine, f: CodeFacts, source: str) -> None:
line.tax_code, line.treatment, line.rate_pct, line.charged_pct = f.tax_code, f.treatment, f.rate_pct, f.charged_pct
line.wording, line.reporting_box, line.source = f.wording, f.reporting_box, source
line.reclaimable_pct, line.input_reporting_box, line.value_box = f.reclaimable_pct, f.input_reporting_box, f.value_box
line.attribution, line.eu_goods = f.attribution, f.eu_goods
def _choose(self, line: DocLine, ctx: Context) -> None:
pin = ctx.pins.get(line.line_no)
if pin is not None: # TAX-47: a line pinned to an earlier frozen document takes ITS code and rate
self._facts_into(line, pin[0], PINNED)
line.rule_ref, line.rule_code = pin[1], pin[0].tax_code
return
if not ctx.registered and ctx.direction == "sale":
nv = sorted((f for f in ctx.codes.values() if f.treatment == Treatment.NOT_REGISTERED.value and f.usable),
key=lambda f: f.tax_code)
if not nv:
raise unmet("nv_code", "a business that is not registered needs an active 'no VAT' code (NV)", "Tax codes")
self._facts_into(line, nv[0], NOT_REGISTERED)
line.rule_ref = "not VAT registered on the tax point"
return
treatment = ctx.treatment_code
if ctx.ship_to_country and ctx.direction == "sale" and line.supply == "goods" and ctx.ship_to_country.upper() != "GB":
treatment = {"uk_business": "overseas_business", "uk_consumer": "overseas_consumer"}.get(treatment or "", treatment)
if not line.consumption and rules.distinguishes(ctx.rules, line.category_code, treatment, line.supply, ctx.tax_point):
raise unmet("consumption_given", f"line {line.line_no}: say whether it is eaten in or taken away", "send consumption eat_in or takeaway")
rule = rules.pick(ctx.rules, line.category_code, treatment, line.supply, ctx.tax_point, line.consumption)
if rule is None:
line.refusal = f"no tax rule for ({line.category_code or 'no category'}, {treatment})"
return
f = ctx.codes.get(rule.tax_code)
line.rule_ref, line.rule_code = rule.describe(), rule.tax_code
if f is not None and f.direction != ctx.direction:
line.refusal = f"the rule chooses {rule.tax_code}, which is a {f.direction} code: a {ctx.direction} takes {ctx.direction} codes only"
return
if f is None or not f.usable:
why = "does not exist" if f is None else (f"is {f.lifecycle}" if f.lifecycle != CodeState.ACTIVE.value
else f"has no rate on {ctx.tax_point}")
line.refusal = f"the rule chooses {rule.tax_code}, which {why}"
return
self._facts_into(line, f, RULE)
def _keep_decision(self, line: DocLine, old: DocLine, ctx: Context) -> None:
"""A person decided about this line before: the decision stands on the recompute."""
line.suggestion_state = old.suggestion_state
if old.override_state == NONE or old.source not in (OVERRIDE, SUGGESTION):
return
f = ctx.codes.get(old.tax_code)
if f is None or not f.usable:
line.refusal = f"the code {old.tax_code} chosen for this line is no longer usable"
return
self._facts_into(line, f, old.source)
line.override_reason, line.override_by, line.override_by_sub = old.override_reason, old.override_by, old.override_by_sub
line.lowers_tax = _lowers(f, line.rule_code, ctx)
same_line = ((old.amount, old.category_code, old.supply, old.description)
== (line.amount, line.category_code, line.supply, line.description))
if not line.lowers_tax:
line.override_state = APPROVED
elif old.lowers_tax and old.override_state == APPROVED and same_line: # an approval is for THIS line, not a changed one
line.override_state, line.override_approved_by = APPROVED, old.override_approved_by
line.override_approval_reason = old.override_approval_reason
else:
line.override_state = WAITING
# ── a person's decisions (TAX-06, TAX-09) ─────────────────────────────────────────────────────────────────
def line(self, line_no: int) -> DocLine:
found = next((ln for ln in self.lines if ln.line_no == line_no), None)
if found is None:
raise RuleBroken(f"{self.document_ref} has no line {line_no}")
return found
def _editable(self) -> None:
if self.status == DocState.FROZEN.value:
raise RuleBroken(f"the tax on {self.document_ref} is frozen at issue: issued: credit it instead")
def override(self, line_no: int, facts: Optional[CodeFacts], reason: str, ctx: Context, *, by: str, source: str = OVERRIDE,
by_sub: str = "") -> DocLine:
"""Change a line's code, with a reason and who. Where it lowers the tax it waits for approval."""
self._editable()
line = self.line(line_no)
if not self.registered:
raise RuleBroken("this business is not VAT registered on the tax point: every line is NV and nothing is overridden")
reason = (reason or "").strip()
if not reason:
raise RuleBroken("an override needs a reason")
if facts is None or not facts.usable:
raise RuleBroken("that code is not an active code with a rate on the tax point")
if facts.tax_code == line.tax_code and line.source == RULE and not line.refusal:
raise RuleBroken(f"{facts.tax_code} is already the code the rule chose")
self._facts_into(line, facts, source)
line.refusal = ""
line.override_reason, line.override_by, line.override_by_sub = reason, by, by_sub
line.override_approved_by = line.override_approval_reason = ""
line.lowers_tax = _lowers(facts, line.rule_code, ctx)
line.override_state = WAITING if line.lowers_tax else APPROVED
self.events.append(LineOverridden(self.document_ref, line_no, facts.tax_code, by))
return line
def approve_override(self, line_no: int, *, approver: str, mode: str, reason: str = "", approver_ids: tuple = ()) -> DocLine:
"""A holder of tax:document_tax:approve_override who is not the person who overrode — unless the owner's control
(override_approval) allows self-approval with a reason."""
self._editable()
line = self.line(line_no)
if line.override_state != WAITING:
raise RuleBroken(f"line {line_no} has no override waiting for approval")
if approver == line.override_by or schedule.same_person((approver, *approver_ids), (line.override_by, line.override_by_sub)):
if mode != "self_with_reason":
raise unmet("override_approved", "an override that lowers tax is approved by a second person, never the one "
"who made it", "ask a holder of tax:document_tax:approve_override")
if not (reason or "").strip():
raise unmet("override_approved", "approving your own override needs a reason", "give the reason")
line.override_state, line.override_approved_by = APPROVED, approver
line.override_approval_reason = (reason or "").strip()
self.events.append(OverrideApproved(self.document_ref, line_no, approver))
return line
def suggest(self, line_no: int, facts: CodeFacts, *, reason: str, confidence: Optional[int], model: str, inputs: str, by: str) -> Suggestion:
"""The AI proposes a code for a line no rule covers. Nothing is applied: the line stays as it is."""
self._editable()
line = self.line(line_no)
if line.resolved and line.source != NOT_REGISTERED and not line.refusal:
raise RuleBroken(f"line {line_no} already has a code: the AI only proposes where no rule covers a line")
if not self.registered:
raise RuleBroken("this business is not VAT registered on the tax point: there is nothing to suggest")
if any(s.line_no == line_no and s.state == OPEN for s in self.suggestions):
raise RuleBroken(f"line {line_no} already has a suggestion waiting for a person")
if not facts.usable:
raise RuleBroken(f"the suggested code {facts.tax_code} is not an active code with a rate on the tax point")
s = Suggestion(line_no, 1 + sum(1 for x in self.suggestions if x.line_no == line_no), facts.tax_code,
reason, confidence, model, inputs, OPEN, by)
self.suggestions.append(s)
line.suggestion_state = OPEN
self.events.append(CodeSuggested(self.document_ref, line_no, facts.tax_code, by))
return s
def decide_suggestion(self, line_no: int, accept: bool, facts: Optional[CodeFacts], ctx: Context, *, by: str, now: datetime,
note: str = "", by_sub: str = "") -> Suggestion:
"""A person accepts (the code becomes an override with the suggestion's reason, the model and who accepted) or
rejects (the line stays refused until a person chooses a code)."""
self._editable()
line = self.line(line_no)
s = next((x for x in self.suggestions if x.line_no == line_no and x.state == OPEN), None)
if s is None:
raise RuleBroken(f"line {line_no} has no suggestion waiting")
s.decided_by, s.decided_at, s.decision_reason = by, now, (note or "").strip()
if accept:
s.state = ACCEPTED
self.override(line_no, facts, f"AI suggestion accepted: {s.reason}", ctx, by=by, source=SUGGESTION, by_sub=by_sub)
else:
s.state = REJECTED
line.suggestion_state = s.state
self.events.append(SuggestionDecided(self.document_ref, line_no, accept, by))
return s
# ── totals ────────────────────────────────────────────────────────────────────────────────────────────────
def complete(self) -> bool:
return all(ln.resolved for ln in self.lines)
def computed(self) -> Optional[rounding.Computed]:
if self.status == DocState.FROZEN.value and self.sealed is not None:
return self.sealed
if not self.complete():
return None
items = [rounding.Item(ln.line_no, ln.tax_code, ln.amount, ln.rate_pct, ln.charged_pct) for ln in self.lines]
return rounding.compute(items, self.rounding_method, self.prices_include_tax)
def _gbp(self) -> Optional[Decimal]:
c = self.computed()
if c is None:
return None
if self.currency == "GBP":
return c.vat
return (c.vat * self.fx_rate).quantize(P2, ROUND_HALF_UP) if self.fx_rate else None
def purchase_split(self) -> Optional[tuple]:
"""A purchase's VAT split by each code's reclaimable share: ({line_no: (reclaimable, non)}, {code: (reclaimable, non)},
(reclaimable, non)). The VAT is what the supplier charged, or — a reverse charge, charged 0 — the VAT due, accounted as
output and as input. Nothing is reclaimable by a business not VAT registered. reclaimable + non-reclaimable is the
VAT exactly, per line, per code and in total. None for a sale or a document not yet complete."""
c = self.computed()
if self.direction != "purchase" or c is None:
return None
def cut(vat: Decimal, due: Decimal, charged_pct: Decimal, rate_pct: Decimal, pct: Optional[Decimal]) -> tuple:
base = due if charged_pct == 0 and rate_pct > 0 else vat
share = Decimal(0) if not self.registered else (Decimal(100) if pct is None else pct)
rec = (base * share / Decimal(100)).quantize(P2, ROUND_HALF_UP)
return rec, base - rec
by_line = {ln.line_no: cut(c.lines[ln.line_no].vat, c.lines[ln.line_no].vat_due, ln.charged_pct, ln.rate_pct, ln.reclaimable_pct)
for ln in self.lines}
pct = {ln.tax_code: ln for ln in self.lines}
by_code = {s.code: cut(s.vat, s.vat_due, s.charged_pct, s.rate_pct, pct[s.code].reclaimable_pct) for s in c.summary}
return by_line, by_code, (sum((v[0] for v in by_code.values()), Decimal(0)), sum((v[1] for v in by_code.values()), Decimal(0)))
def set_fx(self, fx_rate: Optional[Decimal]) -> None:
self.fx_rate = fx_rate
self.vat_total_gbp = self._gbp()
# ── readiness (the gates of determine / freeze) ───────────────────────────────────────────────────────────
def gates(self) -> list:
refused = [f"line {ln.line_no}: {ln.refusal}" for ln in self.lines if ln.refusal]
open_s = [str(s.line_no) for s in self.suggestions if s.state == OPEN]
waiting = [str(ln.line_no) for ln in self.lines if ln.override_state == WAITING]
return [
Gate("rule_covers_pair", not refused, "; ".join(refused) if refused else "every line has a code",
"Tax rules, or override the line"),
Gate("no_open_suggestion", not open_s,
f"a person has not yet accepted or rejected the AI's code on line {', '.join(open_s)}" if open_s else "no suggestion is waiting",
"Document tax > the line"),
Gate("override_approved", not waiting,
f"an override that lowers tax waits for approval on line {', '.join(waiting)}" if waiting else "no override is waiting",
"ask a holder of tax:document_tax:approve_override"),
]
def freeze(self, contents_gates: list, *, by: str, now: datetime) -> bool:
"""At issue: every gate met, then never recomputed. True if this call froze it, False if it already was."""
if self.status == DocState.FROZEN.value:
return False
bad = [g for g in self.gates() + contents_gates if not g.met]
if bad:
raise GateUnmet(bad)
self.sealed = self.computed()
self.status, self.frozen_by, self.frozen_at = DocState.FROZEN.value, by, now
self.vat_total_gbp = self._gbp()
self.events.append(TaxFrozen(self.document_ref, by))
return True
def _lowers(f: CodeFacts, rule_code: str, ctx: Context) -> bool:
"""Does choosing `f` charge less tax than the rule's code, or (no rule covered the line) than the standard code?"""
base = ctx.codes.get(rule_code) if rule_code else None
if base is None or base.charged_pct is None:
standard = [c.charged_pct for c in ctx.codes.values() if c.treatment == Treatment.STANDARD.value and c.usable]
base_pct = max(standard) if standard else None
else:
base_pct = base.charged_pct
return base_pct is not None and f.charged_pct < base_pct