dodil/erp-brick-fixed-assets

DODIL ERP brick: fixed assets

src/brick_fixed_assets/domain/depreciation.py

"""The depreciation arithmetic (FA-10): planned depreciation per asset and area, derived from rows, never stored.

Everything is an exact Fraction until the one rounding, and that rounding is on the CUMULATIVE figure
(`money(cum)`): the charge of a period is cum(period) minus cum(previous period), so the periods of a life sum to
exactly cost minus scrap and a repeat run posts only the difference (a run posts planned-to-date minus posted).

A plan is a chain of SEGMENTS. A segment starts at a period with a depreciable amount B, a remaining life R (in
periods) and the cumulative planned so far (cum0). A later event closes the segment and opens the next one on what
is left:
  Add      more cost (a later acquisition, a settlement into the asset): B grows, R is what remains
  Reset    a new dated area row (a new life, key, scrap or total units): the remaining NBV over the NEW remaining
           life (IAS 8.36: prospective); posted periods are never recomputed
  SetNbv   a revaluation or an impairment: depreciation continues on the new carrying amount
  Retire   all or part leaves: the closing period is charged by the key's retirement convention
  Open     a takeover or a transfer in: values and remaining life carried in
The first segment of a plan is its first Add (or Open).

Conventions (key.acq_conv): full_month (from the month of acquisition), next_month, mid_month (after the 15th: from
next month), daily (days in use / days in the month in the first period), half_year (the first fiscal year gets half
a year's charge, spread over the periods from acquisition to year end).
"""
from dataclasses import dataclass, replace
from datetime import date
from decimal import Decimal
from fractions import Fraction
from typing import Iterable, Optional

from brick_fixed_assets.domain.errors import RuleBroken
from brick_fixed_assets.domain.values import (
    CONVENTIONS,
    METHODS,
    days_in,
    fiscal_year,
    frac,
    fy_first_period,
    fy_last_period,
    money,
    period_add,
    period_index,
    period_of,
)

ONE = Fraction(1)
ZERO = Fraction(0)


@dataclass(frozen=True)
class DepKey:
    key_code: str
    method: str = "straight_line"
    rate_pct: Optional[Decimal] = None
    life_periods: Optional[int] = None
    base: str = "net_book_value"            # cost | net_book_value (declining balance)
    acq_conv: str = "full_month"
    ret_conv: str = "full_month"
    stop_at: str = "scrap"                  # zero | scrap

    def __post_init__(self):
        if self.method not in METHODS:
            raise RuleBroken(f"method is one of {', '.join(METHODS)}")
        for c in (self.acq_conv, self.ret_conv):
            if c not in CONVENTIONS:
                raise RuleBroken(f"a convention is one of {', '.join(CONVENTIONS)}")
        if self.method in ("declining_balance", "declining_then_straight") and self.rate_pct is None:
            raise RuleBroken("a declining-balance key needs rate_pct")


@dataclass(frozen=True)
class Params:
    """How one area depreciates one asset from a date (an fa_asset_area row)."""
    key: DepKey
    life_periods: Optional[int] = None
    total_units: Optional[Decimal] = None
    scrap_value: Decimal = Decimal("0.00")

    @property
    def life(self) -> Optional[int]:
        return self.life_periods or self.key.life_periods

    @property
    def scrap_eff(self) -> Fraction:
        return frac(self.scrap_value) if self.key.stop_at == "scrap" else ZERO


# ── events ────────────────────────────────────────────────────────────────────────────────────

@dataclass(frozen=True)
class Add:
    on: date
    amount: Decimal
    order: int = 0


@dataclass(frozen=True)
class Open:
    on: date
    depreciable: Decimal
    remaining: int
    first_period: str
    order: int = 0


@dataclass(frozen=True)
class Reset:
    on: date
    params: Params
    order: int = 0


@dataclass(frozen=True)
class SetNbv:
    on: date
    nbv: Decimal
    order: int = 0


@dataclass(frozen=True)
class Retire:
    on: date
    fraction: Fraction = ONE
    order: int = 0


