OpenFin/backend/services/earnings_fetcher.py

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