dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/statements.py
"""Financial statement versions (GL-53), the statements run on them (GL-06), the cash-flow statement (GL-54).
A version says which accounts make up each line of a statement, so the same ledger yields the IFRS statements, the statutory
format and a management view. A PUBLISHED version is never edited (a change is version_no + 1) and a report names the version
and the ledger it was run on. Publishing needs every account with a balance assigned EXACTLY once (gate fsv_complete, the
accounts named); a bank account can show under Cash or Overdraft by the sign of its balance (debit/credit-dependent assignment).
Amounts: debit positive in the functional currency; a node's `sign` turns a credit-natured line (liabilities, income) positive
for presentation.
"""
from datetime import date
from decimal import Decimal
from typing import Annotated, Optional
from erp_platform.bricks import Refused
from pydantic import Field
from sqlalchemy import func, select, type_coerce
from brick_gl.application import carry
from brick_gl.application.brick import (
FSV_WRITE,
GL,
REPORT_READ,
company_row,
decide,
ledger_for,
real_now,
repository,
sub_of,
today,
who,
)
from brick_gl.application.inputs import Code, CompanyCode, IsoDate, LedgerCode, PeriodId, _In
from brick_gl.application.reports import _window, trial_balance_data
from brick_gl.domain import allocation as alloc
from brick_gl.domain import dates
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.money import ZERO
from brick_gl.domain.readiness import Gate, met, require, unmet
from brick_gl.domain.values import BALANCE_SHEET
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import _Keyed, utc
from brick_gl.infrastructure.tables import MONEY
KINDS = ("balance_sheet_pl", "cash_flow")
CF_CLASSES = ("operating", "investing", "financing", "cash", "non_cash")
class NodeModel(_In):
node_id: Code
parent_node_id: Optional[Code] = None
seq: Annotated[int, Field(ge=0, le=100000)] = 0
label: Annotated[str, Field(max_length=200)]
sign: Annotated[int, Field(ge=-1, le=1)] = 1
cash_flow_class: Optional[Code] = None
class AssignmentModel(_In):
from_code: Code
to_code: Optional[Code] = None
node_id: Code
dc_rule: Code = "always"
other_node_id: Optional[Code] = None
basis: Code = "balance"
_ver = lambda repo: _Keyed(repo.conn, t.gl_statement_version, ("fsv_id", "version_no")) # noqa: E731
_node = lambda repo: _Keyed(repo.conn, t.gl_statement_node, ("fsv_id", "version_no", "node_id")) # noqa: E731
_asg = lambda repo: _Keyed(repo.conn, t.gl_statement_assignment, ("fsv_id", "version_no", "from_code", "to_code")) # noqa: E731
def _rows(repo, table, fsv_id: str, version_no: int) -> list:
return [dict(r._mapping) for r in repo.conn.execute(table.select().where(table.c.fsv_id == fsv_id, table.c.version_no == version_no))]
def _version(repo, fsv_id: str, version_no: Optional[int] = None, published: bool = False) -> dict:
q = t.gl_statement_version.select().where(t.gl_statement_version.c.fsv_id == fsv_id).order_by(t.gl_statement_version.c.version_no)
rows = [dict(r._mapping) for r in repo.conn.execute(q)]
if published:
rows = [r for r in rows if r["status"] == "published"]
if version_no is not None:
rows = [r for r in rows if r["version_no"] == version_no]
if not rows:
raise Refused(404, f"no {'published ' if published else ''}statement version {fsv_id}")
return rows[-1]
def _view(repo, v: dict) -> dict:
nodes = sorted(_rows(repo, t.gl_statement_node, v["fsv_id"], v["version_no"]), key=lambda n: (n["parent_node_id"] or "", n["seq"], n["node_id"]))
asg = sorted(_rows(repo, t.gl_statement_assignment, v["fsv_id"], v["version_no"]), key=lambda a: (a["from_code"], a["to_code"]))
return {"fsv_id": v["fsv_id"], "version_no": v["version_no"], "name": v["name"], "kind": v["kind"], "chart_id": v["chart_id"], "status": v["status"],
"nodes": [{k: n[k] for k in ("node_id", "parent_node_id", "seq", "label", "sign", "cash_flow_class")} for n in nodes],
"assignments": [{k: a[k] for k in ("from_code", "to_code", "node_id", "dc_rule", "other_node_id", "basis")} for a in asg]}
def _check_tree(kind: str, nodes: list, asg: list) -> None:
ids = {n.node_id for n in nodes}
if len(ids) != len(nodes):
raise RuleBroken("node ids are unique within a version", status=422)
for n in nodes:
if n.parent_node_id and n.parent_node_id not in ids:
raise RuleBroken(f"node {n.node_id} names a parent {n.parent_node_id} that is not in the version", status=422)
if kind == "cash_flow" and n.cash_flow_class not in CF_CLASSES:
raise RuleBroken(f"cash-flow node {n.node_id}: cash_flow_class is one of {', '.join(CF_CLASSES)}", status=422)
for n in nodes: # no loops
seen, cur = set(), n
while cur and cur.parent_node_id:
if cur.node_id in seen:
raise RuleBroken("the node tree loops", status=422)
seen.add(cur.node_id)
cur = next((x for x in nodes if x.node_id == cur.parent_node_id), None)
for a in asg:
if a.node_id not in ids or (a.other_node_id and a.other_node_id not in ids):
raise RuleBroken(f"assignment {a.from_code}-{a.to_code} names a node that is not in the version", status=422)
if a.dc_rule not in ("always", "debit_balance", "credit_balance"):
raise RuleBroken("dc_rule is always, debit_balance or credit_balance", status=422)
if a.basis not in ("balance", "movement"):
raise RuleBroken("basis is balance or movement", status=422)
@GL.use_case(kind="command", route="POST /statement-versions", permission=FSV_WRITE)
def draft_statement_version(caller: dict, fsv_id: Code, name: Annotated[str, Field(max_length=200)], kind: Code, chart_id: Code,
nodes: Annotated[list[NodeModel], Field(max_length=2000)],
assignments: Annotated[list[AssignmentModel], Field(max_length=5000)]) -> dict:
"""Draft a financial statement version (kind balance_sheet_pl, or cash_flow for the indirect-method cash-flow statement): its
tree of nodes (label, sign, cash_flow_class) and which accounts make each line (code ranges; always, or debit-balance /
credit-balance dependent with an other node; basis balance or movement). Drafting an existing draft replaces it; a published
version is never edited - supersede_statement_version drafts version_no + 1."""
with repository() as repo, decide():
if kind not in KINDS:
raise RuleBroken(f"kind is one of {', '.join(KINDS)}", status=422)
if repo.charts.get(chart_id) is None:
raise Refused(404, f"no chart {chart_id}")
if not nodes:
raise RuleBroken("a statement version has at least one node", status=422)
_check_tree(kind, nodes, assignments)
existing = [dict(r._mapping) for r in repo.conn.execute(t.gl_statement_version.select().where(
t.gl_statement_version.c.fsv_id == fsv_id).order_by(t.gl_statement_version.c.version_no))]
if existing and existing[-1]["status"] == "published":
raise RuleBroken(f"{fsv_id} version {existing[-1]['version_no']} is published and never edited: supersede it to draft the next", status=409)
vno = existing[-1]["version_no"] if existing else 1
if existing:
for table in (t.gl_statement_node, t.gl_statement_assignment):
repo.conn.execute(table.delete().where(table.c.fsv_id == fsv_id, table.c.version_no == vno))
row = {"fsv_id": fsv_id, "version_no": vno, "name": name, "kind": kind, "chart_id": chart_id, "status": "draft", "drafted_by": who(caller),
"published_by": None, "published_at": None}
_ver(repo).put(row)
for n in nodes:
_node(repo).add({"fsv_id": fsv_id, "version_no": vno, **n.model_dump()})
for a in assignments:
d = a.model_dump()
d["to_code"] = d["to_code"] or d["from_code"]
_asg(repo).add({"fsv_id": fsv_id, "version_no": vno, **d})
return _view(repo, row)
def _covers(code: str, lo: str, hi: str) -> bool:
return alloc.account_in_range(code, lo, hi)
def completeness(repo, v: dict) -> list:
"""The gate fsv_complete: every account of the chart with a balance (a statement account, or any movement for a cash-flow
version's balance-sheet accounts) assigned exactly once, the accounts named."""
asg = _rows(repo, t.gl_statement_assignment, v["fsv_id"], v["version_no"])
s = t.gl_split_line.c
comp = [c["company_code"] for c in repo.companies.all() if c["chart_id"] == v["chart_id"]]
if comp:
q = select(s.account_id, type_coerce(func.sum(s.functional_debit - s.functional_credit), MONEY), func.count()).where(
s.company_code.in_(comp)).group_by(s.account_id)
used = {r[0]: (r[1], r[2]) for r in repo.conn.execute(q)}
else:
used = {}
unassigned, doubled = [], []
for aid, (net, n) in used.items():
a = repo.accounts.get(aid)
if a is None or a["chart_id"] != v["chart_id"]:
continue
if v["kind"] == "cash_flow" and a["type"] not in BALANCE_SHEET:
continue
if (net or ZERO) == 0 and v["kind"] != "cash_flow":
pass
hits = [x for x in asg if _covers(a["code"], x["from_code"], x["to_code"])]
if not hits:
if (net or ZERO) != 0 or n:
unassigned.append(a["code"])
elif len(hits) > 1:
doubled.append(a["code"])
gs = []
gs.append(unmet("fsv_complete", "accounts with a balance are unassigned: " + ", ".join(sorted(unassigned)), "assign each exactly once")
if unassigned else met("fsv_complete"))
if doubled:
gs.append(unmet("fsv_complete", "accounts assigned more than once: " + ", ".join(sorted(doubled)), "assign each exactly once"))
return gs
@GL.use_case(kind="command", route="POST /statement-versions/{fsv_id}/publish-statement-version", permission=FSV_WRITE)
def publish_statement_version(caller: dict, fsv_id: Code, version_no: Optional[int] = None) -> dict:
"""Check and publish a draft version: refused while an account with a balance is unassigned or assigned twice (gate
fsv_complete, the accounts named). A published version is never edited again; reports may run on it."""
with repository() as repo, decide():
v = _version(repo, fsv_id, version_no)
if v["status"] == "published":
return _view(repo, v)
if v["status"] != "draft":
raise RuleBroken(f"{fsv_id} version {v['version_no']} is {v['status']}", status=409)
require(f"publish {fsv_id}", completeness(repo, v))
_ver(repo).set((fsv_id, v["version_no"]), status="published", published_by=who(caller), published_at=utc(real_now()))
return _view(repo, _version(repo, fsv_id, v["version_no"]))
@GL.use_case(kind="command", route="POST /statement-versions/{fsv_id}/supersede-statement-version", permission=FSV_WRITE)
def supersede_statement_version(caller: dict, fsv_id: Code, version_no: Optional[int] = None) -> dict:
"""Supersede a published version: it is marked superseded (reports already run keep naming it) and version_no + 1 is drafted as
a copy to change and publish."""
with repository() as repo, decide():
v = _version(repo, fsv_id, version_no, published=True)
_ver(repo).set((fsv_id, v["version_no"]), status="superseded")
new = {**v, "version_no": v["version_no"] + 1, "status": "draft", "drafted_by": who(caller), "published_by": None, "published_at": None}
for k in ("created_at", "updated_at"):
new.pop(k, None)
_ver(repo).add(new)
for n in _rows(repo, t.gl_statement_node, fsv_id, v["version_no"]):
_node(repo).add({**{k: n[k] for k in n if k not in ("created_at", "updated_at")}, "version_no": new["version_no"]})
for a in _rows(repo, t.gl_statement_assignment, fsv_id, v["version_no"]):
_asg(repo).add({**{k: a[k] for k in a if k not in ("created_at", "updated_at")}, "version_no": new["version_no"]})
return _view(repo, new)
@GL.use_case(kind="query", route="GET /statement-versions/{fsv_id}", permission=REPORT_READ)
def get_statement_version(caller: dict, fsv_id: Code, version_no: Optional[int] = None) -> dict:
"""A PUBLISHED statement version with its nodes and assignments (consolidation applies it to its consolidated figures); 404 when
it is not published. Pass version_no for a superseded one."""
with repository() as repo:
v = _version(repo, fsv_id, version_no, published=True) if version_no is None else _version(repo, fsv_id, version_no)
if version_no is not None and v["status"] == "draft":
raise Refused(404, f"statement version {fsv_id} {version_no} is not published")
return _view(repo, v)
@GL.use_case(kind="query", route="GET /statement-versions", permission=REPORT_READ)
def list_statement_versions(caller: dict) -> dict:
"""The statement versions, newest version first per statement, with their kind and status."""
with repository() as repo:
rows = [dict(r._mapping) for r in repo.conn.execute(t.gl_statement_version.select().order_by(
t.gl_statement_version.c.fsv_id, t.gl_statement_version.c.version_no.desc()))]
return {"versions": [{k: r[k] for k in ("fsv_id", "version_no", "name", "kind", "chart_id", "status")} for r in rows]}
# ── running a statement ──────────────────────────────────────────────────────────────────────────────────────────────
def _tree(repo, v: dict, amounts: dict, accounts: dict) -> list:
"""The version's tree with each node's signed total and the accounts behind it. `amounts` {account_id: net debit-positive}."""
nodes = {n["node_id"]: n for n in _rows(repo, t.gl_statement_node, v["fsv_id"], v["version_no"])}
asg = _rows(repo, t.gl_statement_assignment, v["fsv_id"], v["version_no"])
own: dict = {nid: [] for nid in nodes}
for aid, amt in amounts.items():
a = accounts[aid]
hits = [x for x in asg if _covers(a["code"], x["from_code"], x["to_code"])]
if not hits:
continue
x = hits[0]
target = x["node_id"]
if x["dc_rule"] == "debit_balance":
target = x["node_id"] if amt >= 0 else (x["other_node_id"] or x["node_id"])
elif x["dc_rule"] == "credit_balance":
target = x["node_id"] if amt <= 0 else (x["other_node_id"] or x["node_id"])
own[target].append({"account_id": aid, "code": a["code"], "name": a["name"], "amount": amt})
def build(nid: str) -> dict:
n = nodes[nid]
kids = [build(k["node_id"]) for k in sorted((x for x in nodes.values() if x["parent_node_id"] == nid), key=lambda x: (x["seq"], x["node_id"]))]
sign = n["sign"] or 1
leaf = sum((x["amount"] for x in own[nid]), ZERO) * sign
return {"node_id": nid, "label": n["label"], "sign": sign, "total": leaf + sum((k["total"] for k in kids), ZERO),
"accounts": [{**x, "amount": x["amount"] * sign} for x in sorted(own[nid], key=lambda x: x["code"])], "children": kids,
"cash_flow_class": n["cash_flow_class"]}
return [build(n["node_id"]) for n in sorted((x for x in nodes.values() if not x["parent_node_id"]), key=lambda x: (x["seq"], x["node_id"]))]
def _statement_amounts(repo, company: dict, ledger_id: str, year: int, start: date, end: date, *, adjustments: bool = True) -> tuple:
tb = trial_balance_data(repo, company, ledger_id, year, start, end, adjustments=adjustments)
amounts, accounts = {}, {}
for r in tb["rows"]:
accounts[r["account_id"]] = repo.accounts.get(r["account_id"])
amounts[r["account_id"]] = r["closing"] if r["type"] in BALANCE_SHEET else r["debit"] - r["credit"]
return amounts, accounts, tb
def _shift_year(d: date) -> date:
try:
return d.replace(year=d.year - 1)
except ValueError:
return d.replace(year=d.year - 1, day=28)
def _prior_window(repo, company: dict, start: date, end: date) -> tuple:
endm = int(company["financial_year_end_month"])
s, e = _shift_year(start), _shift_year(end)
if end == dates.end_of_month(end) and start == end.replace(day=1):
e = dates.end_of_month(e)
return dates.fiscal_year_of(e, endm), s, e
def _flatten(tree: list, out: dict) -> None:
for n in tree:
out[n["node_id"]] = n["total"]
_flatten(n["children"], out)
def _annotate(tree: list, other: dict, key: str) -> None:
for n in tree:
n[key] = other.get(n["node_id"], ZERO)
n[f"{key}_variance"] = n["total"] - n[key]
_annotate(n["children"], other, key)
@GL.use_case(kind="query", route="GET /reports/statement", permission=REPORT_READ)
def get_financial_statement(caller: dict, fsv_id: Code, company_code: Optional[CompanyCode] = None, ledger_id: Optional[LedgerCode] = None,
as_at: Optional[IsoDate] = None, period_id: Optional[PeriodId] = None, version_no: Optional[int] = None,
compare: Optional[Code] = None, budget_id: Optional[Code] = None, budget_version: Optional[int] = None,
by: Optional[Code] = None, include_adjustments: bool = True) -> dict:
"""Run a published statement version for a company and ledger: the profit and loss of the period alone (filtered by journal date,
not since the last close), the balance sheet as at the end with the year's result as its own line, or - for a cash-flow
version - the indirect-method cash-flow statement. Optionally beside the prior year (compare=prior_year) or an approved budget
(compare=budget, budget_id), and by profit_centre or segment. Each line opens to the accounts behind it. The answer names the
version and the ledger."""
with repository() as repo:
company = company_row(repo, caller, company_code)
v = _version(repo, fsv_id, version_no, published=True)
year, start, end = _window(repo, company, as_at, period_id)
lg = ledger_for(repo, company["company_code"], ledger_id, year)
if v["kind"] == "cash_flow":
return cash_flow_statement(repo, company, v, lg, year, start, end, include_adjustments)
amounts, accounts, tb = _statement_amounts(repo, company, lg, year, start, end, adjustments=include_adjustments)
tree = _tree(repo, v, amounts, accounts)
result = -sum((a for aid, a in amounts.items() if accounts[aid]["type"] not in BALANCE_SHEET), ZERO)
net_bs = sum((a for aid, a in amounts.items() if accounts[aid]["type"] in BALANCE_SHEET), ZERO)
out = {"company_code": company["company_code"], "ledger_id": lg, "fsv_id": fsv_id, "version_no": v["version_no"], "name": v["name"],
"kind": v["kind"], "from": start, "to": end, "fiscal_year": dates.fy_name(year), "functional_currency": company["functional_currency"],
"nodes": tree, "year_result": result, "balance_sheet_net": net_bs, "balanced": net_bs + (-result) == 0,
"unassigned": sorted({accounts[aid]["code"] for aid in amounts if not any(
_covers(accounts[aid]["code"], x["from_code"], x["to_code"]) for x in _rows(repo, t.gl_statement_assignment, fsv_id, v["version_no"]))})}
if compare == "prior_year":
py, ps, pe = _prior_window(repo, company, start, end)
pam, pac, _ = _statement_amounts(repo, company, lg, py, ps, pe, adjustments=include_adjustments)
flat: dict = {}
_flatten(_tree(repo, v, pam, pac), flat)
_annotate(tree, flat, "prior_year")
out["prior_window"] = {"from": ps, "to": pe}
elif compare == "budget":
from brick_gl.application.budgets import plan_by_account
plan, b = plan_by_account(repo, company, lg, budget_id, budget_version, start, end)
pamounts = {aid: p for aid, p in plan.items()}
pacc = {aid: repo.accounts.get(aid) for aid in pamounts}
flat = {}
_flatten(_tree(repo, v, {aid: (p if pacc[aid]["normal_balance"] == "debit" else -p) for aid, p in pamounts.items()}, pacc), flat)
_annotate(tree, flat, "budget")
out["budget"] = {"budget_id": b["budget_id"], "version": b["version"]}
if by in ("profit_centre", "segment"):
groups = {}
tbd = trial_balance_data(repo, company, lg, year, start, end, adjustments=include_adjustments, by=by)
vals = sorted({r[by] for r in tbd["rows"]})
for val in vals:
am = {}
for r in tbd["rows"]:
if r[by] == val:
am[r["account_id"]] = am.get(r["account_id"], ZERO) + (r["closing"] if r["type"] in BALANCE_SHEET else r["debit"] - r["credit"])
groups[val or ""] = _tree(repo, v, am, accounts)
out["by"] = {by: groups}
_ = today
return out
def cash_flow_statement(repo, company: dict, v: dict, ledger_id: str, year: int, start: date, end: date, adjustments: bool) -> dict:
"""The indirect-method cash-flow statement: from the period's result, add back non-cash items, working-capital movements,
investing and financing; the net cash flow equals the movement on the cash accounts to the penny, or the report lists the
unassigned movements that make the difference (gate cash_flow_complete). Unrealised FX on cash is its own line (IAS 7.28)."""
code = company["company_code"]
tb = trial_balance_data(repo, company, ledger_id, year, start, end, adjustments=adjustments)
nodes = {n["node_id"]: n for n in _rows(repo, t.gl_statement_node, v["fsv_id"], v["version_no"])}
asg = _rows(repo, t.gl_statement_assignment, v["fsv_id"], v["version_no"])
accounts = {r["account_id"]: repo.accounts.get(r["account_id"]) for r in tb["rows"]}
result = ZERO
cash_movement = ZERO
assigned: dict = {}
unassigned = []
for r in tb["rows"]:
a = accounts[r["account_id"]]
net = r["debit"] - r["credit"]
if a["type"] not in BALANCE_SHEET:
result -= net
continue
hit = next((x for x in asg if _covers(a["code"], x["from_code"], x["to_code"])), None)
if hit is None:
if net != 0:
unassigned.append({"code": a["code"], "movement": net})
continue
cls = nodes[hit["node_id"]]["cash_flow_class"]
if cls == "cash":
cash_movement += net
continue
assigned.setdefault(hit["node_id"], []).append({"account_id": a["account_id"], "code": a["code"], "name": a["name"], "amount": -net})
# unrealised FX (revaluation journals) on cash is its own line; its P&L side is not an operating cash item
s = t.gl_split_line.c
j = t.gl_journal.c
cash_ids = [r["account_id"] for r in tb["rows"] if accounts[r["account_id"]]["type"] in BALANCE_SHEET and any(
_covers(accounts[r["account_id"]]["code"], x["from_code"], x["to_code"]) and nodes[x["node_id"]]["cash_flow_class"] == "cash" for x in asg)]
fx_cash = ZERO
fx_pl = ZERO
if cash_ids:
q = (select(type_coerce(func.sum(s.functional_debit - s.functional_credit), MONEY)).select_from(
t.gl_split_line.join(t.gl_journal, j.journal_id == s.journal_id)).where(
s.company_code == code, s.journal_date >= start, s.journal_date <= end, j.event == "revaluation", s.account_id.in_(cash_ids)))
fx_cash = repo.conn.execute(q).scalar() or ZERO
pl_ids = [aid for aid, a in accounts.items() if a["type"] not in BALANCE_SHEET]
if pl_ids:
q = (select(type_coerce(func.sum(s.functional_debit - s.functional_credit), MONEY)).select_from(
t.gl_split_line.join(t.gl_journal, j.journal_id == s.journal_id)).where(
s.company_code == code, s.journal_date >= start, s.journal_date <= end, j.event == "revaluation", s.account_id.in_(pl_ids)))
fx_pl = repo.conn.execute(q).scalar() or ZERO
sections: dict = {}
for nid, items in assigned.items():
cls = nodes[nid]["cash_flow_class"]
sections.setdefault(cls, []).append({"node_id": nid, "label": nodes[nid]["label"], "amount": sum((i["amount"] for i in items), ZERO), "accounts": items})
adj_fx = fx_pl # remove the unrealised P&L effect from the result; it returns as the FX line
operating = result + adj_fx + sum((n["amount"] for n in sections.get("non_cash", []) + sections.get("operating", [])), ZERO)
investing = sum((n["amount"] for n in sections.get("investing", [])), ZERO)
financing = sum((n["amount"] for n in sections.get("financing", [])), ZERO)
net_flow = operating + investing + financing + fx_cash
difference = cash_movement - net_flow
return {"company_code": code, "ledger_id": ledger_id, "fsv_id": v["fsv_id"], "version_no": v["version_no"], "name": v["name"], "kind": "cash_flow",
"from": start, "to": end, "result": result, "unrealised_fx_adjustment": adj_fx,
"non_cash": sections.get("non_cash", []), "working_capital": sections.get("operating", []),
"operating_total": operating, "investing": sections.get("investing", []), "investing_total": investing,
"financing": sections.get("financing", []), "financing_total": financing,
"unrealised_fx_on_cash": fx_cash, "net_cash_flow": net_flow, "cash_movement": cash_movement, "difference": difference,
"unassigned_movements": unassigned,
"gate": (met("cash_flow_complete") if difference == 0 and not unassigned else unmet(
"cash_flow_complete", f"the net cash flow differs from the movement on the cash accounts by {difference}; unassigned: "
+ (", ".join(u["code"] for u in unassigned) or "none"), "assign each balance-sheet account's movement to a cash-flow node")).as_dict()}
_ = (Gate, sub_of, Decimal, carry)