""" Trailing-twelve-month (TTM) aggregation (StockLab final engineering pass, Part A9). ## Why this module exists `docs/AUDIT_METRICS.md` cross-cutting finding #2 recorded a real correctness problem: **"current" is not true TTM.** `build_snapshot_from_db()` selects `period_type == "FY"` only and takes the most recent row verbatim, and `FMPAdapter` defaults to `period="annual"` — so every metric this platform labels TTM is in fact "most recent completed fiscal year", which can be up to 15 months stale by the time it is displayed. For a fast-moving company that is not a cosmetic difference: a P/E built on a year-old EPS against today's price is simply a different number from a real trailing P/E. This module is the missing aggregation step. It is deliberately pure — stdlib only, no sqlalchemy, no provider, no DB — so its correctness can be genuinely unit-tested in this environment, the same separation used by `app/engines/scoring/wiring.py`, `app/api/v1/aggregation.py` and `app/adapters/resolver.py::resolve_line_items()`. ## The rules it implements, and why each one matters 1. **Flow vs. stock.** Income-statement and cash-flow items are *flows* — they accumulate over a period, so a TTM value is the SUM of the four quarters. Balance-sheet items and share counts are *stocks* — they are a photograph at one instant, so the TTM value is the MOST RECENT quarter's value, never a sum and never an average. Summing four quarters of `total_assets` would produce a number roughly four times too large; averaging it would silently invent a figure that appears in no filing. Both mistakes are easy to make and neither is visible in the output, which is why the field classification is explicit and exhaustive below. 2. **A missing quarter poisons the whole sum.** If any of the four quarters is missing a flow field, the TTM value for that field is `None` — NOT the sum of the three that are present. A three-quarter sum presented as TTM understates the metric by roughly 25% while looking completely plausible. This is the single most dangerous failure mode in TTM aggregation and the rule is enforced field-by-field, not just at the period level. 3. **Four quarters means four consecutive quarters.** The input is checked for exactly four distinct period ends spaced like real quarters (`_MIN_QUARTER_GAP_DAYS`..`_MAX_QUARTER_GAP_DAYS` apart). A gap that looks like a skipped quarter is rejected with a reason rather than summed — summing Q1, Q2, Q4 of one year and Q1 of the next would produce a "TTM" spanning fifteen months. 4. **`filing_date` is the LATEST quarter's filing date, not the earliest.** A TTM figure does not exist, and cannot be known by anyone, until the final quarter in it has been filed. Using any earlier date would reintroduce exactly the look-ahead bias `app/engines/metrics/point_in_time.py` exists to prevent. 5. **Mixed currencies are refused.** Cross-currency summation is silently wrong. There is no FX normalization layer in this codebase (a known, documented gap), so this returns a reason instead of a number. ## What this module does NOT do - **It does not implement the "annual + interim stub" method** (`FY + latest YTD − prior-year same YTD`), which is what you must use when a market publishes half-year rather than quarterly interim reports. Every provider adapter in this repo targets US-style quarterly reporting; building the stub method without a non-US filing adapter to feed it would be untestable scaffolding. Declared here, not silently omitted. - **It does not restate for discontinued operations or mid-year acquisitions.** A four-quarter sum across an acquisition close date mixes pre- and post-acquisition scope. That is a real limitation of every simple TTM, disclosed rather than papered over. - **It does not fetch, store, or schedule anything.** Wiring it into ingestion and recompute is a separate, DB-touching step — see `docs/AUDIT_TTM_A9.md` for the exact integration points and their honest test status. """ from __future__ import annotations from dataclasses import replace from typing import Iterable, Optional from app.engines.types import LineItems # A calendar quarter is ~91 days. The window below tolerates 52/53-week fiscal calendars, leap # years, and the odd 13-week/14-week quarter, while still rejecting a skipped quarter (~182 days). _MIN_QUARTER_GAP_DAYS = 75 _MAX_QUARTER_GAP_DAYS = 110 QUARTERS_IN_TTM = 4 # --- Field classification. Exhaustive over LineItems: every field is in exactly one bucket. --- #: Flow fields — accumulate over a reporting period, so TTM = sum of the four quarters. FLOW_FIELDS: tuple[str, ...] = ( "revenue", "cogs", "gross_profit", "operating_income", "ebit", "ebitda", "net_income", "eps_diluted", "tax_expense", "pretax_income", "interest_expense", "operating_cash_flow", "capital_expenditure", "dividends_paid", "buybacks", "stock_issuance", "depreciation_and_amortization", "stock_based_compensation", "adjusted_net_income", ) #: Stock fields — a balance at one instant, so TTM = the most recent quarter's value. STOCK_FIELDS: tuple[str, ...] = ( "cash_and_equivalents", "short_term_investments", "total_debt", "short_term_debt", "long_term_debt", "lease_liabilities", "total_assets", "current_assets", "current_liabilities", "shareholders_equity", "minority_interest", "preferred_equity", "goodwill", "intangible_assets", "receivables", "inventory", "shares_outstanding", "diluted_shares", # Market data joined in at calculation time — a price is a point value, never a sum. "price", "market_cap", "beta", ) #: Identity fields, handled explicitly by build_ttm_line_items() rather than copied by bucket. _IDENTITY_FIELDS: tuple[str, ...] = ( "security_id", "period_end", "period_type", "filing_date", "currency", ) class TTMResult: """Either a TTM `LineItems`, or `None` plus a machine-readable reason it could not be built. A reason string rather than a bare `None` so callers can log/report *why* a security has no TTM basis — "we silently fell back to annual" is exactly the kind of invisible degradation this module exists to eliminate. """ __slots__ = ("line_items", "reason", "quarters_used") def __init__( self, line_items: Optional[LineItems], reason: Optional[str] = None, quarters_used: int = 0, ): self.line_items = line_items self.reason = reason self.quarters_used = quarters_used @property def ok(self) -> bool: return self.line_items is not None def __repr__(self) -> str: # pragma: no cover - debugging aid return f"TTMResult(ok={self.ok}, reason={self.reason!r}, quarters_used={self.quarters_used})" def validate_field_classification() -> list[str]: """Return LineItems fields that are in neither bucket (or in both). Called by a test, not at import time. If someone adds a field to `LineItems` and forgets to classify it, that field would otherwise be silently dropped from every TTM row — present and populated on an annual snapshot, `None` on a TTM one, with nothing anywhere saying so. """ all_fields = set(LineItems.__dataclass_fields__) classified = set(FLOW_FIELDS) | set(STOCK_FIELDS) | set(_IDENTITY_FIELDS) overlap = set(FLOW_FIELDS) & set(STOCK_FIELDS) return sorted((all_fields - classified) | (classified - all_fields) | overlap) def _check_consecutive(quarters: list[LineItems]) -> Optional[str]: """Return a reason string if these are not four consecutive quarters, else None.""" ends = [q.period_end for q in quarters] if len(set(ends)) != len(ends): return "duplicate_period_end" for newer, older in zip(ends, ends[1:]): gap = (newer - older).days if gap < _MIN_QUARTER_GAP_DAYS: return f"period_gap_too_small:{gap}d" if gap > _MAX_QUARTER_GAP_DAYS: return f"period_gap_too_large:{gap}d" return None def build_ttm_line_items(quarterly: Iterable[LineItems]) -> TTMResult: """Aggregate quarterly `LineItems` into a single TTM `LineItems`. `quarterly` may be in any order and may contain more than four quarters; the four most recent by `period_end` are used. Every rule in the module docstring is enforced here. """ rows = [q for q in quarterly if q is not None] if not rows: return TTMResult(None, "no_quarters_supplied", 0) non_quarterly = [q.period_type for q in rows if not str(q.period_type).upper().startswith("Q")] if non_quarterly: # Refusing rather than filtering: a caller that hands this an "FY" row has a bug, and # silently ignoring the row would hide it. return TTMResult(None, f"non_quarterly_period_type:{sorted(set(non_quarterly))}", 0) rows.sort(key=lambda q: q.period_end, reverse=True) window = rows[:QUARTERS_IN_TTM] if len(window) < QUARTERS_IN_TTM: return TTMResult(None, f"insufficient_quarters:{len(window)}_of_{QUARTERS_IN_TTM}", len(window)) reason = _check_consecutive(window) if reason is not None: return TTMResult(None, reason, len(window)) currencies = {q.currency for q in window} if len(currencies) > 1: return TTMResult(None, f"mixed_currencies:{sorted(currencies)}", len(window)) security_ids = {q.security_id for q in window} if len(security_ids) > 1: return TTMResult(None, f"mixed_securities:{sorted(security_ids)}", len(window)) latest = window[0] values: dict[str, Optional[float]] = {} for name in FLOW_FIELDS: parts = [getattr(q, name) for q in window] # Rule 2: any missing quarter poisons the sum for that field. values[name] = None if any(p is None for p in parts) else sum(parts) for name in STOCK_FIELDS: values[name] = getattr(latest, name) ttm = replace( latest, period_type="TTM", period_end=latest.period_end, filing_date=latest.filing_date, # Rule 4 **values, ) return TTMResult(ttm, None, QUARTERS_IN_TTM)