dashboard/note_data.py

# -*- coding: utf-8 -*-
"""Everything the note needs from the databases, shaped for the page.

Kept apart from ``dashboard/note.py`` so the document builder is markup and
this file is arithmetic. Nothing here touches Streamlit, which is what lets
the whole series be asserted in tests.

The one piece of real work is fourth quarters. A company files three 10-Qs and
one 10-K a year, so Q4 exists nowhere as a filing: it is the annual figure
minus the three quarters. Flow items subtract; balance-sheet items are
point-in-time and come straight off the 10-K. Rows built that way are marked
``derived`` and the note greys them out and footnotes them, exactly as the
published note did — a derived quarter is a calculation, not a disclosure.
"""
from __future__ import annotations

import re

# Flow items: annual minus the three filed quarters.
_FLOW = (
    "revenue", "net_income", "operating_cash_flow", "capex", "cogs",
    "research_development", "sga", "sbc", "dividends_paid", "buybacks",
    "depreciation_amortization", "interest_expense",
)

# Point-in-time items: the 10-K's own closing value is the quarter's value.
_STOCK = (
    "receivables", "inventory", "payables", "total_debt", "debt_current",
    "cash_and_st_investments", "stockholders_equity", "total_assets",
    "deferred_revenue", "goodwill", "intangibles", "shares_diluted",
    "operating_lease_liabilities",
)


def normalize_period(code: object) -> str:
    """``Q1FY2027`` / ``q1 2027`` / ``Q12027`` all collapse to ``Q12027``.

    The guidance columns in metrics.db were written by more than one parser
    generation and disagree on the spelling of a fiscal quarter. Matching a
    guidance row to the quarter it guided depends on this.
    """
    text = str(code or "").upper().replace(" ", "").replace("-", "")
    m = re.match(r"^Q([1-4])(?:FY)?(\d{2,4})$", text)
    if not m:
        return text
    quarter, year = m.group(1), m.group(2)
    if len(year) == 2:
        year = f"20{year}"
    return f"Q{quarter}{year}"


def period_parts(code: object) -> tuple[int, int] | None:
    """Return ``(fiscal_quarter, fiscal_year)`` or None if *code* is annual."""
    m = re.match(r"^Q([1-4])(\d{4})$", normalize_period(code))
    if not m:
        return None
    return int(m.group(1)), int(m.group(2))


def fiscal_year(code: object) -> int | None:
    text = str(code or "").upper().replace(" ", "")
    m = re.match(r"^FY?(\d{4})$", text) or re.match(r"^FY(\d{2})$", text)
    if m:
        year = m.group(1)
        return int(year) if len(year) == 4 else 2000 + int(year)
    parts = period_parts(code)
    return parts[1] if parts else None


def _float(row: dict, key: str) -> float | None:
    value = row.get(key)
    try:
        return None if value is None else float(value)
    except (TypeError, ValueError):
        return None


def _derive_q4(annual: dict, filed: dict[int, dict]) -> dict | None:
    """Annual row minus its Q1-Q3, or None when a quarter is missing.

    Refusing to derive from two quarters is deliberate: the subtraction would
    silently roll a missing quarter into Q4 and print a fabricated number in
    the same style as a filed one.
    """
    if sorted(filed) != [1, 2, 3]:
        return None

    out = {
        "ticker": annual.get("ticker"),
        "company_name": annual.get("company_name"),
        "period": f"Q4{fiscal_year(annual.get('period'))}",
        "period_basis": "quarter",
        "form_type": annual.get("form_type"),
        "filing_date": annual.get("filing_date"),
        "report_date": annual.get("report_date"),
        "accession": annual.get("accession"),
        "source_url": annual.get("source_url"),
        "derived": True,
    }

    for key in _FLOW:
        total = _float(annual, key)
        if total is None:
            out[key] = None
            continue
        parts = [_float(filed[q], key) for q in (1, 2, 3)]
        out[key] = None if any(p is None for p in parts) else total - sum(parts)

    for key in _STOCK:
        out[key] = _float(annual, key)

    revenue, cogs = out.get("revenue"), out.get("cogs")
    if revenue and cogs is not None:
        out["gross_margin"] = (revenue - cogs) / revenue
        rd, sga = out.get("research_development"), out.get("sga")
        if rd is not None and sga is not None:
            out["operating_margin"] = (revenue - cogs - rd - sga) / revenue
    net_income, shares = out.get("net_income"), out.get("shares_diluted")
    if net_income is not None and shares:
        out["eps"] = net_income / shares
    # A derived quarter carries no guidance of its own: guidance is spoken on
    # a call, and there is no Q4 call to read.
    out["guidance_period"] = None
    out["guidance_revenue_low"] = None
    out["guidance_revenue_high"] = None
    out["guidance_gross_margin_low"] = None
    out["guidance_gross_margin_high"] = None
    out["guidance_source_quote"] = annual.get("guidance_source_quote")
    return out


