momentum-system/backfill_gapper_days.py

#!/usr/bin/env python3
"""Targeted historical backfill: minute bars (premarket included, SIP) for the
year of recorded gapper-days in ``historical_gappers``.

Purpose: fatten n for the sample-starved queues — premarket-structure H1/H2,
pyramid_1r_v1, orderflow experiments — by giving eod_replay and the premarket
labeler real bar paths for ~260 historical sessions. Pulls ONLY each session's
recorded gappers (the population we trade), never the whole market.

    PYTHONPATH=. python backfill_gapper_days.py --limit 20      # newest 20 days
    PYTHONPATH=. python backfill_gapper_days.py                 # all recorded days

Per session: fetch 04:00-20:00 ET SIP minute bars for that day's gappers,
upsert, fetch ~40 calendar days of daily bars ending that session (the daily
screen needs prior-20-day volume), then build premarket_structure for the day.
Idempotent: sessions whose gappers already have premarket bars are skipped.
Replay/experiments consume the bars via their own runs afterwards:

    for d in $(dates); do PYTHONPATH=. python eod_replay.py $d; done
"""

from __future__ import annotations

import argparse
import sys
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

from dotenv import load_dotenv

from runtime.env_overrides import load as _load_cfg_overrides

_load_cfg_overrides()
load_dotenv()

from alpaca_paper.client import AlpacaPaperClient          # noqa: E402
from alpaca_paper.settings import AlpacaPaperSettings      # noqa: E402
from research.ingestion.market_data import (               # noqa: E402
    upsert_daily_bars,
    upsert_minute_bars,
)
from research.multi_schema import open_research_db         # noqa: E402
from research.premarket_structure import build as build_pm  # noqa: E402

ET = ZoneInfo("America/New_York")


def backfill_session(con, client, sess, symbols: list[str], feed: str) -> dict:
    start = datetime(sess.year, sess.month, sess.day, 4, 0, tzinfo=ET)
    end = datetime(sess.year, sess.month, sess.day, 20, 0, tzinfo=ET)
    by_sym = client.get_minute_bars(symbols, start_iso=start.isoformat(),
                                    end_iso=end.isoformat(), feed=feed)
    minute_rows = sum(upsert_minute_bars(con, s, b, source_provider="alpaca_sip_backfill")
                      for s, b in by_sym.items() if b)
    d_start = (start - timedelta(days=40)).isoformat()
    daily = client.get_daily_bars(symbols, start_iso=d_start,
                                  end_iso=end.isoformat(), feed=feed)
    daily_rows = sum(upsert_daily_bars(con, s, b) for s, b in daily.items() if b)
    pm = build_pm(con, session_date=str(sess))
    return {"minute": minute_rows, "daily": daily_rows, "pm_rows": pm["rows"]}


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--limit", type=int, help="only the newest N recorded sessions")
    ap.add_argument("--force", action="store_true", help="re-fetch even if bars exist")
    args = ap.parse_args(argv)

    con = open_research_db("market")
    client = AlpacaPaperClient(AlpacaPaperSettings.from_env())
    feed = client.settings.backfill_feed
    lim = f" LIMIT {int(args.limit)}" if args.limit else ""
    days = con.execute(
        "SELECT session_date, array_agg(DISTINCT symbol) FROM historical_gappers "
        f"GROUP BY session_date ORDER BY session_date DESC{lim}").fetchall()
    print(f"{len(days)} recorded gapper sessions; feed={feed}")
    done = skipped = 0
    for sess, symbols in days:
        symbols = [s for s in (symbols or []) if s]
        if not symbols:
            continue
        if not args.force:
            have = con.execute(
                "SELECT count(DISTINCT symbol) FROM minute_bars "
                "WHERE session_date = ? AND is_premarket AND symbol = ANY(?)",
                [str(sess), symbols]).fetchall()[0][0]
            if have >= max(1, len(symbols) // 2):   # most already present -> skip
                skipped += 1
                continue
        try:
            out = backfill_session(con, client, sess, symbols, feed)
            done += 1
            print(f"{sess}: {len(symbols)} syms, {out['minute']} minute rows, "
                  f"{out['daily']} daily rows, {out['pm_rows']} pm-structure rows", flush=True)
        except Exception as exc:  # noqa: BLE001 — one bad day must not kill the sweep
            print(f"{sess}: FAILED {exc}", flush=True)
    print(f"done: {done} sessions backfilled, {skipped} already present")
    return 0


if __name__ == "__main__":
    sys.exit(main())