File size: 4,868 Bytes
7880373 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | """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
|