From a549942afe9a2f4246e92d25f435f5867365b0a3 Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Wed, 22 Jul 2026 20:10:52 +0200 Subject: [PATCH] =?UTF-8?q?feat(fundamentals):=20A4a=20=E2=80=94=20pure=20?= =?UTF-8?q?read-time=20metric=20derivation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- app/services/fundamentals_derivation.py | 245 +++++++++++++++++++++ tests/unit/test_fundamentals_derivation.py | 137 ++++++++++++ 2 files changed, 382 insertions(+) create mode 100644 app/services/fundamentals_derivation.py create mode 100644 tests/unit/test_fundamentals_derivation.py diff --git a/app/services/fundamentals_derivation.py b/app/services/fundamentals_derivation.py new file mode 100644 index 0000000..0f1d42f --- /dev/null +++ b/app/services/fundamentals_derivation.py @@ -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) diff --git a/tests/unit/test_fundamentals_derivation.py b/tests/unit/test_fundamentals_derivation.py new file mode 100644 index 0000000..f079d35 --- /dev/null +++ b/tests/unit/test_fundamentals_derivation.py @@ -0,0 +1,137 @@ +"""Tests for pure read-time derivation of fundamentals from YTD snapshots.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date, datetime, timezone + +import pytest + +from app.services import fundamentals_derivation as fd + +UTC = timezone.utc + + +@dataclass +class Snap: + fiscal_year: int + fiscal_period: str + period_end: date + filed_date: date + accepted_at: datetime + revenue: float | None = None + net_income: float | None = None + operating_income: float | None = None + diluted_eps: float | None = None + cfo: float | None = None + capex: float | None = None + depreciation_amortization: float | None = None + cash_and_st_investments: float | None = None + total_debt: float | None = None + shares_outstanding: float | None = None + + +_FP = ["Q1", "Q2", "Q3", "FY"] +_ENDS = { # period_end per (fy, quarter index 0..3) + 2025: [date(2024, 12, 31), date(2025, 3, 31), date(2025, 6, 30), date(2025, 9, 30)], + 2026: [date(2025, 12, 31), date(2026, 3, 31), date(2026, 6, 30), date(2026, 9, 30)], +} + + +def _year(fy, discretes: dict[str, list[float]], instants: dict[str, list] | None = None): + """Build 4 snapshot rows (Q1,Q2,Q3,FY) with YTD-cumulative flow fields from the + given per-quarter discrete values; instants set as-is per quarter.""" + rows = [] + for i, fp in enumerate(_FP): + r = Snap(fy, fp, _ENDS[fy][i], _ENDS[fy][i], datetime(fy, 1 + i, 1, tzinfo=UTC)) + for fname, ds in discretes.items(): + setattr(r, fname, round(sum(ds[: i + 1]), 4)) # cumulative YTD + for fname, vals in (instants or {}).items(): + setattr(r, fname, vals[i]) + rows.append(r) + return rows + + +def _two_years(): + rev25 = [100, 110, 120, 130] + rev26 = [110, 121, 132, 143] # +10% each quarter YoY + rows = _year(2025, { + "revenue": rev25, + "operating_income": [x * 0.2 for x in rev25], + "diluted_eps": [1.0, 1.1, 1.2, 1.3], + "cfo": [x * 0.25 for x in rev25], + "capex": [x * 0.05 for x in rev25], + "depreciation_amortization": [x * 0.05 for x in rev25], + }, instants={"shares_outstanding": [1000, 1000, 1000, 1000], "cash_and_st_investments": [40] * 4, "total_debt": [140] * 4}) + rows += _year(2026, { + "revenue": rev26, + "operating_income": [x * 0.2 for x in rev26], + "diluted_eps": [1.1, 1.21, 1.32, 1.43], + "cfo": [x * 0.25 for x in rev26], + "capex": [x * 0.05 for x in rev26], + "depreciation_amortization": [x * 0.05 for x in rev26], + }, instants={"shares_outstanding": [900, 900, 900, 900], "cash_and_st_investments": [50] * 4, "total_debt": [150] * 4}) + return rows + + +def test_revenue_growth_yoy_and_q4_derivation(): + d = fd.derive(_two_years()) + # TTM revenue FY2026 = 110+121+132+143 = 506; FY2025 = 460 -> +10% + assert d.metrics["revenue_growth_yoy"].value == pytest.approx(10.0, abs=1e-6) + # latest period is FY2026 + assert d.latest_period_end == date(2026, 9, 30) + # tape has 4 points, newest last, each carrying a period_end + hist = d.metrics["revenue_growth_yoy"].history + assert len(hist) == 4 and hist[-1].period_end == date(2026, 9, 30) + + +def test_operating_and_fcf_margin(): + d = fd.derive(_two_years()) + assert d.metrics["operating_margin"].value == pytest.approx(20.0, abs=1e-6) + # FCF margin = (TTM cfo - TTM capex)/TTM rev = (0.25 - 0.05) = 20% + assert d.metrics["fcf_margin"].value == pytest.approx(20.0, abs=1e-6) + + +def test_net_debt_leverage_and_share_dilution(): + d = fd.derive(_two_years()) + # net debt = total_debt - cash = 150 - 50 = 100 (latest instant) + assert d.metrics["net_debt"].value == pytest.approx(100.0) + # EBITDA TTM = TTM operating_income + TTM D&A; net_debt/ebitda + op_ttm = 506 * 0.2 # 101.2 + da_ttm = 506 * 0.05 # 25.3 + assert d.metrics["net_debt_to_ebitda"].value == pytest.approx(100.0 / (op_ttm + da_ttm), rel=1e-6) + # shares 900 vs 1000 a year earlier -> -10% (buyback) + assert d.metrics["share_count_change_yoy"].value == pytest.approx(-10.0, abs=1e-6) + + +def test_valuation_inputs(): + d = fd.derive(_two_years()) + # TTM diluted EPS FY2026 = 1.1+1.21+1.32+1.43 = 5.06 + assert d.ttm_diluted_eps == pytest.approx(5.06, abs=1e-6) + # TTM FCF = TTM cfo - TTM capex = 506*0.25 - 506*0.05 = 101.2 + assert d.ttm_fcf == pytest.approx(506 * 0.20, abs=1e-6) + assert d.shares_outstanding == 900 + + +def test_missing_period_yields_null_never_partial(): + rows = _two_years() + # drop FY2026 Q3 -> discrete Q3 and Q4 (needs YTD Q3) become underivable, + # so TTM at FY2026 is null -> revenue growth null (not a partial sum) + rows = [r for r in rows if not (r.fiscal_year == 2026 and r.fiscal_period == "Q3")] + d = fd.derive(rows) + assert d.metrics["revenue_growth_yoy"].value is None + assert d.ttm_diluted_eps is None + + +def test_amendment_selection_newest_accepted_wins(): + rows = _two_years() + # an amendment to FY2026 FY restates revenue YTD higher, accepted later + amended = Snap(2026, "FY", date(2026, 9, 30), date(2026, 11, 1), + datetime(2027, 1, 1, tzinfo=UTC), revenue=999999, + operating_income=100, diluted_eps=1.43, cfo=100, capex=10, + depreciation_amortization=25, shares_outstanding=900, + cash_and_st_investments=50, total_debt=150) + d = fd.derive(rows + [amended]) + # Q4 revenue discrete now uses the amended YTD(FY)=999999 minus YTD(Q3)=363 + # so TTM/growth reflects the amendment, proving newest accepted_at won. + assert d.metrics["revenue_growth_yoy"].value != pytest.approx(10.0, abs=1e-6)