dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/domain/splitting.py
"""The universal line (SAP's ACDOCA analogue) and document splitting (GL-22).
Every journal line resolves to one or more SPLIT lines carrying profit centre, segment, partner company and movement type.
With splitting OFF a line gives one split line with its derived profit centre. With splitting ON:
* ACTIVE split: a receivable, payable, tax or bank line (an "item" line) is shared across the profit centres of the
journal's base lines, in proportion to their amounts - an invoice of revenue 600 PC-LON and 400 PC-WHS puts a
receivable of 1,200 as 720 / 480 and VAT of 200 as 120 / 80;
* PASSIVE split: an item line that names `against_document` follows how that document's line on the same account was
split (a receipt follows the invoice it pays); without it, it takes the company's default profit centre and the
period-end re-split (GL-25) moves it;
* ZERO-BALANCE: whatever still does not net to zero per profit centre (a transfer between two profit centres) gets a
clearing line on the `split_clearing` account per profit centre, so every segment balances on every journal.
Every share is cut with the largest-remainder method, so the shares sum to the line EXACTLY, in both currencies.
"""
from dataclasses import dataclass
from decimal import Decimal
from typing import Callable, Optional
from brick_gl.domain.money import ZERO, largest_remainder
from brick_gl.domain.readiness import refusal
ITEM_SUBTYPES = ("receivable", "payable", "tax", "bank")
@dataclass
class SplitLine:
line_no: int
split_no: int
account_id: int
debit: Decimal
credit: Decimal
f_debit: Decimal
f_credit: Decimal
profit_centre: str = ""
segment: str = ""
cost_centre: str = ""
partner_company: str = ""
movement_type: str = ""
bp_id: Optional[int] = None
asset_id: str = ""
is_clearing: bool = False
@property
def net(self) -> Decimal:
return self.debit - self.credit
@property
def f_net(self) -> Decimal:
return self.f_debit - self.f_credit
def _row(ln, split_no: int, **kw) -> SplitLine:
base = dict(line_no=ln.line_no, split_no=split_no, account_id=ln.account_id, debit=ln.debit, credit=ln.credit,
f_debit=ln.f_debit, f_credit=ln.f_credit, profit_centre=ln.profit_centre, segment=ln.segment,
cost_centre=ln.cost_centre, partner_company=ln.partner_company, movement_type=ln.movement_type,
bp_id=ln.bp_id, asset_id=ln.asset_id)
base.update(kw)
return SplitLine(**base)
def split(lines: list, subtype_of: Callable[[int], str], *, splitting: bool, default_pc: str,
segment_of: Callable[[str], str], passive: Callable[[str, int], Optional[dict]],
clearing_account: Callable[[], Optional[int]]) -> list:
"""The split lines of a journal's resolved lines."""
if not splitting:
return [_row(ln, 1) for ln in lines]
is_item = {ln.line_no: (subtype_of(ln.account_id) in ITEM_SUBTYPES) for ln in lines}
base_net: dict = {}
for ln in lines:
if not is_item[ln.line_no] and not ln.rounding:
base_net[ln.profit_centre] = base_net.get(ln.profit_centre, ZERO) + ln.f_net
base_weights = {pc: abs(v) for pc, v in base_net.items() if v != 0}
# item lines that follow a cleared document (passive) give the weights for the journal's other item lines
followed: dict = {}
for ln in lines:
if is_item[ln.line_no] and ln.against_document:
w = passive(ln.against_document, ln.account_id)
if w:
followed[ln.line_no] = w
journal_weights = base_weights
if not journal_weights and followed:
agg: dict = {}
for w in followed.values():
tot = sum(w.values(), Decimal(0))
for pc, v in w.items():
agg[pc] = agg.get(pc, Decimal(0)) + (v / tot if tot else Decimal(0))
journal_weights = agg
out: list = []
for ln in lines:
weights = None
if is_item[ln.line_no]:
if ln.line_no in followed:
weights = followed[ln.line_no]
elif ln.against_document:
weights = None # nothing to follow: the company default
else:
weights = journal_weights or None
if ln.against_document and is_item[ln.line_no] and ln.line_no not in followed:
pc = default_pc
out.append(_row(ln, 1, profit_centre=pc, segment=segment_of(pc) if pc else ""))
continue
if not weights or len([w for w in weights.values() if w > 0]) < 1:
out.append(_row(ln, 1))
continue
pcs = sorted(pc for pc, w in weights.items() if w > 0)
ws = [weights[pc] for pc in pcs]
fd, fc = largest_remainder(ln.f_debit, ws), largest_remainder(ln.f_credit, ws)
dd, dc = largest_remainder(ln.debit, ws), largest_remainder(ln.credit, ws)
n = 0
for i, pc in enumerate(pcs):
if dd[i] == 0 and dc[i] == 0 and fd[i] == 0 and fc[i] == 0:
continue
n += 1
out.append(_row(ln, n, debit=dd[i], credit=dc[i], f_debit=fd[i], f_credit=fc[i], profit_centre=pc,
segment=segment_of(pc) if pc else ""))
if n == 0:
out.append(_row(ln, 1))
# zero-balance per profit centre
nets: dict = {}
for s in out:
d, f = nets.get(s.profit_centre, (ZERO, ZERO))
nets[s.profit_centre] = (d + s.net, f + s.f_net)
if len(nets) > 1 and any(d != 0 or f != 0 for d, f in nets.values()):
account = clearing_account()
if account is None:
raise refusal("purposes_mapped", "document splitting needs the purpose split_clearing to balance every profit "
"centre, and it has no account in force on the journal date",
"Account map: map split_clearing (GL-12)", status=422)
k = 0
for pc in sorted(nets):
d, f = nets[pc]
if d == 0 and f == 0:
continue
k += 1
out.append(SplitLine(
line_no=0, split_no=k, account_id=account, debit=-d if d < 0 else ZERO, credit=d if d > 0 else ZERO,
f_debit=-f if f < 0 else ZERO, f_credit=f if f > 0 else ZERO, profit_centre=pc,
segment=segment_of(pc) if pc else "", is_clearing=True))
return out
def nets_by(splits: list, key: str) -> dict:
"""{key value: (doc net, functional net)} - the check that a journal balances per profit centre and per segment."""
out: dict = {}
for s in splits:
k = getattr(s, key)
d, f = out.get(k, (ZERO, ZERO))
out[k] = (d + s.net, f + s.f_net)
return out