fix: replay universe = scanner-discovered + historical gappers; add year-scale targeted backfill

eod_replay's universe was today's top-100 by ABSOLUTE volume — small-cap
gappers never crack that list, so 2026-07-28 replay found 1 gapper on a
day the live loop signaled 8 names. Universe is now the union of (a) the
session's symbol_discovered events, (b) historical_gappers for that date,
(c) most-actives only on a same-day replay (past-date most-actives would
be look-ahead). Verified: 7/28 replay now screens 6 gappers / 4 trades,
matching the live day. Historical sessions also fetch that day's
04:00-20:00 ET SIP bars explicitly instead of a from-now lookback window
that pulled the wrong day.

backfill_gapper_days.py: for each of the ~260 recorded gapper sessions,
pull SIP minute bars (premarket incl.) + 40d of daily bars for that day's
gappers only, then build premarket_structure. Idempotent (skips sessions
already covered). Fattens n for premarket H1/H2, pyramid_1r_v1, and the
orderflow experiments.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backfill_gapper_days.py

@@ -0,0 +1,102 @@
+#!/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())

eod_replay.py

@@ -38,6 +38,7 @@ from research.ingestion.market_data import (
     classify_session,
     ingest_daily_history,
     ingest_live_minute_bars,
+    upsert_minute_bars,
 )
 from research.ingestion.signals import scan_gappers
 from research.multi_schema import open_research_db
@@ -86,6 +87,41 @@ def _daily_gappers(con, sess, universe):
     return out
 
 
+def _session_universe(con, client, sess) -> list[str]:
+    """Candidate names for a session, from three sources union-ed:
+
+    (a) what the LIVE scanner discovered that day (events) — the top-100
+        absolute-volume list misses small-cap gappers entirely: 2026-07-28 the
+        replay found 1 gapper on a day the live loop signaled 8 names, because
+        MUZ/DAMD/BIYA-class names never crack the most-actives list;
+    (b) historical_gappers — the year-scale gapper record, covers dates before
+        the live loop existed (backfill path);
+    (c) today's most-actives — ONLY for a same-day replay (the API returns the
+        CURRENT list; unioning it into a past date would be look-ahead).
+    The shared population screen still runs after this — the union only widens
+    candidates, it never bypasses the gate."""
+    syms: set[str] = set()
+    try:
+        rows = con.execute(
+            "SELECT DISTINCT payload_json::json->>'symbol' FROM events "
+            "WHERE event_type = 'symbol_discovered' AND timestamp::date = ?",
+            [str(sess)]).fetchall()
+        syms |= {r[0] for r in rows if r and r[0]}
+    except Exception as exc:  # noqa: BLE001 — events table may not exist in research-only DBs
+        print(f"[universe] discovered-symbols lookup skipped: {exc}")
+    try:
+        rows = con.execute(
+            "SELECT DISTINCT symbol FROM historical_gappers WHERE session_date = ?",
+            [str(sess)]).fetchall()
+        syms |= {r[0] for r in rows if r and r[0]}
+    except Exception as exc:  # noqa: BLE001
+        print(f"[universe] historical_gappers lookup skipped: {exc}")
+    if sess == datetime.now(timezone.utc).date():
+        actives = client.get_most_actives(top=100, by="volume")
+        syms |= {a["symbol"] for a in actives if a.get("symbol")}
+    return sorted(syms)
+
+
 def _exit_cfg() -> ExitConfig:
     return ExitConfig(
         target_r=float(os.environ.get("TRADING_REWARD_MULTIPLE", "10")),
@@ -104,11 +140,12 @@ def main(argv=None):
     con = open_research_db("market")
     client = AlpacaPaperClient(AlpacaPaperSettings.from_env())
 
-    # 1) full active universe (vs the ~20 the live loop watched)
-    actives = client.get_most_actives(top=100, by="volume")
-    universe = sorted({a["symbol"] for a in actives if a.get("symbol")})
+    # 1) session universe: scanner-discovered + historical gapper record +
+    #    (same-day only) most-actives — see _session_universe
+    universe = _session_universe(con, client, sess)
     print(f"=== EOD REPLAY {sess} ===")
-    print(f"universe: {len(universe)} most-active names (live loop watches ~20)")
+    print(f"universe: {len(universe)} candidates (scanner-discovered + "
+          f"historical_gappers + same-day most-actives)")
 
     # 2) cheap DAILY backfill -> screen gappers from daily bars (no need to pull
     #    minute bars for all 100; only the gappers need them)
@@ -123,10 +160,28 @@ def main(argv=None):
     # dataset. Use the SIP (full-volume) feed so rvol/$-volume in the replay
     # reflect REAL volume — the live IEX feed sees only ~2-5% of it, which was
     # silently corrupting every volume-based backtest conclusion.
-    ing = ingest_live_minute_bars(con, client, gapper_syms, lookback_minutes=440,
-                                  feed=client.settings.backfill_feed)
-    print(f"backfilled {ing.minute_rows} minute rows ({client.settings.backfill_feed}) "
-          f"across {len(ing.symbols)} gappers\n")
+    if sess == datetime.now(timezone.utc).date():
+        ing = ingest_live_minute_bars(con, client, gapper_syms, lookback_minutes=440,
+                                      feed=client.settings.backfill_feed)
+        n_rows, n_syms = ing.minute_rows, len(ing.symbols)
+    else:
+        # historical session: a from-now lookback window would fetch the WRONG
+        # day — fetch that session's 04:00-20:00 ET explicitly (premarket incl.)
+        from zoneinfo import ZoneInfo
+        et = ZoneInfo("America/New_York")
+        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(
+            gapper_syms, start_iso=start.isoformat(), end_iso=end.isoformat(),
+            feed=client.settings.backfill_feed) if gapper_syms else {}
+        n_rows, n_syms = 0, 0
+        for sym, bars in by_sym.items():
+            if bars:
+                n_rows += upsert_minute_bars(con, sym, bars,
+                                             source_provider="alpaca_sip_backfill")
+                n_syms += 1
+    print(f"backfilled {n_rows} minute rows ({client.settings.backfill_feed}) "
+          f"across {n_syms} gappers\n")
     keep = set(gapper_syms)
     # Population membership was already decided by the shared gate in
     # _daily_gappers; scan_gappers here only computes gap/rvol/vwap for ranking, so