dodil/erp-brick-consolidation

DODIL ERP brick: consolidation

src/brick_consolidation/application/ownership.py

"""CON-16 / CON-17: a sale of a subsidiary, a further stake bought without a change of control, and taking control of an associate.

Proposing computes the figures from gl's rate on the day and from the group's own translated data of the period, and stores them as shown
(structure_change.figures_json); approving by a second person writes the dated structure row. The consolidated books get the effect through
the eliminations of each period (runs.eliminate), recomputed from the stored inputs, keyed per task, company and run like every other entry.
"""
import json
from datetime import date
from decimal import Decimal

from brick_consolidation.domain import consolidation as C
from brick_consolidation.domain.errors import Invalid, RuleBroken
from brick_consolidation.infrastructure.seams import ports

ACQUIRING = ("acquisition", "ownership_change")


def day_of(value) -> date:
    return value if isinstance(value, date) else date.fromisoformat(str(value))


def acquisition_basis(changes: list, company: str):
    """The change that fixed the company's goodwill and equity at acquisition: the latest effective acquisition, or an ownership change that
    took control (it carries a goodwill)."""
    mine = [c for c in changes if c["company_code"] == company and c["status"] == "effective" and c["kind"] in ACQUIRING
            and (c["kind"] == "acquisition" or c.get("goodwill_group") is not None)]
    return max(mine, key=lambda c: (str(c["effective_date"]), c["change_id"]), default=None)


def goodwill_carried(group, acq: dict, period_id: str) -> Decimal:
    """The goodwill in group currency that is still carried at the period: after the amortisation a local GAAP requires (none under IFRS)."""
    gw = acq["goodwill_group"] or C.ZERO
    if group.goodwill_life_years and gw:
        gw -= C.goodwill_amortisation(gw, group.goodwill_life_years, C.months_between(C.period_of(day_of(acq["effective_date"])), period_id))
    return gw


def goodwill_at_closing(group, acq: dict, period_id: str, closing: Decimal) -> Decimal:
    """v3 (CON-16): the goodwill's CLOSING carrying amount: held in the acquired company's currency since acquisition (IAS 21.47), less the
    amortisation a local GAAP requires, retranslated at the closing rate on the disposal date - never the acquisition rate."""
    local = Decimal(acq.get("goodwill") or 0)
    if group.goodwill_life_years and local:
        local -= C.goodwill_amortisation(local, group.goodwill_life_years, C.months_between(C.period_of(day_of(acq["effective_date"])), period_id))
    return C.q2(local * closing)


def step_up_net(acq, rate=None) -> Decimal:
    """The fair-value step-up still carried, net of its deferred tax, in group currency at `rate` (default: the acquisition rate)."""
    fvs = json.loads(acq.get("fair_value_adjustments_json") or "[]") if acq else []
    return C.step_up_lines(fvs, acq["rate"] if rate is None else rate, "-", "-", C.ZERO)["net"] if fvs else C.ZERO


def _sets(group) -> tuple:
    return ({r.group_account_id for r in group.account_rules if r.rate_type == "average"},
            {r.group_account_id for r in group.account_rules if r.equity == "true"})


# ═══ CON-16 divestiture ═════════════════════════════════════════════════════════════════════════════════════════════
def divestiture_figures(group, change: dict, acq, rows: list, members: dict, period_id: str) -> dict:
    """The deconsolidation figures of a stored divestiture at its period, from the company's translated rows now."""
    pl, eq = _sets(group)
    company = change["company_code"]
    closing = Decimal(change["rate"])                       # v3: every term at gl's closing rate on the disposal date
    gw_close = goodwill_at_closing(group, acq, period_id, closing) if acq else C.ZERO
    su_close = step_up_net(acq, closing) if acq else C.ZERO
    # what the books already hold in the translation reserve for the goodwill and the step-up since acquisition: recycled with it (IAS 21.48)
    retranslated = (gw_close - (goodwill_carried(group, acq, period_id) if acq else C.ZERO)) + (su_close - (step_up_net(acq) if acq else C.ZERO))
    return C.disposal_figures(rows=[r for r in rows if r["company_code"] == company], pl=pl, equity=eq,
                              proceeds_group=C.q2(Decimal(change["proceeds"]) * closing), share=C.effective_ownership(company, members),
                              goodwill_group=gw_close, step_up_net=su_close, retranslation=retranslated)