def quarter_series(ticker: str, limit: int = 14) -> list[dict]:
    """Chronological quarters, oldest first, Q4s derived, capped at *limit*.

    Every row carries ``derived`` (bool), ``fq``/``fy`` (fiscal quarter and
    year) and ``guidance_midpoint`` — the revenue midpoint guided for *this*
    quarter, read off whichever earlier row guided it.
    """
    from storage import metrics_db

    try:
        rows = metrics_db.get_all_metrics(ticker) or []
    except Exception:
        return []

    quarters: dict[str, dict] = {}
    annuals: list[dict] = []
    for row in rows:
        row = dict(row)
        if str(row.get("period_basis") or "").lower() == "quarter":
            code = normalize_period(row.get("period"))
            if period_parts(code):
                row["derived"] = False
                quarters[code] = row
        else:
            annuals.append(row)

    # Fill each fiscal year's Q4 from its 10-K.
    for annual in annuals:
        fy = fiscal_year(annual.get("period"))
        if fy is None or f"Q4{fy}" in quarters:
            continue
        filed = {}
        for q in (1, 2, 3):
            row = quarters.get(f"Q{q}{fy}")
            if row is not None:
                filed[q] = row
        derived = _derive_q4(annual, filed)
        if derived:
            quarters[derived["period"]] = derived

    # Guidance is spoken one quarter ahead: index it by the quarter it guided.
    guided: dict[str, tuple[float, float]] = {}
    for row in list(quarters.values()) + annuals:
        target = normalize_period(row.get("guidance_period"))
        low, high = _float(row, "guidance_revenue_low"), _float(row, "guidance_revenue_high")
        if target and low is not None and high is not None:
            guided.setdefault(target, (low, high))

    ordered = sorted(quarters.values(), key=lambda r: period_parts(r["period"]) [::-1])
    for row in ordered:
        fq, fy = period_parts(row["period"])
        row["fq"], row["fy"] = fq, fy
        band = guided.get(row["period"])
        row["guidance_band"] = band
        row["guidance_midpoint"] = None if band is None else (band[0] + band[1]) / 2

    return ordered[-limit:] if limit else ordered


# ── Derived measures, each one a named calculation the note can cite ────────
def free_cash_flow(row: dict) -> float | None:
    explicit = _float(row, "free_cash_flow")
    if explicit is not None:
        return explicit
    ocf, capex = _float(row, "operating_cash_flow"), _float(row, "capex")
    if ocf is None:
        return None
    return ocf - (capex or 0.0)


def cash_conversion(row: dict) -> float | None:
    """Free cash flow over net income — the note's headline quality ratio."""
    fcf, net_income = free_cash_flow(row), _float(row, "net_income")
    if fcf is None or not net_income:
        return None
    return fcf / net_income


def dso(row: dict) -> float | None:
    """Days sales outstanding on the quarter's own revenue (91.25-day quarter)."""
    receivables, revenue = _float(row, "receivables"), _float(row, "revenue")
    if receivables is None or not revenue:
        return None
    return receivables / revenue * 91.25


def shareholder_returns(row: dict) -> float | None:
    buybacks, dividends = _float(row, "buybacks"), _float(row, "dividends_paid")
    if buybacks is None and dividends is None:
        return None
    return (buybacks or 0.0) + (dividends or 0.0)


def opex(row: dict) -> float | None:
    rd, sga = _float(row, "research_development"), _float(row, "sga")
    if rd is None and sga is None:
        return None
    return (rd or 0.0) + (sga or 0.0)


def guidance_beat(row: dict) -> tuple[float, float] | None:
    """``(absolute_gap, percent_gap)`` against the guided revenue midpoint."""
    revenue, midpoint = _float(row, "revenue"), row.get("guidance_midpoint")
    if revenue is None or not midpoint:
        return None
    gap = revenue - midpoint
    return gap, gap / midpoint * 100.0


def growth(current: dict | None, past: dict | None, key: str = "revenue") -> float | None:
    now = _float(current or {}, key)
    then = _float(past or {}, key)
    if now is None or not then:
        return None
    return (now - then) / abs(then) * 100.0


def find_quarter(series: list[dict], fq: int, fy: int) -> dict | None:
    for row in series:
        if row.get("fq") == fq and row.get("fy") == fy:
            return row
    return None


def year_ago(series: list[dict], row: dict) -> dict | None:
    if not row.get("fq"):
        return None
    return find_quarter(series, row["fq"], row["fy"] - 1)


def prior_quarter(series: list[dict], row: dict) -> dict | None:
    try:
        index = series.index(row)
    except ValueError:
        return None
    return series[index - 1] if index > 0 else None