"""The Quote aggregate — the rules of a priced offer, and nothing else.
No database, no HTTP, no identity: a rule here holds whoever calls it and however it is stored, which is
what lets every one of them be tested in milliseconds and read by someone who sells for a living.
WHAT IS INSIDE THE AGGREGATE, and why there. A boundary encloses what must change together:
INSIDE versions, their lines, each line's revenue schedule. Sending a version must freeze its
lines in the same act, or "a sent quote cannot change" is a comment, not a rule.
OUTSIDE the customer (a business partner), the product, the price book. A quote holds their IDs and never
loads them. A price changing later must not change a quote already sent — which is why
the price is COPIED onto the line, with where it came from.
The design is a general CRM quote model: magazine-specific fields (issue, placement, event seat, campaigns) are left to an extension, and
the discount gate follows the CRM spec's two thresholds (modules/crm/module.yaml, CRM-06).
THE DISCOUNT THE POLICY READS (0.2.0, the CRM audit's H2). The policy applies to the DEEPEST LINE discount as
well as to the net, whichever is larger: one line at +80% and one at 40% off netted to 20% and was sent, a 40%
discount nobody approved. A price above list is allowed, and never offsets another line's discount — the net is
the sum of each line's discount (zero for a line at or above list) over the list total. A figure is never shown
below zero.
"""
from __future__ import annotations
from dataclasses import dataclass, field, replace
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform.money import Money, pct, totals_by_currency
ZERO_PCT = Decimal("0.00")
def _clamp(p: Decimal) -> Decimal:
"""A percentage shown to a person: never below zero, and never "-0.00"."""
return p if p > 0 else ZERO_PCT
# States are identifiers, UPPERCASE in rows, code and on the wire. EXPIRED is never stored: it is
# derived from the date, so no nightly job can be late about it.
DRAFT, SENT, ACCEPTED, DECLINED, SUPERSEDED, EXPIRED = (
"DRAFT", "SENT", "ACCEPTED", "DECLINED", "SUPERSEDED", "EXPIRED")
class QuoteRuleBroken(Exception):
"""The quote's state forbids this. → 409: try something else with this quote."""
class QuoteIncomplete(QuoteRuleBroken):
"""The quote is not ready — add what is missing, then retry. → 422."""
class NeedsApproval(QuoteRuleBroken):
"""Someone with more authority has to act first. → 403, with the figures in the message, so a
person or an agent reads "18.00% … 15.00%" without a second lookup."""
def __init__(self, message: str, discount_pct: Decimal, threshold_pct: Decimal):
super().__init__(message)
self.discount_pct, self.threshold_pct = discount_pct, threshold_pct
# ── value objects ──────────────────────────────────────────────────────────────────────────
@dataclass(frozen=True, slots=True)
class DiscountPolicy:
"""What a quote may discount — a configuration ROW the deal desk changes without a deploy.
up to `auto_approve_pct` sent by the rep
above it, up to `max_pct` needs an approval for THAT discount
above `max_pct` refused outright, approval or not
"""
auto_approve_pct: Decimal = Decimal("15.00")
max_pct: Decimal = Decimal("30.00")
def __post_init__(self):
object.__setattr__(self, "auto_approve_pct", pct(self.auto_approve_pct))
object.__setattr__(self, "max_pct", pct(self.max_pct))
if not (0 <= self.auto_approve_pct <= self.max_pct <= 100):
raise QuoteRuleBroken("a discount policy needs 0 ≤ auto-approve ≤ maximum ≤ 100")
@dataclass(frozen=True, slots=True)
class PricedAt:
"""Where a line's list price came from, so "why is this the price" is answerable from the row a year
later without re-running a lookup against a price book that has since changed."""
price_book_id: str
priced_on: str # ISO date the price was resolved AS OF
@dataclass(frozen=True, slots=True)
class Approval:
"""A discount approval BOUND TO THE FIGURE it was given for. Recording only who approved would
license any discount forever: approve at 18%, re-price to 29%, send."""
approved_by: str
approved_pct: Decimal
@dataclass(frozen=True, slots=True)
class Line:
"""One sold thing. Immutable: a change produces a new line, so a revision can carry lines forward
with no risk of the old version's lines changing under it."""
line_no: int
product_id: str
quantity: int
list_price: Money
unit_price: Money
priced_at: PricedAt
description: str = ""
revenue_months: tuple = () # 'YYYY-MM' the line total is recognised across
def __post_init__(self) -> None:
if self.quantity < 1:
raise QuoteRuleBroken(f"line {self.line_no}: quantity must be at least 1")
if self.list_price.currency != self.unit_price.currency:
raise QuoteRuleBroken(f"line {self.line_no}: a discount cannot change the currency")
if self.unit_price.amount < 0:
raise QuoteRuleBroken(f"line {self.line_no}: a negative price is not a discount")
months = tuple(sorted(set(self.revenue_months)))
for m in months:
try:
date.fromisoformat(f"{m}-01")
except ValueError:
raise QuoteRuleBroken(f"line {self.line_no}: {m!r} is not a month (YYYY-MM)") from None
object.__setattr__(self, "revenue_months", months)
@property
def currency(self) -> str:
return self.unit_price.currency
@property
def line_total(self) -> Money:
return self.unit_price * self.quantity
@property
def list_total(self) -> Money:
return self.list_price * self.quantity
@property
def discount(self) -> Money:
"""DERIVED, never stored, so it cannot disagree with the two prices it comes from. Zero for a line at
or above list: a markup is allowed, but it is not a negative discount that offsets another line's."""
d = self.list_total - self.line_total
return d if d.amount > 0 else Money.zero(self.currency)
@property
def discount_pct(self) -> Decimal:
"""This line's discount below list, as a percentage — never negative."""
if self.list_total.amount <= 0:
return ZERO_PCT
return _clamp(self.discount.percent_of(self.list_total))
def schedule(self) -> dict:
"""month → amount. The parts sum back to the line total (Money.split); no months, no schedule."""
if not self.revenue_months:
return {}
return dict(zip(self.revenue_months, self.line_total.split(len(self.revenue_months))))
# ── a version: one offer ───────────────────────────────────────────────────────────────────
@dataclass(slots=True)
class Version:
"""Editable while a draft; frozen the moment it is sent — the customer is reading it."""
version: int
status: str = DRAFT
lines: list = field(default_factory=list)
valid_until: Optional[str] = None
approval: Optional[Approval] = None
sent_at: Optional[str] = None
@property
def totals(self) -> dict:
return totals_by_currency(ln.line_total for ln in self.lines)
@property
def list_totals(self) -> dict:
return totals_by_currency(ln.list_total for ln in self.lines)
def net_discount_pct(self) -> Decimal:
"""The discount of the whole version: each line's discount below list (a markup counts as none) over
the list total, per currency, then the deepest currency's."""
out = ZERO_PCT
for c, listed in self.list_totals.items():
if listed.amount <= 0:
continue
off = Money.zero(c)
for ln in self.lines:
if ln.currency == c:
off = off + ln.discount
out = max(out, _clamp(off.percent_of(listed)))
return out
def deepest_line_discount_pct(self) -> Decimal:
return max((ln.discount_pct for ln in self.lines), default=ZERO_PCT)
def deepest_line(self):
return max(self.lines, key=lambda ln: (ln.discount_pct, -ln.line_no), default=None)
def discount_pct(self) -> Decimal:
"""The discount the POLICY reads (H2): the deepest line's or the net's, whichever is larger — so no
line can carry a discount the quote's total hides."""
return max(self.net_discount_pct(), self.deepest_line_discount_pct())
def state(self, today: Optional[str] = None) -> str:
"""What is TRUE, as opposed to what is stored: a sent quote past its date is EXPIRED."""
if self.status == SENT and self.valid_until and \
self.valid_until < (today or date.today().isoformat()):
return EXPIRED
return self.status
def _line(self, line_no: int) -> Line:
for ln in self.lines:
if ln.line_no == line_no:
return ln
raise QuoteRuleBroken(f"version {self.version} has no line {line_no}")
# ── the root ───────────────────────────────────────────────────────────────────────────────
@dataclass(slots=True)
class Quote:
"""Every change to a version or a line goes through a method here."""
quote_id: str
bp_id: int # the customer: a business partner (modules/shared.yaml)
currency: str
owner: str
opportunity_id: Optional[str] = None
reference: Optional[str] = None
versions: list = field(default_factory=list)
revision: int = 0 # 0.2.0: saves so far — two people at once, one wins (409)
@classmethod
def open(cls, quote_id: str, bp_id: int, currency: str, owner: str, *,
opportunity_id: Optional[str] = None, reference: Optional[str] = None,
valid_until: Optional[str] = None) -> "Quote":
q = cls(quote_id=quote_id, bp_id=int(bp_id), currency=Money.zero(currency).currency,
owner=owner, opportunity_id=opportunity_id, reference=reference)
q.versions.append(Version(version=1, valid_until=valid_until))
return q
def version(self, n: Optional[int] = None) -> Version:
if not self.versions:
raise QuoteRuleBroken(f"quote {self.quote_id} has no versions")
if n is None:
return max(self.versions, key=lambda v: v.version)
for v in self.versions:
if v.version == n:
return v
raise QuoteRuleBroken(f"quote {self.quote_id} has no version {n}")
@property
def latest(self) -> Version:
return self.version()
def accepted(self) -> Optional[Version]:
"""The accepted version. More than one would be a contract that cannot say what was signed."""
acc = [v for v in self.versions if v.status == ACCEPTED]
if len(acc) > 1:
raise QuoteRuleBroken(f"quote {self.quote_id} has {len(acc)} accepted versions")
return acc[0] if acc else None
def is_open(self) -> bool:
"""Still being quoted: the latest version is a draft or out with the customer."""
return self.latest.status in (DRAFT, SENT)
# ── editing a draft ──────────────────────────────────────────────────────────────────
def _editable(self, n: int) -> Version:
v = self.version(n)
if v.status != DRAFT:
raise QuoteRuleBroken(f"version {n} is {v.status.lower()}, so it cannot change — the "
f"customer is reading it. Revise the quote to make a new version.")
return v
def add_line(self, n: int, line: Line, policy: DiscountPolicy) -> Line:
v = self._editable(n)
if line.currency != self.currency:
raise QuoteRuleBroken(f"this quote is in {self.currency}; a {line.currency} price has to "
f"be converted first, or the quote shows two totals")
numbered = replace(line, line_no=max((ln.line_no for ln in v.lines), default=0) + 1)
self._within_maximum(v, v.lines + [numbered], policy)
v.lines.append(numbered)
self._withdraw_outgrown_approval(v)
return numbered
def edit_line(self, n: int, line_no: int, policy: DiscountPolicy, **fields) -> tuple:
"""Change quantity, agreed price, description or months on a draft. The product and its list
price are not editable: a different product is a different line, priced on its own date.
Returns (before, after) for the history."""
allowed = {"quantity", "unit_price", "description", "revenue_months"}
unknown = set(fields) - allowed
if unknown:
raise QuoteRuleBroken(f"{', '.join(sorted(unknown))} cannot be changed on a line — "
f"remove it and add the right one")
v = self._editable(n)
before = v._line(line_no)
changes = dict(fields)
if "unit_price" in changes:
changes["unit_price"] = (before.list_price if changes["unit_price"] is None
else Money.of(changes["unit_price"], before.currency))
if "revenue_months" in changes:
changes["revenue_months"] = tuple(changes["revenue_months"] or ())
after = replace(before, **changes)
lines = [after if ln.line_no == line_no else ln for ln in v.lines]
self._within_maximum(v, lines, policy)
v.lines = lines
self._withdraw_outgrown_approval(v)
return before, after
def remove_line(self, n: int, line_no: int) -> Line:
v = self._editable(n)
gone = v._line(line_no)
v.lines = [ln for ln in v.lines if ln.line_no != line_no]
self._withdraw_outgrown_approval(v)
return gone
@staticmethod
def _within_maximum(v: Version, lines: list, policy: DiscountPolicy) -> None:
"""The maximum is checked on EVERY edit, not only at send (spec: 'the discount cap is checked on
every edit AND again when the quote is sent') — a rep learns at the line, not at the end. Per line
as well as net (H2)."""
draft = Version(v.version, lines=lines)
deepest = draft.deepest_line()
if deepest is not None and deepest.discount_pct > policy.max_pct:
raise QuoteRuleBroken(f"line {deepest.line_no} would be {deepest.discount_pct}% below list, past the "
f"{policy.max_pct}% maximum — no approval can cover that")
d = draft.net_discount_pct()
if d > policy.max_pct:
raise QuoteRuleBroken(f"this would discount the quote {d}%, past the {policy.max_pct}% "
f"maximum — no approval can cover that")
@staticmethod
def _withdraw_outgrown_approval(v: Version) -> None:
"""An approval covers the discount it was given for. Once the draft discounts more, it no
longer describes this version and is withdrawn — rather than left on the record looking as if
it covered something it never saw. (Send would refuse anyway; this keeps the record honest.)"""
if v.approval and v.discount_pct() > v.approval.approved_pct:
v.approval = None
# ── approval, sending, the customer's answer ─────────────────────────────────────────
def approve_discount(self, n: int, approver: str, policy: DiscountPolicy) -> Approval:
"""Approve the discount this draft carries NOW.
Refused when nothing needs approving — an approval nobody needed sits on the record and could
later cover a discount that does. Refused for the quote's own owner: whoever prices a deal does
not also sign off its margin.
"""
v = self._editable(n)
if not v.lines:
raise QuoteIncomplete("nothing to approve — this version has no lines")
if approver == self.owner:
raise QuoteRuleBroken("the person who owns a quote cannot approve its discount")
d = v.discount_pct()
if d <= policy.auto_approve_pct:
raise QuoteRuleBroken(f"this version discounts {d}%, within the {policy.auto_approve_pct}% "
f"that needs no approval")
if d > policy.max_pct:
raise QuoteRuleBroken(f"this version discounts {d}%, past the {policy.max_pct}% maximum")
v.approval = Approval(approver, d)
return v.approval
def needs_approval(self, policy: DiscountPolicy) -> bool:
"""True when the latest version is a draft whose discount is not yet covered."""
v = self.latest
d = v.discount_pct()
return v.status == DRAFT and bool(v.lines) and d > policy.auto_approve_pct and (
v.approval is None or d > v.approval.approved_pct)
def send(self, n: int, policy: DiscountPolicy, when: str) -> Version:
"""Freeze the version: from now on it is what the customer was offered."""
v = self._editable(n)
if not v.lines:
raise QuoteIncomplete("a quote with no lines cannot be sent")
d = v.discount_pct()
if d > policy.max_pct:
raise QuoteRuleBroken(f"this quote discounts {d}%, past the {policy.max_pct}% maximum")
if d > policy.auto_approve_pct:
if v.approval is None:
raise NeedsApproval(f"this quote discounts {d}%, past the {policy.auto_approve_pct}% "
f"that sends without approval", d, policy.auto_approve_pct)
if d > v.approval.approved_pct:
raise NeedsApproval(f"this quote now discounts {d}%, more than the "
f"{v.approval.approved_pct}% that was approved", d,
v.approval.approved_pct)
v.status, v.sent_at = SENT, when
return v
def decide(self, n: int, outcome: str, today: Optional[str] = None) -> Version:
"""The customer accepts or declines a sent version.
Accepting SUPERSEDES every other open version, so a quote can never hold two accepted ones.
Deciding the same way twice changes nothing — a retried click is not an error.
"""
if outcome not in (ACCEPTED, DECLINED):
raise QuoteRuleBroken(f"the outcome is {ACCEPTED} or {DECLINED}, not {outcome!r}")
v = self.version(n)
if v.status == outcome:
return v
if v.status != SENT:
raise QuoteRuleBroken(f"version {n} is {v.status.lower()} — only a sent quote can be "
f"accepted or declined")
if outcome == ACCEPTED and v.state(today) == EXPIRED:
raise QuoteRuleBroken(f"version {n} expired on {v.valid_until} — revise it with a new date")
if outcome == ACCEPTED:
for other in self.versions:
if other.version != n and other.status in (DRAFT, SENT):
other.status = SUPERSEDED
v.status = outcome
return v
def revise(self, valid_until: Optional[str] = None) -> Version:
"""A new draft carrying the lines forward; the sent one stays as the customer read it."""
current = self.latest
if current.status == DRAFT:
raise QuoteRuleBroken(f"version {current.version} is still a draft — edit it instead")
if current.status == ACCEPTED:
raise QuoteRuleBroken(f"version {current.version} is accepted — a revision would leave the "
f"agreement pointing at a version the customer never saw. Open a new "
f"quote.")
nxt = Version(version=current.version + 1, lines=list(current.lines),
valid_until=valid_until or current.valid_until)
if current.status == SENT:
current.status = SUPERSEDED
self.versions.append(nxt)
return nxt