signal-platform/scripts/backfill_earnings_events.py

"""Bulk-only historical earnings backfill for a local SQLite snapshot.

The job uses FMP's date-range earnings-calendar endpoint. One request covers all
symbols in a date window; per-symbol endpoints are intentionally not available
in this task runner. Successful windows are committed independently so a later
run resumes after a daily quota boundary without repeating completed windows.

Example:
  python scripts/backfill_earnings_events.py --snapshot backtest_snapshots/prod.sqlite \
      --from-date 2012-01-01 --window-days 30 --limit 250
"""

from __future__ import annotations

import argparse
import asyncio
import json
import math
import sys
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any

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"
EVENTS_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)
)
"""
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
)
"""
WINDOW_DDL = """
CREATE TABLE IF NOT EXISTS earnings_backfill_windows (
    from_date TEXT NOT NULL,
    to_date TEXT NOT NULL,
    status TEXT NOT NULL,
    requests INTEGER NOT NULL DEFAULT 0,
    rows_raw INTEGER NOT NULL DEFAULT 0,
    rows_universe INTEGER NOT NULL DEFAULT 0,
    duplicate_rows INTEGER NOT NULL DEFAULT 0,
    restated_rows INTEGER NOT NULL DEFAULT 0,
    updated_at TEXT NOT NULL,
    note TEXT,
    PRIMARY KEY(from_date, to_date)
)
"""


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--snapshot", default="backtest_snapshots/prod.sqlite")
    parser.add_argument("--from-date", default="2012-01-01")
    parser.add_argument("--to-date", default=None)
    parser.add_argument("--window-days", type=int, default=30)
    parser.add_argument("--limit", type=int, default=250)
    parser.add_argument("--sleep", type=float, default=0.35)
    parser.add_argument(
        "--refetch-windows",
        action="store_true",
        help="Re-fetch date windows already logged as done.",
    )
    return parser.parse_args()


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


def _number(value: Any) -> float | None:
    if value is None or value == "":
        return None
    try:
        result = float(value)
    except (TypeError, ValueError):
        return None
    return result if math.isfinite(result) else None


def _normalise_session(value: Any) -> str | None:
    if value is None:
        return None
    cleaned = str(value).strip().lower().replace("_", " ").replace("-", " ")
    aliases = {
        "bmo": "bmo",
        "before market open": "bmo",
        "before open": "bmo",
        "amc": "amc",
        "after market close": "amc",
        "after close": "amc",
        "during market hours": "during",
        "dmh": "during",
    }
    return aliases.get(cleaned, cleaned or None)


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


def _windows(start: date, end: date, window_days: int) -> list[tuple[date, date]]:
    if window_days < 1:
        raise ValueError("window_days must be positive")
    result: list[tuple[date, date]] = []
    cursor = start
    while cursor <= end:
        window_end = min(end, cursor + timedelta(days=window_days - 1))
        result.append((cursor, window_end))
        cursor = window_end + timedelta(days=1)
    return result


def _dedupe_bulk_rows(rows: list[dict]) -> tuple[list[dict], int, int]:
    """Prefer the most complete duplicate; use the later row as the tie-break."""
    fields = (
        "announce_time",
        "eps_estimate",
        "eps_actual",
        "revenue_estimate",
        "revenue_actual",
    )
    chosen: dict[tuple[str, str], dict] = {}
    duplicate_extras = 0
    restated = 0
    for row in rows:
        key = (str(row["symbol"]), str(row["announce_date"]))
        previous = chosen.get(key)
        if previous is None:
            chosen[key] = row
            continue
        duplicate_extras += 1
        if any(
            previous.get(field) is not None
            and row.get(field) is not None
            and previous.get(field) != row.get(field)
            for field in fields
        ):
            restated += 1
        previous_score = sum(previous.get(field) is not None for field in fields)
        new_score = sum(row.get(field) is not None for field in fields)
        if new_score >= previous_score:
            chosen[key] = row
    return list(chosen.values()), duplicate_extras, restated


