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