""" 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("$", "$")