""" premarket_backfill.py — pre-market volume history for Bird's Eye, and the baseline it feeds. WHY THIS EXISTS RVOL as the dashboard computes it today is today's volume against a 20-session average of FULL days. Pre-market that ratio is 0.02-0.05x for everything, because it is comparing an hour of trading to whole sessions. Useless as a signal. A pre-market RVOL needs a same-time-of-day denominator: volume by 08:45 against the median volume by 08:45 over the trailing sessions. That needs history nobody was keeping -- so this backfills it from IBKR, which serves pre-market bars whenever TWS is up, market open or not. WHY 5-MINUTE BUCKETS IBKR caps 1-minute history at roughly 1-2 days per request, so 20 sessions x 50 names would be ~1000 requests against a 60-per-10-minutes pacing limit. 5-minute bars return a month per request: one pass, ~50 requests, about ten minutes. 66 buckets from 04:00 to 09:25 is ample resolution for "is this name unusually busy this morning". SHAPE SQLite is the store, JSON is the interface -- the same split the rest of Bird's Eye uses. `--backfill` fills the database; `--baseline` writes pm_baseline.json; `--today` scores this morning against it and writes pm_rvol.json. Neither the receiver nor the dashboard ever opens the database -- they read the JSON, exactly as they do for float_cache.json. RUN python premarket_backfill.py --backfill # ~10 min for 50 names, paced python premarket_backfill.py --backfill --symbols RKLB,AVAV python premarket_backfill.py --today --symbols INTC,HPE,HOOD # score this morning, ~12s/name python premarket_backfill.py --baseline # rewrite pm_baseline.json from the db python premarket_backfill.py --status # what is in the db python premarket_backfill.py --selftest # no TWS needed NOTES * TWS restarts nightly, roughly 23:45-00:45 ET. Do not run across that window -- it is what killed the v1.12 feed. * clientId 33, distinct from the feed's 17 and float_probe's 31. Historical requests are not market-data lines, so the 100-line cap is not in play. * Safe to re-run: rows are written with INSERT OR REPLACE, keyed on (symbol, session, bucket), so a second pass corrects what is there rather than duplicating it. A run that failed on some names can be retried with --symbols for just those; the rest are untouched. * Bar times are converted to ET and stored as ET. Storing UTC and converting later is exactly the bug that made v1.12 show the wrong session on 2026-09-10. """ import argparse, json, os, sqlite3, statistics, sys, time from datetime import datetime, date, timedelta, timezone HERE = os.path.dirname(os.path.abspath(__file__)) DB = os.path.join(HERE, "premarket.db") BASELINE_OUT = os.path.join(HERE, "pm_baseline.json") ROSTER_CACHE = os.path.join(os.path.dirname(HERE), "Bernie_s Shortlist", "core50_roster_cache.json") BUCKET_MIN = 5 PM_START = 4 * 60 # 04:00 ET, in minutes from midnight RTH_OPEN = 9 * 60 + 30 # 09:30 ET — bars at or after this are not pre-market SESSIONS = 20 # trailing sessions in the denominator MIN_SESSIONS = 5 # below this a median is not worth quoting try: from zoneinfo import ZoneInfo ET = ZoneInfo("America/New_York") except Exception: # pragma: no cover ET = timezone(timedelta(hours=-4)) # -------------------------------------------------------------------------- # pure functions — everything testable without TWS # -------------------------------------------------------------------------- def et_minute(ts) -> int | None: """Bar start as minutes from ET midnight. Returns None outside the pre-market window. `ts` may be tz-aware or naive. A naive stamp is assumed to be UTC, which is what ib_insync hands back when formatDate=2; assuming local time here is how sessions get misattributed. """ if isinstance(ts, str): ts = datetime.fromisoformat(ts.replace("Z", "+00:00")) if ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) t = ts.astimezone(ET) m = t.hour * 60 + t.minute return m if PM_START <= m < RTH_OPEN else None def session_date(ts) -> str: if isinstance(ts, str): ts = datetime.fromisoformat(ts.replace("Z", "+00:00")) if ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) return ts.astimezone(ET).date().isoformat() def to_rows(symbol: str, bars) -> list[tuple]: """Bars -> (symbol, session_date, et_minute, volume, cum_volume) for pre-market bars only. cum_volume is the running total WITHIN its own session, so a session that starts late does not inherit the previous day's total. Bars are sorted first: IBKR returns them in order, but a resumed or merged fetch need not be. """ picked = [] for b in bars: ts = getattr(b, "date", None) if not isinstance(b, dict) else b.get("date") vol = getattr(b, "volume", None) if not isinstance(b, dict) else b.get("volume") if ts is None or vol is None: continue m = et_minute(ts) if m is None: continue picked.append((session_date(ts), m, int(vol))) picked.sort() out, cur_day, running = [], None, 0 for day, m, vol in picked: if day != cur_day: cur_day, running = day, 0 running += vol out.append((symbol, day, m, vol, running)) return out def baseline_from_rows(rows: list[tuple], sessions: int = SESSIONS, min_sessions: int = MIN_SESSIONS) -> dict: """(session_date, et_minute, cum_volume) -> {et_minute: median cum_volume}. Median, not mean: one gap-up morning at 50x normal volume would drag a mean far enough to make every ordinary day look quiet. Only the most recent `sessions` days are used, and a bucket seen in fewer than `min_sessions` of them is omitted rather than quoted thin. """ by_day = {} for day, m, cum in rows: by_day.setdefault(day, {})[m] = cum recent = sorted(by_day)[-sessions:] per_bucket = {} for day in recent: for m, cum in by_day[day].items(): per_bucket.setdefault(m, []).append(cum) return {m: {"median": int(statistics.median(v)), "n": len(v)} for m, v in sorted(per_bucket.items()) if len(v) >= min_sessions} def pm_rvol(today_cum: int | None, baseline_median: int | None) -> float | None: """Pre-market RVOL: today's cumulative volume at a bucket over the median for that bucket. 1.0 is a normal morning by this time of day. This is the number the full-day `av` cannot give you pre-market -- against whole sessions everything reads 0.02-0.05x. """ if not today_cum or not baseline_median: return None return today_cum / baseline_median LAST_PM_BUCKET = ((RTH_OPEN - BUCKET_MIN) // BUCKET_MIN) * BUCKET_MIN # 09:25 def bucket_for(now_et: datetime) -> int: """The bucket a moment falls in — rounded DOWN, so 08:47 reads against the 08:45 baseline. Clamped to the pre-market window at both ends. Past the open it returns the final 09:25 bucket, so `--today` run at lunchtime or in the evening still scores the morning that happened rather than asking for a baseline that does not exist. Before 04:00 it returns the first bucket, which will simply have no volume yet. """ m = now_et.hour * 60 + now_et.minute b = (m // BUCKET_MIN) * BUCKET_MIN return max(PM_START, min(b, LAST_PM_BUCKET)) # -------------------------------------------------------------------------- # storage # -------------------------------------------------------------------------- def connect(path: str = DB) -> sqlite3.Connection: cx = sqlite3.connect(path) cx.execute("""CREATE TABLE IF NOT EXISTS pm_volume ( symbol TEXT NOT NULL, session_date TEXT NOT NULL, et_minute INTEGER NOT NULL, volume INTEGER NOT NULL, cum_volume INTEGER NOT NULL, PRIMARY KEY (symbol, session_date, et_minute))""") cx.execute("CREATE INDEX IF NOT EXISTS ix_sym_min ON pm_volume(symbol, et_minute)") cx.commit() return cx def store(cx, rows) -> int: # REPLACE, not IGNORE: a re-fetch of the same session should correct it, not be discarded. cx.executemany("INSERT OR REPLACE INTO pm_volume VALUES (?,?,?,?,?)", rows) cx.commit() return len(rows) def stored_sessions(cx, symbol: str) -> set: return {r[0] for r in cx.execute( "SELECT DISTINCT session_date FROM pm_volume WHERE symbol=?", (symbol,))} def rows_for(cx, symbol: str) -> list[tuple]: return list(cx.execute( "SELECT session_date, et_minute, cum_volume FROM pm_volume WHERE symbol=? " "ORDER BY session_date, et_minute", (symbol,))) def symbols_in(cx) -> list[str]: return [r[0] for r in cx.execute("SELECT DISTINCT symbol FROM pm_volume ORDER BY symbol")] def write_baseline(cx, path: str = BASELINE_OUT, sessions: int = SESSIONS) -> dict: """pm_baseline.json — what the receiver and dashboard actually read. Atomic, because the receiver polls and must never catch a half-written file. """ names = {} for sym in symbols_in(cx): b = baseline_from_rows(rows_for(cx, sym), sessions) if b: names[sym] = {"buckets": {str(m): v["median"] for m, v in b.items()}, "sessions": max(v["n"] for v in b.values())} doc = {"asof": datetime.now(timezone.utc).isoformat(timespec="seconds"), "source": "premarket_backfill.py", "bucket_minutes": BUCKET_MIN, "window_sessions": sessions, "count": len(names), "names": names} tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(doc, f, separators=(",", ":")) os.replace(tmp, path) return {"path": path, "symbols": len(names)} # -------------------------------------------------------------------------- # IBKR # -------------------------------------------------------------------------- def roster() -> list[str]: try: with open(ROSTER_CACHE, encoding="utf-8") as f: return json.load(f)["tickers"] except Exception as e: sys.exit(f"could not read {ROSTER_CACHE}: {e}\nPass --symbols instead.") def backfill(symbols, days: int, pace: float, db: str, tws: int, client_id: int) -> dict: import asyncio asyncio.set_event_loop(asyncio.new_event_loop()) # eventkit wants a loop at import time try: from ib_insync import IB, Stock except ImportError: sys.exit("ib_insync not installed — run: pip install ib_insync") cx = connect(db) ib = IB() try: ib.connect("127.0.0.1", tws, clientId=client_id, readonly=True, timeout=15) except Exception as e: sys.exit(f"could not connect to TWS on {tws}: {e}\n" f"Is TWS running with the API enabled, and is it outside the nightly restart?") print(f"connected to TWS on {tws} (clientId {client_id})") print(f"{len(symbols)} symbols, {days} calendar days of 5-min bars, {pace:.0f}s apart " f"— roughly {len(symbols)*pace/60:.0f} min\n") done, failed, total_rows = [], [], 0 for i, sym in enumerate(symbols, 1): try: c = Stock(sym, "SMART", "USD") [c] = ib.qualifyContracts(c) bars = ib.reqHistoricalData( c, endDateTime="", durationStr=f"{days} D", barSizeSetting="5 mins", whatToShow="TRADES", useRTH=False, formatDate=2) rows = to_rows(sym, bars) n = store(cx, rows) if rows else 0 total_rows += n sessions = len({r[1] for r in rows}) print(f" [{i:>2}/{len(symbols)}] {sym:<6} {n:>5} pre-market bars over {sessions:>2} sessions") done.append(sym) except Exception as e: print(f" [{i:>2}/{len(symbols)}] {sym:<6} FAILED: {e}") failed.append(sym) if i < len(symbols): time.sleep(pace) ib.disconnect() res = write_baseline(cx, sessions=SESSIONS) cx.close() print(f"\nstored {total_rows:,} rows · baseline written for {res['symbols']} symbols -> {res['path']}") if failed: print(f"failed ({len(failed)}): {', '.join(failed)} — rerun with --symbols to retry just these") return {"done": len(done), "failed": failed, "rows": total_rows, "baseline": res} def today_snapshot(symbols, db: str, tws: int, client_id: int, pace: float, out: str | None = None) -> dict: """Today's pre-market volume for `symbols`, scored against the stored baseline. Why this exists: the baseline is only the denominator. The numerator -- live pre-market volume -- would normally need the extended-session Pine plugin, which is on hold so that pre-market can never touch the geometry feed. This gets the same answer from the same historical API the backfill uses, as a one-shot morning read rather than a live feed. Twenty names at 12s is about four minutes, so run it once around 09:00, not on a timer. It writes pm_rvol.json and does NOT touch the regular-session feed, quotes.json or the geometry in any way. """ import asyncio asyncio.set_event_loop(asyncio.new_event_loop()) try: from ib_insync import IB, Stock except ImportError: sys.exit("ib_insync not installed — run: pip install ib_insync") cx = connect(db) base = {sym: baseline_from_rows(rows_for(cx, sym)) for sym in symbols} missing = [s for s in symbols if not base.get(s)] if missing: print(f" no baseline yet for: {', '.join(missing)} — run --backfill first") ib = IB() try: ib.connect("127.0.0.1", tws, clientId=client_id, readonly=True, timeout=15) except Exception as e: sys.exit(f"could not connect to TWS on {tws}: {e}") now = datetime.now(ET) b_now = bucket_for(now) late = (now.hour * 60 + now.minute) > LAST_PM_BUCKET print(f"connected · {now:%H:%M} ET · reading against the {b_now//60:02d}:{b_now%60:02d} bucket" + (" (past the open — scoring this morning's completed pre-market)" if late else "") + "\n") names = {} for i, sym in enumerate(symbols, 1): try: c = Stock(sym, "SMART", "USD") [c] = ib.qualifyContracts(c) bars = ib.reqHistoricalData(c, endDateTime="", durationStr="1 D", barSizeSetting="5 mins", whatToShow="TRADES", useRTH=False, formatDate=2) rows = [r for r in to_rows(sym, bars) if r[1] == now.date().isoformat()] store(cx, rows) # today is history the moment it happens cum = max((r[4] for r in rows if r[2] <= b_now), default=0) med = (base.get(sym) or {}).get(b_now, {}).get("median") rv = pm_rvol(cum, med) names[sym] = {"cum_volume": cum, "baseline": med, "pm_rvol": rv, "bucket": b_now, "sessions": (base.get(sym) or {}).get(b_now, {}).get("n")} flag = "" if rv is None else (" <-- HOT" if rv >= 3 else " <- busy" if rv >= 1.5 else "") print(f" [{i:>2}/{len(symbols)}] {sym:<6} {cum:>10,} vs median " f"{('—' if not med else f'{med:,}'):>10} " f"{('—' if rv is None else f'{rv:.2f}x'):>7}{flag}") except Exception as e: print(f" [{i:>2}/{len(symbols)}] {sym:<6} FAILED: {e}") if i < len(symbols): time.sleep(pace) ib.disconnect() doc = {"asof": datetime.now(timezone.utc).isoformat(timespec="seconds"), "session_date": now.date().isoformat(), "bucket": b_now, "bucket_et": f"{b_now//60:02d}:{b_now%60:02d}", "source": "premarket_backfill.py --today", "count": len(names), "names": names} path = out or os.path.join(HERE, "pm_rvol.json") tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(doc, f, separators=(",", ":")) os.replace(tmp, path) cx.close() ranked = sorted((v["pm_rvol"], s) for s, v in names.items() if v["pm_rvol"]) if ranked: print("\nbusiest pre-market: " + ", ".join(f"{s} {r:.2f}x" for r, s in ranked[-5:][::-1])) print(f"wrote {path}") return doc # -------------------------------------------------------------------------- # self-test # -------------------------------------------------------------------------- def selftest() -> int: fails, n = [], [0] def check(name, got, want): n[0] += 1 if got != want: fails.append(f"{name}: got {got!r}, want {want!r}") class B: def __init__(self, d, v): self.date, self.volume = d, v def utc(day, hh, mm): return datetime(2026, 9, day, hh, mm, tzinfo=timezone.utc) # --- window and timezone -------------------------------------------------- check("08:00Z is 04:00 ET, first bucket", et_minute(utc(10, 8, 0)), 240) check("13:25Z is 09:25 ET, last bucket", et_minute(utc(10, 13, 25)), 565) check("13:30Z is the open, not pre-market", et_minute(utc(10, 13, 30)), None) check("07:55Z is before 04:00 ET", et_minute(utc(10, 7, 55)), None) check("18:00Z is afternoon, not pre-market", et_minute(utc(10, 18, 0)), None) check("naive stamps are read as UTC", et_minute(datetime(2026, 9, 10, 12, 0)), 480) check("session date is the ET date", session_date(utc(10, 12, 0)), "2026-09-10") # 01:00Z is the previous ET evening — must not be filed under the UTC date check("late-UTC bar keeps its ET date", session_date(utc(11, 1, 0)), "2026-09-10") # --- cumulative within a session ----------------------------------------- bars = [B(utc(10, 8, 0), 100), B(utc(10, 8, 5), 50), B(utc(10, 13, 0), 25), B(utc(10, 14, 0), 9999), # RTH, must be dropped B(utc(11, 8, 0), 7)] # next session, resets rows = to_rows("TEST", bars) check("RTH bars dropped", len(rows), 4) check("first bar cum", rows[0][4], 100) check("second bar cum", rows[1][4], 150) check("third bar cum", rows[2][4], 175) check("new session resets cum", rows[3][4], 7) check("bar volume kept alongside cum", rows[1][3], 50) check("rows carry the symbol", rows[0][0], "TEST") # out-of-order input must still produce a correct running total check("unsorted input still cumulates", [r[4] for r in to_rows("T", list(reversed(bars)))], [100, 150, 175, 7]) # --- median baseline ------------------------------------------------------ # six sessions, bucket 240: 10,20,30,40,50,1000 -> median 35, immune to the outlier hist = [(f"2026-09-{d:02d}", 240, v) for d, v in zip(range(1, 7), [10, 20, 30, 40, 50, 1000])] b = baseline_from_rows(hist, sessions=20, min_sessions=5) check("median ignores the outlier", b[240]["median"], 35) check("session count reported", b[240]["n"], 6) thin = [(f"2026-09-{d:02d}", 300, 5) for d in range(1, 4)] check("a thin bucket is omitted", baseline_from_rows(thin, min_sessions=5), {}) # only the most recent N sessions count long_hist = [(f"2026-09-{d:02d}", 240, 1) for d in range(1, 10)] + \ [(f"2026-09-{d:02d}", 240, 100) for d in range(10, 15)] check("window keeps the recent sessions", baseline_from_rows(long_hist, sessions=5)[240]["median"], 100) # --- pre-market RVOL ------------------------------------------------------ check("twice the median reads 2.00x", pm_rvol(200, 100), 2.0) check("a quiet morning reads below 1", pm_rvol(50, 100), 0.5) check("no baseline -> no number, not a 1.0", pm_rvol(200, None), None) check("no volume yet -> no number", pm_rvol(0, 100), None) check("neither -> None", pm_rvol(None, None), None) # --- bucket rounding ------------------------------------------------------ check("08:47 reads against 08:45", bucket_for(datetime(2026, 9, 11, 8, 47)), 525) check("08:45 is its own bucket", bucket_for(datetime(2026, 9, 11, 8, 45)), 525) check("04:00 is the first bucket", bucket_for(datetime(2026, 9, 11, 4, 0)), 240) # Past the open, score the morning that happened rather than a bucket with no baseline. check("09:30 clamps to the 09:25 bucket", bucket_for(datetime(2026, 9, 11, 9, 30)), 565) check("evening clamps to 09:25 too", bucket_for(datetime(2026, 9, 11, 19, 40)), 565) check("03:00 clamps up to the first bucket", bucket_for(datetime(2026, 9, 11, 3, 0)), 240) # --- storage round trip --------------------------------------------------- import tempfile db = os.path.join(tempfile.mkdtemp(), "t.db") cx = connect(db) store(cx, to_rows("AAA", bars)) check("symbols listed", symbols_in(cx), ["AAA"]) check("sessions listed", stored_sessions(cx, "AAA"), {"2026-09-10", "2026-09-11"}) check("rows read back", len(rows_for(cx, "AAA")), 4) store(cx, to_rows("AAA", bars)) # idempotent check("re-storing does not duplicate", len(rows_for(cx, "AAA")), 4) store(cx, [("AAA", "2026-09-10", 240, 999, 999)]) # corrects in place check("re-fetch corrects a row", rows_for(cx, "AAA")[0][2], 999) cx.close() if fails: print("SELFTEST FAILED") for f in fails: print(" -", f) return 1 print(f"SELFTEST PASSED — {n[0]} assertions: window / timezone / cumulation / median / " f"bucketing / pre-market RVOL / storage") return 0 if __name__ == "__main__": ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--backfill", action="store_true") ap.add_argument("--baseline", action="store_true", help="rewrite pm_baseline.json from the db") ap.add_argument("--today", action="store_true", help="score today's pre-market against the baseline; writes pm_rvol.json") ap.add_argument("--status", action="store_true") ap.add_argument("--selftest", action="store_true") ap.add_argument("--symbols", help="comma-separated; default is the CORE50 roster cache") ap.add_argument("--days", type=int, default=30, help="calendar days to request (≈20 sessions)") ap.add_argument("--pace", type=float, default=12.0, help="seconds between symbols") ap.add_argument("--db", default=DB) ap.add_argument("--tws", type=int, default=7496) ap.add_argument("--client-id", type=int, default=33) a = ap.parse_args() if a.selftest: sys.exit(selftest()) if a.status: cx = connect(a.db) syms = symbols_in(cx) print(f"{a.db}\n{len(syms)} symbols") for s in syms: ss = stored_sessions(cx, s) b = baseline_from_rows(rows_for(cx, s)) print(f" {s:<6} {len(ss):>2} sessions {min(ss) if ss else '—'} .. {max(ss) if ss else '—'}" f" {len(b):>2} usable buckets") sys.exit(0) if a.backfill: syms = [s.strip().upper() for s in a.symbols.split(",")] if a.symbols else roster() backfill(syms, a.days, a.pace, a.db, a.tws, a.client_id) sys.exit(0) if a.today: syms = [s.strip().upper() for s in a.symbols.split(",")] if a.symbols else roster() today_snapshot(syms, a.db, a.tws, a.client_id, a.pace) sys.exit(0) if a.baseline: cx = connect(a.db) print(json.dumps(write_baseline(cx), indent=2)) sys.exit(0) ap.print_help()