""" Earnings calendar + EPS surprise tracker. Uses yfinance — no API key required. """ import logging import math from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional import pandas as pd import yfinance as yf logger = logging.getLogger(__name__) EARNINGS_TICKERS: Dict[str, List[str]] = { "energy": ["XOM", "CVX", "COP", "SLB", "MPC"], "metals": ["NEM", "FCX", "AA", "GOLD", "X"], "defense": ["RTX", "LMT", "GD", "NOC", "BA"], "banks": ["JPM", "GS", "BAC", "C"], "indices": ["AAPL", "MSFT", "NVDA", "AMZN", "META"], } BEAT_THRESHOLD = 5.0 MISS_THRESHOLD = -5.0 UPCOMING_DAYS = 30 HISTORY_DAYS = 90 def _safe_float(val: Any) -> Optional[float]: try: f = float(val) if math.isnan(f): return None return f except (TypeError, ValueError): return None def _fetch_ticker_earnings(ticker: str) -> Optional[Dict]: try: t = yf.Ticker(ticker) df = t.earnings_dates if df is None or df.empty: return None now = datetime.now(tz=timezone.utc) cutoff_future = now + timedelta(days=UPCOMING_DAYS) cutoff_past = now - timedelta(days=HISTORY_DAYS) upcoming = [] surprises = [] for idx, row in df.iterrows(): try: date_ts = pd.Timestamp(idx) if date_ts.tzinfo is None: date_ts = date_ts.tz_localize("UTC") else: date_ts = date_ts.tz_convert("UTC") date_dt = date_ts.to_pydatetime() date_str = date_ts.strftime("%Y-%m-%d") if date_dt > now and date_dt <= cutoff_future: eps_est = _safe_float(row.get("EPS Estimate")) upcoming.append({ "ticker": ticker, "date": date_str, "eps_estimate": eps_est, }) elif date_dt <= now and date_dt >= cutoff_past: surprise_pct = _safe_float(row.get("Surprise(%)")) reported_eps = _safe_float(row.get("Reported EPS")) eps_est = _safe_float(row.get("EPS Estimate")) if surprise_pct is None: continue surprises.append({ "ticker": ticker, "date": date_str, "surprise_pct": surprise_pct, "reported_eps": reported_eps, "eps_estimate": eps_est, }) except Exception as e: logger.debug(f"[EARNINGS] Row error {ticker}: {e}") continue return {"upcoming": upcoming, "surprises": surprises} except Exception as e: logger.warning(f"[EARNINGS] Failed to fetch {ticker}: {e}") return None def fetch_earnings_report() -> Optional[Dict]: """Fetch earnings calendar and EPS surprises for geo-relevant tickers. Returns None if no data.""" now = datetime.now(tz=timezone.utc) report_date = now.strftime("%Y-%m-%d") all_upcoming: List[Dict] = [] all_surprises: List[Dict] = [] raw_data: Dict[str, Any] = {} for sector, tickers in EARNINGS_TICKERS.items(): for ticker in tickers: result = _fetch_ticker_earnings(ticker) if result is None: continue raw_data[ticker] = {**result, "sector": sector} all_upcoming.extend(result["upcoming"]) all_surprises.extend(result["surprises"]) if not all_upcoming and not all_surprises: logger.warning("[EARNINGS] No earnings data returned") return None beats = [s for s in all_surprises if s["surprise_pct"] > BEAT_THRESHOLD] misses = [s for s in all_surprises if s["surprise_pct"] < MISS_THRESHOLD] n_beats = len(beats) n_misses = len(misses) all_upcoming_sorted = sorted(all_upcoming, key=lambda x: x["date"]) beats_sorted = sorted(beats, key=lambda x: abs(x["surprise_pct"]), reverse=True) misses_sorted = sorted(misses, key=lambda x: abs(x["surprise_pct"]), reverse=True) key_points: List[str] = [] if all_upcoming_sorted: upcoming_strs = [ f"{u['ticker']} ({u['date']})" + (f" EPS est. {u['eps_estimate']:.2f}" if u["eps_estimate"] is not None else "") for u in all_upcoming_sorted[:5] ] key_points.append(f"Upcoming earnings (next 30d): {', '.join(upcoming_strs)}") if beats_sorted: beat_strs = [ f"{b['ticker']} +{b['surprise_pct']:.1f}% ({b['date']})" for b in beats_sorted[:3] ] key_points.append(f"Recent beats: {', '.join(beat_strs)}") if misses_sorted: miss_strs = [ f"{m['ticker']} {m['surprise_pct']:.1f}% ({m['date']})" for m in misses_sorted[:3] ] key_points.append(f"Recent misses: {', '.join(miss_strs)}") if not beats and not misses: key_points.append("No significant EPS surprises (>5%) in the past 90 days") sector_beats: Dict[str, int] = {} sector_misses: Dict[str, int] = {} for b in beats: ticker = b["ticker"] for sector, tickers in EARNINGS_TICKERS.items(): if ticker in tickers: sector_beats[sector] = sector_beats.get(sector, 0) + 1 for m in misses: ticker = m["ticker"] for sector, tickers in EARNINGS_TICKERS.items(): if ticker in tickers: sector_misses[sector] = sector_misses.get(sector, 0) + 1 if n_beats > n_misses * 1.5 and n_beats > 0: signal_indices = "bullish" elif n_misses > n_beats * 1.5 and n_misses > 0: signal_indices = "bearish" else: signal_indices = "neutral" implications: List[str] = [] if all_upcoming_sorted: sectors_upcoming = set() for u in all_upcoming_sorted: for sector, tickers in EARNINGS_TICKERS.items(): if u["ticker"] in tickers: sectors_upcoming.add(sector) implications.append( f"Earnings season active — {len(all_upcoming_sorted)} reports due " f"in next 30d across {', '.join(sorted(sectors_upcoming))}" ) dominant_beat_sector = max(sector_beats, key=sector_beats.get) if sector_beats else None dominant_miss_sector = max(sector_misses, key=sector_misses.get) if sector_misses else None if dominant_beat_sector: implications.append( f"{dominant_beat_sector.capitalize()} sector leading on beats " f"({sector_beats[dominant_beat_sector]} beat(s)) — positive momentum" ) if dominant_miss_sector and dominant_miss_sector != dominant_beat_sector: implications.append( f"{dominant_miss_sector.capitalize()} sector showing misses " f"({sector_misses[dominant_miss_sector]} miss(es)) — watch for guidance cuts" ) if not implications: implications = ["Earnings data broadly in line — no sector concentration signal"] importance = 3 if (n_beats > 3 or n_misses > 3) else 2 ai_summary = ( f"Earnings tracker ({report_date}). " f"Upcoming (30d): {len(all_upcoming_sorted)} events. " f"Beats: {n_beats}, Misses: {n_misses}. " f"Signal: {signal_indices.upper()}. " + implications[0] ) return { "report_type": "earnings", "report_date": report_date, "title": f"Earnings Calendar & EPS Surprises — {report_date}", "source": "yfinance", "importance": importance, "category": "equities", "raw_data": raw_data, "key_points": key_points, "trading_implications": " | ".join(implications), "signal_energy": "neutral", "signal_metals": "neutral", "signal_indices": signal_indices, "signal_forex": "neutral", "ai_summary": ai_summary, }