dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/application/runs.py
"""The three runs of a group close — collect, translate, eliminate — and the structure change that fixes an acquisition.
Each takes the open repository and the Group aggregate the use case loaded, decides in the domain, and writes through `repo.data` in
bulk: the statements a run issues do not grow with the number of companies (tests/test_statement_budget.py pins that). Nothing is
written before every gate has been checked, and nothing side-effecting (gl is only ever read) happens after a write.
Two operations on one group collide on ONE key: each use case saves the Group aggregate, which claims its next version with a plain
INSERT (cons_group_versions); the run's own claim (cons_run_claim: group, period, task, run_no) names the run besides.
"""
import json
from datetime import date
from decimal import Decimal
from erp_platform import work
from erp_platform.seam import derived_id
from brick_consolidation.application import ownership
from brick_consolidation.domain import consolidation as C
from brick_consolidation.domain.errors import RuleBroken
from brick_consolidation.domain.events import GroupPeriodCollected, GroupPeriodEliminated, GroupPeriodTranslated
from brick_consolidation.infrastructure.data import REPORTED_KEY, TRANSLATION_DOCS, zero_fill
from brick_consolidation.infrastructure.seams import Missing, ports
ELIMINATION_DOCS = ("ic_balance", "ic_pl", "dividend", "stock_profit", "investment", "nci", "equity_method", "disposal", "ownership_change")
SETTING_DEFAULTS = {"signoff_second_person": True, "allow_open_period_collect": False, "translation_pl_rate": "monthly_average",
"nci_measurement": "proportionate"}
def settings(repo) -> dict:
out = dict(SETTING_DEFAULTS)
from brick_consolidation.infrastructure import clock
for s in repo.settings.all(clock.today()): # the dated row in force on the business date
if s["value"] is not None and s["setting_key"] in out:
out[s["setting_key"]] = s["value"]
return out
def deferred_tax_rate(company: str, explicit, on):
"""v3, the deferred tax rate of a consolidation adjustment (percent): the entry's own tax_rate_pct, else gl's statutory_tax_rate of the company
that holds the temporary difference in force on the period's last day (a decimal fraction there, x 100 here), else (None, why) and the run is
refused (gate tax_rate_known) - never computed at zero. gl is asked per run, never copied."""
if explicit is not None:
return Decimal(str(explicit)), ""
try:
got = ports.gl.company_row(company, on).get("statutory_tax_rate")
except Missing as m:
return None, f"gl could not be asked ({m.reason})"
if got in (None, ""):
return None, f"gl has no statutory_tax_rate for {company} on {on}"
return Decimal(str(got)) * C.HUNDRED, ""
def gl_call(fn, *args):
"""gl is read; if it refuses or is away the run says so in gl's words and writes nothing."""
try:
return fn(*args)
except Missing as m:
raise RuleBroken(f"gl: {m.reason}") from None
def role_account(group, role: str, period_id: str, *, need: bool = True):
best = None
for r in group.account_roles:
if r.role == role and r.valid_from_period <= period_id and (best is None or r.valid_from_period > best.valid_from_period):
best = r
if best is None and need:
raise RuleBroken(f"the group has no account for the role {role} (set it in the account roles)")
return best.group_account_id if best else None
def the_period(group, period_id: str):
C.check_period(period_id)
for gp in group.group_periods:
if gp.period_id == period_id:
return gp
raise RuleBroken(f"there is no group period {period_id} — open it first")
def _not_signed(gp) -> None:
if gp.status == "signed":
raise RuleBroken(f"{gp.period_id} is signed; its figures do not change until it is reopened")
def stale_later_months(group, period_id: str) -> list:
"""A month's data changed: the year's later months read its movements (monthly-average translation) and carry its cumulative entries,
so their unsigned runs are stale (audit v1 MUST-3b)."""
fy = C.fiscal_year_of(period_id, group.fiscal_year_end_month)
later = [x.period_id for x in group.group_periods if x.period_id > period_id and C.fiscal_year_of(x.period_id, group.fiscal_year_end_month) == fy]
return group.mark_stale(min(later), to_period=max(later)) if later else []
# ═══ CON-03 collect ═════════════════════════════════════════════════════════════════════════════════════════════
def collect(repo, group, period_id: str, *, by: str):
"""Fetch every company in scope from gl and store its figures. Returns the summary, or None when nothing changed."""
gp = the_period(group, period_id)
_not_signed(gp)
members = C.members_at(group, period_id)
companies = sorted(members)
cfg = settings(repo)
day = C.period_end(period_id)
latest = {}
for row in repo.data.collections(group.group_id, period_id):
if row["company_code"] not in latest or row["run_no"] > latest[row["company_code"]]["run_no"]:
latest[row["company_code"]] = row
uploaded = {c for c in companies if latest.get(c, {}).get("source") == "upload"} # an accepted upload stands for gl (CON-12)
# gates, before anything is written: gl's period closed per company; the figures map, net to zero
from_gl = [c for c in companies if c not in uploaded]
not_closed = [c for c in from_gl if gl_call(ports.gl.period_status, c, day).get("status") != "closed"]
if not_closed and not cfg["allow_open_period_collect"]:
raise RuleBroken(f"company_periods_closed: {', '.join(not_closed)} still open in gl for {period_id} — close the period in gl "
"(GL-32), or the owner allows collecting from an open period (allow_open_period_collect); a company kept outside gl sends its "
"figures by upload, accepted by a second person (CON-12)")
fetched, problems = {}, []
for c in uploaded:
fetched[c] = {"rows": [], "digest": latest[c]["gl_digest"], "currency": latest[c]["local_currency"], "provisional": "false"}
for c in from_gl:
try:
data = gl_call(ports.gl.balances, c, group.ledger_id, period_id)
except RuleBroken as broken:
raise RuleBroken(f"{broken} — a company kept outside gl sends its figures by upload, accepted by a second person (CON-12)") from None
if data.get("unmapped"):
problems.append(f"{c} has accounts with no group account: " + ", ".join(str(u.get("code") or u.get("account_id")) for u in data["unmapped"])
+ " — map them in gl (GL-41)")
if group.group_chart_id and data.get("group_chart_id") and data["group_chart_id"] != group.group_chart_id:
problems.append(f"{c} reports on group chart {data['group_chart_id']}, the group's is {group.group_chart_id}")
rows = [{"group_account_id": str(r["group_account_id"]), "partner_company": r.get("partner_company") or "",
"movement_type": r.get("movement_type") or "", "segment": r.get("segment") or "",
"ytd_local": C.amount(r["ytd_amount"], f"{c} year-to-date"), "period_local": C.amount(r["period_amount"], f"{c} period")}
for r in data.get("rows", [])]
if sum((r["ytd_local"] for r in rows), Decimal(0)) != 0:
problems.append(f"{c}'s figures do not net to zero (year to date {sum((r['ytd_local'] for r in rows), Decimal(0))})")
fetched[c] = {"rows": rows, "digest": data.get("digest", ""), "currency": data.get("functional_currency", ""),
"provisional": "true" if c in not_closed else "false"}
if problems:
raise RuleBroken("collection refused: " + "; ".join(problems))
changed = [c for c in companies if c not in latest or latest[c]["gl_digest"] != fetched[c]["digest"]
or latest[c]["provisional"] != fetched[c]["provisional"]]
structure_changed = gp.stale_from_step == "collect"
if not changed and not structure_changed and gp.status != "open":
return None # nothing to write: the digests are the ones already held
if gp.status == "open":
run_no, stale = max(gp.run_no, 1), ""
else:
run_no, stale = gp.run_no + 1, "translate"
write = [c for c in (companies if structure_changed else changed) if c not in uploaded]
repo.data.claim(group.group_id, period_id, "collect", run_no)
existing = {}
for r in repo.data.reported(group.group_id, [period_id], write):
existing.setdefault(r["company_code"], set()).add(tuple(r[k] for k in REPORTED_KEY[2:]))
out, collections = [], []
for c in write:
base = {"group_id": group.group_id, "period_id": period_id, "local_currency": fetched[c]["currency"], "source": "gl"}
fresh = [{**base, "company_code": c, **r} for r in fetched[c]["rows"]]
out += fresh + zero_fill({k for k in existing.get(c, set())}, {tuple(r[k] for k in REPORTED_KEY[2:]) for r in fresh}, base)
collections.append({"group_id": group.group_id, "period_id": period_id, "company_code": c, "run_no": run_no,
"gl_digest": fetched[c]["digest"], "period_status": "open" if c in not_closed else "closed",
"local_currency": fetched[c]["currency"], "provisional": fetched[c]["provisional"],
"collected_at": work.now(), "source": "gl"})
repo.data.write_reported(out)
repo.data.add_collections(collections)
gp.status, gp.run_no, gp.stale_from_step = "collected", run_no, stale
stale_later_months(group, period_id)
group._record(GroupPeriodCollected(group_id=group.group_id, period_id=period_id, by=by))
return {"period_id": period_id, "run_no": run_no, "collected": write, "unchanged": [c for c in companies if c not in write],
"provisional": sorted(not_closed)}
# ═══ CON-04 translate ═══════════════════════════════════════════════════════════════════════════════════════════
def _rates(group, period_id: str, currencies: list, months: list) -> tuple:
"""gl's closing rate at the period end and an average rate for each month so far, per foreign currency. A missing one is named."""
closing, monthly, missing = {}, {}, []
for ccy in currencies:
try:
closing[ccy] = C.rate(ports.gl.fx_rate(ccy, C.period_end(period_id), "closing", group.group_currency), f"closing rate {ccy}")
except Missing as m:
missing.append(f"closing rate {ccy}→{group.group_currency} on {C.period_end(period_id)} ({m.reason})")
monthly[ccy] = {}
for month in months:
try:
monthly[ccy][month] = C.rate(ports.gl.fx_rate(ccy, C.period_end(month), "average", group.group_currency),
f"average rate {ccy} {month}")
except Missing as m:
missing.append(f"average rate {ccy}→{group.group_currency} for {month} ({m.reason})")
if missing:
raise RuleBroken("translation refused, no rate in gl: " + "; ".join(missing) + " — add the rate in gl (GL-33)")
return closing, monthly
def _mirror(entries: list, run_no: int) -> list:
"""The entries of an earlier run, reversed by a mirror entry each (neither deleted)."""
return [{"cons_journal_id": str(derived_id(e["cons_journal_id"], "reversal")), "group_id": e["group_id"], "period_id": e["period_id"],
"document_type": e["document_type"], "task": "reversal", "run_no": run_no, "entry_no": 0, "reverses_id": e["cons_journal_id"],
"memo": f"reverses {e['cons_journal_id']} (run {e['run_no']})", "period_specific": "",
"lines": [{**ln, "amount": -ln["amount"]} for ln in e["lines"]]} for e in entries if not e.get("reverses_id")]
def _entry(group, period_id, document, task, company, partner, run_no, lines, memo) -> dict:
lines = [ln for ln in lines if ln["amount"]]
return {"cons_journal_id": str(derived_id("cons", group.group_id, period_id, task, company, partner, run_no)),
"group_id": group.group_id, "period_id": period_id, "document_type": document, "task": task, "run_no": run_no,
"entry_no": 0, "reverses_id": "", "period_specific": "", "memo": memo, "lines": lines}
def translate(repo, group, period_id: str, *, by: str):
gp = the_period(group, period_id)
_not_signed(gp)
if gp.status in ("translated", "eliminated") and gp.stale_from_step == "translate":
gp.run_no, gp.status = gp.run_no + 1, "collected" # a rule or an earlier month changed since: the next run (MUST-3)
elif gp.status in ("translated", "eliminated"):
return None # a repeat writes nothing
if gp.status != "collected":
raise RuleBroken(f"a {gp.status} group period cannot be translated — collect it first")
if gp.stale_from_step == "collect":
raise RuleBroken("the structure changed since the last collection — collect again before translating")
members = C.members_at(group, period_id)
companies = sorted(members)
cfg = settings(repo)
months = C.fiscal_months(period_id, group.fiscal_year_end_month)
run_no = gp.run_no
held = {}
for r in repo.data.collections(group.group_id, period_id):
if r["company_code"] not in held or r["run_no"] > held[r["company_code"]]["run_no"]:
held[r["company_code"]] = r
absent = [c for c in companies if c not in held]
if absent:
raise RuleBroken(f"nothing collected for {', '.join(absent)} in {period_id} — collect first")
currency = {c: held[c]["local_currency"] for c in companies}
foreign = sorted({ccy for ccy in currency.values() if ccy != group.group_currency})
closing_by, monthly_by = _rates(group, period_id, foreign, months)
every = repo.data.reported(group.group_id, months, companies)
current = {c: [] for c in companies}
earlier: dict = {}
for r in every:
if r["period_id"] == period_id:
current[r["company_code"]].append(r)
else:
earlier.setdefault(r["company_code"], {}).setdefault(
(r["group_account_id"], r["partner_company"], r["movement_type"], r["segment"]), {})[r["period_id"]] = r["period_local"]
missing_rules = C.unruled([r for rows in current.values() for r in rows], group.account_rules, period_id)
if missing_rules:
raise RuleBroken(f"no translation rule for the group accounts {', '.join(missing_rules)} — set how each is translated")
history = C.history_at(repo.data.historical(group.group_id, companies), period_id) # fixed once, from the basis change (CON-02/CON-17)
translated, entries = [], []
for c in companies:
ccy = currency[c]
one = Decimal("1.000000")
rows = C.translate_rows(current[c], rules=group.account_rules, period_id=period_id, closing=closing_by.get(ccy, one),
monthly=monthly_by.get(ccy) or {m: one for m in months}, history=history.get(c, {}),
earlier=earlier.get(c, {}), months=months, pl_mode=cfg["translation_pl_rate"])
translated += [{"group_id": group.group_id, "period_id": period_id, "run_no": run_no, "company_code": c, **r} for r in rows]
diff = sum((r["ytd_group"] for r in rows), Decimal(0))
if diff and members[c].method != "equity": # an associate's translation difference moves to OCI by the equity method (CON-09)
nci_share = C.HUNDRED - C.effective_ownership(c, members)
lines = C.translation_entry(diff, nci_share, role_account(group, "cta", period_id),
role_account(group, "nci_equity", period_id, need=nci_share > 0), c)
if diff + sum((ln["amount"] for ln in lines), Decimal(0)) != 0:
raise RuleBroken(f"{c}'s translated data and its translation entry do not net to zero")
entries.append(_entry(group, period_id, "translation", "translation", c, "", run_no, lines,
f"translation difference {c} {ccy}→{group.group_currency} {period_id}"))
repo.data.claim(group.group_id, period_id, "translate", run_no)
earlier_entries = repo.data.posted_entries(group.group_id, period_id, TRANSLATION_DOCS, run_no)
repo.data.mark_reversed(earlier_entries)
repo.data.post_entries(_mirror(earlier_entries, run_no) + entries, C.fiscal_year_of(period_id, group.fiscal_year_end_month))
repo.data.add_translated(translated)
gp.status = "translated"
gp.stale_from_step = "" if gp.stale_from_step == "translate" else gp.stale_from_step
group._record(GroupPeriodTranslated(group_id=group.group_id, period_id=period_id, by=by))
return {"period_id": period_id, "run_no": run_no, "rows": len(translated), "entries": [e["cons_journal_id"] for e in entries],
"reversed": [e["cons_journal_id"] for e in earlier_entries]}
# ═══ CON-05 .. CON-07 eliminate ═════════════════════════════════════════════════════════════════════════════════
def _break_text(breaks: list) -> str:
return "; ".join(f"{b.get('key')}: ours {b.get('ours')} theirs {b.get('theirs')} {b.get('currency')} ({b.get('reason', '')}"
f"{' — fix: ' + b['fix'] if b.get('fix') else ''})" for b in breaks) or "gl has no match run for the pair"
def eliminate(repo, group, period_id: str, *, by: str):
gp = the_period(group, period_id)
_not_signed(gp)
if gp.status == "eliminated":
return None
if gp.status != "translated":
raise RuleBroken(f"a {gp.status} group period cannot be eliminated — translate it first")
members = C.members_at(group, period_id)
run_no = gp.run_no
day = C.period_end(period_id)
translated = {}
trows = repo.data.translated(group.group_id, period_id, run_no)
for r in trows:
translated[(r["company_code"], r["group_account_id"], r["partner_company"], r["movement_type"], r["segment"])] = r["ytd_group"]
inventory = repo.data.inventory(group.group_id, period_id)
held = {}
for r in repo.data.collections(group.group_id, period_id):
if r["company_code"] not in held or r["run_no"] > held[r["company_code"]]["run_no"]:
held[r["company_code"]] = r
refusals = []
# gate: an approved acquisition for every full and equity company
effective = repo.data.effective_changes(group.group_id) # every effective change of the group: one statement
acq = {c: ownership.acquisition_basis(effective, c) for c in members}
acq = {c: ch for c, ch in acq.items() if ch}
bought = set(acq)
fy = C.fiscal_year_of(period_id, group.fiscal_year_end_month)
sold = [ch for ch in effective if ch["kind"] == "divestiture"]
sold_now = [ch for ch in sold if C.period_of(ownership.day_of(ch["effective_date"])) == period_id]
sold_before = [ch for ch in sold if C.period_of(ownership.day_of(ch["effective_date"])) < period_id
and C.fiscal_year_of(C.period_of(ownership.day_of(ch["effective_date"])), group.fiscal_year_end_month) == fy]
gone = {ch["company_code"] for ch in sold_now} # their balance sheet leaves the group at this period's end
for c, m in sorted(members.items()):
if m.investor and m.method in ("full", "equity") and c not in bought:
refusals.append(f"no approved acquisition for {c} (CON-02): record and approve it")
rules = sorted(group.elimination_rules, key=lambda r: r.rule_id)
companies = sorted(members)
pairs = [(x, y) for x in companies for y in companies if x != y and members[x].method == members[y].method == "full"] # not with an associate
closing_cache: dict = {group.group_currency: Decimal("1.000000")}
def closing_of(ccy: str) -> Decimal:
if ccy not in closing_cache:
closing_cache[ccy] = C.rate(gl_call(ports.gl.fx_rate, ccy, day, "closing", group.group_currency), f"closing rate {ccy}")
return closing_cache[ccy]
# gate: intercompany stock entered or stated none for each selling pair (a seller with intercompany revenue to a buyer)
entered = {(i["holder_company"], i["seller_company"]) for i in inventory}
for rule in (r for r in rules if r.kind == "ic_pl"):
for x, y in pairs:
if C.pair_sums(translated, x, y, C.accounts(rule.left_accounts)) and (y, x) not in entered:
refusals.append(f"ic_inventory_entered: {x} sold to {y} and no intercompany stock is entered for {y} — enter it, or state none")
entries = []
for rule in rules:
left_a, right_a, tol = C.accounts(rule.left_accounts), C.accounts(rule.right_accounts), Decimal(rule.tolerance or 0)
diff_account = role_account(group, rule.difference_role or "ic_difference_bs", period_id, need=False)
for x, y in pairs:
if rule.kind == "ic_balance" and (x in gone or y in gone):
continue # the balance sheet of a company sold in this period is not consolidated: what it owed the group is external now
if rule.kind in ("ic_balance", "ic_pl"):
left, right = C.pair_sums(translated, x, y, left_a), C.pair_sums(translated, y, x, right_a)
if not left and not right:
continue
lines, diff = C.balance_elimination(left, right, x, y)
explained = C.ZERO
if diff and rule.kind == "ic_balance":
rec = gl_call(ports.gl.reconciliation, x, y, day)
breaks = (rec or {}).get("breaks", [])
no_match = rec is None
good = sum((abs(Decimal(str(b["difference"]))) * closing_of(b["currency"]) for b in breaks
if b.get("accepted") or b.get("within_tolerance")), Decimal(0))
explained, residual = C.split_difference(diff, good * (1 if diff > 0 else -1), tol)
if abs(residual) > tol:
refusals.append(f"ic_within_tolerance: {x}/{y} differ by {diff} (tolerance {tol}); gl's breaks: {_break_text(breaks)}"
+ (" — gl has no match run for the pair (404); fix: run match_intercompany in gl, then collect again and eliminate"
if no_match else " — fix the difference in gl (GL-52), collect again, then eliminate"))
continue
other = residual
else:
other = diff
if diff and rule.kind == "ic_pl":
explained, other = diff, C.ZERO
if diff:
if diff_account is None:
raise RuleBroken(f"the group has no account for the role {rule.difference_role}")
lines.append(C.line(diff_account, x, y, explained))
if other:
# an unexplained difference between companies that keep books in different currencies is translation:
# it goes to the translation reserve; between same-currency companies it is a difference like the rest
translation = held[x]["local_currency"] != held[y]["local_currency"]
lines.append(C.line(role_account(group, "cta", period_id) if translation else diff_account, x, y, other))
entries.append(_entry(group, period_id, rule.kind, f"{rule.kind}:{rule.rule_id}", x, y, run_no, lines,
f"{rule.kind} {rule.rule_id} {x}/{y} {period_id}"))
elif rule.kind == "dividend":
income = C.pair_sums(translated, x, y, left_a)
paid = {a: v for a, v in C.pair_sums(translated, y, x, right_a, "dividend").items()}
paid.update({a: v for a, v in C.pair_sums(translated, y, "", right_a, "dividend").items() if a not in paid})
if not income and not paid:
continue
lines, diff = C.balance_elimination(income, paid, x, y)
nci_share = C.HUNDRED - C.effective_ownership(y, members)
nci_amount = C.q2(sum(paid.values(), Decimal(0)) * nci_share / C.HUNDRED)
if nci_amount:
lines.append(C.line(role_account(group, "nci_equity", period_id), y, "", nci_amount))
rest = diff - nci_amount # what the lines so far leave: left + right - nci
rest_acc = role_account(group, rule.difference_role or "ic_difference_pl", period_id) if rest else None
if rest:
lines.append(C.line(rest_acc, x, y, rest))
entries.append(_entry(group, period_id, "dividend", f"dividend:{rule.rule_id}", x, y, run_no, lines,
f"dividend {rule.rule_id} {y}→{x} {period_id}"))
# unrealised profit in stock (CON-07)
for inv in sorted(inventory, key=lambda i: (i["holder_company"], i["seller_company"])):
if inv["none_stated"] == "true" or not inv["inventory_amount"]:
continue
h, s = inv["holder_company"], inv["seller_company"]
rate, why = deferred_tax_rate(h, inv["tax_rate_pct"], day)
if rate is None:
refusals.append(f"tax_rate_known: an intercompany inventory entry of {h} has no tax rate, and {why} — enter the rate on the entry, or set "
"statutory_tax_rate in gl (GL settings)")
continue
fig = C.stock_profit(inv["inventory_amount"], inv["margin_pct"], rate, closing_of(held[h]["local_currency"]))
lines = [C.line(role_account(group, "cost_of_sales", period_id), s, h, fig["profit"]),
C.line(role_account(group, "inventory", period_id), h, s, -fig["profit"])]
if fig["tax"]:
lines += [C.line(role_account(group, "deferred_tax_asset", period_id), h, s, fig["tax"]),
C.line(role_account(group, "tax_expense", period_id), h, s, -fig["tax"])]
nci_share = C.HUNDRED - C.effective_ownership(s, members)
if nci_share > 0: # an upstream sale: the minority of the seller bears its share
nci_amount = C.q2((fig["profit"] - fig["tax"]) * nci_share / C.HUNDRED)
lines += [C.line(role_account(group, "nci_equity", period_id), s, "", nci_amount),
C.line(role_account(group, "nci_result", period_id), s, "", -nci_amount)]
entries.append(_entry(group, period_id, "stock_profit", "stock_profit", h, s, run_no, lines,
f"unrealised profit in {h}'s stock bought from {s} {period_id}"))
# last year's unrealised profit in stock, realised this year (CON-15)
from brick_consolidation.application import carry
realised = carry.realisation_lines(repo, group, period_id)
if realised:
entries.append(_entry(group, period_id, "stock_profit", "stock_profit_realised", "", "", run_no, realised,
f"last year's unrealised profit in stock realised {period_id}"))
# the parent's investment against the subsidiary's equity, the minority (CON-08) and the associates by the equity method (CON-09)
extra = _equity_transactions(group, effective, members, period_id, entries, run_no)
own_entries, shown = _ownership_entries(repo, group, period_id, members, translated, acq, run_no, refusals, gone=gone, adjust=extra)
entries += own_entries
shown = {**shown, "deconsolidated": _deconsolidations(repo, group, period_id, members, translated, trows, effective, run_no, sold_now,
sold_before, entries)}
if refusals:
raise RuleBroken("eliminations refused: " + " | ".join(refusals))
for e in entries:
if sum((ln["amount"] for ln in e["lines"]), Decimal(0)) != 0:
raise RuleBroken(f"entry {e['task']} {e['memo']} does not net to zero")
repo.data.claim(group.group_id, period_id, "eliminate", run_no)
earlier_entries = repo.data.posted_entries(group.group_id, period_id, ELIMINATION_DOCS, run_no)
repo.data.mark_reversed(earlier_entries)
repo.data.post_entries(_mirror(earlier_entries, run_no) + entries, C.fiscal_year_of(period_id, group.fiscal_year_end_month))
gp.status = "eliminated"
gp.stale_from_step = ""
group._record(GroupPeriodEliminated(group_id=group.group_id, period_id=period_id, by=by))
return {"period_id": period_id, "run_no": run_no, "entries": [e["cons_journal_id"] for e in entries],
"reversed": [e["cons_journal_id"] for e in earlier_entries], **shown}
def prior_period(period_id: str) -> str:
y, m = int(period_id[:4]), int(period_id[5:7])
return f"{y - (m == 1):04d}-{(m - 2) % 12 + 1:02d}"
def day_of(value) -> date:
return value if isinstance(value, date) else date.fromisoformat(str(value))
def _pre_acquisition_result(group, period_id: str, change: dict, company: str, pl: set, refusals: list):
"""The result a company bought part-way through the year earned BEFORE the purchase, in group currency (credit positive): each month
before the purchase at that month's own average rate, from gl's movement of the month. Zero when it was bought by the year's first
month. None (and a refusal) when gl cannot give a month it needs."""
months = C.fiscal_months(period_id, group.fiscal_year_end_month)
bought = C.period_of(day_of(change["effective_date"]))
before = [m for m in months if m < bought]
if not before:
return C.ZERO
ccy = change.get("currency") or group.group_currency
last = gl_call(ports.gl.balances, company, group.ledger_id, before[-1])
if not any(C.amount(r["ytd_amount"]) for r in last.get("rows", []) if str(r["group_account_id"]) in pl):
return C.ZERO # nothing earned before the purchase: no month to split
total = Decimal(0)
for m in before:
try:
data = last if m == before[-1] else ports.gl.balances(company, group.ledger_id, m)
avg = C.rate(ports.gl.fx_rate(ccy, C.period_end(m), "average", group.group_currency), f"average rate {ccy} {m}")
except Missing as miss:
refusals.append(f"post_acquisition_result: {company} was bought in {bought}; its result of {m} (before the purchase) is taken out at "
f"{m}'s average rate, and gl cannot give it ({miss.reason}) — add it in gl")
return None
for r in data.get("rows", []):
if str(r["group_account_id"]) in pl:
total -= C.q2(C.amount(r["period_amount"]) * avg)
return total
def _ownership_entries(repo, group, period_id: str, members: dict, translated: dict, acq: dict, run_no: int, refusals: list, gone=frozenset(),
adjust=None) -> tuple:
"""CON-08 (investment + minority) for each fully consolidated subsidiary, CON-09 (equity method) for each associate."""
adjust = adjust or {}
subs = [c for c, m in sorted(members.items()) if m.investor and m.method == "full" and c in acq and c not in gone]
assoc = [c for c, m in sorted(members.items()) if m.investor and m.method == "equity" and c in acq]
if not subs and not assoc:
return [], {}
pl = {r.group_account_id for r in group.account_rules if r.rate_type == "average"}
hist = {}
for h in repo.data.historical(group.group_id, subs):
hist.setdefault((h["company_code"], h["change_id"]), {})[h["group_account_id"]] = (h["amount_local"], h["amount_group"])
entries, investments, associates = [], [], []
def balance(investor, company, account):
return sum(C.pair_sums(translated, investor, company, [account]).values(), Decimal(0))
for c in subs:
m, ch = members[c], acq[c]
inv = role_account(group, "investment", period_id)
held = hist.get((c, ch["change_id"]), {})
b_booked = balance(m.investor, c, inv)
if not b_booked:
refusals.append(f"investment_eliminated: {m.investor} has nothing on {inv} for {c} — book the investment, or there is nothing to eliminate it against")
continue
b = b_booked - adjust.get(c, Decimal(0)) # a later stake bought without control is eliminated by its own entry (CON-17)
if not held:
refusals.append(f"investment_eliminated: no equity at acquisition is fixed for {c} (the acquisition's historical amounts are missing)")
continue
pre = _pre_acquisition_result(group, period_id, ch, c, pl, refusals)
if pre is None:
continue
equity = {a: v[1] for a, v in held.items()}
accounts = {"investment": inv, "goodwill": role_account(group, "goodwill", period_id), "cta": role_account(group, "cta", period_id),
"nci_equity": role_account(group, "nci_equity", period_id), "nci_result": role_account(group, "nci_result", period_id)}
fvs = json.loads(ch.get("fair_value_adjustments_json") or "[]")
step_up = None
if fvs: # the fair-value step-up and its deferred tax enter the books with the elimination (CON-02)
step_up = C.step_up_lines(fvs, Decimal(ch["rate"]), role_account(group, "fair_value_step_up", period_id, need=False),
role_account(group, "deferred_tax_liability", period_id, need=False), C.HUNDRED - Decimal(ch["ownership_pct"]))
ccy = ch.get("currency") or group.group_currency
try: # IAS 21.47: goodwill and the step-up are held in the subsidiary's currency, shown at closing
closing = C.rate(ports.gl.fx_rate(ccy, C.period_end(period_id), "closing", group.group_currency), f"closing rate {ccy}")
except Missing as miss:
refusals.append(f"investment_eliminated: goodwill and fair-value adjustments of {c} are shown at {period_id}'s closing rate, and gl cannot "
f"give it ({miss.reason}) — add it in gl")
continue
goodwill_close = C.q2(Decimal(ch["goodwill"] or 0) * closing) if ch.get("goodwill") is not None else None
step_up_close = None
if step_up:
step_up_close = C.step_up_lines(fvs, closing, role_account(group, "fair_value_step_up", period_id, need=False),
role_account(group, "deferred_tax_liability", period_id, need=False), C.HUNDRED - Decimal(ch["ownership_pct"]))
gain = Decimal(ch["gain_loss"] or 0) if ch["kind"] == "ownership_change" else C.ZERO # IFRS 3.42, taking control in stages
if gain:
accounts["gain"] = role_account(group, "gain_on_disposal", period_id)
inv_lines, nci_lines = C.investment_entries(investor=m.investor, company=c, balance=b, equity=equity, nci_group=ch["nci_group"],
goodwill_group=ch["goodwill_group"], accounts=accounts, step_up=step_up, remeasure_gain=gain,
goodwill_close=goodwill_close, step_up_close=step_up_close)
# goodwill is not amortised under IFRS; where the group sets a life (a local GAAP that requires it) it is, straight line
amortised = C.ZERO
if group.goodwill_life_years and ch["goodwill_group"]:
bought = C.period_of(day_of(ch["effective_date"]))
gross = ch["goodwill_group"] if goodwill_close is None else goodwill_close # the goodwill as shown at this period's closing rate
total = C.goodwill_amortisation(gross, group.goodwill_life_years, C.months_between(bought, period_id))
this_year = C.goodwill_amortisation(gross, group.goodwill_life_years,
len([x for x in C.fiscal_months(period_id, group.fiscal_year_end_month) if x >= bought]))
if total:
amortised = total
inv_lines.append(C.line(accounts["goodwill"], c, "", -total))
inv_lines.append(C.line(role_account(group, "goodwill_amortisation", period_id), c, "", this_year))
if total - this_year:
inv_lines.append(C.line(role_account(group, "retained_earnings", period_id), c, "", total - this_year))
result = C.result_of(translated, c, pl) - pre # the months before the purchase are not the group's (CON-08)
nci_pct = C.HUNDRED - C.effective_ownership(c, members)
nci_lines += C.nci_result_lines(c, result, nci_pct, accounts)
entries.append(_entry(group, period_id, "investment", "investment", m.investor, c, run_no, inv_lines,
f"investment of {m.investor} in {c} against its equity at acquisition {period_id}"))
if nci_lines:
entries.append(_entry(group, period_id, "nci", "nci", c, "", run_no, nci_lines, f"non-controlling interest in {c} {period_id}"))
investments.append({"company_code": c, "investor": m.investor, "eliminated": b, "goodwill_group": ch["goodwill_group"],
"goodwill_closing": goodwill_close if goodwill_close is not None else ch["goodwill_group"],
"nci_at_acquisition": ch["nci_group"], "nci_result": C.q2(result * nci_pct / C.HUNDRED),
"post_acquisition_result": result, "goodwill_amortised": amortised})
reported: dict = {}
if assoc:
for r in repo.data.reported(group.group_id, [period_id], assoc):
if r["group_account_id"] in pl:
reported[r["company_code"]] = reported.get(r["company_code"], Decimal(0)) + r["ytd_local"]
for c in assoc:
m, ch = members[c], acq[c]
inv = role_account(group, "investment", period_id)
if not balance(m.investor, c, inv):
refusals.append(f"investment_booked: {m.investor} has nothing on {inv} for the associate {c}")
continue
pre = _pre_acquisition_result(group, period_id, ch, c, pl, refusals)
if pre is None:
continue
share = C.effective_ownership(c, members)
diff = sum((v for (cc, _a, _p, _m, _s), v in translated.items() if cc == c), Decimal(0))
income = sum(C.pair_sums(translated, m.investor, c, [role_account(group, "dividend_income", period_id)]).values(), Decimal(0))
accounts = {"investment": inv, "share_of_associates": role_account(group, "share_of_associates", period_id),
"oci": role_account(group, "oci", period_id), "dividend_income": role_account(group, "dividend_income", period_id)}
lines, parts = C.equity_method_lines(investor=m.investor, associate=c, share_pct=share, result=C.result_of(translated, c, pl) - pre,
translation_diff=diff, dividend_income=income, accounts=accounts)
if lines:
entries.append(_entry(group, period_id, "equity_method", "equity_method", m.investor, c, run_no, lines,
f"equity method {m.investor} in {c} {period_id}"))
associates.append({"company_code": c, "share_pct": share, "share_of_result_local": C.q2(-reported.get(c, Decimal(0)) * share / C.HUNDRED),
"share_of_result": parts["profit"], "oci": parts["oci"], "dividend": parts["dividend"]})
return entries, {"investments": investments, "associates": associates}
def _equity_transactions(group, effective: list, members: dict, period_id: str, entries: list, run_no: int) -> dict:
"""CON-17 without a change of control: the extra price against the minority given up, the rest to group equity. Appends the entries and
answers the price already eliminated per company (the investment entry leaves it out)."""
if not any(ch["kind"] == "ownership_change" and ch["goodwill_group"] is None for ch in effective):
return {}
accounts = {"investment": role_account(group, "investment", period_id), "nci_equity": role_account(group, "nci_equity", period_id),
"retained_earnings": ownership.retained_earnings_account(group, period_id)}
done, eliminated = ownership.equity_transaction_entries(group, effective, members, period_id, accounts)
for ch, lines in done:
entries.append(_entry(group, period_id, "ownership_change", f"ownership_change:{ch['change_id']}", ch["company_code"], "", run_no, lines,
f"ownership of {ch['company_code']} {ch['ownership_pct']}% from {ch['effective_date']}: the minority given up against the price"))
return eliminated
def _deconsolidations(repo, group, period_id: str, members: dict, translated: dict, trows: list, effective: list, run_no: int, sold_now: list,
sold_before: list, entries: list) -> list:
"""CON-16: the entry of each company sold in this period (its balance sheet leaves, gain and recycled reserve to gain_on_disposal) and, in
the later periods of the same year, the same result without the balance sheet, so the year's figures keep what the company earned up to
the sale. The next year's start carries it (CON-15)."""
shown = []
if not (sold_now or sold_before):
return shown
inv = role_account(group, "investment", period_id)
accounts = {"gain": role_account(group, "gain_on_disposal", period_id), "investment": inv, "cta": role_account(group, "cta", period_id),
"nci_equity": role_account(group, "nci_equity", period_id), "nci_result": role_account(group, "nci_result", period_id),
"retained_earnings": ownership.retained_earnings_account(group, period_id)}
for ch in sold_now + sold_before:
c, investor = ch["company_code"], ch["investor_company_code"]
at = ch in sold_now
p = C.period_of(ownership.day_of(ch["effective_date"]))
if at:
rows_p, members_p = trows, members
else:
gp_p = the_period(group, p)
rows_p, members_p = repo.data.translated(group.group_id, p, gp_p.run_no), C.members_at(group, p)
base = ownership.acquisition_basis(effective, c)
figs = ownership.divestiture_figures(group, ch, base, rows_p, members_p, p)
lines = ownership.deconsolidation_entry(group, ch, base, rows_p, members_p, accounts, at_period=at, period_id=p,
investment_balance=sum(C.pair_sums(translated, investor, c, [inv]).values(), Decimal(0)),
own_gain=sum(C.pair_sums(translated, investor, c, [accounts["gain"]]).values(), Decimal(0)))
entries.append(_entry(group, period_id, "disposal", f"disposal:{c}", c, "", run_no, lines, f"sale of {c} on {ch['effective_date']}"))
shown.append({"company_code": c, "effective_date": str(ch["effective_date"]), **{k: v for k, v in figs.items()}})
return shown