dodil/erp-brick-gl
DODIL ERP brick: general ledger
src/brick_gl/application/revaluation.py
"""Foreign-currency revaluation (GL-09, GL-48) and the period-end re-split of open items by profit centre (GL-25).
Each is a RUN keyed (company, ledger, key date, kind): running it twice posts once, and the journal it posts is ledger-specific
(each ledger revalues by its own method), carries a journal id derived from the run, and REVERSES on the first day of the next
period (GL-24) so the next run starts from the booked rate again. Receivable and payable control accounts are not revalued as
balances but item by item, from the open items the owning module reports (payments' GET /settlement, purchasing's GET
/bills/open) at the rate their own journal booked them.
"""
import hashlib # noqa: F401
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.seam import derived_id
from brick_gl.application import ledgerops
from brick_gl.application.brick import (
GL,
RESPLIT_POST,
REVAL_RUN,
claims,
company_row,
decide,
real_now,
repository,
sub_of,
today,
who,
)
from brick_gl.application.inputs import CompanyCode, LedgerCode, PeriodId
from brick_gl.application.posting import finish
from brick_gl.application.scheduled import post_as_gl
from brick_gl.domain import dates
from brick_gl.domain.errors import RuleBroken
from brick_gl.domain.journal import LineIn
from brick_gl.domain.money import ZERO, convert, q2
from brick_gl.domain.readiness import refusal
from brick_gl.infrastructure import seams
from brick_gl.infrastructure import tables as t
from brick_gl.infrastructure.repositories import _Keyed, control_documents, utc
_run_t = lambda repo: _Keyed(repo.conn, t.gl_fx_reval_run, ("run_id",)) # noqa: E731
def _closing(repo, currency: str, functional: str, on: date) -> Decimal:
r = repo.fx.in_force(currency, functional, "closing", on) or repo.fx.in_force(currency, functional, "spot", on)
if r is None:
raise refusal("fx_rate", f"no closing rate for {currency} to {functional} is in force on {on}",
"Currencies: enter the closing rate for the period end", status=422)
return r["rate"]
def _run_view(r: dict, items: list) -> dict:
return {"run_id": r["run_id"], "company_code": r["company_code"], "ledger_id": r["ledger_id"], "key_date": r["key_date"], "kind": r["kind"],
"method": r["method"], "journal_id": r["journal_id"], "reversal_journal_id": r["reversal_journal_id"], "run_by": r["run_by"],
"items": [{k: i[k] for k in ("account_id", "currency", "document", "amount_currency", "booked_functional", "rate",
"revalued_functional", "gain_loss", "profit_centre")} for i in items]}
def _items(repo, run_id: str) -> list:
return [dict(r._mapping) for r in repo.conn.execute(t.gl_fx_reval_item.select().where(t.gl_fx_reval_item.c.run_id == run_id))]
def _period_end(repo, code: str, period_id: str, key_date: Optional[str]) -> tuple:
p = repo.periods.get(code, period_id)
if p is None:
raise Refused(404, f"no period {period_id} of {code}")
if p["status"] != "open":
raise refusal("period_open", f"the period {period_id} is closed", "reopen it, or revalue before the close")
on = dates.parse(key_date) if key_date else p["ends_on"]
if not (p["starts_on"] <= on <= p["ends_on"]):
raise RuleBroken(f"{on} is not in {period_id}", status=422)
return p, on
def _post_run(repo, caller, company: dict, ledger_id: str, kind: str, on: date, items: list, lines: list, method: str, at):
"""Record a run and post its journal (ledger-specific, reversing next period). Returns (view, (touched, summary) | None)."""
code = company["company_code"]
run_id = f"{code}:{ledger_id}:{on}:{kind}"
jid = derived_id(run_id, "reval") if lines else None
summary = touched = None
if lines:
summary, touched = post_as_gl(
repo, company_code=code, key=jid, source_document=f"reval:{run_id}", event="revaluation", on=on,
currency=company["functional_currency"], lines=lines, ledger_id=ledger_id, memo=f"FX revaluation {kind} {on}", book="fx",
extras={"reval_run_id": run_id}, reverse_next_period=True, actor=who(caller), actor_sub=sub_of(caller), at=at, manual=False)
row = {"run_id": run_id, "company_code": code, "ledger_id": ledger_id, "key_date": on, "kind": kind, "journal_id": jid,
"reversal_journal_id": derived_id(jid, "reversal") if jid else None, "method": method, "run_by": who(caller), "run_at": utc(at)}
_run_t(repo).add(row)
keyed = _Keyed(repo.conn, t.gl_fx_reval_item, ("run_id", "account_id", "currency", "document"))
for i in items:
keyed.add({"run_id": run_id, **i})
return _run_view(row, items), ((touched, summary) if touched else None)
def _existing_run(repo, code: str, ledger_id: str, kind: str, on: date) -> Optional[dict]:
r = _run_t(repo).get(f"{code}:{ledger_id}:{on}:{kind}")
return _run_view(r, _items(repo, r["run_id"])) | {"replay": True} if r else None
@GL.use_case(kind="command", route="POST /periods/{period_id}/run-fx-revaluation-balances", permission=REVAL_RUN, idempotent=True)
def run_fx_revaluation_balances(caller: dict, period_id: PeriodId, company_code: Optional[CompanyCode] = None,
key_date: Optional[str] = None) -> dict:
"""Restate foreign-currency monetary balances (a USD bank account) at the closing rate of the key date (default the period
end), for each ledger of the company by ITS method: closing_rate (IAS 21: gains and losses) or losses_only (as HGB section
256a requires: only a loss is recognised). Posts a ledger-specific journal to the map's fx_gain / fx_loss that REVERSES on the
first day of the next period (posted then, or held until it opens). The run is keyed (company, ledger, key date, balances),
records its journal, and running it twice posts once. Receivable and payable control accounts are revalued per open item
(run_fx_revaluation_open_items). A currency with no closing rate blocks the run, named. Needs the setting revaluation on."""
def unit():
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
p, on = _period_end(repo, code, period_id, key_date)
if not repo.settings.in_force(today(repo), code).revaluation:
raise refusal("revaluation_on", f"revaluation is off for {code}", "Settings: switch revaluation on (set_gl_setting)")
year = dates.fy_number(p["fiscal_year"])
at = real_now()
out, finishers = [], []
for ledger_id in repo.company_ledgers.assigned(code, year):
done = _existing_run(repo, code, ledger_id, "balances", on)
if done:
out.append(done)
continue
method = repo.ledgers.get(ledger_id)["reval_method"] or "closing_rate"
accounts = [a for a in repo.accounts.of_chart(company["chart_id"]) if a["monetary"] and a["currency"]
and a["currency"] != company["functional_currency"] and a["subtype"] not in ("receivable", "payable")]
items, lines, gain, loss = [], [], ZERO, ZERO
if accounts:
sums = repo.ledger.sums(code, ledger_id, upto=on, account_ids=[a["account_id"] for a in accounts],
group_by=("account_id", "currency"))
by = {(r["account_id"], r["currency"]): r for r in sums}
for a in accounts:
r = by.get((a["account_id"], a["currency"]))
if not r:
continue
doc, carrying = r["debit"] - r["credit"], r["f_debit"] - r["f_credit"]
if doc == 0 and carrying == 0:
continue
rate = _closing(repo, a["currency"], company["functional_currency"], on)
revalued = convert(doc, rate)
diff = revalued - carrying
if diff == 0 or (method == "losses_only" and diff > 0):
continue
items.append({"account_id": a["account_id"], "currency": a["currency"], "document": "", "amount_currency": doc,
"booked_functional": carrying, "rate": rate, "revalued_functional": revalued, "gain_loss": diff,
"profit_centre": None})
lines.append(LineIn(account_id=a["account_id"], debit=diff if diff > 0 else ZERO, credit=-diff if diff < 0 else ZERO,
memo=f"revalue {a['currency']}"))
gain, loss = gain + (diff if diff > 0 else ZERO), loss + (-diff if diff < 0 else ZERO)
if gain:
lines.append(LineIn(purpose="fx_gain", credit=gain, memo="unrealised gain"))
if loss:
lines.append(LineIn(purpose="fx_loss", debit=loss, memo="unrealised loss"))
view, fin = _post_run(repo, caller, company, ledger_id, "balances", on, items, lines, method, at)
out.append(view | {"replay": False})
if fin:
finishers.append(fin)
return {"company_code": code, "period_id": period_id, "key_date": on, "runs": out}, finishers
result, finishers = claims(unit)
for touched, summary in finishers:
finish(touched, summary)
return result
@GL.use_case(kind="command", route="POST /periods/{period_id}/run-fx-revaluation-open-items", permission=REVAL_RUN, idempotent=True)
def run_fx_revaluation_open_items(caller: dict, period_id: PeriodId, company_code: Optional[CompanyCode] = None,
key_date: Optional[str] = None) -> dict:
"""Restate each OPEN foreign-currency invoice and bill at the period-end closing rate (IAS 21): gl asks payments' GET
/settlement and purchasing's GET /bills/open what is outstanding at the key date, takes each item's BOOKED rate from the
journal that posted it, and posts per ledger and by its method one journal of lines per item on the control account
(against_document = the item, so the split follows it) to the map's fx_gain / fx_loss. The journal reverses on the first day
of the next period, so the next run starts from the booked rate again. Keyed (company, ledger, key date, open_items): twice
posts once. An item with no closing rate blocks the run with the currency named; an item gl never posted is listed as a break,
not revalued."""
def unit():
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
p, on = _period_end(repo, code, period_id, key_date)
if not repo.settings.in_force(today(repo), code).revaluation:
raise refusal("revaluation_on", f"revaluation is off for {code}", "Settings: switch revaluation on (set_gl_setting)")
fn = company["functional_currency"]
breaks, entries = [], [] # entries: (document, control account, currency, outstanding doc (signed), booked rate)
docs = control_documents(repo, code, "receivable", True, fn)
if docs and seams.PAYMENTS.installed:
ans = seams.neighbours.settlement([d[0] for d in docs], on, who(caller))
if not ans.ok:
raise RuleBroken(f"payments did not answer ({ans.status or 'unreachable'}): {ans.reason}", status=503)
by = {i["invoice_id"]: i for i in ans.data["invoices"]}
for doc, j, acc in docs:
item = by.get(doc)
if item is None:
breaks.append({"document": doc, "kind": "posted_not_in_payments"})
continue
out_ = Decimal(str(item["outstanding"]))
if out_ != 0 and item["currency"] != fn:
entries.append((doc, acc, item["currency"], out_, j["fx_rate"]))
pdocs = {d[0]: d for d in control_documents(repo, code, "payable", True, fn)}
if seams.PURCHASING.installed:
ans = seams.neighbours.open_bills(on, who(caller))
if not ans.ok:
raise RuleBroken(f"purchasing did not answer ({ans.status or 'unreachable'}): {ans.reason}", status=503)
for b in ans.data.get("items", []):
if b["currency"] == fn or Decimal(str(b["outstanding"])) == 0:
continue
d = pdocs.get(b["bill_id"])
if d is None:
breaks.append({"document": b["bill_id"], "kind": "not_posted_by_gl"})
continue
entries.append((b["bill_id"], d[2], b["currency"], -Decimal(str(b["outstanding"])), d[1]["fx_rate"]))
year = dates.fy_number(p["fiscal_year"])
at = real_now()
out, finishers = [], []
for ledger_id in repo.company_ledgers.assigned(code, year):
done = _existing_run(repo, code, ledger_id, "open_items", on)
if done:
out.append(done)
continue
method = repo.ledgers.get(ledger_id)["reval_method"] or "closing_rate"
items, lines, gain, loss = [], [], ZERO, ZERO
for doc, acc, cur, signed, booked_rate in entries:
rate = _closing(repo, cur, fn, on)
booked = convert(signed, booked_rate)
revalued = convert(signed, rate)
diff = revalued - booked
if diff == 0 or (method == "losses_only" and diff > 0):
continue
items.append({"account_id": acc, "currency": cur, "document": doc, "amount_currency": signed, "booked_functional": booked,
"rate": rate, "revalued_functional": revalued, "gain_loss": diff, "profit_centre": None})
lines.append(LineIn(account_id=acc, debit=diff if diff > 0 else ZERO, credit=-diff if diff < 0 else ZERO,
against_document=doc, memo=f"revalue {doc}"))
gain, loss = gain + (diff if diff > 0 else ZERO), loss + (-diff if diff < 0 else ZERO)
if gain:
lines.append(LineIn(purpose="fx_gain", credit=gain, memo="unrealised gain"))
if loss:
lines.append(LineIn(purpose="fx_loss", debit=loss, memo="unrealised loss"))
view, fin = _post_run(repo, caller, company, ledger_id, "open_items", on, items, lines, method, at)
out.append(view | {"replay": False})
if fin:
finishers.append(fin)
return {"company_code": code, "period_id": period_id, "key_date": on, "runs": out, "breaks": breaks}, finishers
result, finishers = claims(unit)
for touched, summary in finishers:
finish(touched, summary)
return result
# ═══ the period-end re-split (GL-25) ═══════════════════════════════════════════════════════════════════════════════════
@GL.use_case(kind="command", route="POST /periods/{period_id}/resplit-open-items", permission=RESPLIT_POST, idempotent=True)
def resplit_open_items(caller: dict, period_id: PeriodId, company_code: Optional[CompanyCode] = None,
ledger_id: Optional[LedgerCode] = None) -> dict:
"""With document splitting on, re-split receivables and payables by profit centre from their OPEN ITEMS at the period end, so
a receipt posted before it was allocated does not leave the receivable in the wrong segment: gl reads the open items (payments'
GET /settlement, purchasing's GET /bills/open), takes each item's target split from the split it was posted with, and posts ONE
adjustment journal on the control accounts (the split_clearing lines make every profit centre balance) so the control
account per profit centre is what its open items say. It reverses on the first day of the next period; keyed (company,
ledger, as_at, resplit), so twice posts once."""
def unit():
with repository() as repo, decide():
company = company_row(repo, caller, company_code)
code = company["company_code"]
p = repo.periods.get(code, period_id)
if p is None:
raise Refused(404, f"no period {period_id} of {code}")
if p["status"] != "open":
raise refusal("period_open", f"the period {period_id} is closed", "reopen it")
on = p["ends_on"]
if not ledgerops.splitting_on(repo, code, on):
raise RuleBroken(f"document splitting is off for {code}: there is nothing to re-split", status=409)
fn = company["functional_currency"]
year = dates.fy_number(p["fiscal_year"])
ledgers = [ledger_id] if ledger_id else repo.company_ledgers.assigned(code, year)
at = real_now()
result, finishers = [], []
for lg in ledgers:
key = derived_id(code, lg, str(on), "resplit")
if repo.journals.get(key) is not None:
result.append({"ledger_id": lg, "journal_id": key, "replay": True})
continue
target: dict = {} # (account, profit centre) -> functional net the open items say
for subtype, side in (("receivable", 1), ("payable", -1)):
docs = control_documents(repo, code, subtype)
if not docs:
continue
outstanding = {}
if subtype == "receivable" and seams.PAYMENTS.installed:
ans = seams.neighbours.settlement([d[0] for d in docs], on, who(caller))
if not ans.ok:
raise RuleBroken(f"payments did not answer ({ans.status or 'unreachable'}): {ans.reason}", status=503)
outstanding = {i["invoice_id"]: Decimal(str(i["outstanding"])) for i in ans.data["invoices"]}
elif subtype == "payable" and seams.PURCHASING.installed:
ans = seams.neighbours.open_bills(on, who(caller))
if not ans.ok:
raise RuleBroken(f"purchasing did not answer ({ans.status or 'unreachable'}): {ans.reason}", status=503)
outstanding = {b["bill_id"]: Decimal(str(b["outstanding"])) for b in ans.data.get("items", [])}
else:
continue
for doc, j, acc in docs:
out_ = outstanding.get(doc, ZERO)
if out_ == 0:
continue
weights = repo.ledger.passive_weights(doc, acc)
if not weights:
continue
total = sum(weights.values(), ZERO)
from brick_gl.domain.money import largest_remainder
pcs = sorted(weights)
shares = largest_remainder(convert(out_, j["fx_rate"] or Decimal(1)) * side, [weights[k] for k in pcs])
for pc, sh in zip(pcs, shares):
target[(acc, pc)] = target.get((acc, pc), ZERO) + sh
_ = total
control = [a["account_id"] for a in repo.accounts.of_chart(company["chart_id"]) if a["subtype"] in ("receivable", "payable")]
rows = repo.ledger.sums(code, lg, upto=on, account_ids=control, group_by=("account_id", "profit_centre"))
current = {(r["account_id"], r["profit_centre"] or ""): r["f_debit"] - r["f_credit"] for r in rows}
lines, total_delta = [], ZERO
for (acc, pc) in sorted(set(current) | set(target)):
delta = target.get((acc, pc), ZERO) - current.get((acc, pc), ZERO)
if delta == 0:
continue
total_delta += delta
lines.append(LineIn(account_id=acc, profit_centre=pc or None, debit=delta if delta > 0 else ZERO,
credit=-delta if delta < 0 else ZERO, memo="re-split by open items"))
if not lines:
result.append({"ledger_id": lg, "journal_id": None, "replay": False, "note": "already split as the open items say"})
continue
if total_delta != 0:
result.append({"ledger_id": lg, "journal_id": None, "replay": False,
"note": f"the control accounts differ from the open items by {total_delta}: reconcile first (GL-08)"})
continue
summary, touched = post_as_gl(
repo, company_code=code, key=key, source_document=f"resplit:{code}:{lg}:{on}", event="resplit", on=on, currency=fn,
lines=lines, ledger_id=lg, memo=f"re-split open items {on}", book="misc", extras={}, reverse_next_period=True,
actor=who(caller), actor_sub=sub_of(caller), at=at, manual=False)
result.append({"ledger_id": lg, "journal_id": key, "replay": summary["replay"], "lines": len(lines)})
if touched:
finishers.append((touched, summary))
return {"company_code": code, "period_id": period_id, "as_at": on, "ledgers": result}, finishers
result, finishers = claims(unit)
for touched, summary in finishers:
finish(touched, summary)
return result
_ = (q2, today)