dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/carry.py
"""The balance carry-forward (GL-23): each year's opening balances are DERIVED from the previous year's closing ones, so the
balance sheet carries forward and the year's result lands in retained earnings WITHOUT a closing entry.
opening(ledger, Y) = carried(ledger, Y-1)
carried(ledger, y) = balance-sheet accounts' (opening(y) + movement(y)), per profit centre, segment and partner company,
+ the SUM of income and expense movement of y on retained_earnings (per segment)
A ledger added later (GL-42) opens its first year from the LEADING ledger's carried figures, then keeps its own books. Rows
are derived (never typed) and written with every value in VALUES, so deriving again writes the same rows; they are stale
when a posting landed in a prior year after they were derived (an audit adjustment, a reopened period).
"""
import hashlib
import json
from datetime import date
from typing import Optional
from brick_gl.domain import dates
from brick_gl.domain.money import ZERO
from brick_gl.domain.readiness import refusal
from brick_gl.domain.values import BALANCE_SHEET
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import utc
Key = tuple # (account_id, profit_centre, segment, partner_company)
def year_bounds(company: dict, year: int) -> tuple:
return dates.fy_bounds(year, int(company["financial_year_end_month"]))
def first_year(repo, company_code: str) -> Optional[int]:
years = repo.years.of_company(company_code)
return dates.fy_number(years[0]["fiscal_year"]) if years else None
def movement(repo, company: dict, ledger_id: str, year: int, *, upto: Optional[date] = None, adjustments: bool = True,
since: Optional[date] = None) -> dict:
"""{key: (f_debit, f_credit, debit, credit)} of split lines in a financial year (optionally up to a day / from a day)."""
start, end = year_bounds(company, year)
rows = repo.ledger.sums(company["company_code"], ledger_id, since=since or start, upto=min(upto, end) if upto else end,
group_by=("account_id", "profit_centre", "segment", "partner_company"), adjustments=adjustments)
return {(r["account_id"], r["profit_centre"] or "", r["segment"] or "", r["partner_company"] or ""):
(r["f_debit"], r["f_credit"], r["debit"], r["credit"]) for r in rows}
def retained_earnings_account(repo, company: dict, on: date) -> Optional[int]:
return repo.account_map.account(company["chart_id"], "retained_earnings", "", on)
def carried(repo, company: dict, ledger_id: str, year: int, memo: dict, *, need_re: bool = True) -> dict:
"""What year `year` of the ledger hands to the next: {key: [opening_f, opening_cur]}."""
k = (company["company_code"], ledger_id, year)
if k in memo:
return memo[k]
out: dict = {key: list(v) for key, v in opening(repo, company, ledger_id, year, memo).items()}
mv = movement(repo, company, ledger_id, year)
types: dict = {}
pl_by_segment: dict = {}
for key, (fd, fc, d, c) in mv.items():
acc = key[0]
if acc not in types:
a = repo.accounts.get(acc)
types[acc] = a["type"] if a else "asset"
if types[acc] in BALANCE_SHEET:
cur = out.setdefault(key, [ZERO, ZERO])
cur[0] += fd - fc
cur[1] += d - c
else:
seg = key[2]
pl_by_segment[seg] = pl_by_segment.get(seg, ZERO) + (fd - fc)
if any(v != 0 for v in pl_by_segment.values()):
_, end = year_bounds(company, year)
re_acc = retained_earnings_account(repo, company, end)
if re_acc is None:
if need_re:
raise refusal("retained_earnings_mapped", f"no retained_earnings account is in the account map on {end}, the last "
f"day of {dates.fy_name(year)}", "Account map (GL-12)", status=409)
else:
for seg, amount in pl_by_segment.items():
if amount != 0:
cur = out.setdefault((re_acc, "", seg, ""), [ZERO, ZERO])
cur[0] += amount
memo[k] = out
return out
def opening(repo, company: dict, ledger_id: str, year: int, memo: Optional[dict] = None) -> dict:
"""The opening balances of a year, derived live from the books before it: {key: [opening_f, opening_cur]}."""
memo = memo if memo is not None else {}
code = company["company_code"]
fy0 = first_year(repo, code)
if fy0 is None or year <= fy0:
return {}
source = ledger_id
link = repo.company_ledgers.get(code, ledger_id)
lg = repo.ledgers.get(ledger_id)
if lg and lg["kind"] == "non_leading" and link and link["from_fiscal_year"] and dates.fy_number(link["from_fiscal_year"]) == year:
lead = repo.ledgers.leading()
source = lead["ledger_id"] if lead else ledger_id
return carried(repo, company, source, year - 1, memo)
def digest(rows: dict) -> str:
body = sorted([list(k) + [str(v[0]), str(v[1])] for k, v in rows.items()])
return hashlib.sha256(json.dumps(body).encode()).hexdigest()
def last_prior_posting(repo, company: dict, year: int):
"""The latest posting into any year BEFORE `year`: the moment a derived carry-forward goes stale."""
from sqlalchemy import func, select
start, _ = year_bounds(company, year)
return repo.conn.execute(select(func.max(t.gl_journal.c.posted_at)).where(
t.gl_journal.c.company_code == company["company_code"], t.gl_journal.c.journal_date < start,
t.gl_journal.c.status.in_(("posted", "reversed")))).scalar()
def derive(repo, company: dict, year: int, at) -> dict:
"""Derive and write the carry-forward INTO `year` for every ledger the company keeps in it: (rows written per ledger)."""
code = company["company_code"]
wrote = {}
for ledger_id in repo.company_ledgers.assigned(code, year):
rows = opening(repo, company, ledger_id, year)
existing = {(r["account_id"], r["profit_centre"] or "", r["segment"] or "", r["partner_company"] or "")
for r in repo.ledger.carry(code, ledger_id, dates.fy_name(year))}
dg = digest(rows)
out = []
for key in sorted(set(rows) | existing):
f, cur = rows.get(key, (ZERO, ZERO))
out.append({"company_code": code, "ledger_id": ledger_id, "fiscal_year": dates.fy_name(year), "account_id": key[0],
"profit_centre": key[1], "segment": key[2], "partner_company": key[3], "opening": f, "opening_currency": cur,
"derived_at": utc(at), "source_digest": dg})
repo.ledger.put_carry(out)
wrote[ledger_id] = len(out)
return wrote
def stored_opening(repo, company: dict, ledger_id: str, year: int) -> Optional[dict]:
"""The stored, CURRENT carry-forward into a year, or None (not derived, or stale)."""
rows = repo.ledger.carry(company["company_code"], ledger_id, dates.fy_name(year))
if not rows:
return None
last = last_prior_posting(repo, company, year)
derived = max(r["derived_at"] for r in rows)
if last is not None and derived is not None and derived < last:
return None
return {(r["account_id"], r["profit_centre"] or "", r["segment"] or "", r["partner_company"] or ""):
[r["opening"] or ZERO, r["opening_currency"] or ZERO] for r in rows}
def opening_for_report(repo, company: dict, ledger_id: str, year: int) -> dict:
s = stored_opening(repo, company, ledger_id, year)
return s if s is not None else {k: list(v) for k, v in opening(repo, company, ledger_id, year).items()}
def is_current(repo, company: dict, year: int) -> tuple:
"""(current?, reason): the stored carry-forward into `year` exists for every ledger and is not older than the last posting
into a prior year. A year with nothing before it is current."""
fy0 = first_year(repo, company["company_code"])
if fy0 is None or year <= fy0:
return True, ""
for ledger_id in repo.company_ledgers.assigned(company["company_code"], year):
rows = repo.ledger.carry(company["company_code"], ledger_id, dates.fy_name(year))
if not rows:
mv_exists = bool(opening(repo, company, ledger_id, year))
if mv_exists:
return False, f"the opening balances of {dates.fy_name(year)} ({ledger_id}) are not derived yet"
continue
if stored_opening(repo, company, ledger_id, year) is None:
return False, f"the opening balances of {dates.fy_name(year)} ({ledger_id}) are older than a posting into a prior year"
return True, ""