signal-platform/scripts/fetch_sector_etfs_to_snapshot.py

"""Fetch the 11 SPDR sector ETFs into a snapshot's ``benchmark_prices``.

Research-only. Sector ETFs are auxiliary series (like SPY) — they must not
enter the tradable ticker universe or candidate replay. Storing them in
``benchmark_prices`` keeps that invariant.

Also refreshes SPY on the same window so residual factors share a calendar.

Example
-------
  python scripts/fetch_sector_etfs_to_snapshot.py \\
      --snapshot backtest_snapshots/prod.sqlite --history-days 2200
"""

from __future__ import annotations

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

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()

from app.services.sector_map import SECTOR_ETFS  # noqa: E402


def _parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--snapshot", default="backtest_snapshots/prod.sqlite")
    p.add_argument(
        "--history-days",
        type=int,
        default=2200,
        help="Lookback calendar days (default ~6y; covers 5y snapshot + cushion).",
    )
    p.add_argument("--sleep", type=float, default=0.25)
    p.add_argument(
        "--symbols",
        default=None,
        help="Comma-separated override (default: SPY + 11 sector ETFs).",
    )
    return p.parse_args()


async def _fetch_and_upsert(
    engine,
    provider,
    symbol: str,
    start: date,
    end: date,
    *,
    sleep_s: float,
) -> int:
    from app.exceptions import ProviderError, RateLimitError

    for attempt in range(5):
        try:
            bars = await provider.fetch_ohlcv(symbol, start, end)
            break
        except RateLimitError:
            wait = min(60.0, 2.0 ** attempt)
            print(f"  rate limited {symbol}; sleep {wait:.0f}s")
            await asyncio.sleep(wait)
            bars = []
        except ProviderError as exc:
            if attempt + 1 >= 5:
                raise
            await asyncio.sleep(1.0)
            print(f"  retry {symbol}: {exc}")
            bars = []
    else:
        bars = []

    if sleep_s > 0:
        await asyncio.sleep(sleep_s)

    if not bars:
        print(f"  {symbol}: empty")
        return 0

    written = 0
    with engine.begin() as conn:
        for bar in bars:
            d = bar.date.isoformat() if hasattr(bar.date, "isoformat") else str(bar.date)
            close = float(bar.close)
            existing = conn.execute(
                text(
                    "SELECT id, close FROM benchmark_prices "
                    "WHERE symbol = :sym AND date = :d"
                ),
                {"sym": symbol, "d": d},
            ).fetchone()
            if existing is None:
                # id is INTEGER PK — let sqlite autoincrement if possible
                conn.execute(
                    text(
                        "INSERT INTO benchmark_prices (symbol, date, close) "
                        "VALUES (:sym, :d, :c)"
                    ),
                    {"sym": symbol, "d": d, "c": close},
                )
                written += 1
            elif abs(float(existing[1]) - close) > 1e-9:
                conn.execute(
                    text(
                        "UPDATE benchmark_prices SET close = :c WHERE id = :id"
                    ),
                    {"c": close, "id": int(existing[0])},
                )
                written += 1
    print(f"  {symbol}: {len(bars)} bars, {written} rows written/updated")
    return written


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
    from app.providers.alpaca import AlpacaOHLCVProvider

    if not settings.alpaca_api_key or not settings.alpaca_api_secret:
        raise SystemExit("ALPACA_API_KEY / ALPACA_API_SECRET required")

    if args.symbols:
        symbols = [s.strip().upper() for s in args.symbols.split(",") if s.strip()]
    else:
        symbols = ["SPY", *SECTOR_ETFS]

    end = date.today()
    start = end - timedelta(days=int(args.history_days))
    provider = AlpacaOHLCVProvider(settings.alpaca_api_key, settings.alpaca_api_secret)
    engine = create_engine(
        f"sqlite:///{snapshot.resolve().as_posix()}",
        future=True,
    )

    print(f"Snapshot: {snapshot}")
    print(f"Window: {start} → {end}")
    print(f"Symbols: {symbols}")

    t0 = time.monotonic()
    total = 0
    try:
        for sym in symbols:
            n = await _fetch_and_upsert(
                engine, provider, sym, start, end, sleep_s=float(args.sleep)
            )
            total += n
    finally:
        engine.dispose()

    # Summary counts
    engine = create_engine(
        f"sqlite:///{snapshot.resolve().as_posix()}",
        future=True,
    )
    try:
        with engine.connect() as conn:
            rows = conn.execute(
                text(
                    "SELECT symbol, COUNT(*), MIN(date), MAX(date) "
                    "FROM benchmark_prices GROUP BY symbol ORDER BY symbol"
                )
            ).fetchall()
    finally:
        engine.dispose()

    print(f"Done in {(time.monotonic() - t0) / 60:.1f}m; rows touched={total}")
    for sym, n, d0, d1 in rows:
        print(f"  {sym}: n={n} {d0}→{d1}")


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