Ratio-corrected IEX volume: free fix for the 2-5% live undercount (estimator + audit)

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 <noreply@anthropic.com>

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

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)

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

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")),

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