fix: one invalid symbol can no longer stop all live bar ingestion

At 09:34 ET on 2026-09-28 Alpaca's movers screener returned 'TSX:STC', a
Toronto listing. The discovery filter only dropped symbols containing a dot,
so it joined the universe. The live ingest fetches bars for every symbol in
one batched request, and Alpaca rejects the WHOLE batch with 400 "invalid
symbol". From 09:35 every ingest pass stored 0 rows and the loop ran on
stale bars ("INGEST STALE: freshest bar 458s old").

- Discovery keeps only plain 1-5 letter US tickers from both screeners.
- The ingest never sends a non-US or invalid symbol. If Alpaca still rejects
  one ("invalid symbol: X"), it drops X and retries the rest (up to 3 times),
  so one bad name costs one symbol, not the session.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01E3S5EFKThFVrjmN4HHsHMN

research/ingestion/market_data.py

@@ -13,6 +13,8 @@ hardcoded ``is_regular_hours = TRUE`` on every row. Here we:
 
 from __future__ import annotations
 
+import re
+
 import logging
 from dataclasses import dataclass, field
 from datetime import date, datetime, time, timedelta, timezone
@@ -199,6 +201,22 @@ def data_feed(client) -> str:
     return getattr(getattr(src, "settings", None), "feed", "iex")
 
 
+_US_TICKER = re.compile(r"^[A-Z]{1,5}$")
+
+
+def is_us_ticker(symbol: str) -> bool:
+    """Plain US common-stock ticker (1-5 letters). Alpaca's movers screener can return
+    foreign listings like 'TSX:STC'. One of those in a batched bars request makes Alpaca
+    400 the WHOLE batch ("invalid symbol"), which stopped all live ingestion from
+    09:35 ET on 2026-09-28."""
+    return bool(symbol) and bool(_US_TICKER.match(symbol))
+
+
+def _invalid_symbol(exc) -> str | None:
+    m = re.search(r"invalid symbol: ([^\"\s]+)", str(exc))
+    return m.group(1) if m else None
+
+
 def ingest_live_minute_bars(
     con,
     client,
@@ -228,15 +246,31 @@ def ingest_live_minute_bars(
     now = datetime.now(timezone.utc)
     start = now - timedelta(minutes=lookback_minutes)
     end = now - timedelta(seconds=30)
-    try:
-        bars_by_symbol = client.get_minute_bars(
-            symbols,
-            start_iso=start.isoformat().replace("+00:00", "Z"),
-            end_iso=end.isoformat().replace("+00:00", "Z"),
-            feed=feed,
-        )
-    except Exception as exc:  # noqa: BLE001 - report, don't crash the loop
-        result.errors.append(f"minute bars fetch failed: {exc}")
+    bad = [s for s in symbols if not is_us_ticker(s)]
+    symbols = [s for s in symbols if is_us_ticker(s)]
+    if bad:
+        result.errors.append(f"skipped non-US/invalid symbols: {bad}")
+    bars_by_symbol = None
+    for _attempt in range(3):          # drop any symbol Alpaca rejects and retry the rest
+        if not symbols:
+            return result
+        try:
+            bars_by_symbol = client.get_minute_bars(
+                symbols,
+                start_iso=start.isoformat().replace("+00:00", "Z"),
+                end_iso=end.isoformat().replace("+00:00", "Z"),
+                feed=feed,
+            )
+            break
+        except Exception as exc:  # noqa: BLE001 - report, don't crash the loop
+            reject = _invalid_symbol(exc)
+            if reject and reject in symbols:
+                symbols = [s for s in symbols if s != reject]
+                result.errors.append(f"dropped invalid symbol {reject} and retried")
+                continue
+            result.errors.append(f"minute bars fetch failed: {exc}")
+            return result
+    if bars_by_symbol is None:
         return result
     # SAFETY: the ratio correction exists ONLY for the thin IEX feed. Once the live
     # feed is SIP (paid plan), volume is already consolidated — applying the ~25-300x
@@ -313,7 +347,8 @@ def discover_active_symbols(
             for g in client.get_movers(top=15):
                 sym = (g.get("symbol") or "").strip()
                 # drop warrants/units/rights noise (BKSY.WS, EVLVW-style)
-                if not sym or "." in sym or (len(sym) >= 5 and sym.endswith("W")):
+                if (not is_us_ticker(sym)
+                        or (len(sym) >= 5 and sym.endswith("W"))):
                     continue
                 gainers.append(sym)
         except Exception:  # noqa: BLE001 — movers is additive; actives still work
@@ -321,7 +356,7 @@ def discover_active_symbols(
         # Alpaca caps the most-actives `top` at 100; never request more.
         actives = client.get_most_actives(top=min(top * 3, 100), by="volume")
         symbols = list(dict.fromkeys(
-            gainers + [a["symbol"] for a in actives if a.get("symbol")]))
+            gainers + [a["symbol"] for a in actives if is_us_ticker(a.get("symbol") or "")]))
         if not symbols:
             return []
         trades = client.get_latest_trades(symbols)  # batched internally; price them all

tests/unit/test_volume_ratio.py

@@ -110,3 +110,29 @@ def test_tradier_wrapper_reads_the_wrapped_feed():
     flat = [v for row in con.rows for v in row]
     assert 2500 in flat and 62500 not in flat
     assert "alpaca_sip" in flat and "alpaca_iex_scaled" not in flat
+
+
+class _RejectingClient(_SipClient):
+    """Alpaca 400s the WHOLE batched request when any symbol is invalid."""
+
+    def __init__(self):
+        super().__init__()
+        self.calls = []
+
+    def get_minute_bars(self, symbols, start_iso, end_iso=None, feed=None, limit=10_000):
+        self.calls.append(list(symbols))
+        if "ZZZZ" in symbols:
+            raise RuntimeError('Alpaca API error 400: {"message":"invalid symbol: ZZZZ"}')
+        return super().get_minute_bars(symbols, start_iso, end_iso, feed, limit)
+
+
+def test_one_bad_symbol_cannot_blind_the_whole_ingest():
+    # 2026-09-28 09:35 ET: 'TSX:STC' from the movers screener 400'd every batched
+    # bars request and live ingestion stopped for the session.
+    client = _RejectingClient()
+    con = _Con()
+    res = ingest_live_minute_bars(con, client, ["GOOD", "TSX:STC", "ZZZZ"])
+    assert "TSX:STC" not in client.calls[0]                  # never sent at all
+    assert client.calls[-1] == ["GOOD"]                     # rejected one dropped, retried
+    assert res.minute_rows > 0
+    assert any("TSX:STC" in e for e in res.errors) and any("ZZZZ" in e for e in res.errors)