def start_of(conv: str, on: date) -> tuple:
    """(first period charged, weight of that period, half-year flag) for a date and an acquisition convention."""
    p = period_of(on)
    if conv == "full_month":
        return p, ONE, False
    if conv == "next_month":
        return period_add(p, 1), ONE, False
    if conv == "mid_month":
        return (p if on.day <= 15 else period_add(p, 1)), ONE, False
    if conv == "daily":
        return p, Fraction(days_in(p) - on.day + 1, days_in(p)), False
    if conv == "half_year":
        return p, ONE, True
    raise RuleBroken(f"unknown convention {conv}")


def retirement_weight(conv: str, on: date) -> Fraction:
    """How much of the retirement period is charged: full_month 1, next_month 0, mid_month 1 after the 15th else 0,
    daily the days up to the date. half_year depends on the asset's plan, so it is `Plan.retirement_weight`."""
    p = period_of(on)
    if conv == "full_month":
        return ONE
    if conv == "half_year":
        raise RuleBroken("half_year retirement is weighed by the plan: use Plan.retirement_weight")
    if conv == "next_month":
        return ZERO
    if conv == "mid_month":
        return ONE if on.day > 15 else ZERO
    if conv == "daily":
        return Fraction(on.day, days_in(p))
    raise RuleBroken(f"unknown convention {conv}")


@dataclass
class Seg:
    start: str
    first_w: Fraction
    half: bool
    B: Fraction                 # depreciable at the start (NBV minus scrap)
    R: Fraction                 # remaining life, in periods, at the start
    cum0: Fraction              # cumulative planned before this segment
    e0: Fraction                # periods already elapsed before this segment (whole asset)
    params: Params
    fy_end: int = 12
    units_base: Fraction = ZERO         # units used before the segment (units of production)
    end: Optional[tuple] = None         # (period, weight of that period) once a later event closes it

    # ── weights: how many periods of life each calendar period uses up ──
    def _fy_last_idx(self) -> int:
        return period_index(fy_last_period(fiscal_year(self.start, self.fy_end), self.fy_end))

    def W(self, p: str) -> Fraction:
        """Uncapped cumulative weight through period p (0 before the start)."""
        n = period_index(p) - period_index(self.start)
        if n < 0:
            return ZERO
        if self.half:
            last = self._fy_last_idx()
            m0 = last - period_index(self.start) + 1
            if period_index(p) <= last:
                return Fraction(6 * (n + 1), m0)
            return Fraction(6) + (period_index(p) - last)
        return self.first_w + n

    def w(self, p: str) -> Fraction:
        return self.W(p) - self.W(period_add(p, -1))


def _units_through(usage: dict, p: str) -> Fraction:
    return sum((frac(u) for q, u in usage.items() if q <= p), ZERO)


