"""ingestion/analyst.py — analyst expectations + post-earnings price reaction. All data fetched via yfinance (free, public). Cached in storage/earnings_cache.py with a 6h TTL to avoid hammering the API. Returns (data, error) tuples consistent with ingestion/alphavantage.py. """ from __future__ import annotations import hashlib import json import re import unicodedata from datetime import datetime, timedelta, timezone from typing import Mapping, Optional from urllib.parse import urlparse from zoneinfo import ZoneInfo _TRUSTED_EVENT_SOURCE_TYPES = {"company_ir", "sec_8-k"} _EVENT_HASH_RE = re.compile(r"^[0-9a-f]{64}$") _MARKET_TIMEZONE = "America/New_York" def _canonical_event_text(value: str) -> str: value = unicodedata.normalize("NFKC", str(value or "")) value = value.replace("\r\n", "\n").replace("\r", "\n") return "\n".join(line.rstrip() for line in value.split("\n")).strip() def _event_content_hash(source_excerpt: str) -> str: return hashlib.sha256(_canonical_event_text(source_excerpt).encode("utf-8")).hexdigest() def _event_provenance_id(identity: Mapping[str, str]) -> str: payload = json.dumps(dict(identity), ensure_ascii=False, sort_keys=True, separators=(",", ":")) return "evt_" + hashlib.sha256(payload.encode("utf-8")).hexdigest()[:24] def _derived_market_timing(published_at: str) -> tuple[Optional[str], Optional[str], Optional[str]]: """Return (event_date, timing, error) derived from an offset-aware timestamp.""" try: parsed = datetime.fromisoformat(str(published_at).replace("Z", "+00:00")) except (TypeError, ValueError): return None, None, "invalid_published_at" if parsed.tzinfo is None or parsed.utcoffset() is None: return None, None, "published_at_requires_timezone" try: local = parsed.astimezone(ZoneInfo(_MARKET_TIMEZONE)) except Exception: return None, None, "market_timezone_unavailable" minutes = local.hour * 60 + local.minute if minutes < 9 * 60 + 30: timing = "before_open" elif minutes >= 16 * 60: timing = "after_close" else: return local.date().isoformat(), None, "publication_during_market_hours" return local.date().isoformat(), timing, None def make_earnings_event_provenance( *, ticker: str, published_at: str, source_type: str, source_url: str, source_excerpt: str, ) -> dict[str, str]: """Create a content-addressed provenance record for an earnings release. This helper guarantees internal integrity only. ``fetch_price_reaction`` still validates the source type, URL, timestamp-derived date/timing, ticker, hash, and ID before enabling any return comparison. """ if not str(ticker).strip(): raise ValueError("ticker is required") if source_type not in _TRUSTED_EVENT_SOURCE_TYPES: raise ValueError(f"unsupported event source_type: {source_type!r}") parsed_url = urlparse(source_url) if parsed_url.scheme != "https" or not parsed_url.hostname: raise ValueError("source_url must be an absolute HTTPS URL") if ( source_type == "sec_8-k" and not (parsed_url.hostname == "sec.gov" or parsed_url.hostname.endswith(".sec.gov")) ): raise ValueError("sec_8-k provenance must use a sec.gov URL") event_date, event_timing, timing_error = _derived_market_timing(published_at) if timing_error or not event_date or not event_timing: raise ValueError(f"Cannot derive an earnings-event window: {timing_error}") excerpt = _canonical_event_text(source_excerpt) if not excerpt: raise ValueError("source_excerpt is required") digest = _event_content_hash(excerpt) identity = { "ticker": ticker.upper(), "event_date": event_date, "event_timing": event_timing, "source_type": source_type, "source_url": source_url, "published_at": published_at, "content_hash": digest, } return { **identity, "source_excerpt": excerpt, "evidence_id": _event_provenance_id(identity), } def _validate_earnings_event_provenance( provenance: Optional[Mapping[str, object]], *, ticker: str, event_date: str, event_timing: str, ) -> tuple[Optional[dict[str, str]], str]: """Validate event provenance without trusting caller-supplied status flags.""" if not isinstance(provenance, Mapping): return None, "missing_event_provenance" required = { "ticker", "event_date", "event_timing", "source_type", "source_url", "published_at", "source_excerpt", "content_hash", "evidence_id", } if any(not provenance.get(field) for field in required): return None, "incomplete_event_provenance" normalized = {field: str(provenance[field]) for field in required} if normalized["ticker"].upper() != ticker.upper(): return None, "event_ticker_mismatch" if normalized["event_date"] != event_date: return None, "event_date_mismatch" if normalized["event_timing"] != event_timing: return None, "event_timing_mismatch" if normalized["source_type"] not in _TRUSTED_EVENT_SOURCE_TYPES: return None, "untrusted_event_source_type" parsed_url = urlparse(normalized["source_url"]) if parsed_url.scheme != "https" or not parsed_url.hostname: return None, "event_source_url_must_be_https" if ( normalized["source_type"] == "sec_8-k" and not (parsed_url.hostname == "sec.gov" or parsed_url.hostname.endswith(".sec.gov")) ): return None, "sec_event_source_must_use_sec_gov" derived_date, derived_timing, timing_error = _derived_market_timing(normalized["published_at"]) if timing_error: return None, timing_error if derived_date != normalized["event_date"]: return None, "published_at_event_date_mismatch" if derived_timing != normalized["event_timing"]: return None, "published_at_event_timing_mismatch" digest = _event_content_hash(normalized["source_excerpt"]) if not _EVENT_HASH_RE.fullmatch(normalized["content_hash"]): return None, "invalid_event_content_hash" if digest != normalized["content_hash"]: return None, "event_content_hash_mismatch" identity = { "ticker": normalized["ticker"].upper(), "event_date": normalized["event_date"], "event_timing": normalized["event_timing"], "source_type": normalized["source_type"], "source_url": normalized["source_url"], "published_at": normalized["published_at"], "content_hash": normalized["content_hash"], } if normalized["evidence_id"] != _event_provenance_id(identity): return None, "event_evidence_id_mismatch" return normalized, "verified" def _safe_float(v) -> Optional[float]: try: f = float(v) return None if f != f else f except (TypeError, ValueError): return None def _pick_period_row(df, prefer: list[str]): """Return the first row from df whose index matches one of the preferred period codes.""" if df is None or getattr(df, "empty", True): return None for code in prefer: if code in df.index: return df.loc[code] return None def _pick_period_row_with_code(df, prefer: list[str]): if df is None or getattr(df, "empty", True): return None, None for code in prefer: if code in df.index: return df.loc[code], code return None, None def fetch_analyst_estimates( ticker: str, target_period: Optional[str] = None, ) -> tuple[Optional[dict], Optional[str]]: """Fetch analyst consensus EPS, revenue estimates, and 30-day revision %. Returns dict with keys: consensus_eps_est, consensus_rev_est, estimate_revision_30d_pct Each may be None if yfinance does not expose it for this ticker. """ from storage.earnings_cache import get as cache_get, set as cache_set cache_key = f"ANALYST:v2:{ticker.upper()}:{target_period or 'unspecified'}" 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: t = yf.Ticker(ticker.upper()) eps_row, eps_code = _pick_period_row_with_code( getattr(t, "earnings_estimate", None), ["0q", "+1q", "0y"] ) rev_row, rev_code = _pick_period_row_with_code( getattr(t, "revenue_estimate", None), ["0q", "+1q", "0y"] ) trend_row, trend_code = _pick_period_row_with_code( getattr(t, "eps_trend", None), ["0q", "+1q", "0y"] ) consensus_eps = _safe_float(eps_row.get("avg")) if eps_row is not None else None consensus_rev = _safe_float(rev_row.get("avg")) if rev_row is not None else None revision_30d_pct: Optional[float] = None if trend_row is not None: current = _safe_float(trend_row.get("current")) ago30 = _safe_float(trend_row.get("30daysAgo")) if current is not None and ago30 is not None and ago30 != 0: revision_30d_pct = round((current - ago30) / abs(ago30) * 100, 2) result = { "consensus_eps_est": consensus_eps, "consensus_rev_est": consensus_rev, # USD raw "estimate_revision_30d_pct": revision_30d_pct, "provider_period_codes": { "eps": eps_code, "revenue": rev_code, "trend": trend_code, }, "target_period": target_period, "as_of": datetime.now(timezone.utc).isoformat(), # yfinance exposes relative forward periods, not the historical # pre-release snapshot required for an actual-vs-consensus claim. "period_aligned": False, "comparison_allowed": False, "alignment_status": "UNVERIFIED_CURRENT_CONSENSUS", } cache_set(cache_key, result) return result, None except Exception as e: return None, f"yfinance analyst fetch failed: {e}" def fetch_price_reaction( ticker: str, filing_date: str, *, event_kind: str = "filing_date_proxy", event_timing: str = "unknown", event_provenance: Optional[Mapping[str, object]] = None, ) -> tuple[Optional[dict], Optional[str]]: """Compute correctly-windowed returns for a verified earnings event. Strings supplied via ``event_kind``/``event_timing`` are never sufficient to mark an event aligned. A content-addressed ``event_provenance`` record is required. Without one, returns are null and the result is explicitly fail-closed. Returns {'d1_pct': float|None, 'd5_pct': float|None, 'since_release_pct': float}. """ from storage.earnings_cache import get as cache_get, set as cache_set try: d = datetime.strptime(filing_date[:10], "%Y-%m-%d") except (TypeError, ValueError, IndexError): return None, f"invalid filing_date format: {filing_date!r}" provenance = None provenance_reason = "event_kind_or_timing_not_earnings_release" if event_kind == "earnings_release" and event_timing in {"before_open", "after_close"}: provenance, provenance_reason = _validate_earnings_event_provenance( event_provenance, ticker=ticker, event_date=filing_date[:10], event_timing=event_timing, ) if provenance is None: return { "d1_pct": None, "d5_pct": None, "since_release_pct": None, "event_date": filing_date[:10], "event_kind": event_kind, "event_timing": event_timing, "event_aligned": False, "comparison_allowed": False, "alignment_status": "UNVERIFIED_EVENT_PROVENANCE", "alignment_reason": provenance_reason, "provenance_verified": False, "event_provenance_id": None, "event_source_type": None, "event_source_url": None, "baseline_trading_date": None, "d1_trading_date": None, "d5_trading_date": None, }, None cache_key = ( f"PRICEREACT:v3:{ticker.upper()}:{filing_date}:" f"{event_timing}:{provenance['evidence_id']}" ) cached = cache_get(cache_key, ttl_hours=24) if cached is not None: return cached, None try: import yfinance as yf import pandas as pd except ImportError as e: return None, f"yfinance/pandas not installed: {e}" # Ten calendar days guarantee room for a prior trading-session baseline. start = (d - timedelta(days=10)).strftime("%Y-%m-%d") end = (datetime.today() + timedelta(days=1)).strftime("%Y-%m-%d") try: hist = yf.download(ticker.upper(), start=start, end=end, progress=False, auto_adjust=True) if hist.empty: return None, "no price history returned" if isinstance(hist.columns, pd.MultiIndex): hist.columns = hist.columns.get_level_values(0) closes = hist["Close"].dropna() if closes.empty: return None, "no close prices in range" event_positions = [i for i, timestamp in enumerate(closes.index) if timestamp.date() == d.date()] if not event_positions: result = { "d1_pct": None, "d5_pct": None, "since_release_pct": None, "event_date": filing_date[:10], "event_kind": event_kind, "event_timing": event_timing, "event_aligned": False, "comparison_allowed": False, "alignment_status": "VERIFIED_EVENT_NON_TRADING_DATE", "alignment_reason": "event_date_not_a_trading_session", "provenance_verified": True, "event_provenance_id": provenance["evidence_id"], "event_source_type": provenance["source_type"], "event_source_url": provenance["source_url"], "baseline_trading_date": None, "d1_trading_date": None, "d5_trading_date": None, } cache_set(cache_key, result) return result, None event_idx = event_positions[0] if event_timing == "before_open": baseline_idx = event_idx - 1 d1_idx = event_idx d5_idx = event_idx + 4 else: # after_close baseline_idx = event_idx d1_idx = event_idx + 1 d5_idx = event_idx + 5 if baseline_idx < 0: return None, "insufficient price history for event baseline" baseline = float(closes.iloc[baseline_idx]) d1 = ( (float(closes.iloc[d1_idx]) / baseline - 1) * 100 if d1_idx < len(closes) else None ) d5 = ( (float(closes.iloc[d5_idx]) / baseline - 1) * 100 if d5_idx < len(closes) else None ) since = (float(closes.iloc[-1]) / baseline - 1) * 100 result = { "d1_pct": round(d1, 2) if d1 is not None else None, "d5_pct": round(d5, 2) if d5 is not None else None, "since_release_pct": round(since, 2), "event_date": filing_date[:10], "event_kind": event_kind, "event_timing": event_timing, "event_aligned": True, "comparison_allowed": d1 is not None, "alignment_status": "VERIFIED_EARNINGS_EVENT", "alignment_reason": "verified_provenance_and_trading_window", "provenance_verified": True, "event_provenance_id": provenance["evidence_id"], "event_source_type": provenance["source_type"], "event_source_url": provenance["source_url"], "baseline_trading_date": closes.index[baseline_idx].date().isoformat(), "d1_trading_date": ( closes.index[d1_idx].date().isoformat() if d1_idx < len(closes) else None ), "d5_trading_date": ( closes.index[d5_idx].date().isoformat() if d5_idx < len(closes) else None ), } cache_set(cache_key, result) return result, None except Exception as e: return None, f"price reaction fetch failed: {e}" def post_earnings_returns_batch( ticker: str, dates: list[str], *, event_provenance_by_date: Optional[Mapping[str, Mapping[str, object]]] = None, ) -> dict[str, dict]: """Compute d1/d5 returns only for provenance-verified earnings dates. Used by dashboard/financials.py to render the surprise history. Returns {date_str: {'d1': float|None, 'd5': float|None}}. Returns are expressed as decimals (NOT percentages — the caller multiplies by 100). Missing/invalid provenance produces no row; filing dates are never treated as earnings dates implicitly. """ if not dates or not isinstance(event_provenance_by_date, Mapping): return {} verified: dict[str, dict[str, str]] = {} for date_str in dates: candidate = event_provenance_by_date.get(date_str) timing = str(candidate.get("event_timing", "")) if isinstance(candidate, Mapping) else "" if timing not in {"before_open", "after_close"}: continue provenance, _ = _validate_earnings_event_provenance( candidate, ticker=ticker, event_date=date_str[:10], event_timing=timing, ) if provenance is not None: verified[date_str] = provenance if not verified: return {} try: import yfinance as yf import pandas as pd except ImportError: return {} try: parsed = [datetime.strptime(date_str[:10], "%Y-%m-%d") for date_str in verified] earliest = min(parsed) latest = max(parsed) start = (earliest - timedelta(days=10)).strftime("%Y-%m-%d") end = (latest + timedelta(days=14)).strftime("%Y-%m-%d") hist = yf.download(ticker.upper(), start=start, end=end, progress=False, auto_adjust=True) if hist.empty: return {} if isinstance(hist.columns, pd.MultiIndex): hist.columns = hist.columns.get_level_values(0) closes = hist["Close"].dropna() if closes.empty: return {} result: dict[str, dict] = {} for date_str, provenance in verified.items(): try: d = datetime.strptime(date_str[:10], "%Y-%m-%d") event_positions = [ i for i, timestamp in enumerate(closes.index) if timestamp.date() == d.date() ] if not event_positions: continue event_idx = event_positions[0] if provenance["event_timing"] == "before_open": baseline_idx, d1_idx, d5_idx = event_idx - 1, event_idx, event_idx + 4 else: baseline_idx, d1_idx, d5_idx = event_idx, event_idx + 1, event_idx + 5 if baseline_idx < 0: continue baseline = float(closes.iloc[baseline_idx]) d1 = float(closes.iloc[d1_idx]) / baseline - 1 if d1_idx < len(closes) else None d5 = float(closes.iloc[d5_idx]) / baseline - 1 if d5_idx < len(closes) else None result[date_str] = { "d1": d1, "d5": d5, "event_aligned": True, "event_timing": provenance["event_timing"], "event_provenance_id": provenance["evidence_id"], } except Exception: continue return result except Exception: return {}