SIP re-backfill: fix the 2-5% IEX volume distortion across the whole dataset

MEASURED 2026-07-02: our entire bar history was IEX-only. IEX sees ~2-5% of consolidated
volume (SNDQ 7/01: IEX 188 bars/1.5M shares vs SIP 391 bars/60.1M — 39x) and misses whole
minutes on thin names. Every volume-based signal (RVOL, liquidity floors, volume bursts,
dollar-volume strata) and several research verdicts ("untradeable dust", "thin premarket")
were computed on that sample. Daily bars had the same defect. Historical SIP is FREE on our
keys (15-min embargo) — only real-time SIP needs the paid plan.

- client.get_daily_bars: feed param (None -> settings.feed; pass "sip" for consolidated).
- research/sip_backfill.py: re-fetch minute sessions (04:00-20:00 ET) + dailies from SIP
  via the existing idempotent upserts; embargo-caps the end when refetching today;
  refresh_rolling_volume rebuild after (this morning's 20d averages were IEX-scaled too).
  CLI: full history < today, or --date D for one session.
- premarket_eval.py: nightly job now SIP-upgrades today's bars BEFORE the pm1 rebuild, so
  every finished session lands on true volume automatically (live loop stays IEX real-time
  by design — free-key SIP is embargoed 15 min).

Chain running: full backfill -> verification -> rebuild liquid/premarket/runner shadow
tracks + re-validate (their volume gates were IEX-calibrated; verdicts may change).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

alpaca_paper/client.py

@@ -397,9 +397,13 @@ class AlpacaPaperClient:
         return payload.get("news") or []
 
     def get_daily_bars(
-        self, symbols: list[str], start_iso: str, end_iso: str | None = None
+        self, symbols: list[str], start_iso: str, end_iso: str | None = None,
+        feed: str | None = None,
     ) -> dict[str, list[dict]]:
-        """Daily bars, batching symbols (cap-safe) + following pagination."""
+        """Daily bars, batching symbols (cap-safe) + following pagination.
+
+        `feed`: None -> settings.feed (iex on free keys, ~2-5% of true volume).
+        Pass "sip" for consolidated-tape volume (free keys allow historical SIP)."""
         all_bars: dict[str, list[dict]] = {}
         for chunk in self._chunked(symbols, 100):
             page_token = None
@@ -413,7 +417,7 @@ class AlpacaPaperClient:
                         "start": start_iso,
                         "end": end_iso,
                         "limit": 10_000,
-                        "feed": self.settings.feed,
+                        "feed": feed or self.settings.feed,
                         "adjustment": "raw",
                         "page_token": page_token,
                     },

research/premarket_eval.py

@@ -54,6 +54,18 @@ def _metrics(con) -> dict:
 
 def main() -> None:
     con = open_research_db("market")
+    # upgrade today's bars to the SIP consolidated tape first (live ingest is IEX-only,
+    # ~2-5% of true volume) so the pm1 rebuild + every downstream signal sees real volume
+    try:
+        from alpaca_paper.client import AlpacaPaperClient
+        from alpaca_paper.settings import AlpacaPaperSettings
+        from datetime import date as _date
+        from research.sip_backfill import refetch_minute_session
+        _client = AlpacaPaperClient(AlpacaPaperSettings.from_env())
+        n, rows = refetch_minute_session(con, _client, _date.today())
+        print(f"SIP refetch today: {n} symbols -> {rows} rows")
+    except Exception as exc:  # noqa: BLE001 — never let the refetch break the eval
+        print(f"SIP refetch skipped: {exc}")
     labeler.build_premarket(con, rebuild=True)          # rebuild from the grown DB
     m = _metrics(con)
     ts = datetime.now(timezone.utc).isoformat(timespec="seconds")

research/sip_backfill.py

