311 lines
11 KiB
Python
311 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
|
||
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
|
||
SPLIT_SUSPECT_SHARE_CHANGE_PCT = 25.0
|
||
SPLIT_SENSITIVE_CAVEAT = (
|
||
"Not comparable: share count changed at least 25%; possible split or "
|
||
"corporate action."
|
||
)
|
||
|
||
# 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
|
||
caveat: str | 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),
|
||
}
|
||
_guard_split_sensitive_metrics(result.metrics)
|
||
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)
|
||
|
||
|
||
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 _guard_split_sensitive_metrics(metrics: dict[str, MetricSeries]) -> None:
|
||
"""Suppress historical comparisons likely distorted by a corporate action.
|
||
|
||
Company Facts has no point-in-time split factors. A large YoY share-count
|
||
move can therefore make both the point-in-time share comparison and
|
||
per-share EPS growth non-comparable. Keep the raw facts in snapshots, but
|
||
expose nulls plus an explicit caveat in the user-facing derived series.
|
||
"""
|
||
shares = metrics.get("share_count_change_yoy")
|
||
eps = metrics.get("eps_growth_yoy")
|
||
if shares is None or eps is None:
|
||
return
|
||
|
||
suspect_periods = {
|
||
point.period_end
|
||
for point in shares.history
|
||
if point.value is not None
|
||
and abs(point.value) >= SPLIT_SUSPECT_SHARE_CHANGE_PCT
|
||
}
|
||
if not suspect_periods:
|
||
return
|
||
|
||
for series in (shares, eps):
|
||
latest_guarded = bool(
|
||
series.history and series.history[-1].period_end in suspect_periods
|
||
)
|
||
for point in series.history:
|
||
if point.period_end in suspect_periods:
|
||
point.value = None
|
||
series.value = series.history[-1].value if series.history else None
|
||
if latest_guarded:
|
||
series.caveat = SPLIT_SENSITIVE_CAVEAT
|
||
|
||
|
||
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)
|