analytics/valuation.py

"""analytics/valuation.py — public trading multiples and calendar catalysts.

Runtime-only (yfinance), cached via storage/earnings_cache.py (same TTL-cache
pattern as ingestion/analyst.py). Deliberately narrow: trading multiples and
scheduled dates only — no DCF, no fair-value opinion, no price target. That
matches the hard constraint in CLAUDE.md: public data only, no buy/sell
recommendations, no price targets, ever.

Streamlit-free and session-free, like dashboard/signal_feed.py — the render
layer (dashboard/verdict.py, dashboard/financials.py) owns all HTML.
"""
from __future__ import annotations

from typing import Optional

# Curated peer sets — small, editable MVP list (CLAUDE.md scopes this app to
# 5-10 manually curated large-caps). A ticker absent from this map simply
# renders without a peer comparison rather than failing.
PEERS: dict[str, list[str]] = {
    "NVDA": ["AMD", "AVGO"],
    "AAPL": ["MSFT", "GOOGL"],
    "MSFT": ["AAPL", "GOOGL"],
    "AMZN": ["MSFT", "GOOGL"],
    "GOOGL": ["MSFT", "META"],
    "META": ["GOOGL", "SNAP"],
    "TSLA": ["GM", "F"],
    "AMD": ["NVDA", "INTC"],
    "AVGO": ["NVDA", "QCOM"],
}


def _safe_float(v) -> Optional[float]:
    try:
        f = float(v)
        return None if f != f else f
    except (TypeError, ValueError):
        return None


def fetch_multiples(ticker: str) -> tuple[Optional[dict], Optional[str]]:
    """Fetch current public trading multiples for *ticker* via yfinance.

    Returns (data, error). data has trailing_pe, forward_pe, ev_to_sales,
    ev_to_ebitda, market_cap — any may be None if yfinance doesn't expose it
    for this ticker. Cached 6h to avoid hammering the API.
    """
    from storage.earnings_cache import get as cache_get, set as cache_set
    cache_key = f"VALUATION:v1:{ticker.upper()}"
    cached = cache_get(cache_key, ttl_hours=6)
    if cached is not None:
        return cached, None

    try:
        import yfinance as yf
    except ImportError as e:
        return None, f"yfinance not installed: {e}"

    try:
        info = yf.Ticker(ticker.upper()).info or {}
        result = {
            "trailing_pe": _safe_float(info.get("trailingPE")),
            "forward_pe": _safe_float(info.get("forwardPE")),
            "ev_to_sales": _safe_float(info.get("enterpriseToRevenue")),
            "ev_to_ebitda": _safe_float(info.get("enterpriseToEbitda")),
            "market_cap": _safe_float(info.get("marketCap")),
        }
        if not any(v is not None for v in result.values()):
            return None, "no valuation data available"
        cache_set(cache_key, result)
        return result, None
    except Exception as e:
        return None, f"yfinance valuation fetch failed: {e}"


def fetch_peer_multiples(ticker: str) -> list[dict]:
    """Fetch multiples for *ticker* plus its curated peers.

    Never raises — a peer whose fetch fails is simply omitted. First row is
    always the primary ticker (if its own fetch succeeded).
    """
    peers = PEERS.get(ticker.upper(), [])
    rows: list[dict] = []
    for t in [ticker.upper()] + peers:
        data, _err = fetch_multiples(t)
        if data:
            rows.append({"ticker": t, **data})
    return rows


def next_catalysts(ticker: str) -> dict:
    """Best-effort next scheduled dates: earnings and ex-dividend.

    Returns {"next_earnings_date": str|None, "next_ex_dividend_date": str|None}.
    Never raises; yfinance's calendar schema varies by version, so every
    lookup is defensive and a miss simply yields None for that field.
    """
    from storage.earnings_cache import get as cache_get, set as cache_set
    cache_key = f"CALENDAR:v1:{ticker.upper()}"
    cached = cache_get(cache_key, ttl_hours=24)
    if cached is not None:
        return cached

    result = {"next_earnings_date": None, "next_ex_dividend_date": None}
    try:
        import yfinance as yf
    except ImportError:
        return result

    try:
        t = yf.Ticker(ticker.upper())
        cal = t.calendar
        earnings_raw = None
        if isinstance(cal, dict):
            raw = cal.get("Earnings Date")
            if isinstance(raw, (list, tuple)) and raw:
                earnings_raw = raw[0]
            ex_div = cal.get("Ex-Dividend Date")
            if ex_div:
                result["next_ex_dividend_date"] = str(ex_div)[:10]
        elif cal is not None and not getattr(cal, "empty", True):
            index = list(getattr(cal, "index", []))
            if "Earnings Date" in index:
                row = cal.loc["Earnings Date"]
                earnings_raw = row.iloc[0] if hasattr(row, "iloc") else row
        if earnings_raw is not None:
            result["next_earnings_date"] = str(earnings_raw)[:10]
    except Exception:
        pass

    try:
        cache_set(cache_key, result)
    except Exception:
        pass
    return result