signal-platform/scripts/backfill_earnings_events.py

"""Backfill historical earnings into a snapshot ``earnings_events`` table.

Prefers FMP bulk date-range ``earnings-calendar`` (one request per window).
On free-tier 402/403, falls back to per-symbol ``/stable/earnings`` with
resume support and request counting (≈250 req/day free tier).

Research only — writes to the local snapshot SQLite, never production Postgres.

Example
-------
  python scripts/backfill_earnings_events.py \\
      --snapshot backtest_snapshots/prod.sqlite --limit 250
"""

from __future__ import annotations

import argparse
import asyncio
import json
import sys
import time
from datetime import date, datetime, timedelta, timezone
from pathlib import Path

import httpx
from sqlalchemy import create_engine, text

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from app.ssl_bootstrap import bootstrap_ssl  # noqa: E402

bootstrap_ssl()

FMP_STABLE = "https://financialmodelingprep.com/stable"
DDL = """
CREATE TABLE IF NOT EXISTS earnings_events (
    id INTEGER PRIMARY KEY,
    symbol TEXT NOT NULL,
    announce_date TEXT NOT NULL,
    announce_time TEXT,
    eps_estimate REAL,
    eps_actual REAL,
    revenue_estimate REAL,
    revenue_actual REAL,
    source TEXT NOT NULL,
    fetched_at TEXT NOT NULL,
    UNIQUE(symbol, announce_date)
)
"""
# Side table tracks which symbols have been fully pulled (resume).
META_DDL = """
CREATE TABLE IF NOT EXISTS earnings_backfill_meta (
    symbol TEXT PRIMARY KEY,
    status TEXT NOT NULL,
    n_events INTEGER NOT NULL DEFAULT 0,
    updated_at TEXT NOT NULL,
    note TEXT
)
"""


def _parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--snapshot", default="backtest_snapshots/prod.sqlite")
    p.add_argument(
        "--from-date",
        default="2020-01-01",
        help="Bulk calendar window start (also filters per-symbol rows).",
    )
    p.add_argument(
        "--to-date",
        default=None,
        help="Bulk calendar window end (default: today).",
    )
    p.add_argument(
        "--limit",
        type=int,
        default=250,
        help="Max FMP requests this run (free-tier cushion).",
    )
    p.add_argument("--sleep", type=float, default=0.35)
    p.add_argument(
        "--force-symbol",
        action="store_true",
        help="Skip bulk attempt; go straight to per-symbol.",
    )
    p.add_argument(
        "--refetch-done",
        action="store_true",
        help="Re-fetch symbols already marked done.",
    )
    p.add_argument(
        "--provider",
        choices=("fmp", "alpha_vantage", "auto"),
        default="auto",
        help="Earnings provider. auto tries FMP bulk then FMP/AV per-symbol.",
    )
    return p.parse_args()


def _ensure_tables(engine) -> None:
    with engine.begin() as conn:
        conn.execute(text(DDL))
        conn.execute(text(META_DDL))


def _upsert_events(conn, rows: list[dict], source: str) -> int:
    if not rows:
        return 0
    now = datetime.now(timezone.utc).isoformat()
    written = 0
    for r in rows:
        conn.execute(
            text(
                """
                INSERT INTO earnings_events (
                    symbol, announce_date, announce_time,
                    eps_estimate, eps_actual, revenue_estimate, revenue_actual,
                    source, fetched_at
                ) VALUES (
                    :symbol, :announce_date, :announce_time,
                    :eps_estimate, :eps_actual, :revenue_estimate, :revenue_actual,
                    :source, :fetched_at
                )
                ON CONFLICT(symbol, announce_date) DO UPDATE SET
                    announce_time=excluded.announce_time,
                    eps_estimate=excluded.eps_estimate,
                    eps_actual=excluded.eps_actual,
                    revenue_estimate=excluded.revenue_estimate,
                    revenue_actual=excluded.revenue_actual,
                    source=excluded.source,
                    fetched_at=excluded.fetched_at
                """
            ),
            {
                "symbol": r["symbol"],
                "announce_date": r["announce_date"],
                "announce_time": r.get("announce_time"),
                "eps_estimate": r.get("eps_estimate"),
                "eps_actual": r.get("eps_actual"),
                "revenue_estimate": r.get("revenue_estimate"),
                "revenue_actual": r.get("revenue_actual"),
                "source": source,
                "fetched_at": now,
            },
        )
        written += 1
    return written