def _upsert_events(conn, rows: list[dict]) -> int:
    if not rows:
        return 0
    fetched_at = datetime.now(timezone.utc).isoformat()
    statement = 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, 'fmp_earnings_calendar', :fetched_at
        )
        ON CONFLICT(symbol, announce_date) DO UPDATE SET
            announce_time=COALESCE(excluded.announce_time, earnings_events.announce_time),
            eps_estimate=COALESCE(excluded.eps_estimate, earnings_events.eps_estimate),
            eps_actual=COALESCE(excluded.eps_actual, earnings_events.eps_actual),
            revenue_estimate=COALESCE(excluded.revenue_estimate, earnings_events.revenue_estimate),
            revenue_actual=COALESCE(excluded.revenue_actual, earnings_events.revenue_actual),
            source=excluded.source,
            fetched_at=excluded.fetched_at
        """
    )
    conn.execute(statement, [{**row, "fetched_at": fetched_at} for row in rows])
    return len(rows)


async def _fetch_bulk_window(
    client: httpx.AsyncClient, api_key: str, start: date, end: date
) -> tuple[list[dict], int, str | None]:
    response = await client.get(
        f"{FMP_STABLE}/earnings-calendar",
        params={"from": start.isoformat(), "to": end.isoformat(), "apikey": api_key},
    )
    if response.status_code in (402, 403):
        return [], response.status_code, "bulk_endpoint_unavailable"
    if response.status_code == 429:
        return [], response.status_code, "daily_limit_reached"
    response.raise_for_status()
    payload = response.json()
    if not isinstance(payload, list):
        return [], response.status_code, f"unexpected_payload:{type(payload).__name__}"
    rows = []
    for item in payload:
        if isinstance(item, dict):
            parsed = _parse_bulk_item(item)
            if parsed:
                rows.append(parsed)
    return rows, response.status_code, None


def _write_window_status(
    engine,
    *,
    start: date,
    end: date,
    status: str,
    raw_n: int = 0,
    universe_n: int = 0,
    duplicate_n: int = 0,
    restated_n: int = 0,
    note: str | None = None,
) -> None:
    with engine.begin() as conn:
        conn.execute(
            text(
                """
                INSERT INTO earnings_backfill_windows(
                    from_date, to_date, status, requests, rows_raw, rows_universe,
                    duplicate_rows, restated_rows, updated_at, note
                ) VALUES (:a, :b, :status, 1, :raw, :uni, :dup, :rest, :now, :note)
                ON CONFLICT(from_date, to_date) DO UPDATE SET
                    status=excluded.status,
                    requests=earnings_backfill_windows.requests + 1,
                    rows_raw=excluded.rows_raw,
                    rows_universe=excluded.rows_universe,
                    duplicate_rows=excluded.duplicate_rows,
                    restated_rows=excluded.restated_rows,
                    updated_at=excluded.updated_at,
                    note=excluded.note
                """
            ),
            {
                "a": start.isoformat(),
                "b": end.isoformat(),
                "status": status,
                "raw": raw_n,
                "uni": universe_n,
                "dup": duplicate_n,
                "rest": restated_n,
                "now": datetime.now(timezone.utc).isoformat(),
                "note": note,
            },
        )


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()
    if start > end:
        raise SystemExit("--from-date must not be after --to-date")

    engine = create_engine(f"sqlite:///{snapshot.resolve().as_posix()}", future=True)
    _ensure_tables(engine)
    all_windows = _windows(start, end, int(args.window_days))
    with engine.connect() as conn:
        symbols = [
            str(row[0]).upper().replace(".", "-")
            for row in conn.execute(text("SELECT symbol FROM tickers ORDER BY symbol"))
        ]
        completed = {
            (str(row[0]), str(row[1]))
            for row in conn.execute(
                text(
                    "SELECT from_date, to_date FROM earnings_backfill_windows "
                    "WHERE status='done'"
                )
            )
        }
    pending = [
        window
        for window in all_windows
        if args.refetch_windows
        or (window[0].isoformat(), window[1].isoformat()) not in completed
    ]
    universe = set(symbols)
    print(f"Snapshot: {snapshot}")
    print(f"Universe: {len(symbols)} symbols")
    print(f"Window: {start} -> {end}")
    print(
        f"Bulk windows: {len(all_windows)} total; "
        f"{len(all_windows) - len(pending)} done; {len(pending)} pending"
    )
    print("Provider: FMP bulk earnings-calendar only")

    requests_this_run = 0
    rows_upserted = 0
    duplicate_rows = 0
    restated_rows = 0
    stop_note: str | None = None
    async with httpx.AsyncClient(timeout=60.0) as client:
        for index, (window_start, window_end) in enumerate(pending, 1):
            if requests_this_run >= int(args.limit):
                stop_note = "request_budget_exhausted"
                break
            try:
                raw_rows, status_code, error = await _fetch_bulk_window(
                    client, settings.fmp_api_key, window_start, window_end
                )
            except Exception as exc:
                raw_rows, status_code = [], 0
                error = f"request_error:{type(exc).__name__}:{exc}"
            requests_this_run += 1
            if error:
                _write_window_status(
                    engine,
                    start=window_start,
                    end=window_end,
                    status="error",
                    note=f"http={status_code} {error}"[:300],
                )
                stop_note = error
                print(
                    f"STOP {window_start}..{window_end}: {error} "
                    f"(http={status_code}, request={requests_this_run})"
                )
                break

            in_universe = [row for row in raw_rows if row["symbol"] in universe]
            deduped, duplicate_n, restated_n = _dedupe_bulk_rows(in_universe)
            with engine.begin() as conn:
                rows_upserted += _upsert_events(conn, deduped)
            _write_window_status(
                engine,
                start=window_start,
                end=window_end,
                status="done",
                raw_n=len(raw_rows),
                universe_n=len(deduped),
                duplicate_n=duplicate_n,
                restated_n=restated_n,
                note="bulk",
            )
            duplicate_rows += duplicate_n
            restated_rows += restated_n
            if index == 1 or index % 10 == 0 or index == len(pending):
                print(
                    f"progress windows={index}/{len(pending)} "
                    f"requests={requests_this_run}/{args.limit} "
                    f"last={window_start}..{window_end} rows={len(deduped)}"
                )
            if args.sleep > 0:
                await asyncio.sleep(float(args.sleep))

    with engine.begin() as conn:
        windows_done = int(
            conn.execute(
                text(
                    "SELECT COUNT(*) FROM earnings_backfill_windows "
                    "WHERE status='done' AND from_date >= :a AND to_date <= :b"
                ),
                {"a": start.isoformat(), "b": end.isoformat()},
            ).scalar_one()
        )
        complete = windows_done >= len(all_windows)
        if complete:
            now = datetime.now(timezone.utc).isoformat()
            for symbol in symbols:
                count = int(
                    conn.execute(
                        text(
                            "SELECT COUNT(*) FROM earnings_events "
                            "WHERE symbol=:symbol AND announce_date BETWEEN :a AND :b"
                        ),
                        {"symbol": symbol, "a": start.isoformat(), "b": end.isoformat()},
                    ).scalar_one()
                )
                conn.execute(
                    text(
                        """
                        INSERT INTO earnings_backfill_meta(symbol, status, n_events, updated_at, note)
                        VALUES (:symbol, 'done', :count, :now, 'bulk_complete')
                        ON CONFLICT(symbol) DO UPDATE SET
                            status='done', n_events=excluded.n_events,
                            updated_at=excluded.updated_at, note=excluded.note
                        """
                    ),
                    {"symbol": symbol, "count": count, "now": now},
                )
        params = {"a": start.isoformat(), "b": end.isoformat()}
        total_events = int(
            conn.execute(
                text(
                    "SELECT COUNT(*) FROM earnings_events "
                    "WHERE symbol IN (SELECT symbol FROM tickers) "
                    "AND announce_date BETWEEN :a AND :b"
                ),
                params,
            ).scalar_one()
        )
        paired_events = int(
            conn.execute(
                text(
                    "SELECT COUNT(*) FROM earnings_events "
                    "WHERE symbol IN (SELECT symbol FROM tickers) "
                    "AND announce_date BETWEEN :a AND :b "
                    "AND eps_actual IS NOT NULL AND eps_estimate IS NOT NULL"
                ),
                params,
            ).scalar_one()
        )
        date_range = conn.execute(
            text(
                "SELECT MIN(announce_date), MAX(announce_date) FROM earnings_events "
                "WHERE symbol IN (SELECT symbol FROM tickers) "
                "AND announce_date BETWEEN :a AND :b"
            ),
            params,
        ).fetchone()
        done_symbols = int(
            conn.execute(
                text("SELECT COUNT(*) FROM earnings_backfill_meta WHERE status='done'")
            ).scalar_one()
        )
        totals = conn.execute(
            text(
                "SELECT COALESCE(SUM(requests),0), COALESCE(SUM(duplicate_rows),0), "
                "COALESCE(SUM(restated_rows),0) FROM earnings_backfill_windows "
                "WHERE from_date >= :a AND to_date <= :b"
            ),
            params,
        ).fetchone()

    summary = {
        "mode": "fmp_bulk_date_range_only",
        "window": {"from": start.isoformat(), "to": end.isoformat()},
        "window_days": int(args.window_days),
        "bulk_windows_total": len(all_windows),
        "bulk_windows_done": windows_done,
        "bulk_requests_this_run": requests_this_run,
        "bulk_requests_logged_total": int(totals[0]),
        "rows_upserted_this_run": rows_upserted,
        "duplicate_rows_this_run": duplicate_rows,
        "restated_rows_this_run": restated_rows,
        "duplicate_rows_logged_total": int(totals[1]),
        "restated_rows_logged_total": int(totals[2]),
        "dedupe_policy": (
            "UNIQUE(symbol, announce_date); prefer more non-null fields, then "
            "the provider's later occurrence; non-null bulk fields replace prior "
            "values while null bulk fields retain existing values"
        ),
        "events_in_window": total_events,
        "events_with_actual_and_estimate": paired_events,
        "symbols_done": done_symbols,
        "symbols_universe": len(symbols),
        "announce_date_range": {"min": date_range[0], "max": date_range[1]},
        "request_budget": int(args.limit),
        "stop_note": stop_note,
        "complete": complete,
    }
    output = Path("reports/earnings-backfill-status.json")
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(summary, indent=2))
    print(f"Wrote {output}")


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