dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/application/reports.py
"""The consolidated books and the group statements (CON-11).
A statement line is the sum of translated data and consolidation entries through a published statement version's assignment (gl's
version, read here, never edited). Each line opens to the companies, to the entries and (as a reference) to gl. A signed period's
statements are materialised at sign-off (cons_statement, keyed report, group, period, node) and read from there: they never change.
"""
import json
from decimal import Decimal
from erp_platform import work
from erp_platform.bricks import Refused
from brick_consolidation.application.runs import gl_call, prior_period
from brick_consolidation.domain import consolidation as C
from brick_consolidation.domain.errors import RuleBroken
from brick_consolidation.infrastructure.seams import ports
META = "@meta"
def consolidated(repo, group, gp, members=None) -> tuple:
"""(trial balance {(company, account, partner, movement, segment): amount}, the entries behind it) of a period at its run: the
translated data of the companies consolidated line by line (not the associates) plus the period's entries and the year's cumulative
manual entries so far."""
members = members or C.members_at(group, gp.period_id)
associates = {c for c, m in members.items() if m.method == "equity"}
tb: dict = {}
from brick_consolidation.application import carry
if carry.has_previous_year(group, gp.period_id): # the year opens with last year's carried balances (CON-15)
for r in carry.rows_for(repo, group, C.fiscal_year_of(gp.period_id, group.fiscal_year_end_month)):
k = (r["company_code"], r["group_account_id"], r["partner_company"], "", r["segment"])
tb[k] = tb.get(k, Decimal(0)) + Decimal(r["opening"])
for r in repo.data.translated(group.group_id, gp.period_id, gp.run_no):
if r["company_code"] not in associates:
k = (r["company_code"], r["group_account_id"], r["partner_company"], r["movement_type"], r["segment"])
tb[k] = tb.get(k, Decimal(0)) + r["ytd_group"]
carried = C.fiscal_months(gp.period_id, group.fiscal_year_end_month)[:-1]
entries = repo.data.books_entries(group.group_id, gp.period_id, carried)
for e in entries:
for ln in e["lines"]:
k = (ln["company_code"], ln["group_account_id"], ln["partner_company"], ln["movement_type"], ln["segment"])
tb[k] = tb.get(k, Decimal(0)) + ln["amount"]
return {k: v for k, v in tb.items() if v}, entries
def _rows(tb: dict) -> list:
return [(*k, v) for k, v in tb.items()]
def _version(group, fsv_id: str) -> dict:
v = gl_call(ports.gl.statement_version, fsv_id)
if v.get("status") not in (None, "published"):
raise RuleBroken(f"statement version {fsv_id} is {v.get('status')}, not published: a report runs on a published version")
return v
def _codes(group) -> dict:
return {str(a["account_id"]): str(a.get("code") or a["account_id"]) for a in gl_call(ports.gl.chart_accounts, group.group_chart_id)}
def _period_before_year(group, period_id: str) -> str:
return prior_period(C.fiscal_months(period_id, group.fiscal_year_end_month)[0])
def _opening(repo, group, period_id: str) -> list:
"""The consolidated trial balance at the previous year's end, when the group has that period; otherwise nothing (the group starts here)."""
before = _period_before_year(group, period_id)
gp0 = next((x for x in group.group_periods if x.period_id == before), None)
return _rows(consolidated(repo, group, gp0)[0]) if gp0 and gp0.status in ("eliminated", "signed") else []
def compute(repo, group, gp, fsv_id: str, tb: dict, entries: list, *, prior: bool = True) -> dict:
version = _version(group, fsv_id)
code_of = _codes(group)
closing = _rows(tb)
opening = _opening(repo, group, gp.period_id) if any(a.get("basis") == "movement" for a in version["assignments"]) else []
lines = C.statement_lines(version, code_of, closing, opening)
by_company = C.statement_lines(version, code_of, closing, opening, by="company")["nodes"]
by_segment = C.statement_lines(version, code_of, closing, opening, by="segment")["nodes"]
year, month = int(gp.period_id[:4]), gp.period_id[5:]
before = f"{year - 1:04d}-{month}"
prior_amounts: dict = {}
if prior:
stored = repo.data.statements(group.group_id, before, [fsv_id])
if stored:
prior_amounts = {r["node_id"]: r["amount"] for r in stored if "@" not in r["node_id"]}
else:
gp_before = next((x for x in group.group_periods if x.period_id == before and x.status in ("eliminated", "signed")), None)
if gp_before:
tb0, e0 = consolidated(repo, group, gp_before)
prior_amounts = {n: v["amount"] for n, v in C.statement_lines(version, code_of, _rows(tb0), [])["nodes"].items()}
node_of = lines["node_of"]
behind: dict = {}
for e in entries:
for ln in e["lines"]:
node = node_of.get(ln["group_account_id"])
if node:
behind.setdefault(node, {}).setdefault(e["cons_journal_id"], {"cons_journal_id": e["cons_journal_id"], "document_type": e["document_type"],
"entry_no": e["entry_no"], "amount": Decimal(0)})["amount"] += ln["amount"]
out = []
for n in sorted(version["nodes"], key=lambda n: (str(n.get("parent_node_id") or ""), n.get("seq") or 0, n["node_id"])):
nid = n["node_id"]
out.append({"node_id": nid, "parent_node_id": n.get("parent_node_id"), "seq": n.get("seq"), "label": n.get("label"),
"cash_flow_class": n.get("cash_flow_class"), "amount": lines["nodes"][nid]["amount"],
"by_company": by_company[nid]["split"], "by_segment": by_segment[nid]["split"],
"prior_year": prior_amounts.get(nid) if prior else None,
"entries": sorted(({**x, "amount": C.q2(x["amount"])} for x in behind.get(nid, {}).values()), key=lambda x: x["cons_journal_id"])})
report = {"report": fsv_id, "kind": version["kind"], "fsv_id": fsv_id, "version_no": version["version_no"], "ledger_id": group.ledger_id,
"run_no": gp.run_no, "group_id": group.group_id, "period_id": gp.period_id, "currency": group.group_currency, "lines": out,
"unassigned": lines["unassigned"], "materialised": False, "tb_digest": C.tb_digest(tb),
"drill_down": {"companies": "by_company on each line", "entries": "entries on each line",
"gl": "a company's figure opens to gl's line history (GL-04) — gl lists no lines yet (no route in gl's signed spec)"}}
if version["kind"] == "cash_flow":
parents = {x["parent_node_id"] for x in out}
total = {}
for line in out:
if line["node_id"] not in parents: # a leaf carries its own rows; a parent only sums its children
total[line["cash_flow_class"]] = total.get(line["cash_flow_class"], Decimal(0)) + line["amount"]
flows = {k: C.q2(total.get(k, Decimal(0))) for k in ("operating", "investing", "financing", "cash")}
report["cash_flow"] = {**flows, "translation_difference_on_cash": C.q2(flows["cash"] - flows["operating"] - flows["investing"] - flows["financing"])}
return report
def run(repo, group, gp, fsv_id: str) -> dict:
"""The statement of a period: from the materialised rows when it is signed, else computed now."""
if gp.status == "signed":
stored = repo.data.statements(group.group_id, gp.period_id, [fsv_id])
if stored:
return _from_stored(group, gp, fsv_id, stored)
tb, entries = consolidated(repo, group, gp)
return compute(repo, group, gp, fsv_id, tb, entries)
def _from_stored(group, gp, fsv_id: str, stored: list) -> dict:
meta = json.loads(next(r for r in stored if r["node_id"] == META)["prior_year"])
lines = {}
for r in stored:
if r["node_id"] == META:
continue
node, _, split = r["node_id"].partition("@")
row = lines.setdefault(node, {"node_id": node, "amount": None, "by_company": {}, "by_segment": {}, "prior_year": None})
if not split:
row["amount"], row["prior_year"] = r["amount"], (Decimal(r["prior_year"]) if r["prior_year"] else None)
else:
kind, _, key = split.partition(":")
row["by_company" if kind == "company" else "by_segment"][key] = r["amount"]
return {**meta, "lines": [lines[n] | {"label": meta["labels"].get(n)} for n in sorted(lines)], "materialised": True,
"produced_at": next(r for r in stored if r["node_id"] == META)["produced_at"], "tb_digest": meta["tb_digest"]}
def statement_rows(repo, group, gp, tb: dict) -> tuple:
"""(rows, versions) of the group's statements of a period as they are materialised: every version the group names, each node with the companies'
and segments' contributions. produced_at is the sign-off's own moment, so producing them again gives the same rows."""
_tb, entries = consolidated(repo, group, gp)
digest, rows, done, now = C.tb_digest(tb), [], [], gp.signed_at or work.now()
for fsv_id in [x for x in (group.fsv_id, group.cash_flow_fsv_id) if x]:
rep = compute(repo, group, gp, fsv_id, tb, entries)
for line in rep["lines"]:
base = {"report": fsv_id, "group_id": group.group_id, "period_id": gp.period_id, "produced_at": now, "digest": digest}
rows.append({**base, "node_id": line["node_id"], "amount": line["amount"],
"prior_year": "" if line["prior_year"] is None else str(line["prior_year"])})
rows += [{**base, "node_id": f"{line['node_id']}@company:{k}", "amount": v, "prior_year": ""} for k, v in line["by_company"].items()]
rows += [{**base, "node_id": f"{line['node_id']}@segment:{k}", "amount": v, "prior_year": ""} for k, v in line["by_segment"].items()]
meta = {k: rep[k] for k in ("report", "kind", "fsv_id", "version_no", "ledger_id", "run_no", "group_id", "period_id", "currency", "tb_digest")}
meta["labels"] = {line["node_id"]: line["label"] for line in rep["lines"]}
rows.append({"report": fsv_id, "group_id": group.group_id, "period_id": gp.period_id, "node_id": META, "amount": Decimal("0.00"),
"prior_year": json.dumps(meta, sort_keys=True), "produced_at": now, "digest": digest})
done.append(fsv_id)
return rows, done
def materialise(repo, group, gp, tb: dict) -> list:
"""At sign-off: the statements of every version the group names, with the companies and segments, as signed. Keyed (report, group,
period, node): the second run writes the same rows."""
rows, done = statement_rows(repo, group, gp, tb)
repo.data.write_statements(rows)
return done
def materialise_signed(repo, group, gp) -> dict:
"""CON-21: the stored statements of a signed period equal a fresh computation at the sign-off digest, and producing them again changes
nothing. A signed period's books cannot move (a later change is a difference until it is reopened); if they have, nothing is written and the
difference is named."""
tb, _entries = consolidated(repo, group, gp)
if C.tb_digest(tb) != gp.tb_digest:
raise Refused(409, f"{gp.period_id}'s consolidated books no longer match the digest it was signed at ({gp.tb_digest[:8]} -> {C.tb_digest(tb)[:8]}): "
"its statements stay as signed", gate="signed_digest", fix="reopen the period (reopen-group-period) to bring the change in")
rows, done = statement_rows(repo, group, gp, tb)
held = {(r["report"], r["node_id"]): (r["amount"], r["prior_year"] or "", r["digest"])
for r in repo.data.statements(group.group_id, gp.period_id, done)}
fresh = {(r["report"], r["node_id"]): (r["amount"], r["prior_year"] or "", r["digest"]) for r in rows}
differing = sorted(k for k in fresh if held.get(k) != fresh[k])
if differing:
repo.data.write_statements([r for r in rows if (r["report"], r["node_id"]) in set(differing)])
return {"period_id": gp.period_id, "tb_digest": gp.tb_digest, "reports": done, "rows": len(rows), "written": len(differing), "changed": bool(differing)}
def equity_changes(repo, group, gp) -> dict:
"""The statement of changes in equity: the equity accounts of the consolidated books by movement type (opening, result, dividend, capital
change, translation, ownership change), and by company."""
tb, _entries = consolidated(repo, group, gp)
equity = {r.group_account_id for r in group.account_rules if r.equity == "true"}
movements, by_company = {}, {}
for (c, a, _p, m, _s), v in tb.items():
if a in equity:
key = m or "unclassified"
movements[key] = movements.get(key, Decimal(0)) + v
by_company.setdefault(c, {})[key] = by_company.get(c, {}).get(key, Decimal(0)) + v
return {"report": "equity_changes", "group_id": group.group_id, "period_id": gp.period_id, "run_no": gp.run_no, "currency": group.group_currency,
"movements": {k: C.q2(v) for k, v in sorted(movements.items())}, "closing": C.q2(sum(movements.values(), Decimal(0))),
"by_company": {c: {k: C.q2(v) for k, v in m.items()} for c, m in sorted(by_company.items())}, "materialised": False}