def propose_divestiture(repo, group, it, members: dict, changes: list, proceeds) -> None:
    from brick_consolidation.application.runs import gl_call
    day = it.effective_date
    if day != C.period_end(C.period_of(day)):
        raise RuleBroken(f"a divestiture dates from the last day of a period ({C.period_end(C.period_of(day))}): the company's profit and loss is "
                         f"consolidated to the end of {C.period_of(day)} and its balance sheet leaves the group there, not part-way through a month")
    if proceeds in (None, ""):
        raise Invalid("a divestiture needs the proceeds")
    price = C.amount(proceeds, "proceeds")
    company, period = it.company_code, C.period_of(day)
    member = members.get(company)
    if member is None or member.method != "full" or member.investor != it.investor_company_code:
        raise RuleBroken(f"{company} is not a fully consolidated subsidiary of {it.investor_company_code} in {period}")
    if member.ownership != it.ownership_pct:
        raise RuleBroken(f"{it.investor_company_code} holds {member.ownership}% of {company}: a divestiture sells all of the holding "
                         "(a partial sale is an ownership change)")
    gp = next((x for x in group.group_periods if x.period_id == period), None)
    if gp is None or gp.status not in ("translated", "eliminated"):
        raise RuleBroken(f"{period} must be open and translated before the sale is recorded: its translated data are the net assets the gain is "
                         "computed from (collect and translate it)")
    spot = C.rate(gl_call(ports.gl.fx_rate, it.currency, day, "closing", group.group_currency), "disposal rate")        # v3: the closing rate on the date
    it.proceeds, it.rate = price, spot
    rows = repo.data.translated(group.group_id, period, gp.run_no)
    acq = acquisition_basis(changes, company)
    figs = divestiture_figures(group, {"company_code": company, "proceeds": price, "rate": spot}, acq, rows, members, period)
    it.gain_loss, it.nci_group, it.goodwill_group = figs["total_gain"], figs["nci_derecognised"], figs["goodwill_group"]
    it.figures_json = json.dumps({"period": period, "run_no": gp.run_no, "rate": str(spot), **{k: str(v) for k, v in figs.items()}}, sort_keys=True)
    it.ownership_pct = member.ownership


def deconsolidation_entry(group, change: dict, acq, rows_p: list, members_p: dict, accounts: dict, *, at_period: bool, period_id: str,
                          investment_balance: Decimal, own_gain: Decimal = C.ZERO) -> list:
    """The lines of the deconsolidation entry of one stored divestiture (see consolidation.disposal_lines). `rows_p` are the company's
    translated rows of the disposal period; `members_p` the structure in force there."""
    figs = divestiture_figures(group, change, acq, rows_p, members_p, C.period_of(day_of(change["effective_date"])))
    company = change["company_code"]
    return C.disposal_lines(figs, rows=[r for r in rows_p if r["company_code"] == company], pl=_sets(group)[0], company=company,
                            investor=change["investor_company_code"], investment_balance=investment_balance, accounts=accounts, at_period=at_period, own_gain=own_gain)


# ═══ CON-17 ownership change ════════════════════════════════════════════════════════════════════════════════════════
def _delta(it, members: dict):
    member = members.get(it.company_code)
    if member is None:
        raise RuleBroken(f"{it.company_code} is not in the group in {C.period_of(it.effective_date)}: an ownership change needs the structure it changes")
    if member.investor != it.investor_company_code:
        raise RuleBroken(f"{it.company_code} is held through {member.investor or 'nobody'}, not {it.investor_company_code}")
    return member, it.ownership_pct - member.ownership