class Plan:
    """The planned depreciation of one asset in one area."""

    def __init__(self, first: Params, events: Iterable, usage: Optional[dict] = None, fy_end_month: int = 12):
        self.usage = dict(usage or {})
        self.fy = fy_end_month
        self.params = first
        self.segs: list = []
        for ev in sorted(events, key=lambda e: (e.on, e.order)):
            self._apply(ev)

    # ── building ──
    def _seg_done(self, seg: Seg, p: str) -> Fraction:
        return self._seg_cum(seg, p)

    def _close(self, prev: Seg, period: str, g: Fraction = ONE) -> None:
        prev.end = (period, g)

    def _next_state(self, prev: Seg, last_period: str, g: Fraction = ONE) -> tuple:
        """(cum through, B left, R left, elapsed) after `prev` is charged through last_period with weight g."""
        done = self._seg_cum(prev, last_period, g, force_end=(last_period, g))
        wl = self._W_eff(prev, last_period, g)
        return prev.cum0 + done, prev.B - done, max(prev.R - wl, ZERO), prev.e0 + min(wl, prev.R)

    def _units_before(self, start: str) -> Fraction:
        return _units_through(self.usage, period_add(start, -1))

    def _apply(self, ev) -> None:
        prev = self.segs[-1] if self.segs else None
        if isinstance(ev, Reset):
            self.params = ev.params
        if isinstance(ev, (Add, Open)) and (prev is None or isinstance(ev, Open) and False):
            if isinstance(ev, Open):
                start, fw, half = ev.first_period, ONE, False
                seg = Seg(start, fw, half, frac(ev.depreciable), frac(ev.remaining), ZERO, ZERO, self.params, self.fy,
                          self._units_before(start))
            else:
                start, fw, half = start_of(self.params.key.acq_conv, ev.on)
                life = self.params.life
                R = frac(life) if life else ZERO
                B = max(frac(ev.amount) - self.params.scrap_eff, ZERO)
                seg = Seg(start, fw, half, B, R, ZERO, ZERO, self.params, self.fy, self._units_before(start))
            self.segs.append(seg)
            return
        if prev is None:
            return                                   # a change before anything was capitalised plans nothing yet
        if isinstance(ev, (Add, Open)):
            amount = frac(ev.amount if isinstance(ev, Add) else ev.depreciable)
            start, fw, half = (start_of(prev.params.key.acq_conv, ev.on) if isinstance(ev, Add)
                               else (ev.first_period, ONE, False))
            if period_index(start) <= period_index(prev.start):
                # at or before the segment's own start: the same segment simply carries more
                self.segs[-1] = replace(prev, B=max(prev.B + amount, ZERO))
                return
            cum, B, R, e = self._next_state(prev, period_add(start, -1))
            self._close(prev, period_add(start, -1))
            self.segs.append(Seg(start, fw, half, max(B + amount, ZERO), R if R > 0 else ONE, cum, e, prev.params, self.fy,
                                 self._units_before(start)))
        elif isinstance(ev, Reset):
            start = period_of(ev.on)
            if period_index(start) <= period_index(prev.start):
                start = prev.start
                new = ev.params
                nbv = prev.B + prev.params.scrap_eff
                life = new.life
                R = max(frac(life) - prev.e0, ONE) if life else prev.R
                self.segs[-1] = replace(prev, params=new, B=max(nbv - new.scrap_eff, ZERO), R=R)
                return
            cum, B, R, e = self._next_state(prev, period_add(start, -1))
            nbv = B + prev.params.scrap_eff
            new = ev.params
            life = new.life
            R_new = max(frac(life) - e, ONE) if life else R
            self._close(prev, period_add(start, -1))
            self.segs.append(Seg(start, ONE, False, max(nbv - new.scrap_eff, ZERO), R_new, cum, e, new, self.fy,
                                 self._units_before(start)))
        elif isinstance(ev, SetNbv):
            at = period_of(ev.on)
            start = period_add(at, 1)
            cum, B, R, e = self._next_state(prev, at)
            self._close(prev, at)
            self.segs.append(Seg(start, ONE, False, max(frac(ev.nbv) - prev.params.scrap_eff, ZERO), R, cum, e,
                                 prev.params, self.fy, self._units_before(start)))
        elif isinstance(ev, Retire):
            at = period_of(ev.on)
            g = self._retirement_weight(prev, prev.params.key.ret_conv, ev.on)
            start = period_add(at, 1)
            cum, B, R, e = self._next_state(prev, at, g)
            self._close(prev, at, g)
            keep = ONE - frac(ev.fraction)
            self.segs.append(Seg(start, ONE, False, max(B * keep, ZERO), R, cum, e, prev.params, self.fy,
                                 self._units_before(start)))

    def _retirement_weight(self, seg: Seg, conv: str, on: date) -> Fraction:
        """The weight of the retirement period. half_year: the retirement fiscal year gets half a year's charge (six periods
        of life) in all, so the retirement period carries what the earlier periods of that year left of the six (never
        less than nothing: charges already planned are not taken back)."""
        if conv != "half_year":
            return retirement_weight(conv, on)
        at = period_of(on)
        before_year = period_add(fy_first_period(fiscal_year(at, self.fy), self.fy), -1)
        used = seg.W(period_add(at, -1)) - (seg.W(before_year) if period_index(before_year) >= period_index(seg.start)
                                            else ZERO)
        wp = seg.w(at)
        return max(Fraction(6) - used, ZERO) / wp if wp > 0 else ZERO

    def retirement_weight(self, conv: str, on: date) -> Fraction:
        """The weight a retirement on `on` gives its own period under convention `conv`, for this plan."""
        at = period_of(on)
        seg = None
        for s in self.segs:
            if period_index(s.start) <= period_index(at):
                seg = s
        return ONE if seg is None else self._retirement_weight(seg, conv, on)

    # ── evaluating ──
    def _W_eff(self, seg: Seg, p: str, g: Fraction = ONE) -> Fraction:
        """Capped cumulative weight through p, the last period weighted g."""
        if seg.end is not None and period_index(p) >= period_index(seg.end[0]):
            p, g = seg.end
        if period_index(p) < period_index(seg.start):
            return ZERO
        w = seg.W(p) - (ONE - g) * seg.w(p)
        return min(w, seg.R) if seg.R > 0 else ZERO

    def _seg_cum(self, seg: Seg, p: str, g: Fraction = ONE, force_end: Optional[tuple] = None) -> Fraction:
        """Exact cumulative of this segment alone through p."""
        if force_end:
            p, g = force_end
        elif seg.end is not None and period_index(p) >= period_index(seg.end[0]):
            p, g = seg.end
        if period_index(p) < period_index(seg.start):
            return ZERO
        prm, m = seg.params, seg.params.key.method
        B, R = seg.B, seg.R
        if m == "none" or B <= 0:
            return ZERO
        if m == "immediate":
            return B
        if m == "units_of_production":
            total = frac(prm.total_units) if prm.total_units is not None else ZERO
            left = total - seg.units_base
            if left <= 0:
                return ZERO
            used = _units_through(self.usage, p) - seg.units_base
            return B * min(max(used, ZERO), left) / left
        if R <= 0:
            return B
        if m == "straight_line":
            return B * self._W_eff(seg, p, g) / R
        return self._declining(seg, p, g)

    def _declining(self, seg: Seg, p: str, g: Fraction) -> Fraction:
        prm = seg.params
        rate = frac(prm.key.rate_pct) / 100
        switch = prm.key.method == "declining_then_straight"
        scrap = prm.scrap_eff
        nbv = seg.B + scrap
        elapsed, cum = ZERO, ZERO
        target = period_index(p)
        fy0 = fiscal_year(seg.start, seg.fy_end)
        fy = fy0
        while elapsed < seg.R:
            first = max(period_index(seg.start), period_index(fy_last_period(fy, seg.fy_end)) - 11)
            last = period_index(fy_last_period(fy, seg.fy_end))
            if first > target:
                break
            ws = []
            for i in range(first, last + 1):
                ws.append((i, seg.w(period_add("0000-01", i))))
            full = sum((w for _, w in ws), ZERO)
            full = min(full, seg.R - elapsed)
            if full <= 0:
                break
            base = (seg.B + scrap) if prm.key.base == "cost" else nbv
            db = rate * base * full / 12
            sl = (nbv - scrap) * full / (seg.R - elapsed) if switch else ZERO
            charge = min(max(db, sl), nbv - scrap)
            if target <= last:
                used = ZERO
                for i, w in ws:
                    if i > target:
                        break
                    used += w * (g if i == target else ONE)
                used = min(used, full)
                cum += charge * used / full if full else ZERO
                break
            cum += charge
            nbv -= charge
            elapsed += full
            fy += 1
        return cum

    def cum_exact(self, p: str) -> Fraction:
        """Planned depreciation to date, exact, through the end of period p."""
        seg = None
        for s in self.segs:
            if period_index(s.start) <= period_index(p):
                seg = s
        if seg is None:
            return ZERO
        return seg.cum0 + self._seg_cum(seg, p)

    def cum(self, p: str) -> Decimal:
        """Planned depreciation to date, rounded once, half up, to the cent."""
        return money(self.cum_exact(p))

    def charge(self, p: str) -> Decimal:
        """The period's own charge: cum(p) minus cum(previous period)."""
        return self.cum(p) - self.cum(period_add(p, -1))

    def remaining_periods(self, p: str) -> Fraction:
        """Periods of life left after period p."""
        seg = None
        for s in self.segs:
            if period_index(s.start) <= period_index(p):
                seg = s
        if seg is None:
            return frac(self.params.life or 0)
        return max(seg.R - self._W_eff(seg, p), ZERO)

    def cum_with(self, p: str, g: Fraction) -> Fraction:
        """Exact planned to date through p with the last period weighted g (a retirement's convention)."""
        seg = None
        for s in self.segs:
            if period_index(s.start) <= period_index(p):
                seg = s
        if seg is None:
            return ZERO
        return seg.cum0 + self._seg_cum(seg, p, g)

    def schedule(self, first: str, last: str) -> list:
        """[(period, charge)] for a forecast."""
        out, prev = [], self.cum(period_add(first, -1))
        for i in range(period_index(first), period_index(last) + 1):
            p = period_add("0000-01", i)
            c = self.cum(p)
            out.append((p, c - prev))
            prev = c
        return out