dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/application/carry.py
"""CON-15: the year-end carry-forward.
The consolidated books of a period are the translated data of its companies plus the consolidation entries. Most entries are recomputed in
full in every period from the companies' cumulative data (translation, intercompany, the investment and the minority at acquisition, ownership
changes): they need no carry. What a new year's data cannot recreate is the effect of entries made on last year's profit and loss, because the
companies' own books start the year with that profit already in their reserves. Those entries are carried:
stock profit the unrealised profit in stock, its deferred tax, and the minority's part
nci the minority's share of earlier years' results (the nci_result lines)
equity method the share of an associate's earlier profits, raised on the investment
manual the cumulative manual entries (CON-10), not the period-specific ones
disposal the result of a company sold during the year, and the gain
dividend last year's intragroup dividend: the payer's reserves restored, the minority bearing its share
(and last year's own carry-forward, rolled on)
Their balance-sheet lines open the new year per group account, company, partner and segment; their profit-and-loss effect opens on group retained
earnings. Rows are keyed [group, fiscal year, account, company, partner, segment], every value in VALUES, so deriving again writes the same rows;
source_digest is the signed digest of the period they came from, so a reopened and re-signed period makes them visibly stale.
"""
from decimal import Decimal
from erp_platform import work
from erp_platform.bricks import Refused
from brick_consolidation.application import ownership, reports
from brick_consolidation.application.runs import prior_period, role_account
from brick_consolidation.domain import consolidation as C
from brick_consolidation.domain.errors import RuleBroken
CARRIED = ("stock_profit", "equity_method", "manual", "disposal", "dividend")
KEY = ("group_account_id", "company_code", "partner_company", "segment")
def first_period_of(group, period_id: str) -> str:
return C.fiscal_months(period_id, group.fiscal_year_end_month)[0]
def closing_period_before(group, period_id: str) -> str:
return prior_period(first_period_of(group, period_id))
def has_previous_year(group, period_id: str) -> bool:
before = closing_period_before(group, period_id)
return any(x.period_id == before for x in group.group_periods)
def rows_for(repo, group, fiscal_year: str) -> list:
return repo.data.carry_forward(group.group_id, fiscal_year)
def derive(repo, group, period_id: str) -> dict:
"""Derive the carry-forward of the financial year that `period_id` starts (or belongs to) from the previous year's last period, which must be
signed. Returns {"fiscal_year", "from_period", "source_digest", "rows", "changed"}."""
before = closing_period_before(group, period_id)
prior = next((x for x in group.group_periods if x.period_id == before), None)
fiscal_year = C.fiscal_year_of(period_id, group.fiscal_year_end_month)
if prior is None:
return {"fiscal_year": fiscal_year, "from_period": None, "rows": 0, "changed": False}
if prior.status != "signed":
raise RuleBroken(f"{before}, the previous year's last period, is {prior.status}: the carry-forward is derived from a signed period")
members = C.members_at(group, before)
_tb, entries = reports.consolidated(repo, group, prior, members)
prior_fy = C.fiscal_year_of(before, group.fiscal_year_end_month)
pl = {r.group_account_id for r in group.account_rules if r.rate_type == "average"}
amounts: dict = {}
to_reserves: dict = {} # company -> the profit-and-loss effect, opened on group retained earnings
def add(account, company, partner, segment, amount):
if not amount:
return
if account in pl:
to_reserves[company] = to_reserves.get(company, Decimal(0)) + amount
else:
k = (account, company, partner, segment)
amounts[k] = amounts.get(k, Decimal(0)) + amount
for r in rows_for(repo, group, prior_fy): # last year's own opening, rolled on
add(r["group_account_id"], r["company_code"], r["partner_company"], r["segment"], Decimal(r["opening"]))
re_account = None
nci_result = role_account(group, "nci_result", before, need=False)
nci_equity = role_account(group, "nci_equity", before, need=False)
for e in entries:
doc = e["document_type"]
if doc == "nci":
for ln in e["lines"]:
if nci_result and ln["group_account_id"] == nci_result:
add(ln["group_account_id"], ln["company_code"], "", "", ln["amount"])
add(nci_equity, ln["company_code"], "", "", -ln["amount"])
elif doc in CARRIED and not (doc == "manual" and e["period_specific"] == "true"):
sold = e["task"].split(":", 1)[1] if doc == "disposal" and ":" in (e["task"] or "") else None
re_account = re_account or (ownership.retained_earnings_account(group, before) if doc == "disposal" else None)
for ln in e["lines"]:
if sold and ln["company_code"] == sold and ln["group_account_id"] not in pl and ln["group_account_id"] != re_account:
continue # the sold company's balance sheet came out in the sale and is not there to carry
add(ln["group_account_id"], ln["company_code"], ln["partner_company"], ln["segment"], ln["amount"])
for ch in repo.data.effective_changes(group.group_id, ("divestiture",)): # a company sold in the closing period: its profit and loss to the sale
if C.period_of(ownership.day_of(ch["effective_date"])) == before:
for r in repo.data.translated(group.group_id, before, prior.run_no):
if r["company_code"] == ch["company_code"] and r["group_account_id"] in pl:
add(r["group_account_id"], r["company_code"], r["partner_company"], r["segment"], r["ytd_group"])
if to_reserves:
re_account = ownership.retained_earnings_account(group, before)
for company, amount in to_reserves.items():
k = (re_account, company, "", "")
amounts[k] = amounts.get(k, Decimal(0)) + amount
_fix_year_end_history(repo, group, prior, before, first_period_of(group, period_id))
held = {(r["group_account_id"], r["company_code"], r["partner_company"], r["segment"]): r["source_digest"] for r in rows_for(repo, group, fiscal_year)}
old = {(r["group_account_id"], r["company_code"], r["partner_company"], r["segment"]): Decimal(r["opening"]) for r in rows_for(repo, group, fiscal_year)}
fresh = {k: C.q2(v) for k, v in amounts.items()}
for k in old.keys() - fresh.keys():
fresh[k] = C.ZERO # a row last time and none now is written as zero, never deleted
changed = fresh != old or any(d != prior.tb_digest for d in held.values())
now = work.now()
repo.data.write_carry_forward([{"group_id": group.group_id, "fiscal_year": fiscal_year, **dict(zip(KEY, k)), "opening": str(v), "derived_at": now,
"source_digest": prior.tb_digest} for k, v in sorted(fresh.items())])
if sum(fresh.values(), Decimal(0)) != 0:
raise RuleBroken(f"the carry-forward of {fiscal_year} does not net to zero ({sum(fresh.values(), Decimal(0))})")
return {"fiscal_year": fiscal_year, "from_period": before, "source_digest": prior.tb_digest, "rows": len([v for v in fresh.values() if v]),
"changed": changed}
def _fix_year_end_history(repo, group, prior, before: str, arose: str) -> None:
"""Each company's year, its result and what moved in its equity, fixed at the amounts it was translated at (IAS 21): next year's
translation holds them at those amounts and retranslates only the year's own movements. Keyed [group, company, account, arose period]:
deriving again writes the same rows. Four statements, whatever the number of companies."""
rows = repo.data.translated(group.group_id, before, prior.run_no)
companies = sorted({r["company_code"] for r in rows})
held = C.history_at(repo.data.historical(group.group_id, companies), before)
companies = [c for c in companies if held.get(c)]
if not companies:
return
reported: dict = {}
for r in repo.data.reported(group.group_id, [before], companies):
reported.setdefault(r["company_code"], []).append(r)
reserves = role_account(group, "retained_earnings", before, need=False)
if reserves is None:
raise Refused(409, f"the year's result of {', '.join(companies)} is fixed at year end on the companies' retained earnings account, and the "
"group has no account for the role retained_earnings", gate="account_roles_complete",
fix="Settings: account role retained_earnings")
out = []
for c in companies:
fixed = C.year_end_history(reported=reported.get(c, []), translated=[r for r in rows if r["company_code"] == c], history=held[c],
rules=group.account_rules, period_id=before, reserves_account=reserves)
out += [{"group_id": group.group_id, "company_code": c, "group_account_id": a, "arose_period": arose, "amount_local": lg[0],
"amount_group": lg[1], "rate": None, "change_id": f"{C.YEAR_END}{C.fiscal_year_of(before, group.fiscal_year_end_month)}"}
for a, lg in sorted(fixed.items())]
repo.data.write_historical(out)
def roles_problems(group, period_id: str) -> list:
"""Readiness account_roles_complete: a year's last period with an acquired subsidiary or associate fixes the year's result as historical
amounts on the role retained_earnings (MUST-1b), never on a guessed account."""
if C.fiscal_months(period_id, group.fiscal_year_end_month)[-1] != period_id or C.period_end(period_id).month != int(group.fiscal_year_end_month or 12):
return []
held = [c for c, m in C.members_at(group, period_id).items() if m.investor and m.method in ("full", "equity")]
if held and role_account(group, "retained_earnings", period_id, need=False) is None:
return [f"no account for the role retained_earnings: the year's result of {', '.join(sorted(held))} is fixed on it at year end"]
return []
def current_problems(repo, group, gp) -> list:
"""The readiness gate carry_forward_current: the year's opening must come from the previous year's last period as it is signed now."""
if not has_previous_year(group, gp.period_id):
return []
before = closing_period_before(group, gp.period_id)
prior = next(x for x in group.group_periods if x.period_id == before)
fiscal_year = C.fiscal_year_of(gp.period_id, group.fiscal_year_end_month)
if prior.status != "signed":
return [f"{before}, the previous year's last period, is {prior.status}: the balances carried into {fiscal_year} are provisional until it is signed again"]
held = rows_for(repo, group, fiscal_year)
if not held:
return [f"the carry-forward of {fiscal_year} is not derived from {before}"]
if any(r["source_digest"] != prior.tb_digest for r in held):
return [f"the carry-forward of {fiscal_year} was derived before {before} was signed again (digest {held[0]['source_digest'][:8]} -> {prior.tb_digest[:8]})"]
return []
def realisation_lines(repo, group, period_id: str) -> list:
"""Last year's stock-profit eliminations realised this year: the opposite of the previous year's closing period's, so the opening moves to cost of
sales (and the deferred tax to tax expense) when the year's elimination is run. The goods still held at this year's end are eliminated again by the
year's own stock-profit entry. Not the realisation entry of the year before, which already closed it."""
if not has_previous_year(group, period_id):
return []
before = closing_period_before(group, period_id)
prior = next(x for x in group.group_periods if x.period_id == before)
out: dict = {}
for e in repo.data.entries(group.group_id, before, prior.run_no):
if e["document_type"] == "stock_profit" and e["task"] not in ("stock_profit_realised", "reversal") and e["status"] == "posted":
for ln in e["lines"]:
k = (ln["group_account_id"], ln["company_code"], ln["partner_company"], ln["movement_type"], ln["segment"])
out[k] = out.get(k, Decimal(0)) - ln["amount"]
return [{"group_account_id": a, "company_code": c, "partner_company": p, "movement_type": m, "segment": s, "amount": v}
for (a, c, p, m, s), v in sorted(out.items()) if v]