def propose_ownership_change(repo, group, it, members: dict, changes: list, consideration, previously_held_fair_value, fair_value_adjustments,
                             acquire_figures) -> None:
    """80% -> 90% without a change of control, or an associate (equity) becoming a subsidiary (step acquisition)."""
    from brick_consolidation.application.runs import gl_call, prior_period
    member, delta = _delta(it, members)
    if consideration in (None, ""):
        raise Invalid("an ownership change needs the price of the stake")
    price = C.amount(consideration, "consideration")
    it.consideration = price
    period = C.period_of(it.effective_date)
    if member.method == "equity" and it.method == "full":                     # IFRS 3.42: control obtained in stages
        if delta <= 0:
            raise RuleBroken("taking control needs a larger holding than the associate's")
        if previously_held_fair_value in (None, ""):
            raise Invalid("taking control of an associate needs the fair value of the interest held before")
        held_fv = C.amount(previously_held_fair_value, "previously held fair value")
        it.previously_held_fair_value = held_fv
        spot = acquire_figures(it, price + held_fv, fair_value_adjustments)           # goodwill, NCI and the equity fixed at the date
        before = next((x for x in group.group_periods if x.period_id == prior_period(period)), None)
        if before is None:
            raise RuleBroken(f"{prior_period(period)} is not a group period: the carrying amount of the interest held before is read from it")
        from brick_consolidation.application import reports
        tb, _entries = reports.consolidated(repo, group, before, C.members_at(group, before.period_id))
        inv = _role(group, "investment", prior_period(period))
        carrying = sum((v for (c, a, p, _m, _s), v in tb.items() if c == it.investor_company_code and a == inv and p == it.company_code), Decimal(0))
        it.gain_loss = C.q2(held_fv * spot - carrying)                              # the gain in profit (IFRS 3.42), group currency
        it.figures_json = json.dumps({"kind": "step_acquisition", "carrying_group": str(carrying), "fair_value_group": str(C.q2(held_fv * spot)),
                                      "gain": str(it.gain_loss), "rate": str(spot)}, sort_keys=True)
        return
    if member.method != "full" or it.method != "full":
        raise RuleBroken("an ownership change is recorded for a subsidiary that stays fully consolidated, or an associate that becomes one; "
                         "a change from or to the equity method that is not taking control is a method change")
    if delta <= 0:
        raise RuleBroken("a smaller holding is a sale of part of the holding: not built — record a divestiture of the whole holding, or the sale in gl")
    if not any(c["company_code"] == it.company_code and c["kind"] in ACQUIRING and c["status"] == "effective" for c in changes):
        raise RuleBroken(f"{it.company_code} cannot change ownership without an approved acquisition (CON-02)")
    before = prior_period(period)
    rules = [r for r in group.account_rules if (r.equity == "true" or r.rate_type == "average") and r.valid_from_period <= before]
    accounts = sorted({r.group_account_id for r in rules})
    data = gl_call(ports.gl.balances, it.company_code, group.ledger_id, before)
    net_local = -sum((C.amount(r["ytd_amount"]) for r in data.get("rows", []) if str(r["group_account_id"]) in accounts), Decimal(0))
    spot = C.rate(gl_call(ports.gl.fx_rate, it.currency, it.effective_date, "spot", group.group_currency), "ownership change rate")
    figs = C.ownership_change_figures(price_group=C.q2(price * spot), net_assets_group=C.q2(net_local * spot), delta=delta)
    it.rate = spot
    it.net_assets = net_local
    it.gain_loss = None                                                                  # an equity transaction is never profit
    it.figures_json = json.dumps({"kind": "equity_transaction", "delta": str(delta), "net_assets_local": str(net_local), "digest": data.get("digest", ""),
                                  **{k: str(v) for k, v in figs.items()}}, sort_keys=True)


def _role(group, role: str, period_id: str):
    from brick_consolidation.application.runs import role_account
    return role_account(group, role, period_id)


def retained_earnings_account(group, period_id: str):
    from brick_consolidation.application.runs import role_account
    return role_account(group, "group_retained_earnings", period_id, need=False) or role_account(group, "retained_earnings", period_id, need=False)


def equity_transaction_entries(group, changes: list, members: dict, period_id: str, accounts: dict) -> tuple:
    """(entries' lines per change, the extra price already eliminated per company): every effective ownership change without a change of
    control up to this period. Cumulative by nature: recomputed in every period from the stored figures, so it needs no carry-forward."""
    out, eliminated = [], {}
    for ch in changes:
        if ch["kind"] != "ownership_change" or ch["status"] != "effective" or ch["goodwill_group"] is not None:
            continue
        if C.period_of(day_of(ch["effective_date"])) > period_id or ch["company_code"] not in members:
            continue
        figs = json.loads(ch["figures_json"] or "{}")
        if figs.get("kind") != "equity_transaction":
            continue
        figs = {k: Decimal(v) for k, v in figs.items() if k in ("price_group", "nci_given_group", "equity_group")}
        out.append((ch, C.ownership_change_lines(figs, company=ch["company_code"], investor=ch["investor_company_code"], accounts=accounts)))
        eliminated[ch["company_code"]] = eliminated.get(ch["company_code"], Decimal(0)) + figs["price_group"]
    return out, eliminated