From f8f85b396385cba153fa37aba4b77103a4cd89e7 Mon Sep 17 00:00:00 2001 From: momentum dev Date: Thu, 2 Jul 2026 13:56:01 -0400 Subject: [PATCH] SIP re-backfill: fix the 2-5% IEX volume distortion across the whole dataset MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- alpaca_paper/client.py | 10 ++-- research/premarket_eval.py | 12 +++++ research/sip_backfill.py | 100 +++++++++++++++++++++++++++++++++++++ 3 files changed, 119 insertions(+), 3 deletions(-) create mode 100644 research/sip_backfill.py diff --git a/alpaca_paper/client.py b/alpaca_paper/client.py index 4f38f43..312214d 100644 --- a/alpaca_paper/client.py +++ b/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, }, diff --git a/research/premarket_eval.py b/research/premarket_eval.py index 1d30420..6da23d1 100644 --- a/research/premarket_eval.py +++ b/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") diff --git a/research/sip_backfill.py b/research/sip_backfill.py new file mode 100644 index 0000000..31387d9 --- /dev/null +++ b/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()