def _parse_bulk_item(item: dict) -> dict | None:
    sym = (item.get("symbol") or "").strip().upper()
    d = item.get("date") or item.get("earningsDate")
    if not sym or not d:
        return None
    return {
        "symbol": sym.replace(".", "-"),
        "announce_date": str(d)[:10],
        "announce_time": item.get("time") or item.get("announceTime"),
        "eps_estimate": _f(item.get("epsEstimated") or item.get("estimatedEarning")),
        "eps_actual": _f(item.get("epsActual") or item.get("eps")),
        "revenue_estimate": _f(item.get("revenueEstimated")),
        "revenue_actual": _f(item.get("revenueActual")),
    }


def _parse_symbol_item(item: dict, symbol: str) -> dict | None:
    d = item.get("date")
    if not d:
        return None
    return {
        "symbol": symbol.replace(".", "-").upper(),
        "announce_date": str(d)[:10],
        "announce_time": item.get("time"),
        "eps_estimate": _f(item.get("epsEstimated")),
        "eps_actual": _f(item.get("epsActual")),
        "revenue_estimate": _f(item.get("revenueEstimated")),
        "revenue_actual": _f(item.get("revenueActual")),
    }


def _f(v) -> float | None:
    if v is None or v == "":
        return None
    try:
        return float(v)
    except (TypeError, ValueError):
        return None


async def _try_bulk(
    client: httpx.AsyncClient,
    api_key: str,
    start: date,
    end: date,
    *,
    window_days: int = 30,
) -> tuple[list[dict], int, str | None]:
    """Return (rows, requests_used, error_note)."""
    rows: list[dict] = []
    reqs = 0
    cur = start
    while cur <= end:
        win_end = min(end, cur + timedelta(days=window_days - 1))
        resp = await client.get(
            f"{FMP_STABLE}/earnings-calendar",
            params={
                "from": cur.isoformat(),
                "to": win_end.isoformat(),
                "apikey": api_key,
            },
        )
        reqs += 1
        if resp.status_code in (402, 403):
            return [], reqs, f"bulk_unavailable status={resp.status_code}"
        if resp.status_code == 429:
            return rows, reqs, "rate_limited"
        resp.raise_for_status()
        data = resp.json()
        if not isinstance(data, list):
            return [], reqs, f"unexpected bulk payload type={type(data)}"
        for item in data:
            if isinstance(item, dict):
                parsed = _parse_bulk_item(item)
                if parsed:
                    rows.append(parsed)
        cur = win_end + timedelta(days=1)
    return rows, reqs, None


async def _fetch_symbol(
    client: httpx.AsyncClient, api_key: str, symbol: str
) -> list[dict]:
    resp = await client.get(
        f"{FMP_STABLE}/earnings",
        params={"symbol": symbol, "apikey": api_key},
    )
    if resp.status_code == 429:
        raise RuntimeError("rate_limited")
    if resp.status_code == 402:
        return []
    resp.raise_for_status()
    data = resp.json()
    if not isinstance(data, list):
        return []
    out: list[dict] = []
    for item in data:
        if isinstance(item, dict):
            parsed = _parse_symbol_item(item, symbol)
            if parsed:
                out.append(parsed)
    return out


async def _fetch_symbol_alpha_vantage(
    client: httpx.AsyncClient, api_key: str, symbol: str
) -> list[dict]:
    """Alpha Vantage EARNINGS — includes reportedDate (announce) + estimate/actual."""
    resp = await client.get(
        "https://www.alphavantage.co/query",
        params={"function": "EARNINGS", "symbol": symbol, "apikey": api_key},
    )
    if resp.status_code == 429:
        raise RuntimeError("rate_limited")
    resp.raise_for_status()
    data = resp.json()
    if not isinstance(data, dict):
        return []
    note = str(data.get("Note") or data.get("Information") or "")
    if "rate limit" in note.lower() or "Thank you for using Alpha Vantage" in note:
        raise RuntimeError("rate_limited")
    if data.get("Error Message"):
        return []
    quarterly = data.get("quarterlyEarnings") or []
    out: list[dict] = []
    for item in quarterly:
        if not isinstance(item, dict):
            continue
        # Prefer announce (reportedDate); fall back to fiscal end (worse PIT).
        ad = item.get("reportedDate") or item.get("fiscalDateEnding")
        if not ad:
            continue
        out.append({
            "symbol": symbol.replace(".", "-").upper(),
            "announce_date": str(ad)[:10],
            "announce_time": item.get("reportTime"),
            "eps_estimate": _f(item.get("estimatedEPS")),
            "eps_actual": _f(item.get("reportedEPS")),
            "revenue_estimate": None,
            "revenue_actual": None,
        })
    return out


