feat(fundamentals): A4a — pure read-time metric derivation

Derives the display metrics from the stored YTD snapshots at read time (no I/O, no DB), per the A3 schema decision. Given an issuer's snapshot rows it produces: - amendment selection (newest accepted_at per fiscal period); - discrete quarters = YTD(Qn) - YTD(Qn-1), Q4 = YTD(FY) - YTD(Q3); - TTM = trailing four discrete quarters; missing period -> null, never partial; - metric series (value + 4-quarter tape, each point dated): revenue_growth_yoy, eps_growth_yoy, operating_margin, fcf_margin, net_debt, net_debt_to_ebitda, share_count_change_yoy; - request-time valuation inputs (ttm_diluted_eps, ttm_fcf, shares_outstanding) for the API to combine with price. Units per app convention (percentages = pp, leverage = multiple, dollars). Tests: 6 (growth+Q4, margins, net-debt/EBITDA+dilution, valuation inputs, missing-period-null, amendment selection). Verified on real Apple snapshots: op margin 32.6%, net-debt/EBITDA 0.10, buyback -1.7%/yr, TTM EPS $8.26. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-22 20:10:52 +02:00
parent 96d3b40560
commit a549942afe
2 changed files with 382 additions and 0 deletions
@@ -0,0 +1,245 @@
+"""Pure read-time derivation of fundamental metrics from stored snapshots.
+
+`fundamental_snapshots` stores one immutable row per accession with **cumulative
+YTD** duration facts and period-end balance-sheet instants (A3). This module
+derives everything the UI/API shows — discrete quarters, Q4, TTM, YoY growth,
+margins, leverage, dilution, and the quarter tape — at read time, per the plan's
+schema decision. No I/O, no DB: it takes an issuer's snapshot rows (ORM rows or
+any objects with the same attributes) and returns structured metrics.
+
+Rules:
+- **Amendment selection:** for each (fiscal_year, fiscal_period), the row with
+  the newest `accepted_at` wins.
+- **Discrete quarter** = YTD(Qn) − YTD(Qn−1); Q1 = YTD(Q1); **Q4 = YTD(FY) −
+  YTD(Q3)**. Any missing period → the derived value is null, never partial.
+- **TTM** = sum of the trailing four discrete quarters ending at a period.
+- Units follow app convention: percentages are percentage points (21.0 = 21%),
+  net-debt/EBITDA is a multiple, net debt is dollars.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import date
+from typing import Any, Iterable
+
+_FP_TO_Q = {"Q1": 1, "Q2": 2, "Q3": 3, "FY": 4}
+_Q_TO_FP = {1: "Q1", 2: "Q2", 3: "Q3", 4: "FY"}
+_PREV_FP = {"Q2": "Q1", "Q3": "Q2", "FY": "Q3"}
+TAPE_LEN = 4  # quarter-tape length
+
+# Duration (flow) fields differenced from YTD into discrete quarters + summed to TTM.
+_FLOW_FIELDS = (
+    "revenue", "net_income", "operating_income", "diluted_eps", "cfo", "capex",
+    "depreciation_amortization",
+)
+
+
+@dataclass
+class MetricPoint:
+    period_end: date
+    value: float | None
+
+
+@dataclass
+class MetricSeries:
+    value: float | None = None
+    history: list[MetricPoint] = field(default_factory=list)  # oldest -> newest, <= TAPE_LEN
+    period_end: date | None = None
+    filed_date: date | None = None
+
+
+@dataclass
+class DerivedFundamentals:
+    metrics: dict[str, MetricSeries] = field(default_factory=dict)
+    # request-time valuation inputs (ratios are computed in the API with price)
+    ttm_diluted_eps: float | None = None
+    ttm_fcf: float | None = None
+    shares_outstanding: float | None = None
+    latest_period_end: date | None = None
+    latest_filed_date: date | None = None
+
+
+def _prev_q(fy: int, q: int) -> tuple[int, int]:
+    return (fy, q - 1) if q > 1 else (fy - 1, 4)
+
+
+def derive(snapshots: Iterable[Any]) -> DerivedFundamentals:
+    selected = _select_latest_per_period(snapshots)
+    result = DerivedFundamentals()
+    if not selected:
+        return result
+
+    # Discrete quarter values per flow field: {field: {(fy, q): value}}.
+    discrete = {f: _discrete_quarters(selected, f) for f in _FLOW_FIELDS}
+    quarters = _ordered_quarters(selected)  # chronological (fy, q) with a row
+    latest = quarters[-1]
+    latest_row = selected[(latest[0], _Q_TO_FP[latest[1]])]
+
+    result.latest_period_end = latest_row.period_end
+    result.latest_filed_date = latest_row.filed_date
+    result.shares_outstanding = getattr(latest_row, "shares_outstanding", None)
+    result.ttm_diluted_eps = _ttm(discrete["diluted_eps"], *latest)
+    ttm_cfo = _ttm(discrete["cfo"], *latest)
+    ttm_capex = _ttm(discrete["capex"], *latest)
+    result.ttm_fcf = None if ttm_cfo is None or ttm_capex is None else ttm_cfo - ttm_capex
+
+    # tape = the last TAPE_LEN quarters that have a row, oldest -> newest
+    tape = quarters[-TAPE_LEN:]
+    result.metrics = {
+        "revenue_growth_yoy": _yoy_growth_series(discrete["revenue"], selected, tape),
+        "eps_growth_yoy": _yoy_growth_series(discrete["diluted_eps"], selected, tape),
+        "operating_margin": _margin_series(discrete["operating_income"], discrete["revenue"], selected, tape),
+        "fcf_margin": _fcf_margin_series(discrete, selected, tape),
+        "net_debt": _instant_series(selected, tape, _net_debt),
+        "net_debt_to_ebitda": _leverage_series(selected, discrete, tape),
+        "share_count_change_yoy": _share_change_series(selected, tape),
+    }
+    for series in result.metrics.values():
+        series.period_end = latest_row.period_end
+        series.filed_date = latest_row.filed_date
+    return result
+
+
+# -- period selection --------------------------------------------------------
+
+def _select_latest_per_period(snapshots: Iterable[Any]) -> dict[tuple[int, str], Any]:
+    best: dict[tuple[int, str], Any] = {}
+    for row in snapshots:
+        fp = getattr(row, "fiscal_period", None)
+        fy = getattr(row, "fiscal_year", None)
+        if fp not in _FP_TO_Q or fy is None:
+            continue
+        key = (fy, fp)
+        cur = best.get(key)
+        if cur is None or _accepted(row) > _accepted(cur):
+            best[key] = row
+    return best
+
+
+def _accepted(row: Any):
+    return getattr(row, "accepted_at", None) or getattr(row, "filed_date", None)
+
+
+def _ordered_quarters(selected: dict[tuple[int, str], Any]) -> list[tuple[int, int]]:
+    return sorted((fy, _FP_TO_Q[fp]) for (fy, fp) in selected)
+
+
+# -- discrete + TTM ----------------------------------------------------------
+
+def _discrete_quarters(selected: dict[tuple[int, str], Any], field_name: str) -> dict[tuple[int, int], float]:
+    out: dict[tuple[int, int], float] = {}
+    for (fy, fp), row in selected.items():
+        val = _discrete_value(selected, fy, fp, field_name)
+        if val is not None:
+            out[(fy, _FP_TO_Q[fp])] = val
+    return out
+
+
+def _discrete_value(selected, fy: int, fp: str, field_name: str) -> float | None:
+    cur = getattr(selected[(fy, fp)], field_name, None)
+    if cur is None:
+        return None
+    if fp == "Q1":
+        return cur
+    prev = selected.get((fy, _PREV_FP[fp]))
+    prev_val = getattr(prev, field_name, None) if prev is not None else None
+    if prev_val is None:
+        return None
+    return cur - prev_val
+
+
+def _ttm(dq: dict[tuple[int, int], float], fy: int, q: int) -> float | None:
+    keys = [(fy, q)]
+    k = (fy, q)
+    for _ in range(3):
+        k = _prev_q(*k)
+        keys.append(k)
+    vals = [dq.get(kk) for kk in keys]
+    if any(v is None for v in vals):
+        return None
+    return sum(vals)
+
+
+def _pct_change(cur: float | None, prior: float | None) -> float | None:
+    if cur is None or prior is None or prior == 0:
+        return None
+    return (cur / prior - 1.0) * 100.0
+
+
+# -- per-metric series (value at latest + tape history) ----------------------
+
+def _period_end(selected, fy: int, q: int) -> date | None:
+    row = selected.get((fy, _Q_TO_FP[q]))
+    return row.period_end if row is not None else None
+
+
+def _yoy_growth_series(dq, selected, tape) -> MetricSeries:
+    pts = []
+    for (fy, q) in tape:
+        cur, prior = _ttm(dq, fy, q), _ttm(dq, fy - 1, q)
+        pts.append(MetricPoint(_period_end(selected, fy, q), _pct_change(cur, prior)))
+    return _series(pts)
+
+
+def _margin_series(num_dq, den_dq, selected, tape) -> MetricSeries:
+    pts = []
+    for (fy, q) in tape:
+        num, den = _ttm(num_dq, fy, q), _ttm(den_dq, fy, q)
+        val = None if num is None or not den else num / den * 100.0
+        pts.append(MetricPoint(_period_end(selected, fy, q), val))
+    return _series(pts)
+
+
+def _fcf_margin_series(discrete, selected, tape) -> MetricSeries:
+    pts = []
+    for (fy, q) in tape:
+        cfo, capex, rev = _ttm(discrete["cfo"], fy, q), _ttm(discrete["capex"], fy, q), _ttm(discrete["revenue"], fy, q)
+        val = None if cfo is None or capex is None or not rev else (cfo - capex) / rev * 100.0
+        pts.append(MetricPoint(_period_end(selected, fy, q), val))
+    return _series(pts)
+
+
+def _instant_series(selected, tape, fn) -> MetricSeries:
+    pts = [MetricPoint(_period_end(selected, fy, q), fn(selected.get((fy, _Q_TO_FP[q])))) for (fy, q) in tape]
+    return _series(pts)
+
+
+def _leverage_series(selected, discrete, tape) -> MetricSeries:
+    pts = []
+    for (fy, q) in tape:
+        row = selected.get((fy, _Q_TO_FP[q]))
+        nd = _net_debt(row)
+        op, da = _ttm(discrete["operating_income"], fy, q), _ttm(discrete["depreciation_amortization"], fy, q)
+        ebitda = None if op is None or da is None else op + da
+        val = None if nd is None or not ebitda else nd / ebitda
+        pts.append(MetricPoint(_period_end(selected, fy, q), val))
+    return _series(pts)
+
+
+def _share_change_series(selected, tape) -> MetricSeries:
+    pts = []
+    for (fy, q) in tape:
+        cur = _shares(selected.get((fy, _Q_TO_FP[q])))
+        prior = _shares(selected.get((fy - 1, _Q_TO_FP[q])))
+        pts.append(MetricPoint(_period_end(selected, fy, q), _pct_change(cur, prior)))
+    return _series(pts)
+
+
+def _net_debt(row: Any) -> float | None:
+    if row is None:
+        return None
+    cash = getattr(row, "cash_and_st_investments", None)
+    debt = getattr(row, "total_debt", None)
+    if cash is None and debt is None:
+        return None
+    return (debt or 0.0) - (cash or 0.0)  # positive = net debt
+
+
+def _shares(row: Any) -> float | None:
+    return getattr(row, "shares_outstanding", None) if row is not None else None
+
+
+def _series(points: list[MetricPoint]) -> MetricSeries:
+    value = points[-1].value if points else None
+    return MetricSeries(value=value, history=points)