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 newest
  `accepted_at` wins **per field**, falling back to the newest row that actually
  reports one. A partial amendment (a 10-K/A adding Part III carries no financial
  facts) must not blank the period.
- **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 types import SimpleNamespace
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",
)
# Reported facts resolved independently across a period's accessions (see
# _merge_amendments); period identity/provenance is taken from the newest one.
_MERGED_FIELDS = (
    *_FLOW_FIELDS,
    "cash_and_st_investments", "total_debt", "shares_outstanding",
    "shares_outstanding_date",
    # period_start is set alongside revenue by the parser, so it follows the same
    # fallback: a bare amendment reports neither and must not blank it.
    "period_start",
)
_CARRIED_FIELDS = (
    "fiscal_year", "fiscal_period", "period_end", "filed_date",
    "accepted_at", "form", "accession", "cik",
)


@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
    # Set when ttm_diluted_eps was suppressed rather than simply unavailable.
    ttm_diluted_eps_caveat: str | None = None
    ttm_fcf: float | None = None
    shares_outstanding: float | None = None
    # True when shares_outstanding came from the weighted-average diluted count
    # because the point-in-time cover-page count was absent (always so for
    # multi-class issuers). Consumers must label anything derived from it as
    # estimated — it is a period average, not a point-in-time count.
    shares_outstanding_estimated: bool = False
    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)
    if result.shares_outstanding is None:
        # Multi-class issuers (META, CMCSA, BRK-B, CHTR, FOXA, NWSA, LEN) report
        # the cover-page count per class, which is dimensional and so absent from
        # companyfacts — leaving market cap and FCF yield silently unavailable for
        # some of the largest names. The weighted-average diluted count is always
        # present and within ~0.6% of the true count where both exist, so fall
        # back to it and mark the result estimated rather than show nothing.
        result.shares_outstanding = getattr(latest_row, "weighted_avg_diluted_shares", None)
        result.shares_outstanding_estimated = result.shares_outstanding is not 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),
    }
    # TTM EPS sums four quarters of *per-share* values, so a split inside that
    # window mixes pre- and post-split units — the same distortion the guard
    # already catches for the series, and the one that produced BKNG's P/E of
    # 1.10. Left unguarded it does not merely mislead: a nonsense-low P/E clamps
    # to a perfect 100 fundamental sub-score, so it must null out like the rest.
    if _guard_split_sensitive_metrics(result.metrics):
        result.ttm_diluted_eps = None
        result.ttm_diluted_eps_caveat = SPLIT_SENSITIVE_CAVEAT
    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]:
    grouped: dict[tuple[int, str], list[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
        grouped.setdefault((fy, fp), []).append(row)
    return {key: _merge_amendments(rows) for key, rows in grouped.items()}


def _merge_amendments(rows: list[Any]) -> Any:
    """Resolve one period from its accessions: newest wins, per field.

    Amendments are frequently partial — a 10-K/A filed only to add Part III
    reports no financial facts at all. Taking the newest accession wholesale
    would blank every field it omits and null the period downstream (and with
    it TTM and YoY, which need an unbroken quarter chain), so each field falls
    back to the newest accession that actually reports it.

    Only rows sharing the newest row's ``period_end`` are merged. A same-key row
    covering a *different* period is a mislabelled filing, not an amendment, and
    blending the two would silently mix fiscal years.
    """
    if len(rows) == 1:
        return rows[0]
    ordered = sorted(rows, key=_amendment_order, reverse=True)  # newest first
    newest = ordered[0]
    same_period = [
        row
        for row in ordered
        if getattr(row, "period_end", None) == getattr(newest, "period_end", None)
    ]
    if len(same_period) == 1:
        return newest
    merged = SimpleNamespace(**{name: getattr(newest, name, None) for name in _CARRIED_FIELDS})
    for name in _MERGED_FIELDS:
        merged_value = None
        for row in same_period:  # newest first
            value = getattr(row, name, None)
            if value is not None:
                merged_value = value
                break
        setattr(merged, name, merged_value)
    return merged


def _amendment_order(row: Any) -> tuple[bool, Any]:
    # (has-timestamp, timestamp) so a row without one sorts oldest instead of
    # raising when compared against a row that has one.
    accepted = _accepted(row)
    return (accepted is not None, accepted)


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]) -> bool:
    """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.

    Returns True when the *latest* period is suspect, so callers can apply the
    same suppression to per-share scalars derived from that window.
    """
    shares = metrics.get("share_count_change_yoy")
    eps = metrics.get("eps_growth_yoy")
    if shares is None or eps is None:
        return False

    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 False

    latest_suspect = False
    for series in (shares, eps):
        latest_guarded = bool(
            series.history and series.history[-1].period_end in suspect_periods
        )
        latest_suspect = latest_suspect or latest_guarded
        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
    return latest_suspect


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)