dodil/erp-brick-gl

DODIL ERP brick: general ledger

src/brick_gl/domain/allocation.py

"""Allocation cycles (GL-46): costs moved from senders to receivers by agreed rules, so each department's result carries its
real cost. Pure arithmetic: the shares are cut with the largest-remainder method and sum to the sender's amount EXACTLY.

An ASSESSMENT credits the sender on an assessment account and debits the receivers on it; a DISTRIBUTION moves the original
accounts (credit the sender's account, debit the receivers' on the same account). Segments run in order and a later one sees
the earlier one's result."""
from decimal import Decimal
from typing import Optional

from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.money import ZERO, largest_remainder

RULES = ("fixed_pct", "fixed_amount", "key_figure", "tracing_account")
KINDS = ("assessment", "distribution")


def check_cycle(segments: list, approval_pct_total: bool = True) -> None:
    """The gates of defining a cycle: every segment has senders, receivers and a rule; percentages total 100."""
    if not segments:
        raise RuleBroken("a cycle has at least one segment", status=422)
    seqs = [s.get("seq") for s in segments]
    if seqs != list(range(1, len(segments) + 1)):
        raise RuleBroken("segments are numbered 1, 2, 3 ... in the order they run", status=422)
    for s in segments:
        n = s["seq"]
        if s.get("kind") not in KINDS:
            raise RuleBroken(f"segment {n}: kind is assessment or distribution", status=422)
        if s.get("rule") not in RULES:
            raise RuleBroken(f"segment {n}: rule is one of {', '.join(RULES)}", status=422)
        snd = s.get("sender") or {}
        if not (snd.get("account_from") or snd.get("cost_centres") or snd.get("profit_centres")):
            raise RuleBroken(f"segment {n} has no senders", status=422)
        rec = s.get("receivers") or []
        if not rec:
            raise RuleBroken(f"segment {n} has no receivers", status=422)
        if any(not (r.get("cost_centre") or r.get("profit_centre")) for r in rec):
            raise RuleBroken(f"segment {n}: every receiver is a cost centre or a profit centre", status=422)
        if s["kind"] == "assessment" and not s.get("assessment_account"):
            raise RuleBroken(f"segment {n}: an assessment names its assessment_account", status=422)
        if s["rule"] == "fixed_pct":
            total = sum((Decimal(str(r.get("pct", 0))) for r in rec), Decimal(0))
            if total != 100:
                raise RuleBroken(f"segment {n}: the percentages total {total}, not 100", status=422)
        if s["rule"] == "fixed_amount" and any(Decimal(str(r.get("amount", 0))) <= 0 for r in rec):
            raise RuleBroken(f"segment {n}: every receiver has a fixed amount above zero", status=422)
        if s["rule"] == "key_figure" and not s.get("key_figure"):
            raise RuleBroken(f"segment {n}: name the key figure", status=422)
        if s["rule"] == "tracing_account" and not s.get("tracing_account"):
            raise RuleBroken(f"segment {n}: name the tracing account", status=422)


def weights(seg: dict, figure_of, traced_of) -> list:
    """The receivers' weights for a segment: fixed percentages, fixed amounts, a statistical key figure, or actual postings on a
    tracing account."""
    rec = seg["receivers"]
    if seg["rule"] == "fixed_pct":
        return [Decimal(str(r["pct"])) for r in rec]
    if seg["rule"] == "fixed_amount":
        return [Decimal(str(r["amount"])) for r in rec]
    if seg["rule"] == "key_figure":
        return [figure_of(r.get("cost_centre") or r.get("profit_centre")) for r in rec]
    return [abs(traced_of(r)) for r in rec]


def share(total: Decimal, ws: list) -> list:
    if sum(ws, Decimal(0)) == 0:
        raise RuleBroken("the receivers' weights total zero: nothing to share by", status=409)
    return largest_remainder(total, ws)


def account_in_range(code: str, lo: Optional[str], hi: Optional[str]) -> bool:
    if not lo:
        return True
    hi = hi or lo
    if code.isdigit() and lo.isdigit() and hi.isdigit():
        return int(lo) <= int(code) <= int(hi)
    return lo <= code <= hi


def empty() -> Decimal:
    return ZERO