320 lines
11 KiB
Python
320 lines
11 KiB
Python
"""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, datetime, timezone
|
||
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 CONSECUTIVE run of up to TAPE_LEN quarters ending at the latest,
|
||
# stopping at a gap — so trend text never compares non-adjacent periods.
|
||
tape = _consecutive_suffix(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
|
||
|
||
|
||
def derive_as_of(snapshots: Iterable[Any], as_of: datetime) -> DerivedFundamentals:
|
||
"""Derive using only SEC filings accepted by the historical cutoff."""
|
||
cutoff = _utc_datetime(as_of)
|
||
visible = (
|
||
row
|
||
for row in snapshots
|
||
if (accepted := getattr(row, "accepted_at", None)) is not None
|
||
and _utc_datetime(accepted) <= cutoff
|
||
)
|
||
return derive(visible)
|
||
|
||
|
||
def _utc_datetime(value: datetime) -> datetime:
|
||
if value.tzinfo is None:
|
||
return value.replace(tzinfo=timezone.utc)
|
||
return value.astimezone(timezone.utc)
|
||
|
||
|
||
# -- 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)
|
||
|
||
|
||
def _consecutive_suffix(
|
||
quarters: list[tuple[int, int]], n: int
|
||
) -> list[tuple[int, int]]:
|
||
"""The run of up to n quarters ending at the latest, walking back only through
|
||
adjacent periods (stop at the first gap). Returned oldest -> newest."""
|
||
if not quarters:
|
||
return []
|
||
present = set(quarters)
|
||
run = [quarters[-1]]
|
||
cur = quarters[-1]
|
||
while len(run) < n:
|
||
prev = _prev_q(*cur)
|
||
if prev not in present:
|
||
break
|
||
run.append(prev)
|
||
cur = prev
|
||
run.reverse()
|
||
return run
|
||
|
||
|
||
# -- 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:
|
||
# A non-positive prior makes a YoY % meaningless (e.g. loss->profit), so null it.
|
||
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
|
||
# Null when EBITDA <= 0: a negative denominator would flip polarity and a
|
||
# "lower is better" read would rank a distressed issuer as favorable.
|
||
val = None if nd is None or ebitda is None or ebitda <= 0 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)
|
||
# Require BOTH components — treating a missing side as zero would produce a
|
||
# partial, misleading value.
|
||
if cash is None or debt is None:
|
||
return None
|
||
return debt - cash # 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)
|