dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/domain/group.py
"""The Group aggregate — the rules, and nothing else.
a set of companies reported together, in one currency, from one gl ledger
No database, no HTTP, no identity: a rule here is true whoever calls it and however it is stored.
Generated by `erp-compose scaffold` from module.yaml; the TODO lines are the work left.
"""
import json
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from typing import Optional
from brick_consolidation.domain.errors import RuleBroken
from brick_consolidation.domain.events import (
GroupPeriodCollected,
GroupPeriodCollectedFromSigned,
GroupPeriodEliminated,
GroupPeriodOpen,
GroupPeriodSigned,
GroupPeriodTranslated,
GroupPeriodUpdated,
GroupPeriodUpdatedFromCollected,
GroupPeriodUpdatedFromOpen,
)
from brick_consolidation.domain.values import (
AccountRuleRateType,
EliminationRuleKind,
GroupPeriodState,
GroupStructureMethod,
ReportedDataSource,
UploadStatus,
)
@dataclass
class GroupStructure:
"""who owns how much of a company and how it is consolidated, dated by period"""
group_id: str
company_code: str
valid_from_period: str
investor_company_code: str = ""
ownership_pct: Decimal = Decimal("0")
voting_pct: Decimal = Decimal("0")
method: str = GroupStructureMethod.FULL.value
change_id: str = ""
source: str = "gl" # v3: gl | upload (a company kept outside gl, CON-12)
local_currency: str = ""
name: str = ""
@dataclass
class AccountRule:
"""how a group account is translated and whether it takes part in an elimination"""
group_id: str
group_account_id: str
valid_from_period: str
rate_type: str = AccountRuleRateType.CLOSING.value
ic: str = ""
equity: str = ""
@dataclass
class AccountRole:
"""which group account receives each kind of consolidation result (translation reserve, NCI, goodwill,
differences)
"""
group_id: str
role: str
valid_from_period: str
group_account_id: str = ""
@dataclass
class EliminationRule:
"""a pair of group-account sets eliminated against each other per pair of companies"""
group_id: str
rule_id: str
kind: str = EliminationRuleKind.IC_BALANCE.value
left_accounts: str = ""
right_accounts: str = ""
difference_role: str = ""
tolerance: Decimal = Decimal("0.00")
@dataclass
class GroupPeriod:
"""one period of the group close and where it stands"""
group_id: str
period_id: str
run_no: int = 0
stale_from_step: str = ""
prepared_by: str = ""
prepared_sub: str = ""
signed_by: str = ""
signed_sub: str = ""
signed_at: Optional[datetime] = None
tb_digest: str = ""
reopened_reason: str = ""
status: str = GroupPeriodState.OPEN.value
preparers: str = "" # JSON list, appended: every step runner of every run {step, run_no, email, sub} (CON-14)
def record_preparer(self, step: str, email: str, sub: str) -> None:
held = json.loads(self.preparers or "[]")
item = {"step": step, "run_no": self.run_no, "email": email or "", "sub": sub or ""}
if item not in held:
held.append(item)
self.preparers = json.dumps(held, sort_keys=True)
def preparer_identities(self) -> set:
"""Every email and sub that ran a step of the period, in any run (and the latest preparer, for rows from before the list)."""
out = {self.prepared_by, self.prepared_sub}
for p in json.loads(self.preparers or "[]"):
out |= {p.get("email"), p.get("sub")}
return {x for x in out if x}
@dataclass
class ReportedData:
"""a company's figures for a period by group account, as gl (or an upload) gave them"""
group_id: str
period_id: str
company_code: str
group_account_id: str
partner_company: str
movement_type: str
segment: str
ytd_local: str = ""
period_local: str = ""
local_currency: str = ""
source: str = ReportedDataSource.GL.value
gl_digest: str = ""
period_status: str = ""
collected_at: Optional[datetime] = None
@dataclass
class Upload:
"""a file of reported data from a company kept elsewhere, and its acceptance"""
group_id: str
period_id: str
company_code: str
upload_no: int
object_key: str = ""
sha256: str = ""
rows: str = ""
status: str = UploadStatus.SUBMITTED.value
submitted_by: str = ""
submitted_sub: str = ""
accepted_by: str = ""
accepted_sub: str = ""
@dataclass
class TranslatedData:
"""a company's figures in the group currency, with the rate that made each"""
group_id: str
period_id: str
run_no: int
company_code: str
group_account_id: str
partner_company: str
movement_type: str
segment: str
ytd_group: str = ""
rate_type: str = ""
rate: str = ""
@dataclass
class IcInventory:
"""stock a company holds that it bought from another group company, and the seller's margin"""
group_id: str
period_id: str
holder_company: str
seller_company: str
inventory_amount: Decimal = Decimal("0.00")
holder: str = ""
margin_pct: Decimal = Decimal("0.0000")
none_stated: str = ""
entered_by: str = ""
entered_sub: str = ""
@dataclass
class HistoricalAmount:
"""an equity account's amount fixed in group currency at the rate of the day it arose"""
group_id: str
company_code: str
group_account_id: str
arose_period: str
amount_local: Decimal = Decimal("0.00")
amount_group: Decimal = Decimal("0.00")
rate: str = ""
change_id: str = ""
@dataclass
class NumberSeries:
"""the consolidation entry numbers of a group and year"""
group_id: str
fiscal_year: str
entry_no: int
cons_journal_id: str = ""
claimed_at: Optional[datetime] = None
@dataclass
class CarryForward:
"""the accumulated balance-sheet effect of consolidation entries at a year's start"""
group_id: str
fiscal_year: str
group_account_id: str
company_code: str
partner_company: str
segment: str
opening: str = ""
derived_at: Optional[datetime] = None
source_digest: str = ""
@dataclass
class Group:
"""a set of companies reported together, in one currency, from one gl ledger"""
group_id: str
name: str = ""
group_currency: str = ""
ledger_id: str = ""
group_chart_id: str = ""
top_company_code: str = ""
fiscal_year_end_month: str = ""
fsv_id: str = ""
goodwill_life_years: str = ""
cash_flow_fsv_id: str = ""
group_structures: list = field(default_factory=list)
account_rules: list = field(default_factory=list)
account_roles: list = field(default_factory=list)
elimination_rules: list = field(default_factory=list)
group_periods: list = field(default_factory=list)
events: list = field(default_factory=list, repr=False, compare=False)
def _record(self, event):
self.events.append(event)
return event
def _group_period(self, period_id: str) -> GroupPeriod:
for it in self.group_periods:
if it.period_id == period_id:
return it
raise RuleBroken(f"there is no group period {period_id}")
# ── the lifecycle (module.yaml workflows) ──────────────────────────────────────────────────────────────────
def mark_stale(self, from_period: str = "", *, to_period: str = "", step: str = "translate") -> list:
"""The periods from `from_period` (to `to_period`) computed under what has just changed are stale: those translated or eliminated are
stale from translate (readiness no_step_stale is not met; translate runs again under the next run number, then eliminate; until then the
reports show the last run's figures, flagged stale); with step collect they are collected again first.
A signed period is never touched: it stays signed until it is reopened. In memory, saved with the aggregate: one save, whatever the
number of periods. Returns the periods marked."""
marked = []
for gp in self.group_periods:
if gp.status == "signed" or gp.period_id < from_period or (to_period and gp.period_id > to_period):
continue
if step == "collect" and gp.status in ("collected", "translated", "eliminated"):
gp.stale_from_step = "collect"
elif gp.status in ("translated", "eliminated"):
gp.stale_from_step = gp.stale_from_step if gp.stale_from_step == "collect" else "translate"
else:
continue
marked.append(gp.period_id)
return marked
def open_group_period(self, period_id: str, *, by: str = "", **attrs) -> Optional[GroupPeriodOpen]:
"""Open the group period: none → open. Gate: the structure is approved for the period; the previous
period is signed (or this is the first). Effect: scope and methods fixed from the structure in force
on the period's last day; for a year's first period the group carry-forward derived (CON-15). Adding
the same one again changes nothing.
"""
if any(x.period_id == period_id for x in self.group_periods):
return None
it = GroupPeriod(group_id=self.group_id, period_id=period_id, **attrs)
# TODO(rule): the gate — the structure is approved for the period; the previous period is signed (or this is the first)
# TODO(rule): the effect — scope and methods fixed from the structure in force on the period's last day; for a year's first period the group carry-forward derived (CON-15)
it.status = GroupPeriodState.OPEN.value
self.group_periods.append(it)
return self._record(GroupPeriodOpen(group_id=it.group_id, period_id=it.period_id, by=by))
def upload_reported_data(self, period_id: str, *, by: str = "") -> Optional[GroupPeriodUpdated]:
"""Upload a company's figures: open → open. Gate: the company is in scope and not kept in gl; the file
validates. Effect: rows stored with source upload, waiting for acceptance (CON-12).
"""
it = self._group_period(period_id)
if it.status == GroupPeriodState.SIGNED:
raise RuleBroken(f"{period_id} is signed; an upload waits until it is reopened")
# the gate and the effect are close.upload (CON-12): a replacement after the collection is allowed, and marks later steps stale on acceptance
return self._record(GroupPeriodUpdated(group_id=it.group_id, period_id=it.period_id, by=by))
def accept_reported_upload(self, period_id: str, *, by: str = "") -> Optional[GroupPeriodUpdatedFromOpen]:
"""Accept the upload: open → open. Gate: the accepter is not the uploader (email AND sub). Effect: the
rows become the company's reported data.
"""
it = self._group_period(period_id)
if it.status == GroupPeriodState.SIGNED:
raise RuleBroken(f"{period_id} is signed; an upload is accepted after it is reopened")
# the gate and the effect are close.accept (CON-12)
return self._record(GroupPeriodUpdatedFromOpen(group_id=it.group_id, period_id=it.period_id, by=by))
def collect_group_data(self, period_id: str, *, by: str = "") -> Optional[GroupPeriodCollected]:
"""Collect the companies' figures: open → collected. Gate: every company in scope collected from gl or
by an accepted upload; each nets to zero; no unmapped accounts; the gl period closed (unless
allow_open_period_collect). Effect: reported data with gl's digest per company (CON-03).
"""
it = self._group_period(period_id)
if it.status == GroupPeriodState.COLLECTED:
return None # already collected: a repeat changes nothing
if it.status not in (GroupPeriodState.OPEN,):
raise RuleBroken(f"a {it.status} group period cannot move to collected")
# TODO(rule): the gate — every company in scope collected from gl or by an accepted upload; each nets to zero; no unmapped accounts; the gl period closed (unless allow_open_period_collect)
# TODO(rule): the effect — reported data with gl's digest per company (CON-03)
it.status = GroupPeriodState.COLLECTED.value
return self._record(GroupPeriodCollected(group_id=it.group_id, period_id=it.period_id, by=by))
def recollect_group_data(self, period_id: str, *, by: str = "") -> Optional[GroupPeriodUpdatedFromCollected]:
"""Collect again: collected, translated, eliminated → collected. Gate: a company's digest changed, an
upload was replaced, or the structure changed. Effect: later steps stale; the next runs take run_no +
1 (CON-20).
"""
it = self._group_period(period_id)
if it.status not in (GroupPeriodState.COLLECTED, GroupPeriodState.TRANSLATED, GroupPeriodState.ELIMINATED,):
raise RuleBroken(f"a {it.status} group period cannot move to collected")
# TODO(rule): the gate — a company's digest changed, an upload was replaced, or the structure changed
# TODO(rule): the effect — later steps stale; the next runs take run_no + 1 (CON-20)
return self._record(GroupPeriodUpdatedFromCollected(group_id=it.group_id, period_id=it.period_id, by=by))
def translate_currencies(self, period_id: str, *, by: str = "") -> Optional[GroupPeriodTranslated]:
"""Translate into the group currency: collected → translated. Gate: a closing rate on the period end and
an average rate for every month of the year so far, for every foreign company's currency. Effect:
translated data and one translation-reserve entry per company, with the NCI share (CON-04).
"""
it = self._group_period(period_id)
if it.status == GroupPeriodState.TRANSLATED:
return None # already translated: a repeat changes nothing
if it.status not in (GroupPeriodState.COLLECTED,):
raise RuleBroken(f"a {it.status} group period cannot move to translated")
# TODO(rule): the gate — a closing rate on the period end and an average rate for every month of the year so far, for every foreign company's currency
# TODO(rule): the effect — translated data and one translation-reserve entry per company, with the NCI share (CON-04)
it.status = GroupPeriodState.TRANSLATED.value
return self._record(GroupPeriodTranslated(group_id=it.group_id, period_id=it.period_id, by=by))
def run_eliminations(self, period_id: str, *, by: str = "") -> Optional[GroupPeriodEliminated]:
"""Run the eliminations: translated → eliminated. Gate: intercompany within tolerance or accepted per
pair; intercompany inventory entered or stated none per selling pair; an approved acquisition for
every full and equity company. Effect: automatic entries for intercompany balances, revenue and
expense, dividends, stock profit, investments, NCI and the equity method, keyed per task, pair and
run (CON-05 to CON-09).
"""
it = self._group_period(period_id)
if it.status == GroupPeriodState.ELIMINATED:
return None # already eliminated: a repeat changes nothing
if it.status not in (GroupPeriodState.TRANSLATED,):
raise RuleBroken(f"a {it.status} group period cannot move to eliminated")
# TODO(rule): the gate — intercompany within tolerance or accepted per pair; intercompany inventory entered or stated none per selling pair; an approved acquisition for every full and equity company
# TODO(rule): the effect — automatic entries for intercompany balances, revenue and expense, dividends, stock profit, investments, NCI and the equity method, keyed per task, pair and run (CON-05 to CON-09)
it.status = GroupPeriodState.ELIMINATED.value
return self._record(GroupPeriodEliminated(group_id=it.group_id, period_id=it.period_id, by=by))
def sign_off_group_close(self, period_id: str, *, by: str = "") -> Optional[GroupPeriodSigned]:
"""Sign off the group close: eliminated → signed. Gate: readiness met (CON-13); the signer is not the
preparer (email AND sub). Effect: who, when and the digest recorded; statements materialised
(CON-21).
"""
it = self._group_period(period_id)
if it.status == GroupPeriodState.SIGNED:
return None # already signed: a repeat changes nothing
if it.status not in (GroupPeriodState.ELIMINATED,):
raise RuleBroken(f"a {it.status} group period cannot move to signed")
# TODO(rule): the gate — readiness met (CON-13); the signer is not the preparer (email AND sub)
# TODO(rule): the effect — who, when and the digest recorded; statements materialised (CON-21)
it.status = GroupPeriodState.SIGNED.value
return self._record(GroupPeriodSigned(group_id=it.group_id, period_id=it.period_id, by=by))
def reopen_group_period(self, period_id: str, reason: str, *, by: str = "") -> Optional[GroupPeriodCollectedFromSigned]:
"""Reopen the group period: signed → collected. Gate: the reopener is not who signed it; the next period
is not signed. Effect: recorded; the figures provisional; the next runs take run_no + 1.
"""
it = self._group_period(period_id)
if it.status == GroupPeriodState.COLLECTED:
return None # already collected: a repeat changes nothing
if it.status not in (GroupPeriodState.SIGNED,):
raise RuleBroken(f"a {it.status} group period cannot move to collected")
# TODO(rule): the gate — the reopener is not who signed it; the next period is not signed
# TODO(rule): the effect — recorded; the figures provisional; the next runs take run_no + 1
if not reason:
raise RuleBroken("reason is required")
it.status = GroupPeriodState.COLLECTED.value
return self._record(GroupPeriodCollectedFromSigned(group_id=it.group_id, period_id=it.period_id, by=by))
# ── invariants: one stub per story; implement it, call it where it guards ───────────────────────────────