palladium/studies/202602_TEI_Amazon_Connect/teicalc/model.py

"""
Finance engine — the single source of truth for every number in the notebook.

Transplanted from the retired shared ``core/calculations`` and
``core/export/report_data.py`` so the study is self-contained (Mercury
Notebook Pattern, Required §2/§7). Conventions match the Forrester TEI
methodology:

* The *Initial* investment is **not** discounted — it occurs at time zero.
* Year-N cash flows are discounted at the end of the year:
  ``PV = CF_n / (1 + r) ** n``.
* Benefits are risk-adjusted *down* (``×(1−rf)``), costs *up* (``×(1+rf)``).
* Payback runs on risk-adjusted **undiscounted** flows (the PDF's
  "<6 months" uses the Cash Flow Analysis table's nominal RA rows).

Everything this module returns for display is keyed by **calendar year**
(Forrester Year 1/2/3 → 2026/2027/2028); ``initial`` stays a Year-0 scalar
and never appears inside a ``*_by_year`` dict.

This module is stdlib-only on purpose — the repo-root test suite imports it
without the study's venv.
"""

from __future__ import annotations

import math
from collections.abc import Iterable, Sequence
from copy import deepcopy

# ── Timeline ─────────────────────────────────────────────────────────

YEARS: list[int] = [2026, 2027, 2028]      # Forrester Year 1/2/3; window opens Jan 2026
YEAR_INDEX: dict[int, int] = {y: i for i, y in enumerate(YEARS, start=1)}
X_LABELS: list[str] = ["Initial"] + [str(y) for y in YEARS]

