dodil/erp-brick-sales-order
DODIL ERP brick: sales order
src/brick_sales_order/domain/pricing.py
"""Condition pricing and order types — the rules, and nothing else (SO-29, SO-30, SO-31).
The base price is ONLY product's find_price (the BASE condition); this module adds to it or takes from it, step by step through a
pricing procedure, and spreads header conditions over the lines. Money is exact until the end: a line's conditions are priced on the
EXACT running subtotal and each row is the difference of the rounded running totals, so the net is rounded ONCE, half up, from exact
arithmetic and the rows add up to it exactly. No float anywhere.
"""
import math
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from fractions import Fraction
from typing import Optional
from brick_sales_order.domain.errors import RuleBroken
from brick_sales_order.domain.values import EPOCH, ZERO, money
D = Decimal
BASE, DISC_PCT, SURCH_PCT, HDR_DISC, FREIGHT = "BASE", "DISC_PCT", "SURCH_PCT", "HDR_DISC", "FREIGHT"
WEIGHT_KG = {"KGM": Fraction(1), "GRM": Fraction(1, 1000)} # the weight units freight can be computed from
def cents(x: Fraction) -> Decimal:
"""An exact amount to the minor unit, half up (away from zero on a tie): the one rounding."""
q = abs(x) * 100
n = math.floor(q + Fraction(1, 2))
return money(D(n if x >= 0 else -n) / 100)
def exact(x) -> Fraction:
return Fraction(D(x)) if not isinstance(x, Fraction) else x
# ── configuration rows: the in-code seed is what an unseeded bucket behaves as ───────────────────────
@dataclass
class OrderType:
order_type_code: str
valid_from: date
name: str = ""
series_id: str = "SO"
direction: str = "out"
billing: str = "invoice"
delivery_relevant: bool = True
reference: str = "none" # none | order_line | invoice | consignment
default_block: str = "none" # none | APPROVAL
pricing_procedure_code: str = "STANDARD"
workflow: str = "sales_order"
priority: str = "normal"
credit_check: bool = True
special_stock: str = "none"
approval_limit: str = ""
status: str = "active"
@dataclass
class ItemCategory:
item_category_code: str
fulfilment: str = "stock" # stock | service | supplier | none
billing_relevant: bool = True
pricing: str = "full" # full | statistical
schedule_lines: bool = False
returns: bool = False
@dataclass
class ItemCategoryRule:
order_type_code: str
product_type_code: str
usage: str
valid_from: date
item_category_code: str = ""
allowed_alternatives: str = ""
@dataclass
class ConditionType:
condition_type_code: str
calculation: str = "percent" # base | percent | per_unit | fixed | per_weight
level: str = "item" # item | header
header_spread: str = "" # net | weight
sign: str = "minus"
access_sequence_code: str = ""
manual_allowed: bool = False
manual_limit_pct: Optional[Decimal] = None
statistical: bool = False
scale_basis: str = "quantity"
account_purpose: str = "revenue"
@dataclass
class AccessStep:
access_sequence_code: str
step: int
key_fields: str
exclusive: bool = True
@dataclass
class ProcStep:
pricing_procedure_code: str
step: int
condition_type_code: str
from_step: int = 0
to_step: int = 0
subtotal_name: str = ""
required: bool = False
statistical: bool = False
@dataclass
class Record:
"""An agreed discount, surcharge or freight rate for one key, from a date after the day it was set."""
condition_type_code: str
key_fields: str
key_value: str
valid_from: date
valid_to: Optional[date] = None
bp_id: Optional[int] = None
price_group: str = ""
product_id: str = ""
category_code: str = ""
ship_to_country: str = ""
rate: Decimal = ZERO
minimum: Optional[Decimal] = None
scale_json: str = "[]"
currency: str = ""
set_by: str = ""
@property
def scale(self) -> list:
import json
return [(D(x["from"]), D(x["rate"])) for x in json.loads(self.scale_json or "[]")]
def rate_for(self, quantity_base: Fraction) -> Decimal:
"""The scale tier reached by the quantity (the highest 'from' at or below it), else the plain rate."""
got = self.rate
for frm, rate in sorted(self.scale):
if quantity_base >= exact(frm):
got = rate
return got
def key_value(key_fields: str, bp_id=None, price_group="", product_id="", category_code="", ship_to_country="") -> str:
parts = {"customer": str(bp_id or ""), "product": product_id, "category": category_code, "price_group": price_group,
"ship_to_country": ship_to_country, "all": "*"}
return "|".join(parts[w.strip()] for w in key_fields.split("+"))
ACCESS_FIELDS = ("customer + product", "customer + category", "price_group + product", "price_group + category", "category",
"customer", "ship_to_country", "all")
def _epoch() -> date:
return date.fromisoformat(EPOCH)
def default_types() -> list:
e = _epoch()
spec = [("STANDARD", "SO", "out", "invoice", True, "none", "none", True), ("RUSH", "RU", "out", "invoice", True, "none", "none", True),
("CASH", "CA", "out", "invoice", True, "none", "none", False), ("RETURNS", "RE", "in", "credit", True, "order_line", "none", True),
("CREDIT_REQ", "CR", "none", "credit", False, "invoice", "APPROVAL", True), ("DEBIT_REQ", "DR", "none", "debit", False, "invoice", "APPROVAL", True),
("FREE_OF_CHARGE", "FC", "out", "none", True, "none", "APPROVAL", True), ("CONS_FILL", "CS", "out", "none", True, "none", "none", True),
("CONS_ISSUE", "CS", "out", "invoice", True, "consignment", "none", True), ("CONS_PICKUP", "CS", "in", "none", True, "consignment", "none", True),
("CONS_RETURN", "CS", "in", "none", True, "consignment", "none", True)]
return [OrderType(c, e, c.replace("_", " ").title(), series, direction, billing, delivery, ref, block, "STANDARD",
"return_order" if c == "RETURNS" else "sales_order", "rush" if c == "RUSH" else "normal", credit,
"to_consignment" if c == "CONS_FILL" else "from_consignment" if c in ("CONS_PICKUP", "CONS_RETURN") else "none")
for c, series, direction, billing, delivery, ref, block, credit in spec]
def default_categories() -> list:
return [ItemCategory("STD_STOCK", "stock", True, "full", True), ItemCategory("STD_SERVICE", "service", True, "full"),
ItemCategory("THIRD_PARTY", "supplier", True, "full", True), ItemCategory("FREE", "stock", False, "statistical", True),
ItemCategory("RETURN", "stock", True, "full", False, True), ItemCategory("CONS_FILL", "stock", False, "statistical", True),
ItemCategory("CONS_ISSUE", "stock", True, "full", True)]
def default_rules() -> list:
e = _epoch()
rows = []
for t in ("STANDARD", "RUSH", "CASH"):
rows += [ItemCategoryRule(t, "TRADING", "normal", e, "STD_STOCK"), ItemCategoryRule(t, "SERVICE", "normal", e, "STD_SERVICE"),
ItemCategoryRule(t, "TRADING", "drop_ship", e, "THIRD_PARTY")]
rows += [ItemCategoryRule("FREE_OF_CHARGE", "*", "*", e, "FREE"), ItemCategoryRule("RETURNS", "*", "*", e, "RETURN"),
ItemCategoryRule("CREDIT_REQ", "*", "*", e, "STD_SERVICE"), ItemCategoryRule("DEBIT_REQ", "*", "*", e, "STD_SERVICE"),
ItemCategoryRule("CONS_FILL", "*", "*", e, "CONS_FILL"), ItemCategoryRule("CONS_ISSUE", "*", "*", e, "CONS_ISSUE"),
ItemCategoryRule("CONS_PICKUP", "*", "*", e, "RETURN"), ItemCategoryRule("CONS_RETURN", "*", "*", e, "RETURN")]
return rows
def default_condition_types() -> list:
return [ConditionType(BASE, "base", "item", "", "plus", "", False, None, False, "quantity", "revenue"),
ConditionType(DISC_PCT, "percent", "item", "", "minus", "DISC", True, D(10), False, "quantity", "discount"),
ConditionType(SURCH_PCT, "percent", "item", "", "plus", "SURCH", False, None, False, "quantity", "revenue"),
ConditionType(HDR_DISC, "percent", "header", "net", "minus", "", True, D(10), False, "value", "discount"),
ConditionType(FREIGHT, "per_weight", "header", "weight", "plus", "FREIGHT", False, None, False, "weight", "freight")]
def default_access() -> list:
return [AccessStep("DISC", 10, "customer + product"), AccessStep("DISC", 20, "customer + category"), AccessStep("DISC", 30, "price_group + category"),
AccessStep("SURCH", 10, "category"), AccessStep("FREIGHT", 10, "ship_to_country")]
def default_procedure() -> list:
return [ProcStep("STANDARD", 10, BASE, 0, 0, "Base price", True), ProcStep("STANDARD", 20, DISC_PCT, 10, 10, "Net after discount"),
ProcStep("STANDARD", 30, SURCH_PCT, 10, 20, "Net"), ProcStep("STANDARD", 40, HDR_DISC, 0, 0), ProcStep("STANDARD", 50, FREIGHT, 0, 0)]
def in_force(rows: list, on: date, *keys) -> Optional[object]:
"""The dated row in force on a date: the latest valid_from on or before it among rows with the same key."""
same = [r for r in rows if all(getattr(r, k) == v for k, v in keys) and r.valid_from <= on]
return max(same, key=lambda r: r.valid_from) if same else None
def item_category(rules: list, order_type: str, product_type: str, usage: str, on: date) -> str:
"""The item category a line gets (SO-29): the most specific rule (its product type and usage named, not '*') in force on the date."""
live = [r for r in rules if r.order_type_code == order_type and r.valid_from <= on
and r.product_type_code in (product_type, "*") and r.usage in (usage, "*")]
if not live:
raise RuleBroken(f"no item category rule for an order of type {order_type} with a {product_type or 'product of no type'} line "
f"(usage {usage}): ask for one in Order types")
best = max(live, key=lambda r: ((r.product_type_code != "*") + (r.usage != "*"), r.valid_from))
return best.item_category_code
@dataclass
class Ctx:
"""What a line is looked up by."""
bp_id: int
price_group: str
product_id: str
category_code: str
ship_to_country: str
quantity_base: Fraction
on: date
def find_record(ctype: ConditionType, access: list, records: list, ctx: Ctx):
"""The first record found, most specific access step first: the search stops at the first hit (SO-30)."""
for st in sorted((a for a in access if a.access_sequence_code == ctype.access_sequence_code), key=lambda a: a.step):
want = key_value(st.key_fields, ctx.bp_id, ctx.price_group, ctx.product_id, ctx.category_code, ctx.ship_to_country)
hits = [r for r in records if r.condition_type_code == ctype.condition_type_code and r.key_fields == st.key_fields
and r.key_value == want and r.valid_from <= ctx.on and (r.valid_to is None or ctx.on <= r.valid_to)]
if hits:
return max(hits, key=lambda r: r.valid_from), st
return None, None
@dataclass
class Row:
step: int
condition_type_code: str
rate: Decimal
amount: Decimal
basis: Decimal
source: str
statistical: bool = False
def item_rows(base_exact: Fraction, base_rate: Decimal, base_source: str, steps: list, types: dict, chosen: dict) -> tuple:
"""(rows, net). `chosen`: {condition_type_code: (rate, source, found)} for each item condition that applies to the line. The BASE row
first; each percent step applies to the exact running subtotal of its from..to steps; each row is the difference of the rounded
running totals, so the net is rounded once and the rows add up to it."""
run: dict = {}
cum, prev = base_exact, cents(base_exact)
rows = [Row(10, BASE, base_rate, prev, ZERO, base_source)]
run[10] = base_exact
for st in sorted(steps, key=lambda s: s.step):
t = types.get(st.condition_type_code)
if t is None or t.level != "item" or t.calculation == "base" or st.condition_type_code not in chosen:
continue
rate, source, qty = chosen[st.condition_type_code]
if t.calculation == "percent":
basis = sum((run.get(s, Fraction(0)) for s in range(st.from_step, st.to_step + 1)), Fraction(0))
amt = exact(rate) * basis / 100
elif t.calculation == "per_unit":
basis, amt = qty, exact(rate) * qty
else:
basis, amt = Fraction(1), exact(rate)
amt = -amt if t.sign == "minus" else amt
stat = t.statistical or st.statistical
run[st.step] = Fraction(0) if stat else amt
if stat:
rows.append(Row(st.step, st.condition_type_code, rate, cents(amt), cents(basis), source, True))
continue
cum += amt
now = cents(cum)
rows.append(Row(st.step, st.condition_type_code, rate, now - prev, cents(basis), source))
prev = now
return rows, prev
def spread(amount: Decimal, weights: dict) -> dict:
"""A header amount over lines by weight: each line but the largest takes its rounded share, the LARGEST takes the remainder, so the
shares add up to the amount exactly (ties: the lowest line number)."""
total = sum(weights.values(), Fraction(0))
if not weights:
return {}
biggest = min(weights, key=lambda k: (-weights[k], k))
out = {}
for k in weights:
if k != biggest:
out[k] = cents(exact(amount) * weights[k] / total) if total else ZERO
out[biggest] = amount - sum(out.values(), ZERO)
return out
def header_amount(ctype: ConditionType, rate: Decimal, base: Fraction, minimum: Optional[Decimal] = None) -> Decimal:
"""One header condition's amount, rounded once, signed: a percent of the lines' net, or a rate per KGM of gross weight (`base`), at
least `minimum`."""
amt = exact(rate) * base / (100 if ctype.calculation == "percent" else 1)
if minimum is not None and ctype.calculation == "per_weight":
amt = max(amt, exact(minimum))
amt = cents(amt)
return -amt if ctype.sign == "minus" else amt
def line_weight_kg(quantity: Decimal, gross_weight, weight_unit: str) -> Optional[Fraction]:
"""A line's gross weight in KGM (the weight is per ONE of the line's unit), or None when the product's weight or its unit is unknown."""
if gross_weight is None or Decimal(gross_weight) <= 0 or weight_unit not in WEIGHT_KG:
return None
return exact(quantity) * exact(gross_weight) * WEIGHT_KG[weight_unit]