Ingest: per-symbol session backfill; never arm without the true open

Evening audit of 2026-07-08's post-fix entries (TVRD +4% true gap,
FRTT — both fired as "20% gappers"): the morning's static-gap fix was
defeated by a data hole. ingest_live_minute_bars' full-session first
pass was a GLOBAL one-shot, so symbols discovered mid-session only ever
received the 20-minute refresh window — their earliest known bar was a
mid-day bar, which became both their "session open" (fake overnight
gap) and the base of their "opening range" (fake ORB trigger). Exactly
the failure class the population gate exists to prevent.

Fixes: (1) step_ingest tracks per-symbol first sight — new names get
the full-session lookback, known names the incremental refresh;
(2) _session_open only trusts a first-RTH bar stamped <= 09:33 ET and
is the sole source for the static gap; (3) if the session is underway
and a candidate's true open is not yet ingested, it cannot arm (trigger
withheld until the backfill lands, typically the next pass).

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

run_live_paper.py

@@ -41,6 +41,7 @@ from alpaca_paper.settings import AlpacaPaperSettings
 from alpaca_paper.sync import AlpacaPaperSync
 from api.main import DashboardState, serve_in_background
 from research.ingestion.market_data import (
+    IngestionResult,
     classify_session,
     discover_active_symbols,
     ingest_daily_history,
@@ -388,6 +389,40 @@ def build_runtime(args: argparse.Namespace) -> dict:
     }
 
 
+
+_SESSION_OPEN_CACHE: dict = {}
+
+
+def _session_open(con, symbol: str, session_date) -> float | None:
+    """First regular-hours bar's open for symbol/day — the TRUE session open.
+
+    query_minute_bars returns a recent WINDOW, so its first bar drifts intraday;
+    this queries the actual 09:30 bar once and caches it (immutable once printed).
+    """
+    key = (symbol, str(session_date))
+    v = _SESSION_OPEN_CACHE.get(key)
+    if v is not None:
+        return v
+    try:
+        r = con.execute(
+            "SELECT open, timestamp FROM minute_bars WHERE symbol=? AND session_date=? "
+            "AND is_regular_hours ORDER BY timestamp LIMIT 1", [symbol, session_date]
+        ).fetchall()
+        if r:
+            # trust it ONLY if it is really the opening bar (<= 09:33 ET = 13:33Z):
+            # mid-day-discovered names briefly have bars starting mid-session, and
+            # treating that first bar as "the open" is how TVRD/FRTT armed as fake
+            # 20% gappers on 2026-07-08.
+            ts = str(r[0][1])
+            hhmm = ts[11:16]
+            if "13:30" <= hhmm <= "13:33":
+                _SESSION_OPEN_CACHE[key] = float(r[0][0])
+                return _SESSION_OPEN_CACHE[key]
+    except Exception:  # noqa: BLE001 — fail-open; gap falls back to window/day_open
+        pass
+    return None
+
+
 def main(argv: list[str] | None = None) -> int:
     args = parse_args(argv)
     rt = build_runtime(args)
@@ -552,16 +587,30 @@ def main(argv: list[str] | None = None) -> int:
         med = sorted(known.values())[len(known) // 2]
         return f"{len(fresh)} fresh / {len(known)} known, median x{med:.0f}"
 
+    _ingested_syms: set = set()
+
     def step_ingest():
         if not client:
             return "skipped (no keys)"
-        if _ingest_first["v"]:
-            lb = _first_pass_lookback()
-            _ingest_first["v"] = False
-        else:
-            lb = refresh_lb
-        res = ingest_live_minute_bars(rt["research_con"], client, symbols, lookback_minutes=lb,
-                                      volume_scale=rt["vol_ratios"])
+        # PER-SYMBOL first pass: a global one-shot meant symbols discovered
+        # mid-session only ever got the 20-min refresh window — no 09:30 bars, so
+        # their opening range and overnight gap were computed from mid-day bars
+        # (the 2026-07-08 TVRD/FRTT fake-gapper hole). New names get the full
+        # session; known names get the incremental refresh.
+        new_syms = [s for s in symbols if s not in _ingested_syms]
+        known_syms = [s for s in symbols if s in _ingested_syms]
+        res = IngestionResult() if not known_syms else ingest_live_minute_bars(
+            rt["research_con"], client, known_syms, lookback_minutes=refresh_lb,
+            volume_scale=rt["vol_ratios"])
+        if new_syms:
+            res_new = ingest_live_minute_bars(
+                rt["research_con"], client, new_syms,
+                lookback_minutes=_first_pass_lookback(),
+                volume_scale=rt["vol_ratios"])
+            res.minute_rows += res_new.minute_rows
+            res.errors.extend(res_new.errors)
+            res.per_symbol.update(res_new.per_symbol)
+            _ingested_syms.update(new_syms)
         # 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
@@ -762,9 +811,23 @@ def main(argv: list[str] | None = None) -> int:
             # definition. The scanner's g.gap_pct is DYNAMIC (latest price vs prev
             # close): intraday rallies masqueraded as gappers through it (TVRD 6%
             # true gap entered 2026-07-07; audit 2026-07-08). None until the open.
+            # day_open above comes from query_minute_bars, which returns a WINDOW —
+            # its "first RTH bar" is a mid-session bar later in the day, which let
+            # TVRD (+4%) and FRTT arm on 2026-07-08 as fake "20% gappers". Use the
+            # authoritative first-RTH-bar open, cached (immutable once set).
+            true_open = _session_open(rt["research_con"], g.symbol, session_date)
+            if true_open:
+                day_open = true_open
             gap_open_pct = None
-            if day_open and g.prev_close and g.prev_close > 0:
-                gap_open_pct = (day_open / g.prev_close - 1.0) * 100.0
+            if true_open and g.prev_close and g.prev_close > 0:
+                gap_open_pct = (true_open / g.prev_close - 1.0) * 100.0
+            elif _is_rth():
+                # session underway but this symbol's opening bars are not ingested
+                # yet (mid-day discovery): its "opening range" would be a fake
+                # mid-day window and its gap unknowable. Do not arm until the
+                # per-symbol backfill lands (usually the next ingest pass).
+                hi = None
+                stop = None
             candidates.append({
                 "symbol": g.symbol,
                 "gap_pct": g.gap_pct,