async def _main() -> None:
    args = _parse_args()
    snapshot = Path(args.snapshot)
    if not snapshot.exists():
        raise SystemExit(f"Snapshot not found: {snapshot}")

    from app.config import settings

    if not settings.fmp_api_key:
        raise SystemExit("FMP_API_KEY required")

    start = date.fromisoformat(args.from_date)
    end = date.fromisoformat(args.to_date) if args.to_date else date.today()
    engine = create_engine(
        f"sqlite:///{snapshot.resolve().as_posix()}",
        future=True,
    )
    _ensure_tables(engine)

    with engine.connect() as conn:
        symbols = [
            str(r[0]).upper().replace(".", "-")
            for r in conn.execute(text("SELECT symbol FROM tickers ORDER BY symbol"))
        ]
        done = set()
        if not args.refetch_done:
            done = {
                str(r[0])
                for r in conn.execute(
                    text(
                        "SELECT symbol FROM earnings_backfill_meta "
                        "WHERE status='done' AND n_events > 0"
                    )
                )
            }

    pending = [s for s in symbols if s not in done]
    print(f"Snapshot: {snapshot}")
    print(f"Universe: {len(symbols)}; pending: {len(pending)}; done: {len(done)}")
    print(f"Window filter: {start} → {end}")
    print(f"Provider: {args.provider}")

    req_budget = int(args.limit)
    reqs_used = 0
    events_written = 0
    mode = "per_symbol"
    use_av = args.provider in ("alpha_vantage", "auto") and bool(
        getattr(settings, "alpha_vantage_api_key", "")
    )
    use_fmp = args.provider in ("fmp", "auto") and bool(settings.fmp_api_key)

    async with httpx.AsyncClient(timeout=60.0) as client:
        if (
            not args.force_symbol
            and req_budget > 0
            and use_fmp
            and args.provider != "alpha_vantage"
        ):
            print("Attempting bulk earnings-calendar…")
            bulk_rows, bulk_reqs, err = await _try_bulk(
                client, settings.fmp_api_key, start, end
            )
            reqs_used += bulk_reqs
            if err:
                print(f"  Bulk unavailable: {err} (requests={bulk_reqs})")
            else:
                # Filter to universe.
                uni = set(symbols)
                bulk_rows = [r for r in bulk_rows if r["symbol"] in uni]
                with engine.begin() as conn:
                    events_written += _upsert_events(conn, bulk_rows, "fmp_earnings_calendar")
                    for sym in symbols:
                        n = conn.execute(
                            text(
                                "SELECT COUNT(*) FROM earnings_events WHERE symbol=:s"
                            ),
                            {"s": sym},
                        ).scalar_one()
                        conn.execute(
                            text(
                                """
                                INSERT INTO earnings_backfill_meta(symbol, status, n_events, updated_at, note)
                                VALUES (:s, 'done', :n, :t, 'bulk')
                                ON CONFLICT(symbol) DO UPDATE SET
                                  status='done', n_events=excluded.n_events,
                                  updated_at=excluded.updated_at, note=excluded.note
                                """
                            ),
                            {
                                "s": sym,
                                "n": int(n),
                                "t": datetime.now(timezone.utc).isoformat(),
                            },
                        )
                mode = "bulk"
                print(f"  Bulk wrote {events_written} events; requests={bulk_reqs}")
                pending = []

        # Per-symbol fallback / completion.
        fmp_limited = False
        for sym in pending:
            if reqs_used >= req_budget:
                print(f"Request budget exhausted ({req_budget}). Resume later.")
                break
            items: list[dict] = []
            source = "fmp_earnings"
            note = "per_symbol"
            try:
                if use_fmp and not fmp_limited and args.provider != "alpha_vantage":
                    items = await _fetch_symbol(client, settings.fmp_api_key, sym)
                    source = "fmp_earnings"
                    note = "fmp_per_symbol"
                    # Empty list may mean soft-limit or no data — try AV if available.
                    if not items and use_av:
                        items = await _fetch_symbol_alpha_vantage(
                            client, settings.alpha_vantage_api_key, sym
                        )
                        source = "alpha_vantage_earnings"
                        note = "av_after_fmp_empty"
                        reqs_used += 1  # count AV call separately below too
                elif use_av:
                    items = await _fetch_symbol_alpha_vantage(
                        client, settings.alpha_vantage_api_key, sym
                    )
                    source = "alpha_vantage_earnings"
                    note = "av_per_symbol"
                else:
                    raise RuntimeError("no provider available")
            except Exception as exc:
                msg = str(exc)
                print(f"  FAIL {sym}: {msg}")
                reqs_used += 1
                if "rate_limited" in msg and note.startswith("fmp"):
                    fmp_limited = True
                with engine.begin() as conn:
                    conn.execute(
                        text(
                            """
                            INSERT INTO earnings_backfill_meta(symbol, status, n_events, updated_at, note)
                            VALUES (:s, 'error', 0, :t, :n)
                            ON CONFLICT(symbol) DO UPDATE SET
                              status='error', updated_at=excluded.updated_at, note=excluded.note
                            """
                        ),
                        {
                            "s": sym,
                            "t": datetime.now(timezone.utc).isoformat(),
                            "n": msg[:200],
                        },
                    )
                if args.sleep > 0:
                    await asyncio.sleep(args.sleep)
                continue

            reqs_used += 1
            # Keep all rows with dates on/before end — SUE needs trailing history.
            filtered = [
                r for r in items if r["announce_date"] <= end.isoformat()
            ]
            # Do NOT mark empty as done — leave pending for another provider/day.
            status = "done" if filtered else "empty"
            with engine.begin() as conn:
                n_w = _upsert_events(conn, filtered, source) if filtered else 0
                events_written += n_w
                conn.execute(
                    text(
                        """
                        INSERT INTO earnings_backfill_meta(symbol, status, n_events, updated_at, note)
                        VALUES (:s, :st, :n, :t, :note)
                        ON CONFLICT(symbol) DO UPDATE SET
                          status=excluded.status, n_events=excluded.n_events,
                          updated_at=excluded.updated_at, note=excluded.note
                        """
                    ),
                    {
                        "s": sym,
                        "st": status,
                        "n": len(filtered),
                        "t": datetime.now(timezone.utc).isoformat(),
                        "note": note,
                    },
                )
            if reqs_used % 10 == 0 or reqs_used == 1:
                print(
                    f"  progress reqs={reqs_used}/{req_budget} last={sym} "
                    f"events_batch={len(filtered)} src={source}"
                )
            # AV free tier is ~5/min or 25/day — be polite when using it.
            sleep_s = float(args.sleep)
            if source.startswith("alpha_vantage"):
                sleep_s = max(sleep_s, 12.0)
            if sleep_s > 0:
                await asyncio.sleep(sleep_s)

    with engine.connect() as conn:
        total_events = int(
            conn.execute(text("SELECT COUNT(*) FROM earnings_events")).scalar_one()
        )
        done_n = int(
            conn.execute(
                text("SELECT COUNT(*) FROM earnings_backfill_meta WHERE status='done'")
            ).scalar_one()
        )
        d_range = conn.execute(
            text("SELECT MIN(announce_date), MAX(announce_date) FROM earnings_events")
        ).fetchone()
        with_actual = int(
            conn.execute(
                text(
                    "SELECT COUNT(*) FROM earnings_events "
                    "WHERE eps_actual IS NOT NULL AND eps_estimate IS NOT NULL"
                )
            ).scalar_one()
        )

    summary = {
        "mode": mode,
        "fmp_requests": reqs_used,
        "events_written_this_run": events_written,
        "total_events": total_events,
        "symbols_done": done_n,
        "symbols_universe": len(symbols),
        "announce_date_range": {"min": d_range[0], "max": d_range[1]},
        "events_with_actual_and_estimate": with_actual,
        "budget": req_budget,
        "complete": done_n >= len(symbols),
    }
    print(json.dumps(summary, indent=2))
    out = Path("reports") / "earnings-backfill-status.json"
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
    print(f"Wrote {out}")


if __name__ == "__main__":
    asyncio.run(_main())