dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/domain/consolidation.py
"""The group's arithmetic — structure in force, ownership, goodwill, translation, eliminations. Pure: no storage, no gl.
Every amount is a Decimal, NUMERIC(18,2), debit positive (gl's sign); every rate a Decimal at gl's six places. Rounding is
ROUND_HALF_UP and happens once per figure, where the figure is made — never in a sum of figures — so a run is reproducible to the
penny and the translation difference is exactly what is left over.
"""
from collections import defaultdict
from dataclasses import dataclass
from datetime import date, timedelta
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
from typing import Optional
from brick_consolidation.domain.errors import Invalid, RuleBroken
CENT = Decimal("0.01")
RATE_Q = Decimal("0.000001")
MAX_AMOUNT = Decimal("9999999999999999.99") # NUMERIC(18,2)
ZERO = Decimal("0.00")
HUNDRED = Decimal(100)
# ── numbers ────────────────────────────────────────────────────────────────────────────────────────────────────
def dec(value, what: str = "amount") -> Decimal:
"""A Decimal from text or a number; never a float's tail, never NaN/Infinity."""
if isinstance(value, float):
raise Invalid(f"{what} must be a decimal string, not a float")
try:
d = Decimal(str(value))
except (InvalidOperation, ValueError):
raise Invalid(f"{what} {value!r} is not a number") from None
if not d.is_finite():
raise Invalid(f"{what} must be a finite number")
return d
def q2(value) -> Decimal:
return Decimal(value).quantize(CENT, ROUND_HALF_UP)
def amount(value, what: str = "amount") -> Decimal:
"""NUMERIC(18,2): at most two decimals, 16 integer digits (refused 422, never a 500 at the database)."""
d = dec(value, what)
if d != d.quantize(CENT, ROUND_HALF_UP) or abs(d) > MAX_AMOUNT:
raise Invalid(f"{what} {value} does not fit NUMERIC(18,2) (two decimals, at most {MAX_AMOUNT})")
return q2(d)
def pct(value, what: str, *, zero_ok: bool = False) -> Decimal:
"""NUMERIC(9,6) above 0 (or from 0) and at most 100."""
d = dec(value, what)
if d != d.quantize(RATE_Q, ROUND_HALF_UP) or d > HUNDRED or d < 0 or (d == 0 and not zero_ok):
low = "from 0" if zero_ok else "above 0"
raise Invalid(f"{what} {value} must be {low} and at most 100, with at most six decimals")
return d.quantize(RATE_Q)
def rate(value, what: str = "rate") -> Decimal:
d = dec(value, what)
if d <= 0 or d > Decimal("999999999999.999999"):
raise Invalid(f"{what} {value} is not a usable rate")
return d.quantize(RATE_Q, ROUND_HALF_UP)
# ── periods (YYYY-MM) ──────────────────────────────────────────────────────────────────────────────────────────
def check_period(period_id: str) -> str:
try:
y, m = int(period_id[:4]), int(period_id[5:7])
if len(period_id) != 7 or period_id[4] != "-" or not 1 <= m <= 12 or y < 1900:
raise ValueError
except (ValueError, TypeError):
raise Invalid(f"period {period_id!r} must be YYYY-MM") from None
return period_id
def period_end(period_id: str) -> date:
check_period(period_id)
y, m = int(period_id[:4]), int(period_id[5:7])
first_next = date(y + (m == 12), m % 12 + 1, 1)
return first_next - timedelta(days=1)
def fiscal_months(period_id: str, fy_end_month) -> list:
"""The months of the group's financial year up to and including `period_id` (a year ending in March starts in April)."""
check_period(period_id)
end = int(fy_end_month or 12)
start_m = end % 12 + 1
y, m = int(period_id[:4]), int(period_id[5:7])
start_y = y if m >= start_m else y - 1
out, cy, cm = [], start_y, start_m
while (cy, cm) <= (y, m):
out.append(f"{cy:04d}-{cm:02d}")
cy, cm = (cy + (cm == 12), cm % 12 + 1)
return out
def fiscal_year_of(period_id: str, fy_end_month) -> str:
"""The group's financial year of a period, named by the year it ends in (gl's convention: FY2027 ends in March 2027)."""
check_period(period_id)
end = int(fy_end_month or 12)
y, m = int(period_id[:4]), int(period_id[5:7])
return f"FY{y + (1 if end != 12 and m > end else 0)}"
def period_of(day: date) -> str:
return f"{day.year:04d}-{day.month:02d}"
# ── structure ──────────────────────────────────────────────────────────────────────────────────────────────────
@dataclass(frozen=True)
class Member:
company_code: str
investor: str
ownership: Decimal
voting: Decimal
method: str
from_period: str
def members_at(group, period_id: str) -> dict:
"""The structure in force on the period's last day: per company the newest row that started on or before the period,
and the top company (always full, 100%). Companies whose method is none are not in scope."""
best: dict = {}
for row in group.group_structures:
if row.valid_from_period <= period_id and (row.company_code not in best or
row.valid_from_period > best[row.company_code].valid_from_period):
best[row.company_code] = row
out = {c: Member(c, r.investor_company_code, Decimal(r.ownership_pct), Decimal(r.voting_pct), r.method, r.valid_from_period)
for c, r in best.items() if r.method != "none"}
if group.top_company_code:
out[group.top_company_code] = Member(group.top_company_code, "", HUNDRED, HUNDRED, "full", "")
return out
def effective_ownership(company: str, members: dict) -> Decimal:
"""The group's share of a company through the chain of investors (UK01 80% of DE01 which owns 100% of DE02: 80)."""
share, seen, cur = HUNDRED, set(), company
while cur in members and members[cur].investor:
if cur in seen:
raise RuleBroken(f"the ownership of {company} runs in a circle")
seen.add(cur)
share = share * members[cur].ownership / HUNDRED
cur = members[cur].investor
return share.quantize(RATE_Q)
# ── acquisition (CON-02) ───────────────────────────────────────────────────────────────────────────────────────
def acquisition_figures(consideration: Decimal, ownership: Decimal, net_assets: Decimal, fair_values: list, rate_at: Decimal) -> dict:
"""goodwill = consideration - share of net assets (adjusted by the fair-value adjustments net of their deferred tax);
NCI = (100% - share) of the same net assets (proportionate share, IFRS 3.19). Local currency, then group at the acquisition rate."""
adjusted = net_assets
for fv in fair_values:
value = amount(fv.get("amount"), "fair-value adjustment")
tax = pct(fv.get("tax_rate_pct", "0"), "fair-value deferred tax rate", zero_ok=True)
adjusted += value - q2(value * tax / HUNDRED) # the uplift less its deferred tax liability
share = q2(adjusted * ownership / HUNDRED)
goodwill = consideration - share
nci = adjusted - share
return {"net_assets": adjusted, "goodwill": q2(goodwill), "nci": q2(nci),
"goodwill_group": q2(goodwill * rate_at), "nci_group": q2(nci * rate_at), "bargain_purchase": goodwill < 0}
# ── translation (CON-04) ───────────────────────────────────────────────────────────────────────────────────────
def rule_at(rules: list, account: str, period_id: str):
best = None
for r in rules:
if r.group_account_id == account and r.valid_from_period <= period_id and (best is None or r.valid_from_period > best.valid_from_period):
best = r
return best
def translate_rows(rows: list, *, rules: list, period_id: str, closing: Decimal, monthly: dict, history: dict, earlier: dict,
months: list, pl_mode: str) -> list:
"""Translate one company's reported rows into the group currency.
rows: dicts with group_account_id, partner_company, movement_type, segment, ytd_local, period_local.
closing: the closing rate on the period end. monthly: {month: average rate}. history: {account: (amount_local, amount_group)}
fixed at acquisition. earlier: {(account, partner, movement, segment): {month: period_local}} for the year's earlier months.
Returns rows with ytd_group, rate_type and rate. Balance sheet at closing; income and expense month by month at each
month's average (what the year's earlier months did not cover at the period's own average); equity at its fixed amounts,
anything since at closing."""
out, by_account = [], defaultdict(list)
for r in rows:
rule = rule_at(rules, r["group_account_id"], period_id)
kind = rule.rate_type if rule else None
if kind == "average":
if pl_mode == "ytd_average":
avg = (sum(monthly[m] for m in months) / len(months)).quantize(RATE_Q, ROUND_HALF_UP)
total = r["ytd_local"] * avg
eff = avg
else:
prior = earlier.get((r["group_account_id"], r["partner_company"], r["movement_type"], r["segment"]), {})
total, covered = Decimal(0), Decimal(0)
for m in months:
p = r["period_local"] if m == period_id else prior.get(m)
if p is not None:
total += p * monthly[m]
covered += p
total += (r["ytd_local"] - covered) * monthly[period_id]
eff = (q2(total) / r["ytd_local"]).quantize(RATE_Q, ROUND_HALF_UP) if r["ytd_local"] else monthly[period_id]
out.append({**_key(r), "ytd_group": q2(total), "rate_type": "average", "rate": eff})
elif kind == "historical":
by_account[r["group_account_id"]].append(r)
else:
out.append({**_key(r), "ytd_group": q2(r["ytd_local"] * closing), "rate_type": "closing", "rate": closing})
for account, group_rows in sorted(by_account.items()):
local_h, group_h = history.get(account, (Decimal(0), Decimal(0)))
total_local = sum((r["ytd_local"] for r in group_rows), Decimal(0))
total_group = q2(group_h + (total_local - local_h) * closing)
eff = (total_group / total_local).quantize(RATE_Q, ROUND_HALF_UP) if total_local else closing
group_rows = sorted(group_rows, key=lambda r: (r["partner_company"], r["movement_type"], r["segment"]))
given = Decimal(0)
for i, r in enumerate(group_rows):
if i == len(group_rows) - 1:
value = total_group - given # the last row takes the remainder: the account totals exactly
elif total_local:
value = q2(total_group * r["ytd_local"] / total_local)
else:
value = ZERO
given += value
out.append({**_key(r), "ytd_group": value, "rate_type": "historical", "rate": eff})
return out
YEAR_END = "year_end:" # the change_id of a year's result fixed as a historical amount (CON-15)
def history_at(rows: list, period_id: str) -> dict:
"""{company: {account: (amount_local, amount_group)}}: the historical amounts in force for a period. Only those from the company's
latest basis change on or before the period (an acquisition, or taking control in stages: the equity is fixed ONCE, at the date control
was obtained) plus each later year's result fixed at its year end; never an amount that arose after the period."""
basis: dict = {}
for h in rows:
if h["arose_period"] <= period_id and not str(h.get("change_id") or "").startswith(YEAR_END):
basis[h["company_code"]] = max(basis.get(h["company_code"], ""), h["arose_period"])
out: dict = {}
for h in rows:
c = h["company_code"]
if c not in basis or not basis[c] <= h["arose_period"] <= period_id:
continue
local, grp = out.setdefault(c, {}).get(h["group_account_id"], (Decimal(0), Decimal(0)))
out[c][h["group_account_id"]] = (local + Decimal(h["amount_local"]), grp + Decimal(h["amount_group"]))
return out
def year_end_history(*, reported: list, translated: list, history: dict, rules: list, period_id: str, reserves_account: Optional[str]) -> dict:
"""One company's year at its end, fixed as historical amounts for the years after (IAS 21: equity is not retranslated): per historical
account what moved during the year at the amount it was translated at, and the year's result at its translated amount on the reserves
account it closes into. {account: (amount_local, amount_group)}, the delta over the history already held."""
local, grp, res_l, res_g = defaultdict(Decimal), defaultdict(Decimal), Decimal(0), Decimal(0)
def kind(account):
rule = rule_at(rules, account, period_id)
return rule.rate_type if rule else None
for r in reported:
k = kind(r["group_account_id"])
if k == "historical":
local[r["group_account_id"]] += r["ytd_local"]
elif k == "average":
res_l += r["ytd_local"]
for r in translated:
k = kind(r["group_account_id"])
if k == "historical":
grp[r["group_account_id"]] += r["ytd_group"]
elif k == "average":
res_g += r["ytd_group"]
accounts = sorted(set(local) | set(grp) | set(history))
target = reserves_account if reserves_account in accounts else (accounts[-1] if accounts else None)
out = {}
for a in accounts:
h_l, h_g = history.get(a, (Decimal(0), Decimal(0)))
d_l, d_g = local[a] - h_l, grp[a] - h_g
if a == target:
d_l, d_g = d_l + res_l, d_g + res_g
out[a] = (q2(d_l), q2(d_g))
return out
def _key(r: dict) -> dict:
return {k: r[k] for k in ("group_account_id", "partner_company", "movement_type", "segment")}
def unruled(rows: list, rules: list, period_id: str) -> list:
return sorted({r["group_account_id"] for r in rows if rule_at(rules, r["group_account_id"], period_id) is None})
def translation_entry(difference: Decimal, nci_share: Decimal, cta_account: Optional[str], nci_account: Optional[str], company: str) -> list:
"""The balancing difference of a company's translated data goes to the translation reserve; the minority's share of it
to the NCI account. Lines add up to minus the difference, so translated data plus this entry net to zero exactly."""
nci_amount = q2(-difference * nci_share / HUNDRED)
lines = [{"group_account_id": cta_account, "company_code": company, "partner_company": "", "movement_type": "", "segment": "",
"amount": -difference - nci_amount}]
if nci_amount:
lines.append({"group_account_id": nci_account, "company_code": company, "partner_company": "", "movement_type": "", "segment": "",
"amount": nci_amount})
return lines
# ── eliminations (CON-05 .. CON-07) ────────────────────────────────────────────────────────────────────────────
def accounts(text: str) -> list:
return [a.strip() for a in (text or "").split(",") if a.strip()]
def line(account, company, partner, amount_, movement="") -> dict:
return {"group_account_id": account, "company_code": company, "partner_company": partner, "movement_type": movement,
"segment": "", "amount": amount_}
def pair_sums(translated: dict, company: str, partner: str, accts: list, movement: Optional[str] = None) -> dict:
"""{account: amount} of `company`'s translated rows towards `partner` on these accounts (all movements, or one)."""
out = {}
for (c, a, p, mv, _seg), v in translated.items():
if c == company and p == partner and a in accts and (movement is None or mv == movement):
out[a] = out.get(a, Decimal(0)) + v
return {a: v for a, v in out.items() if v}
def balance_elimination(left: dict, right: dict, x: str, y: str) -> tuple:
"""Remove both sides of an intercompany pair; returns (lines without the difference, difference). X's side is `left`
(receivable, revenue), Y's side `right` (payable, cost). The difference is what would remain: left + right."""
lines = [line(a, x, y, -v) for a, v in sorted(left.items())] + [line(a, y, x, -v) for a, v in sorted(right.items())]
return lines, sum(left.values(), Decimal(0)) + sum(right.values(), Decimal(0))
def split_difference(difference: Decimal, explained: Decimal, tolerance: Decimal) -> tuple:
"""(explained part, unexplained residual) of an intercompany difference. `explained` is what gl's break list accounts for
(accepted breaks, in group currency); it never exceeds the difference nor takes its opposite sign."""
if not difference:
return ZERO, ZERO
sign = 1 if difference > 0 else -1
covered = min(abs(difference), abs(explained)) * sign if explained * sign > 0 else ZERO
return q2(covered), q2(difference - covered)
def stock_profit(holder_amount: Decimal, margin: Decimal, tax_rate: Decimal, closing: Decimal) -> dict:
"""Unrealised profit in stock held by the buyer: amount x margin in the holder's currency, translated at closing; the deferred
tax at the holder's rate on the translated profit (IAS 12)."""
profit_local = q2(holder_amount * margin / HUNDRED)
profit = q2(profit_local * closing)
return {"profit_local": profit_local, "profit": profit, "tax": q2(profit * tax_rate / HUNDRED)}
# ── consolidation of investments and the equity method (CON-08, CON-09) ───────────────────────────────────────────
def spread(total: Decimal, weights: dict) -> dict:
"""`total` split across the keys in proportion to `weights`, each part rounded once and the last taking the remainder, so the parts
add up to the total exactly."""
keys, wsum, out, given = sorted(weights), sum(weights.values(), Decimal(0)), {}, Decimal(0)
for i, k in enumerate(keys):
value = total - given if i == len(keys) - 1 else (q2(total * weights[k] / wsum) if wsum else ZERO)
out[k] = value
given += value
return out
def step_up_lines(fair_values: list, rate_at: Decimal, default_asset, tax_account, nci_pct: Decimal) -> dict:
"""The fair-value adjustments of an acquisition in group currency (CON-02): per adjustment the asset stepped up and its deferred tax liability,
each translated once at the acquisition rate (asset first, the tax on the translated asset), and the minority's part of what is left.
Returns {"assets": [(account, amount)], "dtl": (account, amount) | None, "net": Decimal, "nci": Decimal}. Nothing is depreciated here: the
unwind of a step-up is a judgement the group accountant books as a manual entry (CON-10)."""
assets, dtl_total = [], Decimal(0)
for fv in fair_values:
value = q2(amount(fv.get("amount"), "fair-value adjustment") * rate_at)
tax = pct(fv.get("tax_rate_pct", "0"), "fair-value deferred tax rate", zero_ok=True)
account = str(fv.get("group_account_id") or default_asset or "")
if not account:
raise RuleBroken("a fair-value adjustment names the group account it steps up (group_account_id), or the group has an account for the "
"role fair_value_step_up")
assets.append((account, value))
dtl_total += q2(value * tax / HUNDRED)
if dtl_total and not tax_account:
raise RuleBroken("the group has no account for the role deferred_tax_liability (set it in the account roles)")
net = sum((v for _a, v in assets), Decimal(0)) - dtl_total
return {"assets": assets, "dtl": (tax_account, dtl_total) if dtl_total else None, "net": net, "nci": q2(net * nci_pct / HUNDRED)}
def investment_entries(*, investor: str, company: str, balance: Decimal, equity: dict, nci_group: Decimal, goodwill_group: Decimal,
accounts: dict, step_up: Optional[dict] = None, remeasure_gain: Decimal = ZERO,
goodwill_close: Optional[Decimal] = None, step_up_close: Optional[dict] = None) -> tuple:
"""The investment against the parent's share of the equity at acquisition (CON-08), and the minority's share of that equity.
balance: the investor's investment in the company as translated (debit positive). equity: {account: the equity fixed at acquisition in
group currency, credit negative}. Returns (investment lines, the NCI-at-acquisition lines), each netting to zero:
investment: Cr investment (balance); Dr the parent's part of each equity account; Dr goodwill; any rest to the translation reserve
nci: Dr the minority's part of each equity account; Cr non-controlling interest
step_up (CON-02): the acquisition's fair-value adjustments enter the consolidated books here: Dr the asset stepped up, Cr its deferred tax
liability, Cr the minority's part of the net uplift on non-controlling interest. The parent's part of the uplift is what the goodwill already
leaves out (goodwill = price - share x adjusted net assets), so the entry nets to zero without a plug. remeasure_gain (CON-17, IFRS 3.42): the
gain on the previously held interest of a step acquisition, in profit.
v2 (IAS 21.47): goodwill and the fair-value adjustments are assets of the foreign operation, held in ITS currency and shown at the period's
closing rate: goodwill_close and step_up_close are the same figures at that rate (default: the acquisition's). What they gain or lose since
acquisition is not profit: it falls out as the translation reserve line (the entry's balancing rest). The minority takes none of the
goodwill's difference (proportionate measurement: the goodwill is the group's alone) and its share of the fair-value adjustments' difference
(they are part of the net assets): the NCI line follows step_up_close."""
total = -sum(equity.values(), Decimal(0))
nci_fv = step_up["nci"] if step_up else ZERO
nci_hist = nci_group - nci_fv # the minority's part of the equity at acquisition itself
nci_part = spread(nci_hist, equity)
parent_part = spread(total - nci_hist, equity)
lines = [line(accounts["investment"], investor, company, -balance)]
lines += [line(a, company, "", v) for a, v in sorted(parent_part.items())]
goodwill_shown = goodwill_group if goodwill_close is None else goodwill_close
if goodwill_shown:
lines.append(line(accounts["goodwill"], company, "", goodwill_shown))
rest = balance - sum(parent_part.values(), Decimal(0)) - goodwill_shown
if step_up:
now = step_up_close or step_up
lines += [line(a, company, "", v) for a, v in now["assets"]]
if now["dtl"]:
lines.append(line(now["dtl"][0], company, "", -now["dtl"][1]))
if now["nci"]:
lines.append(line(accounts["nci_equity"], company, "", -now["nci"]))
rest += now["nci"] - now["net"]
if remeasure_gain:
lines.append(line(accounts["gain"], company, "", -remeasure_gain))
rest += remeasure_gain
if rest:
lines.append(line(accounts["cta"], company, "", rest)) # the investment was booked at another rate than the one fixed at acquisition
nci_lines = [line(a, company, "", v) for a, v in sorted(nci_part.items())]
if nci_hist:
nci_lines.append(line(accounts["nci_equity"], company, "", -nci_hist))
return lines, nci_lines
def result_of(translated: dict, company: str, pl_accounts: set) -> Decimal:
"""A company's profit (credit) for the year to date, translated: minus the sum of its profit-and-loss rows."""
return -sum((v for (c, a, _p, _m, _s), v in translated.items() if c == company and a in pl_accounts), Decimal(0))
def nci_result_lines(company: str, result: Decimal, nci_pct: Decimal, accounts: dict) -> list:
"""The minority's share of the post-acquisition result: Dr the NCI result line, Cr non-controlling interest."""
share = q2(result * nci_pct / HUNDRED)
return [line(accounts["nci_result"], company, "", share), line(accounts["nci_equity"], company, "", -share)] if share else []
def goodwill_amortisation(goodwill_group: Decimal, life_years, months: int) -> Decimal:
"""Straight line over the life, in whole months, never above the goodwill itself."""
if not life_years or not goodwill_group:
return ZERO
return min(goodwill_group, q2(goodwill_group * months / (Decimal(str(life_years)) * 12)))
def months_between(first_period: str, last_period: str) -> int:
"""The months from `first_period` to `last_period`, both counted."""
a, b = (int(first_period[:4]) * 12 + int(first_period[5:7])), (int(last_period[:4]) * 12 + int(last_period[5:7]))
return b - a + 1
def equity_method_lines(*, investor: str, associate: str, share_pct: Decimal, result: Decimal, translation_diff: Decimal, dividend_income: Decimal,
accounts: dict) -> tuple:
"""One entry for an associate (CON-09): the share of its result raises the investment, the dividend received reduces it (and the
investor's dividend income is eliminated), and the share of its translation difference moves to the group's OCI.
result: the associate's profit (credit) translated; translation_diff: the sum of its translated rows (what its translation reserve
balances); dividend_income: the investor's dividend income from it, as a credit (negative) amount. Returns (lines, shares)."""
profit = q2(result * share_pct / HUNDRED)
oci = q2(-translation_diff * share_pct / HUNDRED) # a gain is a credit, as in the translation reserve
lines = []
if profit:
lines += [line(accounts["investment"], investor, associate, profit), line(accounts["share_of_associates"], investor, associate, -profit)]
if oci:
lines += [line(accounts["investment"], investor, associate, -oci), line(accounts["oci"], investor, associate, oci)]
if dividend_income:
lines += [line(accounts["dividend_income"], investor, associate, -dividend_income), line(accounts["investment"], investor, associate, dividend_income)]
return lines, {"profit": profit, "oci": oci, "dividend": -dividend_income}
# ── divestiture and ownership change (CON-16, CON-17) ─────────────────────────────────────────────────────────────
def disposal_figures(*, rows: list, pl: set, equity: set, proceeds_group: Decimal, share: Decimal, goodwill_group: Decimal,
step_up_net: Decimal = ZERO, retranslation: Decimal = ZERO) -> dict:
"""What a sale of a whole holding takes out of the group, in group currency (IFRS 10.B98, IAS 21.48).
rows: the company's translated rows at the disposal period. Net assets = its balance-sheet accounts that are not equity (assets less
liabilities, debit positive) plus the fair-value step-up still carried. gain = proceeds - (the group's share of net assets + goodwill) -
the spec's figure. The translation reserve is the company's translation difference less the minority's part; the parent's part is recycled
to profit within the gain (IAS 21.48). The minority's interest, proportionate (IFRS 3.19), is derecognised with the net assets.
v3: goodwill_group and step_up_net are the CLOSING carrying amounts, retranslated at the disposal date's closing rate; `retranslation` is the
difference that retranslation has put in the reserve since acquisition, recycled with the rest, so gain + recycled reserve is the gain translated once."""
net_assets = sum((r["ytd_group"] for r in rows if r["group_account_id"] not in pl and r["group_account_id"] not in equity), Decimal(0)) + step_up_net
diff = sum((r["ytd_group"] for r in rows), Decimal(0))
nci_pct = HUNDRED - share
nci_translation = q2(-diff * nci_pct / HUNDRED)
cta_line = -diff - nci_translation # the translation reserve line as translate() booked it (credit negative)
share_of_net_assets = q2(net_assets * share / HUNDRED)
gain = q2(proceeds_group - share_of_net_assets - goodwill_group)
result = -sum((r["ytd_group"] for r in rows if r["group_account_id"] in pl), Decimal(0))
cta_goodwill_line = -q2(retranslation) # the reserve's credit for goodwill and step-up retranslated since acquisition (credit negative)
return {"net_assets_group": q2(net_assets), "result_group": q2(result), "nci_result": q2(result * nci_pct / HUNDRED), "share_of_net_assets": share_of_net_assets, "goodwill_group": goodwill_group,
"proceeds_group": proceeds_group, "gain": gain, "cta_line": cta_line, "cta_goodwill_line": cta_goodwill_line, "nci_translation_line": nci_translation,
"cta_recycled": -cta_line - cta_goodwill_line, "total_gain": q2(gain - cta_line - cta_goodwill_line), "nci_derecognised": q2(net_assets * nci_pct / HUNDRED)}
def disposal_lines(figs: dict, *, rows: list, pl: set, company: str, investor: str, investment_balance: Decimal, accounts: dict, at_period: bool,
own_gain: Decimal = ZERO) -> list:
"""The deconsolidation entry. In the period of the sale (`at_period`): the company's balance sheet and its translation entry come out,
its profit and loss stays (it is in the translated data). In a later period of the year the company is no longer a member: its profit
and loss up to the sale is put back. Either way the gain (and the recycled translation reserve) go to gain_on_disposal, the investor's
investment is derecognised, the minority's share of the profit up to the sale is attributed to it (nci_result), and the investor's own
separate gain on the sale (own_gain, a credit on gain_on_disposal against the company) is replaced by the group's. Given the investor
booked the proceeds and the derecognition, the lines net to zero by themselves (the parent's share of the net assets' movement is its
share of the result and of the translation difference); whatever is left over from books that differ goes to group retained earnings, so
the entry always nets to zero and is the same figure in every period."""
out = [line(accounts["gain"], investor, company, -figs["gain"])]
if own_gain:
out.append(line(accounts["gain"], investor, company, -own_gain))
if figs["nci_result"]:
out.append(line(accounts["nci_result"], company, "", figs["nci_result"]))
if figs["cta_line"]:
out.append(line(accounts["gain"], investor, company, figs["cta_line"])) # the recycled reserve is part of the gain
if figs.get("cta_goodwill_line"):
out.append(line(accounts["gain"], investor, company, figs["cta_goodwill_line"])) # v3: and so is what goodwill's retranslation put there
if investment_balance:
out.append(line(accounts["investment"], investor, company, -investment_balance))
for r in sorted(rows, key=lambda r: (r["group_account_id"], r["partner_company"], r["movement_type"], r["segment"])):
isl = r["group_account_id"] in pl
if at_period and not isl:
out.append(line(r["group_account_id"], company, r["partner_company"], -r["ytd_group"], r["movement_type"]) | {"segment": r["segment"]})
elif not at_period and isl:
out.append(line(r["group_account_id"], company, r["partner_company"], r["ytd_group"], r["movement_type"]) | {"segment": r["segment"]})
if at_period:
out.append(line(accounts["cta"], company, "", -figs["cta_line"]))
out.append(line(accounts["nci_equity"], company, "", -figs["nci_translation_line"]))
out = [ln for ln in out if ln["amount"]]
plug = -sum((ln["amount"] for ln in out), Decimal(0))
if plug:
out.append(line(_reserves(accounts), company, "", plug))
return out
def ownership_change_figures(*, price_group: Decimal, net_assets_group: Decimal, delta: Decimal) -> dict:
"""A further stake bought without a change of control (IFRS 10.23): the minority given up is the stake bought x the net assets; the
difference between the price and that is an equity transaction among owners, never profit and never goodwill."""
given = q2(net_assets_group * delta / HUNDRED)
return {"price_group": price_group, "net_assets_group": q2(net_assets_group), "nci_given_group": given, "equity_group": q2(price_group - given)}
def _reserves(accounts: dict) -> str:
if not accounts.get("retained_earnings"):
raise RuleBroken("the group has no account for the role group_retained_earnings (set it in the account roles)")
return accounts["retained_earnings"]
def ownership_change_lines(figs: dict, *, company: str, investor: str, accounts: dict) -> list:
"""Cr the investment for the extra price, Dr the NCI given up, Dr (or Cr) group equity for the difference."""
out = [line(accounts["investment"], investor, company, -figs["price_group"]), line(accounts["nci_equity"], company, "", figs["nci_given_group"]),
line(_reserves(accounts) if figs["equity_group"] else "-", company, "", figs["equity_group"])]
return [ln for ln in out if ln["amount"]]
# ── manual entries (CON-10) ───────────────────────────────────────────────────────────────────────────────────────
MOVEMENTS = ("opening", "result", "dividend", "capital_change", "translation", "ownership_change")
def entry_lines(raw: list, scope: set, rules: list, period_id: str) -> list:
"""Validate the lines of a manual entry and return them as lines: at least two, each naming a company in scope (and a partner in scope
where it names one, required on an intercompany account), a movement type on an equity account, NUMERIC(18,2), and netting to zero
in group currency exactly."""
if not isinstance(raw, list) or len(raw) < 2:
raise Invalid("an entry has at least two lines")
out = []
for n, r in enumerate(raw, 1):
if not isinstance(r, dict) or not r.get("group_account_id"):
raise Invalid(f"line {n} names no group account")
company, partner, movement = str(r.get("company_code") or ""), str(r.get("partner_company") or ""), str(r.get("movement_type") or "")
if company not in scope:
raise RuleBroken(f"line {n}: {company or 'no company'} is not in the group's scope in {period_id}")
if partner and partner not in scope:
raise RuleBroken(f"line {n}: the partner {partner} is not in the group's scope in {period_id}")
rule = rule_at(rules, str(r["group_account_id"]), period_id)
if rule is None:
raise RuleBroken(f"line {n}: the group account {r['group_account_id']} has no account rule")
if rule.ic == "true" and not partner:
raise RuleBroken(f"line {n}: {r['group_account_id']} is an intercompany account — name the partner company")
if rule.equity == "true" and not movement:
raise RuleBroken(f"line {n}: {r['group_account_id']} tracks movements — name the movement type ({', '.join(MOVEMENTS)})")
if movement and movement not in MOVEMENTS:
raise Invalid(f"line {n}: movement type {movement} is not one of {', '.join(MOVEMENTS)}")
value = amount(r.get("amount"), f"line {n} amount")
if value == 0:
raise Invalid(f"line {n} is zero")
out.append(line(str(r["group_account_id"]), company, partner, value, movement) | {"segment": str(r.get("segment") or "")})
total = sum((ln["amount"] for ln in out), Decimal(0))
if total != 0:
raise RuleBroken(f"the entry is not balanced in group currency: debits less credits are {total}")
return out
def second_person(a_email: str, a_sub: str, b_email: str, b_sub: str) -> bool:
"""True when the two are different people: no identity of one (email or sub) is an identity of the other."""
mine, theirs = {x for x in (a_email, a_sub) if x}, {x for x in (b_email, b_sub) if x}
return not (mine & theirs)
# ── digests and statements (CON-11, CON-13, CON-14) ───────────────────────────────────────────────────────────────
def tb_rows(tb: dict) -> list:
return [[*k, str(v)] for k, v in sorted(tb.items()) if v]
def tb_digest(tb: dict) -> str:
import hashlib
import json
return hashlib.sha256(json.dumps(tb_rows(tb), separators=(",", ":")).encode()).hexdigest()
def node_for(version: dict, code: str, basis_total: dict) -> Optional[str]:
"""The node an account lands on: the assignment whose range holds its code, decided by the sign of its balance (or movement) when
the assignment is debit- or credit-dependent (a credit bank balance shows under overdraft)."""
for a in sorted(version["assignments"], key=lambda x: (x["from_code"], x["to_code"])):
if not (a["from_code"] <= code <= a["to_code"]):
continue
total = basis_total[a.get("basis") or "balance"]
rule, node, other = a.get("dc_rule") or "always", a["node_id"], a.get("other_node_id") or None
if rule == "always":
return node
if rule == "debit_balance":
if total >= 0:
return node
elif total < 0:
return node
if other:
return other
return None
def statement_lines(version: dict, code_of: dict, closing: list, opening: list, *, by: str = "") -> dict:
"""The lines of a statement version over the consolidated figures.
closing / opening: rows (company, account, partner, movement, segment, amount) of the consolidated trial balance now and at the
opening of the year (movement basis reads the difference). by: "" (the lines), "company" or "segment" (each line split).
Returns {node_id: {"amount", "split": {key: amount}}, ...} for every node, parents summing their children (each node's own rows
carry its sign), and the accounts no assignment reaches."""
by_account: dict = {}
for (c, a, p, m, s, v) in closing:
by_account.setdefault(a, {"closing": ZERO, "opening": ZERO})["closing"] += v
for (c, a, p, m, s, v) in opening:
by_account.setdefault(a, {"closing": ZERO, "opening": ZERO})["opening"] += v
node_of, unassigned = {}, []
for a, t in sorted(by_account.items()):
node = node_for(version, code_of.get(a, a), {"balance": t["closing"], "movement": t["closing"] - t["opening"]})
if node is None:
if t["closing"] or t["opening"]:
unassigned.append({"group_account_id": a, "amount": t["closing"]})
else:
node_of[a] = node
basis = {x["node_id"]: x.get("basis") or "balance" for x in version["assignments"]}
own: dict = {n["node_id"]: {"amount": ZERO, "split": {}} for n in version["nodes"]}
for sign_, rows in ((1, closing), (-1, opening)):
for (c, a, p, m, s, v) in rows:
node = node_of.get(a)
if node is None or node not in own:
continue
if sign_ < 0 and basis.get(node) != "movement":
continue
own[node]["amount"] += sign_ * v
key = c if by == "company" else s if by == "segment" else None
if key is not None:
own[node]["split"][key] = own[node]["split"].get(key, ZERO) + sign_ * v
nodes = {n["node_id"]: n for n in version["nodes"]}
sign = {n: Decimal(str(nodes[n].get("sign") or 1)) for n in nodes}
def shown(x, n):
v = q2(x * sign[n])
return ZERO if v == 0 else v # a zero is 0.00, never -0.00
out = {n: {"amount": shown(own[n]["amount"], n), "split": {k: shown(x, n) for k, x in own[n]["split"].items()}} for n in own}
children: dict = {}
for n in version["nodes"]:
if n.get("parent_node_id") in nodes:
children.setdefault(n["parent_node_id"], []).append(n["node_id"])
def total(n):
for ch in children.get(n, []):
t = total(ch)
out[n]["amount"] += t["amount"]
for k, x in t["split"].items():
out[n]["split"][k] = out[n]["split"].get(k, ZERO) + x
return out[n]
for n in [n["node_id"] for n in version["nodes"] if n.get("parent_node_id") not in nodes]:
total(n)
return {"nodes": out, "unassigned": unassigned, "node_of": node_of}