dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/reports.py
"""Reading the books: the trial balance, an account's lines, the rollup of the account tree, an account's balance, and the
materialised statements of a closed period. Everything is DERIVED from split lines (and the carry-forward) with SUM; nothing
here writes a balance. A ledger's figures are its common journals plus its own ledger-specific ones, never another ledger's.
Amounts are in the company's functional currency unless a column says it is the account's own. Debit is positive.
"""
from datetime import date
from decimal import Decimal
from typing import Annotated, Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId
from pydantic import Field
from brick_gl.application import carry
from brick_gl.application.brick import (
GL,
JOURNAL_READ,
PERIOD_READ,
REPORT_READ,
STATEMENT_MATERIALISE,
company_row,
decide,
ledger_for,
real_now,
repository,
today,
)
from brick_gl.application.inputs import CompanyCode, IsoDate, LedgerCode, PeriodId
from brick_gl.application.posting import number_of
from brick_gl.domain import dates
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.money import ZERO
from brick_gl.domain.values import BALANCE_SHEET
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import utc
def _window(repo, company: dict, as_at: Optional[str], period_id: Optional[str]) -> tuple:
"""(fiscal year number, window start, window end) of a report: a period, or the financial year to a date."""
code = company["company_code"]
endm = int(company["financial_year_end_month"])
if period_id:
p = repo.periods.get(code, period_id)
if p is None:
raise Refused(404, f"no period {period_id} of {code}")
return dates.fy_number(p["fiscal_year"]), p["starts_on"], p["ends_on"]
on = dates.parse(as_at) if as_at else today(repo)
year = dates.fiscal_year_of(on, endm)
start, _ = dates.fy_bounds(year, endm)
return year, start, on
def trial_balance_data(repo, company: dict, ledger_id: str, year: int, start: date, end: date, *, adjustments: bool = True,
by: Optional[str] = None) -> dict:
"""The trial balance of one ledger for a window of a financial year: per account its opening (the carry-forward and the
year's movement before the window), the window's movement and its closing."""
code = company["company_code"]
fy_start, fy_end = carry.year_bounds(company, year)
dim = by in ("profit_centre", "segment")
cf = carry.opening_for_report(repo, company, ledger_id, year)
grp = ("account_id", "profit_centre", "segment", "partner_company")
before = repo.ledger.sums(code, ledger_id, since=fy_start, upto=start - dates.timedelta(days=1), group_by=grp,
adjustments=adjustments) if start > fy_start else []
window = repo.ledger.sums(code, ledger_id, since=start, upto=end, group_by=grp, adjustments=adjustments)
accounts = {}
def acct(aid):
if aid not in accounts:
accounts[aid] = repo.accounts.get(aid)
return accounts[aid]
cells: dict = {} # (account, dimension value) -> figures
def cell(aid, pc, seg):
dv = {"profit_centre": pc, "segment": seg}.get(by or "", "")
return cells.setdefault((aid, dv), {"opening": ZERO, "debit": ZERO, "credit": ZERO, "opening_cur": ZERO, "debit_cur": ZERO,
"credit_cur": ZERO})
for key, (opening_f, opening_cur) in cf.items():
a = acct(key[0])
if a and a["type"] in BALANCE_SHEET:
c = cell(key[0], key[1], key[2])
c["opening"] += opening_f
c["opening_cur"] += opening_cur
for r in before:
c = cell(r["account_id"], r["profit_centre"] or "", r["segment"] or "")
c["opening"] += r["f_debit"] - r["f_credit"]
c["opening_cur"] += r["debit"] - r["credit"]
for r in window:
c = cell(r["account_id"], r["profit_centre"] or "", r["segment"] or "")
c["debit"] += r["f_debit"]
c["credit"] += r["f_credit"]
c["debit_cur"] += r["debit"]
c["credit_cur"] += r["credit"]
rows = []
total_d = total_c = total_open = ZERO
groups: dict = {}
for (aid, dv), c in sorted(cells.items(), key=lambda kv: (acct(kv[0][0])["code"], kv[0][1])):
closing = c["opening"] + c["debit"] - c["credit"]
if c["opening"] == 0 and c["debit"] == 0 and c["credit"] == 0 and closing == 0:
continue
a = acct(aid)
row = {"account_id": aid, "code": a["code"], "name": a["name"], "type": a["type"], "opening": c["opening"],
"debit": c["debit"], "credit": c["credit"], "closing": closing,
"closing_debit": closing if closing > 0 else ZERO, "closing_credit": -closing if closing < 0 else ZERO}
if a["currency"]:
row["currency"] = a["currency"]
row["closing_currency"] = c["opening_cur"] + c["debit_cur"] - c["credit_cur"]
if dim:
row[by] = dv
g = groups.setdefault(dv, ZERO)
groups[dv] = g + closing
rows.append(row)
total_d += row["closing_debit"]
total_c += row["closing_credit"]
total_open += c["opening"]
out = {"ledger_id": ledger_id, "fiscal_year": dates.fy_name(year), "from": start, "to": end,
"functional_currency": company["functional_currency"], "include_adjustments": adjustments, "rows": rows,
"total_debit": total_d, "total_credit": total_c, "difference": total_d - total_c,
"total_movement_debit": sum((r["debit"] for r in rows), ZERO), "total_movement_credit": sum((r["credit"] for r in rows), ZERO)}
if dim:
out["groups"] = [{by: k, "net": v} for k, v in sorted(groups.items())]
return out
@GL.use_case(kind="query", route="GET /reports/trial-balance", permission=REPORT_READ)
def get_trial_balance(caller: dict, company_code: Optional[CompanyCode] = None, ledger_id: Optional[LedgerCode] = None,
as_at: Optional[IsoDate] = None, period_id: Optional[PeriodId] = None, include_adjustments: bool = True,
by: Optional[str] = None) -> dict:
"""The trial balance of a company and ledger (the leading ledger by default) as at a date, or for a period: per account its
opening (the carry-forward), movement and closing in the functional currency, and in its own currency where the account has
one. Total debits equal total credits exactly (difference 0, no float tail). With document splitting on, `by` =
profit_centre or segment shows each group netting to zero. A ledger's figures are its common journals plus its own
ledger-specific ones. `include_adjustments` false gives the pre-audit trial balance."""
with repository() as repo:
company = company_row(repo, caller, company_code)
year, start, end = _window(repo, company, as_at, period_id)
lg = ledger_for(repo, company["company_code"], ledger_id, year)
if by not in (None, "profit_centre", "segment"):
raise Refused(422, "by is profit_centre or segment")
out = trial_balance_data(repo, company, lg, year, start, end, adjustments=include_adjustments, by=by)
out["company_code"] = company["company_code"]
return out
def balance_at(repo, company: dict, ledger_id: str, account_id: int, on: date) -> tuple:
"""(functional balance, balance in the transaction currencies) of one account up to a day: the sum of its posted split lines."""
rows = repo.ledger.sums(company["company_code"], ledger_id, upto=on, account_ids=[account_id], group_by=("account_id",))
if not rows:
return ZERO, ZERO
r = rows[0]
return r["f_debit"] - r["f_credit"], r["debit"] - r["credit"]
@GL.use_case(kind="query", route="GET /accounts/{account_id}/lines", permission=JOURNAL_READ, aliases={"date_from": "from", "date_to": "to"})
def get_account_lines(caller: dict, account_id: BigId, company_code: Optional[CompanyCode] = None, ledger_id: Optional[LedgerCode] = None,
date_from: Optional[IsoDate] = None, date_to: Optional[IsoDate] = None, group_account_id: Optional[BigId] = None,
limit: Annotated[int, Field(ge=1, le=1000)] = 1000, offset: Annotated[int, Field(ge=0, le=10_000_000)] = 0) -> dict:
"""Every entry behind an account balance, for a date range: each line with its journal and number, its date, its source module
and document, its party and memo and dimensions, oldest to newest, with a running balance that ends on the account's balance
at the end of the range (GL-04). A reversal and its original both show; nothing is hidden. v6 (seam GL-04): with `group_account_id`,
instead of the path account every account mapped to that group account in force at the end of the range; a page at a time (limit,
offset; `more` says another follows) - the running balance carries across pages and closing_balance is the account's balance at
the end of the range, so the last line's running_balance equals it. `entry_no` is the journal's number and `bp_id` the line's party.
v7: each line names `account_id`, the local account it posted to (with `group_account_id`, one of the accounts mapped to it), and the path
then carries the SAME group account id (a different one is refused 422). The range is `from` and `to` on the wire."""
with repository() as repo:
company = company_row(repo, caller, company_code)
a = repo.accounts.get(int(account_id))
if a is None:
raise Refused(404, f"no account {account_id}")
if group_account_id is not None and int(group_account_id) != int(account_id):
raise Refused(422, f"the path carries group account {account_id} and group_account_id is {group_account_id}: they are the same value", gate="group_account_matches_path",
fix="send the group account id in both places")
year, _, _ = _window(repo, company, None, None)
lg = ledger_for(repo, company["company_code"], ledger_id, year)
since, upto = dates.parse(date_from), dates.parse(date_to)
ids = repo.group_map.accounts_of(int(group_account_id), upto or today(repo)) if group_account_id else [a["account_id"]]
opening = (sum((balance_at(repo, company, lg, i, since - dates.timedelta(days=1))[0] for i in ids), ZERO) if since else ZERO)
lines = repo.ledger.lines(company["company_code"], lg, ids if group_account_id else a["account_id"], since, upto) if ids else []
lines.sort(key=lambda ln: (ln["journal_date"], ln["journal_id"], ln["line_no"], ln["split_no"]))
running = opening
out = []
for ln in lines:
net = ln["functional_debit"] - ln["functional_credit"]
running += net
out.append({"account_id": ln["account_id"], "journal_id": ln["journal_id"], "number": number_of(ln), "entry_no": ln["entry_no"], "bp_id": ln["bp_id"], "line_no": ln["line_no"], "journal_date": ln["journal_date"],
"source_module": ln["source_module"], "source_document": ln["source_document"], "event": ln["event"],
"memo": ln["memo"], "debit": ln["functional_debit"], "credit": ln["functional_credit"],
"transaction_debit": ln["debit"], "transaction_credit": ln["credit"], "currency": ln["currency"],
"profit_centre": ln["profit_centre"], "segment": ln["segment"], "partner_company": ln["partner_company"],
"is_clearing": ln["is_clearing"], "running_balance": running})
return {"account_id": a["account_id"], "code": a["code"], "name": a["name"], "company_code": company["company_code"],
"ledger_id": lg, "from": since, "to": upto, "opening_balance": opening, "lines": out[offset:offset + limit],
"more": offset + limit < len(out), "closing_balance": running}
@GL.use_case(kind="query", route="GET /accounts/{account_id}/balance", permission=REPORT_READ)
def get_account_balance(caller: dict, account_id: BigId, as_of: Optional[IsoDate] = None, company_code: Optional[CompanyCode] = None) -> dict:
"""An account's balance as of a date, the sum of its posted lines up to it in the leading ledger (payments reads it as the
book side of 'book equals bank'): {account_id, company_code, as_of, balance, currency}. An unknown account, or a company
the caller may not see, is a 404. Read only."""
with repository() as repo:
company = company_row(repo, caller, company_code)
a = repo.accounts.get(int(account_id))
if a is None:
raise Refused(404, f"no account {account_id}")
on = dates.parse(as_of) if as_of else today(repo)
year = dates.fiscal_year_of(on, int(company["financial_year_end_month"]))
lg = ledger_for(repo, company["company_code"], None, year)
f, cur = balance_at(repo, company, lg, a["account_id"], on)
return {"account_id": a["account_id"], "company_code": company["company_code"], "as_of": on, "balance": f,
"currency": company["functional_currency"]}
@GL.use_case(kind="query", route="GET /accounts/{account_id}/rollup", permission=REPORT_READ)
def get_account_rollup(caller: dict, account_id: BigId, as_at: Optional[IsoDate] = None, company_code: Optional[CompanyCode] = None,
ledger_id: Optional[LedgerCode] = None) -> dict:
"""The balance of an account with everything beneath it however deep (the account tree is walked from any node): the parent
shows the total of its children's totals plus its own, and the answer is the tree with each node's own and rolled-up
balance as at a date."""
with repository() as repo:
company = company_row(repo, caller, company_code)
root = repo.accounts.get(int(account_id))
if root is None:
raise Refused(404, f"no account {account_id}")
on = dates.parse(as_at) if as_at else today(repo)
year = dates.fiscal_year_of(on, int(company["financial_year_end_month"]))
lg = ledger_for(repo, company["company_code"], ledger_id, year)
edges: dict = {}
for p, c in repo.accounts.edges():
edges.setdefault(p, []).append(c)
ids, stack = [], [root["account_id"]]
seen = set()
while stack:
n = stack.pop()
if n in seen:
continue
seen.add(n)
ids.append(n)
stack.extend(edges.get(n, []))
sums = {r["account_id"]: r["f_debit"] - r["f_credit"] for r in
repo.ledger.sums(company["company_code"], lg, upto=on, account_ids=ids, group_by=("account_id",))}
def node(n: int) -> dict:
a = repo.accounts.get(n)
kids = [node(k) for k in sorted(edges.get(n, []))]
own = sums.get(n, ZERO)
return {"account_id": n, "code": a["code"], "name": a["name"], "own": own,
"total": own + sum((k["total"] for k in kids), ZERO), "children": kids}
tree = node(root["account_id"])
return {"company_code": company["company_code"], "ledger_id": lg, "as_at": on, **tree}
# ── the statements of a closed period (GL-21) ────────────────────────────────────────────────────────────────────────
def materialise(repo, company: dict, period: dict, at) -> int:
"""Store the trial balance of a period per ledger (keyed report, company, ledger, period, account); running it again
changes nothing (a row whose figures are unchanged keeps its produced_at)."""
from erp_platform.repository import upsert
code = company["company_code"]
year = dates.fy_number(period["fiscal_year"])
n = 0
for ledger_id in repo.company_ledgers.assigned(code, year):
tb = trial_balance_data(repo, company, ledger_id, year, period["starts_on"], period["ends_on"])
stored = {r["account_id"]: r for r in (dict(x._mapping) for x in repo.conn.execute(
t.gl_statement.select().where(t.gl_statement.c.report == "trial_balance", t.gl_statement.c.company_code == code,
t.gl_statement.c.ledger_id == ledger_id, t.gl_statement.c.period_id == period["period_id"])))}
rows = []
for r in tb["rows"]:
old = stored.get(r["account_id"])
same = old and (old["debit"], old["credit"], old["closing"]) == (r["debit"], r["credit"], r["closing"])
rows.append({"report": "trial_balance", "company_code": code, "ledger_id": ledger_id, "period_id": period["period_id"],
"account_id": r["account_id"], "debit": r["debit"], "credit": r["credit"], "closing": r["closing"],
"produced_at": old["produced_at"] if same else utc(at)})
if rows:
upsert(repo.conn, t.gl_statement, rows, ("report", "company_code", "ledger_id", "period_id", "account_id"),
insert_only=("created_at",))
n += len(rows)
repo.conn.execute(t.gl_period.update().where(t.gl_period.c.company_code == code, t.gl_period.c.period_id == period["period_id"]).values(
statements_stale=False))
return n
@GL.use_case(kind="command", route="POST /periods/{period_id}/materialise-statements", permission=STATEMENT_MATERIALISE, idempotent=True)
def materialise_statements(caller: dict, period_id: PeriodId, company_code: Optional[CompanyCode] = None) -> dict:
"""The system materialises the trial balance of a CLOSED period so that a report of it is fast and stable: the stored figures
match a fresh computation, and re-materialising changes nothing."""
with repository() as repo, decide():
company = company_row(repo, caller, company_code, scoped=False)
p = repo.periods.get(company["company_code"], period_id)
if p is None:
raise Refused(404, f"no period {period_id}")
if p["status"] != "closed":
raise RuleBroken(f"{period_id} is open: only a closed period's statements are materialised", status=409)
return {"company_code": company["company_code"], "period_id": period_id, "rows": materialise(repo, company, p, real_now())}
@GL.use_case(kind="query", route="GET /periods/{period_id}/statements", permission=PERIOD_READ)
def get_materialised_statements(caller: dict, period_id: PeriodId, company_code: Optional[CompanyCode] = None,
ledger_id: Optional[LedgerCode] = None) -> dict:
"""The stored trial balance of a closed period (materialised when it closed, stale after a reopen)."""
with repository() as repo:
company = company_row(repo, caller, company_code)
p = repo.periods.get(company["company_code"], period_id)
if p is None:
raise Refused(404, f"no period {period_id}")
year = dates.fy_number(p["fiscal_year"])
lg = ledger_for(repo, company["company_code"], ledger_id, year)
rows = [dict(r._mapping) for r in repo.conn.execute(t.gl_statement.select().where(
t.gl_statement.c.report == "trial_balance", t.gl_statement.c.company_code == company["company_code"],
t.gl_statement.c.ledger_id == lg, t.gl_statement.c.period_id == period_id))]
return {"company_code": company["company_code"], "period_id": period_id, "ledger_id": lg, "stale": bool(p["statements_stale"]),
"rows": [{k: r[k] for k in ("account_id", "debit", "credit", "closing", "produced_at")} for r in rows]}
_ = Decimal