dodil/erp-brick-sales-order
DODIL ERP brick: sales order
src/brick_sales_order/domain/rebate.py
"""The rebate agreement aggregate (SO-32): what a customer earns back on what they buy, agreed before it is earned.
An agreement names a customer (or a price group), a basis (products, categories or all), a validity period and a scale of tiers; it
is opened as a draft, activated from a day AFTER today, and its scale cannot change once an accrual exists: the agreement the customer
was shown is the agreement that settles. Accruals and settlements (SO-33, SO-34) append to the same aggregate, so a scale change and a
first accrual racing for the agreement meet on its version and one of them is told to retry.
"""
import json
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from typing import Optional
from brick_sales_order.domain.errors import GateUnmet, OutOfBounds, RuleBroken
from brick_sales_order.domain.readiness import no, ok, require
from brick_sales_order.domain.values import ZERO, money, num
D = Decimal
BASES = ("products", "categories", "all")
@dataclass
class RebateAccrual:
"""What one invoice or credit-note line added to the agreement's accrual, keyed (agreement, document, line): written once."""
rebate_agreement_id: str
document_id: str
line_no: int
kind: str = "invoice" # invoice | credit_note
issue_date: Optional[date] = None
basis_amount: Decimal = ZERO # signed: a credit note takes from the basis
accrual_amount: Decimal = ZERO
tax_category: str = "" # the billed line's tax category: a settlement's credit note has a line per category
journal_id: str = "" # derived from (source document, event): the same document never posts twice
posting_status: str = "pending" # posted | pending | reported
@dataclass
class RebateSettlement:
"""What a customer is paid back at one point: the rate the scale reached on the whole basis so far (retroactive), less what was
settled before. Keyed (agreement, settlement_no); asked by one person, approved by another, credited once in invoicing."""
rebate_agreement_id: str
settlement_no: int
kind: str = "partial" # partial | final
basis_amount: Decimal = ZERO
rate_pct: Decimal = ZERO
earned_amount: Decimal = ZERO
settled_before: Decimal = ZERO
amount: Decimal = ZERO
true_up_amount: Decimal = ZERO # final only: what was earned less what was accrued (signed)
requested_by_email: str = ""
requested_by_sub: str = ""
approved_by_email: str = ""
approved_by_sub: str = ""
credit_id: str = ""
status: str = "requested" # requested | approved | credited
reason: str = ""
lines_json: str = "[]" # the credit note's lines: [{tax_category, net_amount}] adding up to amount exactly
requested_on: Optional[date] = None
approved_on: Optional[date] = None
@property
def lines(self) -> list:
return json.loads(self.lines_json or "[]")
@dataclass
class RebateAgreement:
rebate_agreement_id: str
bp_id: int = 0 # the customer, or 0 when the agreement names a price group
price_group: str = ""
basis: str = "all"
basis_values_json: str = "[]"
valid_from: Optional[date] = None
valid_to: Optional[date] = None
scale_json: str = "[]"
retroactive: bool = True
accrual_rate_pct: Decimal = ZERO
currency: str = ""
lifecycle: str = "draft" # draft | active | ended | settled | cancelled
set_by_email: str = ""
set_by_sub: str = ""
accruals: list = field(default_factory=list)
settlements: list = field(default_factory=list)
@property
def basis_values(self) -> list:
return json.loads(self.basis_values_json or "[]")
@property
def scale(self) -> list:
return json.loads(self.scale_json or "[]")
# ── making one ────────────────────────────────────────────────────────────────────────────
@classmethod
def open(cls, rebate_agreement_id: str, *, bp_id: int, price_group: str, basis: str, basis_values: list, valid_from: date,
valid_to: date, scale: list, accrual_rate_pct: Decimal, currency: str, by_email: str, by_sub: str,
retroactive: bool = True) -> "RebateAgreement":
"""A draft agreement. It names a customer or a price group (not both), a basis and a period; the scale is checked when it is activated
(gate scale_valid), so a draft can be worked on."""
if bool(bp_id) == bool(price_group):
raise RuleBroken("an agreement names a customer or a price group, one of them")
if not (D(0) <= D(accrual_rate_pct) <= D(100)):
raise OutOfBounds("an accrual rate is between 0 and 100 percent")
if basis not in BASES:
raise RuleBroken(f"the basis is one of {', '.join(BASES)}")
if (basis == "all") != (not basis_values):
raise RuleBroken("a basis of products or categories names which; a basis of all names none")
if valid_to < valid_from:
raise RuleBroken("an agreement ends on or after the day it starts")
a = cls(rebate_agreement_id, int(bp_id or 0), price_group, basis, json.dumps(sorted(set(basis_values))), valid_from, valid_to,
json.dumps(_scale(scale)), retroactive, accrual_rate_pct, currency, "draft", by_email, by_sub)
return a
# ── the gates ─────────────────────────────────────────────────────────────────────────────
def activate_gates(self, today: date) -> list:
gates = [ok("starts_after_today") if self.valid_from > today else
no("starts_after_today", f"an agreement starts after today ({today}): it cannot start on {self.valid_from}",
"open a new agreement from a later date")]
problems = scale_problems(self.scale)
gates.append(no("scale_valid", "; ".join(problems), "set the scale: tiers ascending, rates from 0 to 100 percent") if problems else ok("scale_valid"))
return gates
def activate(self, today: date) -> bool:
"""Draft to active; a repeat changes nothing. Invoices from its start accrue against it."""
if self.lifecycle == "active":
return False
if self.lifecycle != "draft":
raise RuleBroken(f"the agreement is {self.lifecycle}: only a draft is activated")
require(self.activate_gates(today))
self.lifecycle = "active"
return True
def change_scale(self, scale: list) -> None:
"""The scale of a draft, or of an active agreement nothing has accrued on yet. After the first accrual it is the agreement the
customer was shown: end it and open a new one from a later date."""
if self.lifecycle not in ("draft", "active"):
raise RuleBroken(f"the agreement is {self.lifecycle}: its scale is fixed")
if self.accruals:
raise RuleBroken("an accrual exists on this agreement: end it and open a new agreement from a later date")
new = _scale(scale)
problems = scale_problems(new)
if problems:
raise GateUnmet([no("scale_valid", "; ".join(problems), "tiers ascending, rates from 0 to 100 percent")])
self.scale_json = json.dumps(new)
def cancel(self) -> bool:
"""A draft agreement is cancelled; a repeat changes nothing. An active one is ended, never cancelled."""
if self.lifecycle == "cancelled":
return False
if self.lifecycle != "draft":
raise RuleBroken(f"the agreement is {self.lifecycle}: only a draft is cancelled")
self.lifecycle = "cancelled"
return True
# ── what accrued, what was settled ────────────────────────────────────────────────────────
def basis_total(self) -> Decimal:
return sum((x.basis_amount for x in self.accruals), ZERO)
def accrued_total(self) -> Decimal:
return sum((x.accrual_amount for x in self.accruals), ZERO)
def settled_total(self) -> Decimal:
"""What approved settlements have paid out (their credit notes debit the accrued-rebates account)."""
return sum((s.amount for s in self.settlements if s.status in ("approved", "credited")), ZERO)
def balance(self) -> Decimal:
"""The agreement's accrual account: accrued, plus the final true-up, less what was credited. Exactly 0.00 once settled."""
trued = sum((s.true_up_amount for s in self.settlements if s.status in ("approved", "credited")), ZERO)
return self.accrued_total() + trued - self.settled_total()
def earned(self, basis: Decimal) -> tuple:
"""(rate_pct, earned) on `basis`: retroactive pays the tier reached on the whole basis; otherwise each slice at its own tier."""
scale = sorted(self.scale, key=lambda t: D(str(t["from_value"])))
if basis <= 0 or not scale:
return ZERO, ZERO
reached = [t for t in scale if D(str(t["from_value"])) <= basis]
if not reached:
return ZERO, ZERO
if self.retroactive:
rate = D(str(reached[-1]["rate_pct"]))
return rate, money(basis * rate / 100)
total = ZERO
for i, t in enumerate(reached):
upper = D(str(reached[i + 1]["from_value"])) if i + 1 < len(reached) else basis
total += (upper - D(str(t["from_value"]))) * D(str(t["rate_pct"])) / 100
return D(str(reached[-1]["rate_pct"])), money(total)
def accrue(self, found: list, rate_pct: Decimal) -> list:
"""Add the accrual of each billed line not yet accrued, keyed (agreement, document, line): written once, a re-run adds nothing.
`found`: [{document_id, line_no, kind, issue_date, net_amount, tax_category}]. A credit note takes from the basis."""
if self.lifecycle not in ("active", "ended"):
raise RuleBroken(f"the agreement is {self.lifecycle}: only an active or ended agreement accrues")
have = {(x.document_id, x.line_no) for x in self.accruals}
added = []
for f in found:
key = (str(f["document_id"]), int(f["line_no"]))
if key in have:
continue
sign = -1 if f["kind"] == "credit_note" else 1
basis = money(sign * abs(D(str(f["net_amount"]))))
row = RebateAccrual(self.rebate_agreement_id, key[0], key[1], f["kind"], f["issue_date"], basis, money(basis * rate_pct / 100),
f.get("tax_category") or "")
self.accruals.append(row)
have.add(key)
added.append(row)
if self.accrued_total() > max(self.basis_total(), ZERO):
raise RuleBroken("accruals would exceed the agreement's basis: the accrual rate is a percentage of what was invoiced")
return added
def end(self, today: date) -> bool:
"""Active to ended once valid_to is before today; a repeat changes nothing."""
if self.lifecycle in ("ended", "settled"):
return False
if self.lifecycle != "active":
raise RuleBroken(f"the agreement is {self.lifecycle}: only an active agreement ends")
if not self.valid_to < today:
raise GateUnmet([no("valid_to_passed", f"the agreement runs to {self.valid_to}: it ends the day after", "wait until valid_to is before today")])
self.lifecycle = "ended"
return True
def _lines_for(self, amount: Decimal) -> list:
"""The credit note's lines: one per tax category of the basis, shares by category net, the remainder on the largest, adding up to
`amount` exactly."""
by: dict = {}
for x in self.accruals:
by[x.tax_category] = by.get(x.tax_category, ZERO) + x.basis_amount
by = {k: v for k, v in by.items() if v > 0} or {"": D(1)}
total, out, left = sum(by.values(), ZERO), [], amount
for k in sorted(by, key=lambda c: (-by[c], c)):
share = money(amount * by[k] / total)
out.append({"tax_category": k, "net_amount": num(share, 2)})
left -= share
out[0]["net_amount"] = num(D(out[0]["net_amount"]) + left, 2)
return out
def request_settlement(self, kind: str, reason: str, *, by_email: str, by_sub: str, today: date) -> RebateSettlement:
"""A settlement draft: the rate reached on the basis so far, less what earlier settlements paid. One open request at a time."""
want = "active" if kind == "partial" else "ended"
if self.lifecycle != want:
raise RuleBroken(f"the agreement is {self.lifecycle}: a {kind} settlement is asked while it is {want}")
if not self.bp_id:
raise RuleBroken("a price-group agreement settles per customer: name the customer on the agreement")
if not reason.strip():
raise RuleBroken("a settlement request needs a reason")
if any(s.status == "requested" for s in self.settlements):
raise RuleBroken("a settlement is already waiting for approval: approve it first")
if kind == "final" and any(s.kind == "final" for s in self.settlements):
raise RuleBroken("the final settlement was already asked")
basis = self.basis_total()
rate, earned = self.earned(basis)
before = self.settled_total()
amount = earned - before
if amount < 0 or (amount == 0 and kind == "partial"):
raise GateUnmet([no("something_to_settle", f"earned {num(earned, 2)} on {num(basis, 2)}, already settled {num(before, 2)}: nothing to pay",
"wait for more invoices")])
true_up = (earned - self.accrued_total()) if kind == "final" else ZERO
highest = max((s.settlement_no for s in self.settlements), default=0) # a child number inside one aggregate: the version (and the claim) guards it
st = RebateSettlement(self.rebate_agreement_id, highest + 1, kind, basis, rate, earned, before, amount, true_up, by_email, by_sub,
reason=reason.strip(), lines_json=json.dumps(self._lines_for(amount) if amount > 0 else []), requested_on=today)
self.settlements.append(st)
return st
def settlement(self, settlement_no: int) -> RebateSettlement:
st = next((s for s in self.settlements if s.settlement_no == int(settlement_no)), None)
if st is None:
raise RuleBroken(f"the agreement has no settlement {settlement_no}")
return st
def approve_settlement(self, settlement_no: int, kind: str, *, ids: tuple, email: str, sub: str, today: date) -> tuple:
"""Approve by a second person: neither the requester nor the agreement's author, by email OR sign-in subject. Returns (settlement,
newly): a repeat finds it approved and changes nothing."""
st = self.settlement(settlement_no)
if st.kind != kind:
raise RuleBroken(f"settlement {settlement_no} is a {st.kind} settlement")
if st.status != "requested":
return st, False
mine = {x.strip().lower() for x in ids if x and x.strip()}
theirs = {x.strip().lower() for x in (st.requested_by_email, st.requested_by_sub, self.set_by_email, self.set_by_sub) if x and x.strip()}
if mine & theirs:
raise GateUnmet([no("approver_not_requester", "a settlement is never approved by the person who asked for it or opened the agreement",
"ask another holder of so:rebate:approve")])
st.status, st.approved_by_email, st.approved_by_sub, st.approved_on = "approved", email, sub, today
return st, True
def mark_credited(self, settlement_no: int, credit_id: str) -> None:
st = self.settlement(settlement_no)
st.credit_id, st.status = credit_id, "credited"
if st.kind == "final":
self.lifecycle = "settled"
def mark_posting(self, document_id: str, status: str) -> None:
for x in self.accruals:
if x.document_id == document_id:
x.posting_status = status
def view(self) -> dict:
return {"rebate_agreement_id": self.rebate_agreement_id, "bp_id": self.bp_id or None, "price_group": self.price_group, "basis": self.basis,
"basis_values": self.basis_values, "valid_from": self.valid_from, "valid_to": self.valid_to, "scale": self.scale,
"retroactive": self.retroactive, "accrual_rate_pct": num(self.accrual_rate_pct), "currency": self.currency, "lifecycle": self.lifecycle,
"set_by_email": self.set_by_email, "accruals": len(self.accruals),
"basis_total": num(self.basis_total(), 2), "accrued_total": num(self.accrued_total(), 2), "settled_total": num(self.settled_total(), 2),
"accrual_balance": num(self.balance(), 2),
"settlements": [{k: (num(v, 0 if k == "rate_pct" else 2) if isinstance(v, Decimal) else v) for k, v in s.__dict__.items() if k != "lines_json"}
| {"lines": s.lines} for s in self.settlements]}
def _scale(scale: list) -> list:
for t in scale:
if D(str(t["from_value"])) >= Decimal("1e16"):
raise OutOfBounds(f"a tier cannot start at {t['from_value']}: an amount is below 10,000,000,000,000,000.00")
return [{"from_value": num(D(str(t["from_value"])), 2), "rate_pct": num(D(str(t["rate_pct"])))} for t in scale]
def scale_problems(scale: list) -> list:
"""Why a scale is not valid: tiers strictly ascending from zero or more, rates between 0 and 100 percent."""
if not scale:
return ["the scale has no tier"]
out, last = [], None
for i, t in enumerate(scale, 1):
frm, rate = D(str(t["from_value"])), D(str(t["rate_pct"]))
if frm < 0:
out.append(f"tier {i} starts below zero")
if last is not None and frm <= last:
out.append(f"tier {i} does not start above the tier before it")
if not (D(0) <= rate <= D(100)):
out.append(f"tier {i}'s rate is not between 0 and 100 percent")
last = frm
return out