_MONTHS = ["Jan", "Feb", "Mar", "Apr", "May", "Jun",
           "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]


def month_label(m: int) -> str:
    """Calendar label for a 1-indexed month from Jan of YEARS[0]."""
    return f"{_MONTHS[(m - 1) % 12]} {YEARS[0] + (m - 1) // 12}"


def by_calendar(year_values: dict[str, float]) -> dict[int, float]:
    """Map Forrester's ``{"1": v, …}`` year-index keys to calendar years."""
    return {YEARS[int(k) - 1]: float(v or 0) for k, v in year_values.items()}


# ── Discounting primitives ───────────────────────────────────────────


def discount_factor(year_index: int, discount_rate: float) -> float:
    """``1 / (1 + r) ** n``. Year 0 → 1.0 (no discount)."""
    if year_index < 0:
        raise ValueError("year_index must be >= 0")
    return 1.0 / ((1.0 + discount_rate) ** year_index)


def present_value(amount: float, year_index: int, discount_rate: float) -> float:
    """Discount ``amount`` from end-of-year ``year_index`` to present."""
    return amount * discount_factor(year_index, discount_rate)


def npv(cashflows: Iterable[float], discount_rate: float,
        initial: float = 0.0) -> float:
    """``initial + Σ CF_n / (1 + r)^n`` — initial undiscounted (TEI)."""
    return initial + sum(
        present_value(float(cf), n, discount_rate)
        for n, cf in enumerate(cashflows, start=1)
    )


def roi_pct(benefits_pv: float, costs_pv: float) -> float:
    """``(Benefits − Costs) / Costs`` as a percentage; 0 when costs ≤ 0."""
    if costs_pv <= 0:
        return 0.0
    return (benefits_pv - costs_pv) / costs_pv * 100.0


# ── Payback ──────────────────────────────────────────────────────────


def payback_years(initial_cost: float,
                  yearly_net: Sequence[float]) -> float | None:
    """
    Years until cumulative net benefits cover the initial cost, with linear
    interpolation inside the crossing year. ``None`` if never reached.
    """
    remaining = float(initial_cost)
    if remaining <= 0:
        return 0.0
    for i, cf in enumerate(yearly_net):
        cf = float(cf)
        if cf <= 0:
            remaining += -cf            # a net-loss year widens the gap
            continue
        if cf >= remaining:
            return i + remaining / cf
        remaining -= cf
    return None


def payback_months(initial_cost: float,
                   yearly_net: Sequence[float]) -> float | None:
    """Same as :func:`payback_years`, in months."""
    yrs = payback_years(initial_cost, yearly_net)
    return yrs * 12.0 if yrs is not None else None


def payback_label(months: float | None) -> str:
    """Human label: ``"0.7 months (~Jan 2026)"`` / ``"immediate"`` / ``"beyond 2028"``."""
    if months is None:
        return f"beyond {YEARS[-1]}"
    if months <= 0:
        return "immediate"
    return f"{months:.1f} months (~{month_label(max(1, math.ceil(months)))})"


# ── Risk adjustment (TEI: benefits down, costs up) ───────────────────


def risk_adjust_benefit(amount: float, risk_factor: float) -> float:
    """``amount × (1 − rf)``, rf clamped to [0, 1]."""
    rf = max(0.0, min(1.0, float(risk_factor)))
    return amount * (1.0 - rf)


def risk_adjust_cost(amount: float, risk_factor: float) -> float:
    """``amount × (1 + rf)``, rf clamped to [0, 1]."""
    rf = max(0.0, min(1.0, float(risk_factor)))
    return amount * (1.0 + rf)


def risk_adjusted_rows(rows: list[dict], table: str) -> list[dict]:
    """Deep-copied rows with the per-row risk factor applied to every value."""
    adjust = risk_adjust_benefit if table == "benefits" else risk_adjust_cost
    out: list[dict] = []
    for raw in rows:
        row = deepcopy(raw)
        rf = float(row.get("risk_adjustment") or 0.0)
        row["year_values"] = {
            k: adjust(float(v or 0), rf) for k, v in row["year_values"].items()
        }
        if row.get("initial"):
            # Only costs carry an initial; TEI adjusts it upward like the years.
            row["initial"] = risk_adjust_cost(float(row["initial"]), rf) \
                if table == "costs" else float(row["initial"])
        out.append(row)
    return out


# ── Aggregation (calendar-keyed) ─────────────────────────────────────


def _totals_by_year(ra_rows: list[dict]) -> dict[int, float]:
    totals = {y: 0.0 for y in YEARS}
    for row in ra_rows:
        for y, v in by_calendar(row["year_values"]).items():
            totals[y] += v
    return totals


def benefits_by_year(rows: list[dict]) -> dict[int, float]:
    """Risk-adjusted benefit totals per calendar year."""
    return _totals_by_year(risk_adjusted_rows(rows, "benefits"))


def costs_by_year(rows: list[dict]) -> dict[int, float]:
    """Risk-adjusted cost totals per calendar year (excludes ``initial``)."""
    return _totals_by_year(risk_adjusted_rows(rows, "costs"))


def initial_costs(rows: list[dict]) -> float:
    """Risk-adjusted Year-0 outlay (undiscounted)."""
    return sum(
        float(row.get("initial") or 0)
        for row in risk_adjusted_rows(rows, "costs")
    )


# ── Composite summary ────────────────────────────────────────────────


def compute_summary(benefits: list[dict], costs: list[dict],
                    discount_rate: float = 0.10) -> dict:
    """
    The full business-case readout for one set of value rows.

    Returns KPIs (``benefits_pv``/``costs_pv``/``npv``/``roi_pct``/
    ``payback_months``/``payback_label``/``initial_costs``/nominal totals),
    calendar-keyed schedules (``benefits_by_year``/``costs_by_year``/
    ``net_by_year``/``cumulative_net_by_year`` — cumulative subtracts the
    initial outlay), and a per-row breakdown under ``rows``.
    """
    ben_ra = risk_adjusted_rows(benefits, "benefits")
    cost_ra = risk_adjusted_rows(costs, "costs")

    ben_by = _totals_by_year(ben_ra)
    cost_by = _totals_by_year(cost_ra)
    initial = sum(float(r.get("initial") or 0) for r in cost_ra)

    benefits_pv = npv([ben_by[y] for y in YEARS], discount_rate)
    costs_pv = npv([cost_by[y] for y in YEARS], discount_rate, initial=initial)

    net_by = {y: ben_by[y] - cost_by[y] for y in YEARS}
    cum, cum_by = -initial, {}
    for y in YEARS:
        cum += net_by[y]
        cum_by[y] = cum

    pb_months = payback_months(initial, [net_by[y] for y in YEARS])

    def _row_breakdown(ra_rows: list[dict], table: str) -> list[dict]:
        out = []
        for row in ra_rows:
            ra_by = by_calendar(row["year_values"])
            init_ra = float(row.get("initial") or 0)
            entry = {
                "field_key": row["field_key"],
                "label": row["label"],
                "category": row["category"],
                "risk_adjustment": row["risk_adjustment"],
                "ra_by_year": ra_by,
                "three_yr_ra": sum(ra_by.values()),
                "pv": npv([ra_by[y] for y in YEARS], discount_rate,
                          initial=init_ra if table == "costs" else 0.0),
            }
            if table == "costs":
                entry["initial_ra"] = init_ra
            out.append(entry)
        return out

    return {
        "discount_rate": discount_rate,
        "benefits_pv": benefits_pv,
        "costs_pv": costs_pv,
        "npv": benefits_pv - costs_pv,
        "roi_pct": roi_pct(benefits_pv, costs_pv),
        "payback_months": pb_months,
        "payback_label": payback_label(pb_months),
        "initial_costs": initial,
        "benefits_nominal": sum(ben_by.values()),
        "costs_nominal": sum(cost_by.values()) + initial,
        "benefits_by_year": ben_by,
        "costs_by_year": cost_by,
        "net_by_year": net_by,
        "cumulative_net_by_year": cum_by,
        "rows": {
            "benefits": _row_breakdown(ben_ra, "benefits"),
            "costs": _row_breakdown(cost_ra, "costs"),
        },
    }


# ── Display helpers ──────────────────────────────────────────────────


def money(v: float) -> str:
    sign, a = ("-" if v < 0 else ""), abs(v)
    return f"{sign}${a/1e6:,.1f}M" if a >= 1e6 else f"{sign}${a/1e3:,.0f}K"


def html_money(v: float) -> str:
    """Plotly text with two or more bare ``$`` triggers MathJax math mode —
    annotations holding several amounts must use the HTML entity instead."""
    return money(v).replace("$", "&#36;")