From 1210ac8ef6f6c4b503d94c91cf0b414c7d5dab6e Mon Sep 17 00:00:00 2001 From: momentum dev Date: Thu, 2 Jul 2026 22:58:01 -0400 Subject: [PATCH] Ratio-corrected IEX volume: free fix for the 2-5% live undercount (estimator + audit) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With the nightly SIP backfill making all BASELINES consolidated-scale, the still-IEX live ingest would make Monday's RVOL read iex_intraday/sip_baseline ~= 25-50x too LOW and choke the scanner. Free-key SIP is only 15-min DELAYED, so each morning the true per-name capture ratio is directly measurable: - research/ingestion/volume_ratio.py: compute_ratios fetches BOTH feeds for the identical embargo-capped window (never vs our DB bars — those are scaled; that would double-correct); guardrails (min samples, clamp [1,300]); persist_ratios records per-day ratios. - ingest_live_minute_bars(volume_scale=...): scales bar volumes per symbol at ingest, marks source_provider=alpaca_iex_scaled. Missing symbol -> 1.0 (conservative: reads LOW, blocks). Constant per-symbol scale leaves cumulative VWAP unchanged (cancels in the weighted mean). - run_live_paper: "vratio" step every 300s (VOLUME_RATIO_ENABLED/INTERVAL) merges fresh ratios (a name gone thin keeps its last good ratio); step_ingest applies them. - premarket_eval nightly audit: snapshots the day's scaled ESTIMATES before the SIP truth overwrites, then logs median/worst error to data/volume_estimator_audit.tsv — the estimator's accuracy is self-measured daily, which is exactly the evidence that decides whether the $99/mo real-time SIP plan is ever needed. Live check (2026-07-02 full session): ratios are hugely name-dependent — DSY x300(clamped), CETX x286, SOFI x45, SNDQ x42, NOK x17 — validating per-name daily measurement over any global constant. 3 pure unit tests (ratio math + thin-sample skip; ingest scaling + provider marking; no-ratio unchanged). Co-Authored-By: Claude Opus 4.8 --- research/ingestion/market_data.py | 15 +++++- research/ingestion/volume_ratio.py | 75 ++++++++++++++++++++++++++++++ research/premarket_eval.py | 25 +++++++++- run_live_paper.py | 26 ++++++++++- tests/unit/test_volume_ratio.py | 55 ++++++++++++++++++++++ 5 files changed, 193 insertions(+), 3 deletions(-) create mode 100644 research/ingestion/volume_ratio.py create mode 100644 tests/unit/test_volume_ratio.py diff --git a/research/ingestion/market_data.py b/research/ingestion/market_data.py index b3e2a09..a4eae97 100644 --- a/research/ingestion/market_data.py +++ b/research/ingestion/market_data.py @@ -194,6 +194,7 @@ def ingest_live_minute_bars( symbols: list[str], lookback_minutes: int = 240, feed: str | None = None, + volume_scale: dict[str, float] | None = None, ) -> IngestionResult: """Pull recent 1-minute bars for `symbols` into the research database. @@ -201,6 +202,13 @@ def ingest_live_minute_bars( Free Alpaca keys can only see IEX data, and the SIP feed additionally embargoes the most recent 15 minutes; we end the window slightly in the past so the request never 403s. + + `volume_scale`: optional per-symbol IEX->consolidated correction (see + research.ingestion.volume_ratio) — IEX carries ~2-5% of true volume, and the + (nightly-SIP) 20d baselines are consolidated-scale, so unscaled live RVOL + reads ~25-50x low. A missing symbol scales 1.0 (conservative). Scaled rows + are marked source_provider="alpaca_iex_scaled"; the nightly SIP re-fetch + replaces the estimates with the true tape. """ result = IngestionResult() if not symbols: @@ -221,7 +229,12 @@ def ingest_live_minute_bars( return result for symbol, bars in bars_by_symbol.items(): try: - count = upsert_minute_bars(con, symbol, bars) + scale = float((volume_scale or {}).get(symbol, 1.0)) + provider = "alpaca_iex" + if scale > 1.0 and bars: + bars = [{**b, "v": int(round((b.get("v") or 0) * scale))} for b in bars] + provider = "alpaca_iex_scaled" + count = upsert_minute_bars(con, symbol, bars, source_provider=provider) result.per_symbol[symbol] = count if count: result.minute_rows += count diff --git a/research/ingestion/volume_ratio.py b/research/ingestion/volume_ratio.py new file mode 100644 index 0000000..af55685 --- /dev/null +++ b/research/ingestion/volume_ratio.py @@ -0,0 +1,75 @@ +"""Ratio-corrected IEX volume — the free fix for the 2-5% IEX undercount. + +MEASURED 2026-07-02: free-key live ingestion is IEX-only (~2-5% of consolidated volume), +while the nightly SIP backfill makes every BASELINE (20d rolling volume) consolidated-scale. +Unscaled, Monday's live RVOL = iex_intraday / sip_baseline reads ~25-50x too LOW and chokes +the scanner. Free-key SIP is only 15-min DELAYED though — so each morning we can measure +today's per-name capture ratio directly: + + ratio(sym) = sip_volume(04:00 -> now-16min) / iex_volume(same window) + +and scale live IEX bar volumes by it at ingest. Both feeds are fetched from the API for the +IDENTICAL window (never against our DB bars, which are already scaled — that would +double-correct). The nightly SIP re-fetch then replaces estimates with truth, and the +estimator's daily error is auditable (see premarket_eval). Cumulative VWAP is unaffected by +a constant per-symbol scale (it cancels in the weighted mean).""" + +from __future__ import annotations + +import logging +from datetime import date, datetime, timedelta, timezone + +logger = logging.getLogger(__name__) + +# guardrails: don't trust a ratio measured on dust, and cap the correction +MIN_IEX_SHARES = 300.0 +MIN_SIP_SHARES = 3_000.0 +RATIO_CLAMP = (1.0, 300.0) + + +def _cum_volume(bars_by_symbol: dict, symbol: str) -> float: + return float(sum(b.get("v") or 0 for b in (bars_by_symbol.get(symbol) or []))) + + +def compute_ratios(client, symbols: list[str]) -> dict[str, float]: + """Per-symbol SIP/IEX volume ratio over today's session so far (embargo-capped). + Symbols whose sample is too thin to trust are omitted (caller keeps prior/default 1.0 + — conservative: an uncorrected name reads LOW and blocks, never over-trades).""" + if not symbols: + return {} + now = datetime.now(timezone.utc) + start = f"{date.today().isoformat()}T08:00:00Z" # 04:00 ET + end = (now - timedelta(minutes=16)).strftime("%Y-%m-%dT%H:%M:%SZ") + if end <= start: + return {} + try: + iex = client.get_minute_bars(symbols, start_iso=start, end_iso=end, feed="iex") + sip = client.get_minute_bars(symbols, start_iso=start, end_iso=end, feed="sip") + except Exception as exc: # noqa: BLE001 — estimator is best-effort, never fatal + logger.warning("volume-ratio fetch failed: %s", exc) + return {} + out: dict[str, float] = {} + for s in symbols: + iv, sv = _cum_volume(iex, s), _cum_volume(sip, s) + if iv < MIN_IEX_SHARES or sv < MIN_SIP_SHARES: + continue + out[s] = max(RATIO_CLAMP[0], min(RATIO_CLAMP[1], sv / iv)) + return out + + +def persist_ratios(con, ratios: dict[str, float]) -> None: + """Record today's ratios for the nightly accuracy audit. Best-effort.""" + try: + con.execute( + """CREATE TABLE IF NOT EXISTS volume_ratios ( + symbol TEXT NOT NULL, session_date DATE NOT NULL, + ratio DOUBLE PRECISION NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (symbol, session_date))""") + for s, r in ratios.items(): + con.execute( + "INSERT INTO volume_ratios (symbol, session_date, ratio) VALUES (?, ?, ?) " + "ON CONFLICT (symbol, session_date) DO UPDATE SET ratio=EXCLUDED.ratio, " + "updated_at=now()", [s, date.today(), r]) + except Exception as exc: # noqa: BLE001 + logger.warning("volume-ratio persist failed: %s", exc) diff --git a/research/premarket_eval.py b/research/premarket_eval.py index 6da23d1..7c2231f 100644 --- a/research/premarket_eval.py +++ b/research/premarket_eval.py @@ -62,8 +62,31 @@ def main() -> None: 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()) + today = _date.today() + # estimator audit (pre): today's stored volumes are ratio-SCALED IEX estimates — + # snapshot them before the SIP truth overwrites, then measure the error. + est = dict(con.execute( + "SELECT symbol, sum(volume) FROM minute_bars WHERE session_date=? " + "AND source_provider='alpaca_iex_scaled' GROUP BY symbol", [today]).fetchall()) + n, rows = refetch_minute_session(con, _client, today) print(f"SIP refetch today: {n} symbols -> {rows} rows") + if est: + errs = [] + for sym, ev in est.items(): + tv = con.execute("SELECT sum(volume) FROM minute_bars WHERE symbol=? " + "AND session_date=?", [sym, today]).fetchone()[0] + if tv: + errs.append((sym, float(ev) / float(tv) - 1.0)) + if errs: + errs.sort(key=lambda x: abs(x[1])) + med = errs[len(errs) // 2][1] + worst = max(errs, key=lambda x: abs(x[1])) + line = (f"{today}\t{len(errs)}\t{med:+.3f}\t{worst[0]}\t{worst[1]:+.3f}") + print(f"volume-estimator audit: n={len(errs)} median_err={med:+.1%} " + f"worst={worst[0]} {worst[1]:+.1%}") + with open(os.path.join(os.path.dirname(__file__), "..", "data", + "volume_estimator_audit.tsv"), "a") as f: + f.write(line + "\n") 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 diff --git a/run_live_paper.py b/run_live_paper.py index 6be720f..6a2cd5f 100644 --- a/run_live_paper.py +++ b/run_live_paper.py @@ -366,6 +366,10 @@ def build_runtime(args: argparse.Namespace) -> dict: return { "symbols": symbols, "symbols_source": symbols_source, + # per-symbol IEX->consolidated volume correction (research.ingestion.volume_ratio): + # measured each morning from the delayed free SIP feed, applied at ingest so live + # RVOL/liquidity signals are consolidated-scale like their (nightly-SIP) baselines + "vol_ratios": {}, "client": client, "has_keys": has_keys, "store": store, @@ -524,6 +528,23 @@ def main(argv: list[str] | None = None) -> int: # lag can't flag the whole book (pure + unit-tested in runtime.halt_guard) return is_halt_gap(_symbol_bar_age(symbol), _freshest_bar_age(), gap_s) + def step_vratio(): + """Measure today's per-name IEX->SIP volume capture ratio from the delayed + (free) SIP feed and keep it for ingest scaling. Merge, never clear: a name + whose sample got too thin keeps its last good ratio for the session.""" + if not client: + return "skipped (no keys)" + from research.ingestion.volume_ratio import compute_ratios, persist_ratios + fresh = compute_ratios(client, list(symbols)) + if fresh: + rt["vol_ratios"].update(fresh) + persist_ratios(rt["research_con"], fresh) + known = rt["vol_ratios"] + if not known: + return "no ratios yet" + med = sorted(known.values())[len(known) // 2] + return f"{len(fresh)} fresh / {len(known)} known, median x{med:.0f}" + def step_ingest(): if not client: return "skipped (no keys)" @@ -532,7 +553,8 @@ def main(argv: list[str] | None = None) -> int: _ingest_first["v"] = False else: lb = refresh_lb - res = ingest_live_minute_bars(rt["research_con"], client, symbols, lookback_minutes=lb) + res = ingest_live_minute_bars(rt["research_con"], client, symbols, lookback_minutes=lb, + volume_scale=rt["vol_ratios"]) # Freshness watchdog: 45% of passes historically returned 0 rows with NO # error — indistinguishable from a feed outage. Make staleness VISIBLE # (emit + display); hard-block arming only if explicitly enabled, so a @@ -1050,6 +1072,8 @@ def main(argv: list[str] | None = None) -> int: scheduler.add("discover", step_discover, float(os.environ.get("DISCOVER_INTERVAL_SECONDS", "300")), enabled=_flag("DISCOVER_ENABLED", "1")) + scheduler.add("vratio", step_vratio, float(os.environ.get("VOLUME_RATIO_INTERVAL_SECONDS", "300")), + enabled=_flag("VOLUME_RATIO_ENABLED", "1")) scheduler.add("ingest", step_ingest, float(os.environ.get("LIVE_BARS_INTERVAL_SECONDS", "60")), enabled=_flag("LIVE_BARS_ENABLED", "1")) scheduler.add("watch", step_watch, float(os.environ.get("WATCHER_INTERVAL_SECONDS", "30")), diff --git a/tests/unit/test_volume_ratio.py b/tests/unit/test_volume_ratio.py new file mode 100644 index 0000000..e5b4085 --- /dev/null +++ b/tests/unit/test_volume_ratio.py @@ -0,0 +1,55 @@ +"""Ratio-corrected IEX volume: the free fix for the 2-5% IEX undercount. Pure tests — +a stub client provides both feeds, a stub con captures the upserted rows.""" + +from research.ingestion.market_data import ingest_live_minute_bars +from research.ingestion.volume_ratio import compute_ratios + + +def _bars(n, vol, t0="2026-07-06T13:3{i}:00Z", px=5.0): + return [{"t": f"2026-07-06T13:3{i}:00Z", "o": px, "h": px + .05, "l": px - .05, + "c": px, "v": vol, "vw": px} for i in range(n)] + + +class _Client: + """Serves distinct volumes per feed (iex thin, sip fat) + per symbol.""" + def __init__(self): + self.by_feed = { + "iex": {"GOOD": _bars(5, 100), "DUST": _bars(5, 10)}, + "sip": {"GOOD": _bars(5, 2500), "DUST": _bars(5, 200)}, + } + def get_minute_bars(self, symbols, start_iso, end_iso=None, feed=None, limit=10_000): + src = self.by_feed.get(feed or "iex", {}) + return {s: src.get(s, []) for s in symbols} + + +class _Con: + """Captures executemany rows; enough surface for upsert_minute_bars.""" + def __init__(self): + self.rows = [] + def executemany(self, sql, rows): + self.rows.extend(rows) + def execute(self, *a, **k): + return self + + +def test_compute_ratios_measures_and_skips_thin(): + r = compute_ratios(_Client(), ["GOOD", "DUST", "MISSING"]) + assert r["GOOD"] == 25.0 # 12500 sip / 500 iex + assert "DUST" not in r # sip sample below floor -> untrusted + assert "MISSING" not in r + + +def test_ingest_scales_volume_and_marks_provider(): + con = _Con() + ingest_live_minute_bars(con, _Client(), ["GOOD"], volume_scale={"GOOD": 25.0}) + flat = [v for row in con.rows for v in row] + assert 2500 in flat and 100 not in flat # 100-share bars scaled to 2500 + assert "alpaca_iex_scaled" in flat + + +def test_ingest_without_ratio_is_unchanged(): + con = _Con() + ingest_live_minute_bars(con, _Client(), ["GOOD"]) # no scale dict + flat = [v for row in con.rows for v in row] + assert 100 in flat and 2500 not in flat + assert "alpaca_iex" in flat and "alpaca_iex_scaled" not in flat