"""Price resolution (PRD-06, PRD-08, PRD-11): derived on read from dated rows and rules, never stored.
Plain Python over a `PriceBook` (anything that can say what a list is, which live rows it holds for a product
and which live rules it holds). The same question always gets the same answer, and the answer names where it
came from, so an audit can say what a price WAS on a date.
The order, inside one list:
1. its rules, in the list's stated order (seq); the FIRST that matches wins. A rule matches on scope (one
product, a category and everything under it, or all), the quantity from which it applies and its dates.
A rule whose base price cannot be found does not match, and the next one is tried.
2. else its own price rows: the latest valid_from on or before the date (one price per product from a date; quantity
breaks are rules, not rows).
3. else the list it is based on, asked the same question.
Across lists: the customer's assigned list for the currency (when it is published and the assignment has
started), then the currency's default list. A price is never converted from another currency.
"""
from dataclasses import dataclass, field
from datetime import date
from decimal import ROUND_HALF_UP, Decimal
from fractions import Fraction
from typing import Optional, Protocol, Sequence
from brick_product.domain.errors import RuleBroken
from brick_product.domain.units import round_half_up
from brick_product.domain.values import RuleMethod, RuleScope
MAX_DEPTH = 8 # lists based on lists based on lists: a chain longer than this is a mistake
RATE = Decimal("0.0001")
CENTS = Decimal("0.01")
PRICE_LIMIT = Decimal("1e14") # NUMERIC(19,4): fifteen digits before the point
ROUNDINGS = ("none", "0.001", "0.01", "0.05", "0.10", "0.50", "1.00", "x.99")
def rate(value) -> Decimal:
"""A price per unit as the doors show it: four places at most, and two when that says the same ("102.00"). An exact
Fraction is rounded half up to four places here, at the end, never before."""
q = round_half_up(value, 4) if isinstance(value, Fraction) else Decimal(value).quantize(RATE, rounding=ROUND_HALF_UP)
if abs(q) >= PRICE_LIMIT:
raise RuleBroken(f"a price of that size ({str(q)[:24]}…) cannot be stored: a unit price is below 100,000,000,000,000")
c = q.quantize(CENTS)
return c if c == q else q
def round_price(value, rounding: str) -> Decimal:
"""Round a COMPUTED price (a Decimal or an exact Fraction) to the list's rule: a step ("0.05" is the nearest 5p) or
"x.99" (the nearest whole unit, less one cent; never below 0.99 — under 0.50 it is rounded to the cent instead,
since .99 of nothing is a negative price). "none" keeps the four places a rate carries."""
v = Fraction(value)
if rounding == "none" or not rounding:
return rate(v)
if rounding == "x.99":
whole = round_half_up(v, 0)
if whole < 1: # a price under 0.50 would round to -0.01: below that there is no .99, only the cent
return rate(round_half_up(v, 2))
return rate(whole - CENTS)
step = Fraction(Decimal(rounding))
return rate(Fraction(round_half_up(v / step, 0)) * step)
@dataclass(frozen=True)
class ListHead:
price_list_id: str
currency: str
based_on: str
rounding: str
status: str
includes_tax: bool = False
retired_on: Optional[date] = None # the business day it was retired: it answers dates up to then, not after
@dataclass(frozen=True)
class PriceRow:
price_list_id: str
product_id: str
valid_from: date
unit_price: Decimal
valid_to: Optional[date] = None
unit_code: str = "" # the unit this price is per: one row per product, unit and start date
pricing_quantity: int = 1 # v4: the price is per this many of unit_code (2.4500 per 100 C62)
@dataclass(frozen=True)
class PriceRule:
price_list_id: str
seq: int
applies_to: str = RuleScope.ALL.value
target: str = ""
min_quantity: Decimal = Decimal(0)
valid_from: Optional[date] = None
valid_to: Optional[date] = None
method: str = RuleMethod.DISCOUNT_PCT.value
discount_pct: Decimal = Decimal(0)
fixed_price: Optional[Decimal] = None
surcharge_amount: Decimal = Decimal(0)
base_list: str = ""
unit_code: str = "" # the unit a fixed price or a surcharge is stated in — set when written, never re-read
pricing_quantity: int = 1 # v4: the quantity a fixed price or surcharge is per
@dataclass(frozen=True)
class Subject:
"""What a price is asked for: the PRODUCT (a SKU, never a template), its category with every category above it on
that date, its template (a rule can cover all a template's variants) and the UNIT asked — with how many base units
one of it is (`factor`); `factors` holds every unit the product is counted in, so a rule's fixed price or surcharge can be
converted from the unit it was WRITTEN in (rule.unit_code) to the unit asked."""
product_id: str
category_path: tuple = ()
template_id: str = ""
unit_code: str = ""
base_unit_code: str = ""
factor: Fraction = Fraction(1)
factors: dict = field(default_factory=dict)
class PriceBook(Protocol):
def head(self, price_list_id: str) -> Optional[ListHead]: ...
def rows(self, price_list_id: str, product_id: str) -> Sequence[PriceRow]: ... # live rows only
def rules(self, price_list_id: str) -> Sequence[PriceRule]: ... # live rules only
@dataclass
class Resolved:
"""One answer: the price, and the list, rule and row that made it."""
unit_price: Decimal
price_list_id: str
via: str # "rule" | "row" | "based_on"
rule: Optional[PriceRule] = None
row: Optional[PriceRow] = None
base: Optional["Resolved"] = None # what a percentage or an inheriting list started from
note: str = field(default="")
exact: Fraction = Fraction(0) # the price before it was rounded for show (rules compute from this)
converted_from: str = "" # the unit of the row it was converted from, when the asked unit had none
pricing_quantity: int = 1 # v4: unit_price is per this many of pricing_unit (the row's own, unless re-expressed)
pricing_unit: str = "" # the unit unit_price is stated in; "" = the unit asked
def rule_matches(rule: PriceRule, subject: Subject, qty, on: date) -> bool:
"""`qty` is the quantity in the BASE unit: a break compares the same whether it was asked in packs or bottles."""
if Fraction(qty) < Fraction(rule.min_quantity or 0):
return False
if (rule.valid_from and on < rule.valid_from) or (rule.valid_to and on > rule.valid_to):
return False
if rule.applies_to == RuleScope.PRODUCT.value:
return rule.target == subject.product_id
if rule.applies_to == RuleScope.TEMPLATE.value:
return bool(subject.template_id) and rule.target == subject.template_id
if rule.applies_to == RuleScope.CATEGORY.value:
return rule.target in subject.category_path
return rule.applies_to == RuleScope.ALL.value
def best_row(rows: Sequence[PriceRow], on: date) -> Optional[PriceRow]:
"""The row in force on `on`: the latest valid_from on or before it (and not past its valid_to)."""
live = [r for r in rows if r.valid_from <= on and (r.valid_to is None or on <= r.valid_to)]
return max(live, key=lambda r: r.valid_from, default=None)
def express(price: Fraction, quantity: int, unit: str, subject: Subject, stated_factor: Optional[Fraction]) -> tuple:
"""(price, pricing quantity, pricing unit, the unit it was converted from) for a price of `price` per `quantity` of `unit`.
The pricing unit stays the row's own (v4). Only a price per ONE unit, which converts into the asked unit exactly at four
decimals, is re-expressed in the asked unit (the v1 behaviour: a pack with no price of its own costs its bottles)."""
if unit == subject.unit_code or quantity != 1 or stated_factor is None:
return price, quantity, unit, ""
converted = price * subject.factor / stated_factor
if (converted * 10000).denominator == 1:
return converted, 1, subject.unit_code, unit
return price, quantity, unit, ""
def row_price(rows: Sequence[PriceRow], subject: Subject, on: date) -> Optional[tuple]:
"""(row, the exact price per the pricing quantity of the pricing unit, that quantity, that unit, the unit it was converted from
or ''): the row in the asked unit if there is one; otherwise the base-unit row, in the base unit, unless it is a price per one
that converts exactly (see `express`)."""
own = best_row([r for r in rows if r.unit_code == subject.unit_code], on)
if own is not None:
return own, Fraction(own.unit_price), own.pricing_quantity or 1, own.unit_code, ""
if subject.unit_code != subject.base_unit_code:
base = best_row([r for r in rows if r.unit_code == subject.base_unit_code], on)
if base is not None:
price, pq, unit, was = express(Fraction(base.unit_price), base.pricing_quantity or 1, base.unit_code, subject, Fraction(1))
return base, price, pq, unit, was
return None
def _usable(book: PriceBook, price_list_id: str, currency: str, on: date) -> Optional[ListHead]:
head = book.head(price_list_id)
if head is None or head.currency != currency:
return None
if head.status == "published" or (head.status == "retired" and head.retired_on is not None and on <= head.retired_on):
return head
return None
def resolve_in(book: PriceBook, price_list_id: str, subject: Subject, qty, on: date,
currency: str, _seen: tuple = ()) -> Optional[Resolved]:
"""The price in ONE list (rules, then rows, then the list it is based on), or None. `qty` is in the base unit."""
head = _usable(book, price_list_id, currency, on)
if head is None or price_list_id in _seen or len(_seen) >= MAX_DEPTH:
return None
seen = _seen + (price_list_id,)
for rule in sorted(book.rules(price_list_id), key=lambda r: r.seq):
if rule_matches(rule, subject, qty, on):
got = _apply(book, head, rule, subject, qty, on, seen)
if got is not None:
return got
return _own_or_inherited(book, head, subject, qty, on, seen)
def _own_or_inherited(book: PriceBook, head: ListHead, subject: Subject, qty, on: date,
seen: tuple) -> Optional[Resolved]:
found = row_price(book.rows(head.price_list_id, subject.product_id), subject, on)
if found is not None:
row, exact_price, pq, unit, converted = found
return Resolved(rate(exact_price), head.price_list_id, "row", row=row, exact=exact_price, converted_from=converted,
pricing_quantity=pq, pricing_unit=unit)
if head.based_on:
up = resolve_in(book, head.based_on, subject, qty, on, head.currency, seen)
if up is not None:
return Resolved(up.unit_price, head.price_list_id, "based_on", base=up,
note=f"inherited from {head.based_on}", exact=up.exact, converted_from=up.converted_from,
pricing_quantity=up.pricing_quantity, pricing_unit=up.pricing_unit)
return None
def _apply(book: PriceBook, head: ListHead, rule: PriceRule, subject: Subject, qty, on: date,
seen: tuple) -> Optional[Resolved]:
stated = subject.factors.get(rule.unit_code) if rule.unit_code else None # a fixed price or surcharge is per the unit it was WRITTEN in
rq = rule.pricing_quantity or 1
if rule.method == RuleMethod.FIXED.value:
if rule.fixed_price is None or stated is None:
return None # the product is not counted in that unit: the rule has no price
price, pq, unit, was = express(Fraction(rule.fixed_price), rq, rule.unit_code, subject, stated)
return Resolved(rate(price), head.price_list_id, "rule", rule=rule, exact=Fraction(rate(price)), converted_from=was,
pricing_quantity=pq, pricing_unit=unit)
if rule.base_list:
base = resolve_in(book, rule.base_list, subject, qty, on, head.currency, seen)
else:
base = _own_or_inherited(book, head, subject, qty, on, seen)
if base is None:
return None
price = base.exact * (100 - Fraction(rule.discount_pct or 0)) / 100 # a percentage applies to the price per pricing quantity
if rule.method == RuleMethod.FORMULA.value and rule.surcharge_amount:
here = subject.factors.get(base.pricing_unit or subject.unit_code)
if stated is None or here is None:
return None
price += Fraction(rule.surcharge_amount) * (here / stated) * Fraction(base.pricing_quantity, rq)
price = max(price, Fraction(0))
shown = round_price(price, head.rounding)
return Resolved(shown, head.price_list_id, "rule", rule=rule, base=base, exact=Fraction(shown),
converted_from=base.converted_from, pricing_quantity=base.pricing_quantity, pricing_unit=base.pricing_unit)
@dataclass
class Asked:
"""A full resolution: what was answered, and which lists were tried before it."""
resolved: Optional[Resolved]
tried: list = field(default_factory=list) # price_list_ids in the order they were asked
assigned: str = "" # the customer's list, when they have one for the currency
fell_back: bool = False # the answer is not from the customer's assigned list
def resolve(book: PriceBook, candidates: Sequence[str], subject: Subject, qty, on: date,
currency: str, assigned: str = "") -> Asked:
"""Ask `candidates` in order (the customer's list first, then the currency's default) and take the first
list that has a price. Never converts: only lists in `currency` are ever considered."""
out = Asked(None, assigned=assigned)
basis = None
for list_id in candidates:
if list_id in out.tried:
continue
head = book.head(list_id)
if basis is not None and head is not None and head.includes_tax != basis:
continue # a gross customer list never falls back to a net default, nor the reverse
if basis is None and head is not None:
basis = head.includes_tax
out.tried.append(list_id)
got = resolve_in(book, list_id, subject, qty, on, currency)
if got is not None:
out.resolved = got
out.fell_back = bool(assigned) and list_id != assigned
break
return out
def deviation_pct(price: Decimal, reference: Optional[Decimal]) -> Optional[Decimal]:
"""How far `price` sits from `reference`, in percent (negative = below). None when there is nothing to compare."""
if reference is None or reference <= 0:
return None
return ((Decimal(price) - reference) / reference * 100).quantize(CENTS, rounding=ROUND_HALF_UP)
def is_unusual(price: Decimal, reference: Optional[Decimal], threshold_pct: int) -> bool:
d = deviation_pct(price, reference)
return d is not None and abs(d) > threshold_pct