@@ -0,0 +1,100 @@
+"""Re-ingest bar history from the SIP consolidated tape (free on our keys, 15-min delayed).
+
+WHY (measured 2026-07-02): our entire dataset was IEX-only — ~2-5%% of true consolidated
+volume, and IEX even misses whole minutes on thin names (SNDQ 2026-07-01: IEX 188 bars /
+1.5M shares vs SIP 391 bars / 60.1M). Every volume-based signal (RVOL, liquidity floors,
+volume bursts, dollar-volume strata) and several past research verdicts were computed on
+that 2-5%% sample. Daily bars had the same defect (get_daily_bars used the default feed).
+
+  python -m research.sip_backfill              # all history BEFORE today
+  python -m research.sip_backfill --date D     # one session (used by the nightly job)
+
+Safe by design: pure upserts through the existing ingestion helpers (ON CONFLICT rewrites
+the row), so re-running is idempotent; the live loop's IEX real-time ingest for the CURRENT
+session is intentionally untouched (SIP has a 15-min embargo on free keys) — the nightly
+job upgrades each finished session to SIP after the close."""
+
+import argparse
+import time
+from datetime import date, datetime, timedelta
+
+from dotenv import load_dotenv
+import os
+load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env"))
+
+from alpaca_paper.client import AlpacaPaperClient          # noqa: E402
+from alpaca_paper.settings import AlpacaPaperSettings      # noqa: E402
+from research.ingestion.market_data import (               # noqa: E402
+    refresh_rolling_volume, upsert_daily_bars, upsert_minute_bars)
+from research.multi_schema import open_research_db         # noqa: E402
+
+
+def refetch_minute_session(con, client, d) -> tuple[int, int]:
+    """Re-fetch one session's minute bars (04:00-20:00 ET) from SIP for every symbol
+    that already has bars that day. Returns (symbols, rows)."""
+    d = d if isinstance(d, date) else date.fromisoformat(str(d))
+    syms = [r[0] for r in con.execute(
+        "SELECT DISTINCT symbol FROM minute_bars WHERE session_date=?", [d]).fetchall()]
+    if not syms:
+        return 0, 0
+    start = f"{d.isoformat()}T08:00:00Z"                    # 04:00 ET
+    end = f"{(d + timedelta(days=1)).isoformat()}T00:00:00Z"  # 20:00 ET
+    # cap the end inside the free-key SIP embargo when refetching TODAY
+    now_embargo = datetime.utcnow() - timedelta(minutes=16)
+    if d == date.today() and now_embargo.isoformat() + "Z" < end:
+        end = now_embargo.strftime("%Y-%m-%dT%H:%M:%SZ")
+    bars_by_symbol = client.get_minute_bars(syms, start_iso=start, end_iso=end, feed="sip")
+    rows = 0
+    for sym, bars in bars_by_symbol.items():
+        if bars:
+            rows += upsert_minute_bars(con, sym, bars, source_provider="alpaca_sip")
+    return len(syms), rows
+
+
+def refetch_daily(con, client, days: int = 60) -> int:
+    """Re-fetch daily bars from SIP for every symbol in daily_bars, then rebuild the
+    20d rolling volume (which was computed off IEX-thin dailies)."""
+    syms = [r[0] for r in con.execute("SELECT DISTINCT symbol FROM daily_bars").fetchall()]
+    start = (datetime.utcnow() - timedelta(days=days * 2)).strftime("%Y-%m-%dT00:00:00Z")
+    rows = 0
+    for i in range(0, len(syms), 100):
+        chunk = syms[i:i + 100]
+        try:
+            by_symbol = client.get_daily_bars(chunk, start_iso=start, feed="sip")
+        except Exception as exc:  # noqa: BLE001
+            print(f"  ! daily chunk {i}: {exc}")
+            continue
+        for sym, bars in by_symbol.items():
+            if bars:
+                rows += upsert_daily_bars(con, sym, bars)
+    refresh_rolling_volume(con)     # full rebuild on true volume
+    return rows
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description="SIP re-backfill of bar history")
+    ap.add_argument("--date", help="one session date (YYYY-MM-DD); default = all before today")
+    ap.add_argument("--skip-daily", action="store_true")
+    args = ap.parse_args()
+    con = open_research_db("market")
+    client = AlpacaPaperClient(AlpacaPaperSettings.from_env())
+    t0 = time.time()
+    if args.date:
+        dates = [date.fromisoformat(args.date)]
+    else:
+        dates = [r[0] for r in con.execute(
+            "SELECT DISTINCT session_date FROM minute_bars WHERE session_date < ? "
+            "ORDER BY session_date", [date.today()]).fetchall()]
+    total = 0
+    for d in dates:
+        n, rows = refetch_minute_session(con, client, d)
+        total += rows
+        print(f"  {d}: {n} symbols -> {rows} SIP rows  ({time.time()-t0:.0f}s)", flush=True)
+    if not args.skip_daily:
+        drows = refetch_daily(con, client)
+        print(f"daily bars re-fetched: {drows} rows + rolling volume rebuilt", flush=True)
+    print(f"DONE: {total} minute rows across {len(dates)} sessions in {time.time()-t0:.0f}s")
+
+
+if __name__ == "__main__":
+    main()