dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
src/brick_consolidation/application/structure.py
"""CON-01 / CON-02: record who owns whom, and fix an acquisition's goodwill, minority and equity at its date.
Proposing computes everything from gl (the company exists, the equity before the date, the rate on the day) and shows it; approving
by a second person writes the dated structure row and fixes the equity at its historical rate. Nothing is typed in: the figures that
make the goodwill come from gl's route or are refused.
"""
import json
import re
from datetime import timedelta
from decimal import Decimal
from brick_consolidation.application import ownership
from erp_platform.bricks import Refused
from brick_consolidation.application.runs import gl_call, prior_period, settings
from brick_consolidation.domain import consolidation as C
from brick_consolidation.domain.errors import Invalid, RuleBroken
from brick_consolidation.domain.group import GroupStructure
from brick_consolidation.domain.structure_change import StructureChange
from brick_consolidation.infrastructure.seams import Missing, ports
KINDS = ("acquisition", "ownership_change", "divestiture", "method_change")
METHODS = ("full", "equity", "none")
def change_id_of(group_id: str, company: str, effective_period: str, kind: str) -> str:
return f"{group_id}:{company}:{effective_period}:{kind}"
def _no_signed_from(group, period_id: str) -> None:
signed = sorted(gp.period_id for gp in group.group_periods if gp.status == "signed" and gp.period_id >= period_id)
if signed:
raise RuleBroken(f"{signed[0]} is signed: a change takes effect from a period not yet signed, never back into a signed one "
f"(it dates from {period_id})")
def _statutory_pct(company: str, on):
"""gl's statutory_tax_rate of the company in force on `on`, as a percent; (None, why) when gl has none or cannot be asked."""
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 with_tax_rates(company: str, effective_date, fair_value_adjustments) -> list:
"""v3, one rule for every adjustment that carries deferred tax: the adjustment's own tax_rate_pct wins; absent, gl's statutory_tax_rate of the
company that holds the temporary difference (the acquired company) in force on the effective date; absent both, the change is refused (gate
tax_rate_known), never computed at zero. The rate used is written onto each adjustment, so the stored figures are reproducible."""
out = []
for fv in fair_value_adjustments or []:
if fv.get("tax_rate_pct") in (None, ""):
rate, why = _statutory_pct(company, effective_date)
if rate is None:
raise RuleBroken(f"tax_rate_known: a fair-value adjustment has no tax rate, and {why} — enter the rate on the adjustment, or set statutory_tax_rate in gl")
fv = {**fv, "tax_rate_pct": str(rate)}
out.append(fv)
return out
def _acquire(repo, group, it, price, fair_value_adjustments) -> Decimal:
"""Goodwill, minority and the equity fixed at the date (CON-02): the equity before the date from gl, the rate on the day. Sets the figures on
`it` and returns the rate."""
fair_value_adjustments = with_tax_rates(it.company_code, it.effective_date, fair_value_adjustments)
period = C.period_of(it.effective_date)
equity_period = prior_period(period)
rules = [r for r in group.account_rules if r.equity == "true" and r.valid_from_period <= equity_period]
accounts = sorted({r.group_account_id for r in rules})
if not accounts:
raise RuleBroken("no equity accounts are flagged in the account rules: the net assets at acquisition cannot be found")
data = gl_call(ports.gl.balances, it.company_code, group.ledger_id, equity_period)
equity = {}
for r in data.get("rows", []):
if str(r["group_account_id"]) in accounts:
equity[str(r["group_account_id"])] = equity.get(str(r["group_account_id"]), Decimal(0)) + C.amount(r["ytd_amount"])
net_assets = -sum(equity.values(), Decimal(0)) # equity is a credit balance in gl's sign
today_rate = C.rate(gl_call(ports.gl.fx_rate, it.currency, it.effective_date, "spot", group.group_currency), "acquisition rate")
fig = C.acquisition_figures(price, it.ownership_pct, net_assets, fair_value_adjustments or [], today_rate)
it.net_assets_json = json.dumps({"period": equity_period, "currency": it.currency, "equity": {a: str(v) for a, v in equity.items()},
"digest": data.get("digest", "")}, sort_keys=True)
it.fair_value_adjustments_json = json.dumps(fair_value_adjustments or [], sort_keys=True)
it.net_assets, it.goodwill, it.nci = fig["net_assets"], fig["goodwill"], fig["nci"]
it.goodwill_group, it.nci_group, it.rate = fig["goodwill_group"], fig["nci_group"], today_rate
it.bargain_purchase = "true" if fig["bargain_purchase"] else "false"
return today_rate
def next_period(period_id: str) -> str:
return C.period_of(C.period_end(period_id) + timedelta(days=1))
def propose(repo, group, *, kind, company_code, investor_company_code, ownership_pct, voting_pct, method, effective_date, consideration,
fair_value_adjustments, by, sub, proceeds=None, previously_held_fair_value=None, source="gl", local_currency=None, name=None) -> StructureChange:
if kind not in KINDS:
raise Invalid(f"kind must be one of {', '.join(KINDS)}")
if kind == "divestiture":
method = "none" # the structure row of a sold company is method none
if method not in METHODS:
raise Invalid(f"method must be one of {', '.join(METHODS)}")
own = C.pct(ownership_pct, "ownership")
vote = C.pct(voting_pct if voting_pct not in (None, "") else ownership_pct, "voting")
period = C.period_of(effective_date)
_no_signed_from(group, period)
if source not in ("gl", "upload"):
raise Invalid("source is gl or upload")
known = {c["company_code"]: c for c in gl_call(ports.gl.companies)}
if investor_company_code not in known:
raise RuleBroken(f"the investor {investor_company_code} is not a company in gl (GET /companies) — a company exists only in gl")
if source == "gl" and company_code not in known:
raise Refused(409, f"source_matches_gl: source gl names {company_code}, which gl does not know (GET /companies)", gate="source_matches_gl",
fix="source upload for a company kept outside gl (CON-12), source gl for one of gl's companies")
if source == "upload": # CON-12: the one way a company gl does not know enters the structure
if company_code in known:
raise Refused(409, f"source_matches_gl: source upload names {company_code}, which gl knows", gate="source_matches_gl",
fix="source gl for one of gl's companies (a company is kept in one place)")
if not re.match(r"^[A-Z0-9]{1,4}$", company_code or ""):
raise Invalid("an upload company's code is 1 to 4 upper-case letters or digits")
if any(r.company_code == company_code for r in group.group_structures) or repo.data.changes(group.group_id, [company_code]):
raise RuleBroken(f"{company_code} is already in the group: a company code is unique in the group")
if not (len(local_currency or "") == 3 and local_currency.isalpha() and local_currency.isupper()):
raise Invalid("a company kept outside gl names its local_currency (ISO 4217)")
if not (name or "").strip():
raise Invalid("a company kept outside gl names its name")
known = {**known, company_code: {"company_code": company_code, "functional_currency": local_currency, "name": name}}
members = C.members_at(group, period)
if investor_company_code not in members:
raise RuleBroken(f"the investor {investor_company_code} is not in the group in {period}")
if company_code == group.top_company_code or company_code == investor_company_code:
raise RuleBroken("the top company is not acquired, and a company does not own itself")
change_id = change_id_of(group.group_id, company_code, period, kind)
it = StructureChange(change_id=change_id, kind=kind, effective_date=effective_date, company_code=company_code,
investor_company_code=investor_company_code, ownership_pct=own, voting_pct=vote, method=method,
currency=known[company_code]["functional_currency"], proposed_by=by, proposed_sub=sub, source=source, name=(name or "").strip())
if kind == "acquisition":
if consideration in (None, ""):
raise Invalid("an acquisition needs the consideration paid")
price = C.amount(consideration, "consideration")
it.consideration = price
_acquire(repo, group, it, price, fair_value_adjustments)
elif kind == "divestiture":
ownership.propose_divestiture(repo, group, it, members, repo.data.changes(group.group_id, [company_code]), proceeds)
elif kind == "ownership_change":
ownership.propose_ownership_change(repo, group, it, members, repo.data.changes(group.group_id, [company_code]), consideration,
previously_held_fair_value, fair_value_adjustments,
lambda ch, price, fvs: _acquire(repo, group, ch, price, fvs))
elif method == "full":
have = [c for c in repo.data.changes(group.group_id, [company_code]) if c["kind"] == "acquisition" and c["status"] == "effective"]
if not have:
raise RuleBroken(f"{company_code} cannot be consolidated in full without an approved acquisition (CON-02)")
return it
def _different_people(a_email: str, a_sub: str, b_email: str, b_sub: str) -> bool:
mine = {x for x in (a_email, a_sub) if x}
theirs = {x for x in (b_email, b_sub) if x}
return not (mine & theirs)
def approve(repo, group, change: StructureChange, *, by: str, sub: str) -> None:
if change.status != "proposed":
raise RuleBroken(f"a {change.status} structure change cannot be approved")
if settings(repo)["signoff_second_person"] and not _different_people(by, sub, change.proposed_by, change.proposed_sub):
raise RuleBroken("the approver cannot be the person who proposed the change (matched on email and sub)")
period = C.period_of(change.effective_date)
_no_signed_from(group, period)
from_period = next_period(period) if change.kind == "divestiture" else period # a sale leaves the structure from the period after its date
row = GroupStructure(group_id=group.group_id, company_code=change.company_code, valid_from_period=from_period,
investor_company_code=change.investor_company_code, ownership_pct=change.ownership_pct,
voting_pct=change.voting_pct, method=change.method, change_id=change.change_id, source=change.source or "gl",
local_currency=change.currency if change.source == "upload" else "", name=change.name or "")
group.group_structures = [r for r in group.group_structures
if not (r.company_code == change.company_code and r.valid_from_period == from_period)] + [row]
if change.net_assets_json: # an acquisition, or taking control in stages: the equity at the date, at that day's rate
eq = json.loads(change.net_assets_json)["equity"]
repo.data.write_historical([{"group_id": group.group_id, "company_code": change.company_code, "group_account_id": a,
"arose_period": period, "amount_local": Decimal(v), "amount_group": C.q2(Decimal(v) * change.rate),
"rate": change.rate, "change_id": change.change_id} for a, v in sorted(eq.items())])
# the periods already collected from here on were collected under another structure: collect again
for gp in group.group_periods:
if gp.period_id >= period and gp.status in ("collected", "translated", "eliminated"):
gp.stale_from_step = "collect"
change.approved_by, change.approved_sub, change.status = by, sub, "effective"