signal-platform/app/services/fundamentals_derivation.py

"""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 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


# -- 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)