feat: whole-market premarket squeeze scan via Alpaca snapshots

Discovery finds premarket squeezes far too late. Across 53 trades by a
professional small-cap momentum trader, our scanner had found only 4 of his
names before his entry, and 27 of his 28 entries were premarket (07:00-09:30
ET). KITT on 2026-09-25 ran $3.50 -> $8.39 at 09:00-09:01 ET (prior close
$0.49); the loop first "Discovered KITT via manual" at 09:35. The existing
sources (Alpaca movers / most-actives screeners) only reflect the regular
session, so a premarket squeeze is invisible until the open.

research/ingestion/snapshot_scan.py sweeps every active, tradable, listed US
equity (~11.9k names; OTC, warrants/units/rights and DISCOVER_EXCLUDE dropped;
asset list cached per ET day) through GET /v2/stocks/snapshots in 2,000-symbol
batches. The endpoint is bound by URL length rather than symbol count (8,000
worked; ~12.7k returned 414). A name qualifies when:
  - the LATEST TRADE price is inside WATCHER_PRICE_MIN/MAX (never the prior
    close, which is routinely < $1 on these names),
  - % change vs the prior close >= DISCOVER_SNAPSHOT_MIN_PCT (20),
  - it is trading now: last trade <= 180s old, latest minute bar <= 300s old
    with >= 5,000 shares, and >= 100,000 shares since 04:00 ET (summed from
    minute bars for the short list, one extra request).
Qualifiers are ranked by % change, capped per sweep, and announced once per
day. dailyBar OHLC ignores extended-hours prints (KITT's dailyBar high was
$3.24 against a premarket high of $8.39), so in premarket it can still be
yesterday's bar. The prior close is prevDailyBar.c only when dailyBar is dated
today, otherwise dailyBar.c.

run_live_paper.py runs it on its own daemon thread (04:00-16:00 ET weekdays,
every 45s), using the plan-019 pattern: its own DB connection, the thread-safe
Watchlist and the generic interval driver. Each new name emits
symbol_discovered with source="snapshot_scan" and price/prev_close/pct/minute
and cumulative volume in symbol_data, so first-sighting latency can be
measured. It gets a daily-history backfill and joins the watchlist, so the
normal new-symbol ingest path gives it its first full-session pull. This is
discovery only; no trading gate changes.

Gated by DISCOVER_SNAPSHOT_SCAN_ENABLED (default 1); knobs in .env.example.
Live read-only sweep at 13:10 ET: 11,871 names in 8 requests / 6.3s cold
(7 requests / 3.5s warm); top hits KITT +376%, MSGY +262%, INLF +60%.

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

.env.example

@@ -84,6 +84,22 @@ LIVE_BARS_INTERVAL_SECONDS=20
 LIVE_BARS_REFRESH_MINUTES=20
 WATCHER_INTERVAL_SECONDS=15
 DISCOVER_INTERVAL_SECONDS=120
+# --- whole-market premarket squeeze scan (research/ingestion/snapshot_scan.py) ---
+# The movers/most-actives screeners only see the regular session, so premarket
+# squeezes were invisible until 09:30 (KITT 2026-09-25 ran at 09:00, found 09:35).
+# This sweeps every listed, tradable US equity through Alpaca snapshots
+# 04:00-16:00 ET on its own thread and adds qualifying names to the watchlist
+# (symbol_discovered source=snapshot_scan). Discovery only; all trading gates apply.
+# Price band = WATCHER_PRICE_MIN/MAX on the LATEST trade (not the prior close).
+DISCOVER_SNAPSHOT_SCAN_ENABLED=1
+DISCOVER_SNAPSHOT_INTERVAL_SECONDS=45     # one full sweep (~7 requests)
+DISCOVER_SNAPSHOT_MIN_PCT=20              # % change vs prior close to qualify
+DISCOVER_SNAPSHOT_MAX_PER_SWEEP=5         # new names injected per sweep (top by %)
+DISCOVER_SNAPSHOT_MIN_BAR_VOLUME=5000     # shares in the latest minute bar
+DISCOVER_SNAPSHOT_MIN_CUM_VOLUME=100000   # shares traded today since 04:00 ET
+DISCOVER_SNAPSHOT_MAX_TRADE_AGE_SECONDS=180
+DISCOVER_SNAPSHOT_MAX_BAR_AGE_SECONDS=300
+DISCOVER_SNAPSHOT_BATCH=2000              # symbols per request (URL-length bound; 414 at ~12.7k)
 # Active setups. The backtest sweep found the opening-range breakout is the only
 # profitable one (others lose and fire later); empty = all setups. See learn_params.py.
 STRATEGY_SETUPS=opening_range_break

research/ingestion/snapshot_scan.py

@@ -0,0 +1,351 @@
+"""Whole-market premarket squeeze scanner (Alpaca snapshots).
+
+Why this exists (measured 2026-09-25): the screener-based discovery
+(``discover_active_symbols`` -> Alpaca movers / most-actives) only reflects the
+REGULAR session, so a premarket squeeze is invisible until 09:30. Against 53
+trades by a professional small-cap momentum trader, discovery had found only 4
+of his names before his entry, and 27 of his 28 entries were premarket. KITT on
+2026-09-25 ran $3.50 -> $8.39 at 09:00 ET (prior close $0.49); we first
+"discovered" it at 09:35.
+
+This sweeps EVERY active, tradable, listed US equity through
+``GET /v2/stocks/snapshots`` (latestTrade / minuteBar / dailyBar /
+prevDailyBar), which is live from 04:00 ET, and qualifies a name when:
+
+* the LATEST TRADE price is inside the watcher price band (never the prior
+  close: a squeeze's prior close is routinely < $1),
+* % change vs the prior session close >= ``min_pct``,
+* it is actually trading now: last trade within ``max_trade_age_s``, the
+  latest minute bar within ``max_bar_age_s`` with >= ``min_bar_volume``
+  shares, and >= ``min_cum_volume`` shares traded today (from 04:00 ET).
+
+Discovery/ingestion only: a qualifying name is added to the watchlist, and
+every trading gate still decides whether it is ever traded.
+
+Snapshot quirk handled here: ``dailyBar`` OHLC ignores extended-hours prints
+(KITT 2026-09-25: dailyBar high $3.24, premarket high $8.39 at 09:01 ET), so
+before today's first regular-session print it may still be the PREVIOUS
+session's bar. The prior close is therefore ``prevDailyBar.c`` only when
+``dailyBar`` is dated today, otherwise ``dailyBar.c``. Cumulative volume for the
+short list comes from today's minute bars since 04:00 ET, which do include
+premarket (one extra batched request per sweep).
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import re
+import time
+from dataclasses import dataclass, field
+from datetime import date, datetime, timezone
+
+from research.ingestion.market_data import EASTERN, parse_alpaca_timestamp
+
+logger = logging.getLogger(__name__)
+
+SOURCE = "snapshot_scan"
+# Alpaca caps snapshots by URL length, not symbol count: 8,000 symbols per
+# request worked and ~12,700 returned HTTP 414 (probed 2026-09-25). 2,000 keeps
+# the URL around 12 KB and a full sweep at ~7 requests.
+DEFAULT_BATCH = 2000
+_DERIV_NAME = re.compile(r"\bwarrants?\b|\brights\b|\bunits?\s*$|\bwhen[- ]issued\b", re.I)
+_SCAN_START_ET = (4, 0)
+_SCAN_END_ET = (16, 0)
+
+
+def _env_float(name: str, default: float) -> float:
+    raw = os.environ.get(name, "").strip()
+    try:
+        return float(raw) if raw else default
+    except ValueError:
+        return default
+
+
+@dataclass
+class SnapshotScanConfig:
+    price_min: float = 1.0
+    price_max: float = 20.0
+    min_pct: float = 20.0             # % change vs prior close
+    max_per_sweep: int = 5            # new names injected per sweep (top by %)
+    min_bar_volume: int = 5_000       # shares in the latest minute bar
+    min_cum_volume: int = 100_000     # shares traded today since 04:00 ET
+    max_trade_age_s: float = 180.0    # latest trade must be this fresh
+    max_bar_age_s: float = 300.0      # latest minute bar must be this fresh
+    batch_size: int = DEFAULT_BATCH
+    interval_s: float = 45.0
+
+    @classmethod
+    def from_env(cls) -> "SnapshotScanConfig":
+        return cls(
+            price_min=_env_float("WATCHER_PRICE_MIN", 1.0),
+            price_max=_env_float("WATCHER_PRICE_MAX", 20.0),
+            min_pct=_env_float("DISCOVER_SNAPSHOT_MIN_PCT", 20.0),
+            max_per_sweep=int(_env_float("DISCOVER_SNAPSHOT_MAX_PER_SWEEP", 5)),
+            min_bar_volume=int(_env_float("DISCOVER_SNAPSHOT_MIN_BAR_VOLUME", 5_000)),
+            min_cum_volume=int(_env_float("DISCOVER_SNAPSHOT_MIN_CUM_VOLUME", 100_000)),
+            max_trade_age_s=_env_float("DISCOVER_SNAPSHOT_MAX_TRADE_AGE_SECONDS", 180.0),
+            max_bar_age_s=_env_float("DISCOVER_SNAPSHOT_MAX_BAR_AGE_SECONDS", 300.0),
+            batch_size=int(_env_float("DISCOVER_SNAPSHOT_BATCH", DEFAULT_BATCH)),
+            interval_s=_env_float("DISCOVER_SNAPSHOT_INTERVAL_SECONDS", 45.0),
+        )
+
+
+@dataclass
+class SqueezeHit:
+    symbol: str
+    price: float
+    prev_close: float
+    pct_change: float
+    minute_volume: int
+    cum_volume: int | None = None
+    trade_age_s: float | None = None
+
+    def as_data(self) -> dict:
+        return {
+            "source": SOURCE,
+            "price": round(self.price, 4),
+            "prev_close": round(self.prev_close, 4),
+            "pct_change": round(self.pct_change, 1),
+            "minute_volume": self.minute_volume,
+            "cum_volume": self.cum_volume,
+            "trade_age_s": None if self.trade_age_s is None else round(self.trade_age_s, 1),
+        }
+
+
+@dataclass
+class ScanResult:
+    scanned: int = 0
+    requests: int = 0
+    duration_s: float = 0.0
+    hits: list[SqueezeHit] = field(default_factory=list)       # all qualifiers, ranked
+    injected: list[SqueezeHit] = field(default_factory=list)   # new this sweep (<= cap)
+    errors: list[str] = field(default_factory=list)
+
+
+def data_client(client):
+    """The Alpaca client that serves market data (unwraps TradierClient)."""
+    return getattr(client, "data_client", None) or client
+
+
+def _is_derivative_listing(symbol: str, name: str) -> bool:
+    """Warrants / units / rights / preferreds / when-issued — not the common."""
+    if not symbol or "." in symbol or "/" in symbol or "-" in symbol:
+        return True
+    # Nasdaq 5th-letter convention: W = warrant, U = unit, R = rights
+    if len(symbol) >= 5 and symbol[-1] in "WUR":
+        return True
+    # Name check is deliberately narrow: "Common Units" (MLPs like ET/PAA) and
+    # ADS "the right to receive" are ordinary equity (checked against the live
+    # asset list 2026-09-25).
+    return bool(_DERIV_NAME.search(name or ""))
+
+
+def build_universe(assets: list[dict], excluded: set[str] | None = None) -> list[str]:
+    """Active, tradable, listed (non-OTC) US equities minus derivatives/excludes."""
+    excluded = excluded or set()
+    out = []
+    for a in assets:
+        sym = (a.get("symbol") or "").strip().upper()
+        if not sym or sym in excluded:
+            continue
+        if a.get("status", "active") != "active" or not a.get("tradable"):
+            continue
+        if (a.get("exchange") or "").upper() in {"OTC", ""}:
+            continue
+        if _is_derivative_listing(sym, a.get("name") or ""):
+            continue
+        out.append(sym)
+    return sorted(set(out))
+
+
+def _et_date(ts_utc: datetime) -> date:
+    return ts_utc.astimezone(EASTERN).date()
+
+
+def prior_close(snap: dict, today_et: date) -> float | None:
+    """The previous session's close. ``dailyBar`` OHLC ignores extended-hours
+    prints, so before today's first RTH print it can still be yesterday's bar."""
+    daily = snap.get("dailyBar") or {}
+    prev = snap.get("prevDailyBar") or {}
+    try:
+        if daily.get("t") and _et_date(parse_alpaca_timestamp(daily["t"])) < today_et:
+            return float(daily["c"]) or None
+        return float(prev["c"]) if prev.get("c") else None
+    except (KeyError, TypeError, ValueError):
+        return None
+
+
+def qualify(snapshots: dict, cfg: SnapshotScanConfig, now: datetime) -> list[SqueezeHit]:
+    """Pure: snapshot map -> qualifying names ranked by % change (desc).
+    Cumulative volume is NOT checked here (see ``apply_cum_volume``)."""
+    today = _et_date(now)
+    hits: list[SqueezeHit] = []
+    for sym, snap in (snapshots or {}).items():
+        if not snap:
+            continue
+        trade = snap.get("latestTrade") or {}
+        bar = snap.get("minuteBar") or {}
+        try:
+            price = float(trade.get("p") or 0.0)
+            if not (cfg.price_min <= price <= cfg.price_max):
+                continue
+            if not trade.get("t") or not bar.get("t"):
+                continue
+            trade_age = (now - parse_alpaca_timestamp(trade["t"])).total_seconds()
+            bar_age = (now - parse_alpaca_timestamp(bar["t"])).total_seconds()
+            if trade_age > cfg.max_trade_age_s or bar_age > cfg.max_bar_age_s:
+                continue
+            bar_vol = int(bar.get("v") or 0)
+            if bar_vol < cfg.min_bar_volume:
+                continue
+            pc = prior_close(snap, today)
+            if not pc or pc <= 0:
+                continue
+            pct = (price - pc) / pc * 100.0
+            if pct < cfg.min_pct:
+                continue
+        except (TypeError, ValueError, KeyError):
+            continue
+        hits.append(SqueezeHit(sym, price, pc, pct, bar_vol, trade_age_s=max(0.0, trade_age)))
+    hits.sort(key=lambda h: h.pct_change, reverse=True)
+    return hits
+
+
+def fetch_snapshots(client, symbols: list[str], batch_size: int = DEFAULT_BATCH):
+    """(snapshots, requests, errors). One failed batch doesn't sink the sweep."""
+    dc = data_client(client)
+    feed = getattr(getattr(dc, "settings", None), "feed", "sip")
+    out: dict = {}
+    requests = 0
+    errors: list[str] = []
+    step = max(1, int(batch_size))
+    for i in range(0, len(symbols), step):
+        chunk = symbols[i:i + step]
+        requests += 1
+        try:
+            payload = dc._data("GET", "/v2/stocks/snapshots",
+                               params={"symbols": ",".join(chunk), "feed": feed})
+        except Exception as exc:  # noqa: BLE001
+            errors.append(f"snapshots batch {i // step}: {exc}"[:300])
+            continue
+        if isinstance(payload, dict):
+            # multi-symbol response is {SYM: snapshot}; tolerate a wrapped shape
+            inner = payload.get("snapshots")
+            out.update(inner if isinstance(inner, dict) else payload)
+    return out, requests, errors
+
+
+def apply_cum_volume(client, hits: list[SqueezeHit], cfg: SnapshotScanConfig,
+                     now: datetime) -> tuple[list[SqueezeHit], int, list[str]]:
+    """Fill today's cumulative volume (04:00 ET -> now) from minute bars for the
+    short list and drop names under the floor. One batched request. If the bar
+    fetch fails the floor is skipped (recency + minute volume still applied),
+    so an API blip can't blind premarket discovery."""
+    if not hits or cfg.min_cum_volume <= 0:
+        return hits, 0, []
+    now_et = now.astimezone(EASTERN)
+    start = now_et.replace(hour=4, minute=0, second=0, microsecond=0).astimezone(timezone.utc)
+    try:
+        bars = client.get_minute_bars(
+            [h.symbol for h in hits],
+            start_iso=start.isoformat().replace("+00:00", "Z"),
+            end_iso=now.astimezone(timezone.utc).isoformat().replace("+00:00", "Z"),
+        ) or {}
+    except Exception as exc:  # noqa: BLE001
+        return hits, 1, [f"cum-volume bars fetch failed (floor skipped): {exc}"[:300]]
+    kept = []
+    for h in hits:
+        h.cum_volume = int(sum(int(b.get("v") or 0) for b in bars.get(h.symbol) or []))
+        if h.cum_volume >= cfg.min_cum_volume:
+            kept.append(h)
+    return kept, 1, []
+
+
+def in_scan_window(now: datetime) -> bool:
+    et = now.astimezone(EASTERN)
+    if et.weekday() >= 5:
+        return False
+    return _SCAN_START_ET <= (et.hour, et.minute) < _SCAN_END_ET
+
+
+class SnapshotScanner:
+    """Stateful sweep driver: per-day universe cache + per-day dedupe of names
+    already injected, so each squeeze is announced once (its first sighting)."""
+
+    def __init__(self, client, cfg: SnapshotScanConfig | None = None,
+                 excluded: set[str] | None = None):
+        self.client = client
+        self.cfg = cfg or SnapshotScanConfig.from_env()
+        self.excluded = excluded
+        self._universe: list[str] = []
+        self._universe_day: date | None = None
+        self._announced: set[str] = set()
+        self._announced_day: date | None = None
+
+    def universe(self, today: date) -> tuple[list[str], int]:
+        """(universe, requests made). The asset list is fetched once per ET day."""
+        if self._universe_day == today and self._universe:
+            return self._universe, 0
+        from research.ingestion.discovery import _excluded_symbols
+        excluded = self.excluded if self.excluded is not None else _excluded_symbols()
+        assets = data_client(self.client).get_assets()
+        self._universe = build_universe(assets, excluded)
+        self._universe_day = today
+        return self._universe, 1
+
+    def sweep(self, now: datetime | None = None,
+              already_watched: set[str] | frozenset = frozenset()) -> ScanResult:
+        now = now or datetime.now(timezone.utc)
+        today = _et_date(now)
+        if self._announced_day != today:
+            self._announced, self._announced_day = set(), today
+        res = ScanResult()
+        t0 = time.monotonic()
+        try:
+            universe, n_req = self.universe(today)
+            res.requests += n_req
+        except Exception as exc:  # noqa: BLE001
+            res.errors.append(f"assets fetch failed: {exc}"[:300])
+            res.duration_s = time.monotonic() - t0
+            return res
+        snaps, n_req, errs = fetch_snapshots(self.client, universe, self.cfg.batch_size)
+        res.scanned = len(snaps)
+        res.requests += n_req
+        res.errors.extend(errs)
+        hits = qualify(snaps, self.cfg, now)
+        fresh = [h for h in hits
+                 if h.symbol not in self._announced and h.symbol not in already_watched]
+        # volume-floor a bounded short list (cap x3) so one request covers it
+        short = fresh[: max(1, self.cfg.max_per_sweep) * 3]
+        short, n_req, errs = apply_cum_volume(data_client(self.client), short, self.cfg, now)
+        res.requests += n_req
+        res.errors.extend(errs)
+        res.hits = hits
+        res.injected = short[: self.cfg.max_per_sweep]
+        self._announced.update(h.symbol for h in res.injected)
+        res.duration_s = time.monotonic() - t0
+        return res
+
+
+def emit_discoveries(store, hits: list[SqueezeHit], *, mode, correlation_id=None,
+                     scan_duration_s: float | None = None) -> None:
+    """One symbol_discovered per injected name, source=snapshot_scan, with the
+    evidence (pct/price/volume) so first-sighting latency is measurable."""
+    from storage.event_schema import SymbolDiscoveredEvent
+
+    for rank, h in enumerate(hits, 1):
+        data = {**h.as_data(), "rank": rank}
+        if scan_duration_s is not None:
+            data["scan_duration_s"] = round(scan_duration_s, 2)
+        store.emit(SymbolDiscoveredEvent(
+            timestamp=datetime.now(),
+            mode=mode,
+            correlation_id=correlation_id,
+            message=(f"Discovered {h.symbol} via {SOURCE}: +{h.pct_change:.0f}% "
+                     f"${h.price:g} (prev ${h.prev_close:g}), "
+                     f"min vol {h.minute_volume:,}"),
+            symbol=h.symbol,
+            symbol_data=data,
+            payload={"symbol": h.symbol, "source": SOURCE},
+        ))

run_live_paper.py

@@ -1551,6 +1551,73 @@ def main(argv: list[str] | None = None) -> int:
 
         threading.Thread(target=discover_loop, name="discover", daemon=True).start()
         print(f"[boot] discovery thread started (every {_disc_interval:.0f}s) — scheduler discover step OFF")
+    # Whole-market premarket squeeze scan (Alpaca snapshots) on its OWN light
+    # thread. The movers/most-actives screeners only see the regular session, so
+    # premarket squeezes stayed invisible until 09:30 (KITT 2026-09-25: ran at
+    # 09:00, found 09:35). Discovery only: names join the watchlist and flow
+    # through the normal new-symbol ingest (full-session first pass); every
+    # trading gate still applies. Same thread-safety pattern as plan 019: own DB
+    # connection, thread-safe Watchlist, single writer per step.
+    if (_flag("DISCOVER_SNAPSHOT_SCAN_ENABLED", "1") and _flag("DISCOVER_ENABLED", "1")
+            and client is not None):
+        from research.ingestion.snapshot_scan import (
+            SnapshotScanConfig, SnapshotScanner, emit_discoveries, in_scan_window)
+        from runtime.exit_thread import run_exit_loop as _run_scan_loop  # generic driver
+        _scan_cfg = SnapshotScanConfig.from_env()
+        _scanner = SnapshotScanner(client, _scan_cfg)
+        _scan_con: dict = {"c": None}
+
+        def step_snapshot_scan():
+            now = datetime.now(timezone.utc)
+            if not in_scan_window(now):   # 04:00-16:00 ET, weekdays
+                return
+            res = _scanner.sweep(now, already_watched=set(symbols.snapshot()))
+            names = [h.symbol for h in res.injected]
+            if names:
+                emit_discoveries(rt["store"], res.injected, mode=EventMode.PAPER,
+                                 correlation_id=rt["session_id"],
+                                 scan_duration_s=res.duration_s)
+                # prior close / 20d volume baseline BEFORE the name is watched,
+                # so its first gap/RVOL read isn't blank
+                try:
+                    ingest_daily_history(_scan_con["c"] or rt["research_con"], client,
+                                         names, days=30)
+                except Exception as exc:  # noqa: BLE001
+                    res.errors.append(f"daily backfill failed: {exc}"[:300])
+                symbols.extend(names)   # next step_ingest: new-symbol full-session pass
+                symbols.trim(int(os.environ.get("WATCHER_MAX_UNIVERSE", "80")))
+            if names or res.errors:
+                rt["store"].emit(ModuleTickEvent(
+                    timestamp=datetime.now(), mode=EventMode.PAPER,
+                    correlation_id=rt["session_id"],
+                    message=(f"snapshot scan: {res.scanned} scanned, {len(res.hits)} "
+                             f"qualify, +{len(names)} new"),
+                    module="discovery", stage="snapshot_scan",
+                    duration_ms=res.duration_s * 1000.0,
+                    input_count=res.scanned, output_count=len(names),
+                    metrics={"requests": res.requests, "injected": names,
+                             "top": [{"symbol": h.symbol, **h.as_data()}
+                                     for h in res.hits[:10]]},
+                    errors=[{"error": e} for e in res.errors],
+                ))
+            if names:
+                print(f"[snapscan] +{', '.join(names)} ({res.scanned} scanned, "
+                      f"{res.duration_s:.1f}s, {res.requests} req)", flush=True)
+
+        def snapshot_scan_loop():
+            try:
+                _scan_con["c"] = open_research_db("market")
+            except Exception as exc:  # noqa: BLE001
+                print(f"[snapscan-thread] dedicated DB unavailable, using shared: {exc}", flush=True)
+
+            class _ScanStep:
+                def manage(_self):        # adapt to the generic loop driver
+                    step_snapshot_scan()
+            _run_scan_loop(_ScanStep(), stop_event, _scan_cfg.interval_s)
+
+        threading.Thread(target=snapshot_scan_loop, name="snapscan", daemon=True).start()
+        print(f"[boot] snapshot squeeze scan started (every {_scan_cfg.interval_s:.0f}s, "
+              f">= {_scan_cfg.min_pct:g}% , cap {_scan_cfg.max_per_sweep}/sweep, 04:00-16:00 ET)")
     # news enrichment on its OWN thread (never blocks the scheduler's trade steps)
     if ollama_cfg.enabled and ollama_cfg.enrichment_enabled:
         threading.Thread(target=enrich_loop, name="enrich", daemon=True).start()

tests/unit/research/test_snapshot_scan.py

@@ -0,0 +1,213 @@
+"""Whole-market premarket squeeze scanner (research/ingestion/snapshot_scan.py).
+
+The screener-based discovery only sees the regular session: KITT ran
+$3.50 -> $8.39 at 09:00 ET on 2026-09-25 (prior close $0.49) and we first saw
+it at 09:35. These pin the qualification rules with a fake client (no network).
+"""
+
+from datetime import datetime, timedelta, timezone
+
+from research.ingestion.snapshot_scan import (
+    SOURCE,
+    SnapshotScanConfig,
+    SnapshotScanner,
+    build_universe,
+    emit_discoveries,
+    fetch_snapshots,
+    in_scan_window,
+    prior_close,
+    qualify,
+)
+from storage.event_schema import EventMode, EventType
+
+# 2026-09-25 09:00 ET (EDT = UTC-4) — premarket
+NOW = datetime(2026, 9, 25, 13, 0, 0, tzinfo=timezone.utc)
+
+
+def _iso(dt):
+    return dt.isoformat().replace("+00:00", "Z")
+
+
+def snap(price, prev_close, *, trade_age=5, bar_age=60, bar_vol=50_000,
+         daily_today=False, prev2_close=None):
+    """Premarket-shaped snapshot: dailyBar is YESTERDAY's (regular-session-only)
+    unless daily_today, in which case prevDailyBar is yesterday's."""
+    today_bar = {"c": price, "t": "2026-09-25T04:00:00Z", "v": 1_000_000}
+    yday_bar = {"c": prev_close, "t": "2026-09-24T04:00:00Z", "v": 700_000}
+    older = {"c": prev2_close or prev_close * 0.9, "t": "2026-09-23T04:00:00Z", "v": 1}
+    return {
+        "latestTrade": {"p": price, "t": _iso(NOW - timedelta(seconds=trade_age))},
+        "minuteBar": {"c": price, "v": bar_vol, "t": _iso(NOW - timedelta(seconds=bar_age))},
+        "dailyBar": today_bar if daily_today else yday_bar,
+        "prevDailyBar": yday_bar if daily_today else older,
+    }
+
+
+CFG = SnapshotScanConfig(price_min=1.0, price_max=20.0, min_pct=20.0, max_per_sweep=2,
+                         min_bar_volume=5_000, min_cum_volume=100_000,
+                         max_trade_age_s=180, max_bar_age_s=300)
+
+
+def test_price_band_uses_latest_trade_not_prior_close():
+    # KITT: prior close $0.49 (below the $1 band floor), trading $3.50 premarket
+    hits = qualify({"KITT": snap(3.50, 0.49)}, CFG, NOW)
+    assert [h.symbol for h in hits] == ["KITT"]
+    assert round(hits[0].pct_change) == 614
+    # latest price outside the band is rejected even with a huge move
+    assert qualify({"BIG": snap(25.0, 10.0)}, CFG, NOW) == []
+    assert qualify({"PENNY": snap(0.80, 0.30)}, CFG, NOW) == []
+
+
+def test_pct_threshold():
+    assert qualify({"A": snap(5.9, 5.0)}, CFG, NOW) == []       # +18%
+    assert [h.symbol for h in qualify({"A": snap(6.0, 5.0)}, CFG, NOW)] == ["A"]  # +20%
+
+
+def test_prior_close_premarket_vs_rth():
+    # premarket: dailyBar is still yesterday's -> prior close = dailyBar.c
+    assert prior_close(snap(6.0, 5.0), NOW.date()) == 5.0
+    # after the first RTH print: dailyBar is today's -> prior close = prevDailyBar.c
+    s = snap(6.0, 5.0, daily_today=True)
+    assert prior_close(s, NOW.date()) == 5.0
+    assert [h.symbol for h in qualify({"A": s}, CFG, NOW)] == ["A"]
+
+
+def test_stale_or_dead_names_rejected():
+    snaps = {
+        "OLDTRADE": snap(6.0, 4.0, trade_age=3600),     # last print an hour ago
+        "OLDBAR": snap(6.0, 4.0, bar_age=900),          # no minute bar for 15 min
+        "THIN": snap(6.0, 4.0, bar_vol=300),            # 300 shares last minute
+        "LIVE": snap(6.0, 4.0),
+    }
+    snaps["NOTRADE"] = {**snap(6.0, 4.0), "latestTrade": {}}
+    snaps["NOBAR"] = {**snap(6.0, 4.0), "minuteBar": None}
+    assert [h.symbol for h in qualify(snaps, CFG, NOW)] == ["LIVE"]
+
+
+def test_universe_excludes_warrants_units_rights_otc_and_excludes():
+    assets = [
+        {"symbol": "KITT", "exchange": "NASDAQ", "status": "active", "tradable": True,
+         "name": "Nauticus Robotics, Inc. Common Stock"},
+        {"symbol": "ABCDW", "exchange": "NASDAQ", "status": "active", "tradable": True,
+         "name": "ABCD Corp Warrant"},
+        {"symbol": "ABCDU", "exchange": "NASDAQ", "status": "active", "tradable": True,
+         "name": "ABCD Corp Unit"},
+        {"symbol": "ABCDR", "exchange": "NASDAQ", "status": "active", "tradable": True,
+         "name": "ABCD Corp Right"},
+        {"symbol": "FOO.WS", "exchange": "NYSE", "status": "active", "tradable": True,
+         "name": "Foo Redeemable warrants"},
+        {"symbol": "BCTXZ", "exchange": "NASDAQ", "status": "active", "tradable": True,
+         "name": "BriaCell Therapeutics Corp. Warrant expiring 2030"},
+        {"symbol": "ET", "exchange": "NYSE", "status": "active", "tradable": True,
+         "name": "Energy Transfer LP Common Units representing limited partner interests"},
+        {"symbol": "OTCX", "exchange": "OTC", "status": "active", "tradable": True, "name": "x"},
+        {"symbol": "HALT", "exchange": "NYSE", "status": "active", "tradable": False, "name": "x"},
+        {"symbol": "SOXL", "exchange": "ARCA", "status": "active", "tradable": True, "name": "x"},
+    ]
+    assert build_universe(assets, excluded={"SOXL"}) == ["ET", "KITT"]
+
+
+def test_ranking_cap_dedupe_and_cum_volume():
+    class FakeData:
+        def __init__(self):
+            self.settings = type("S", (), {"feed": "sip"})()
+            self.snapshot_calls = 0
+            self.bar_calls = []
+            self.snaps = {
+                "AAA": snap(3.0, 1.0),     # +200%
+                "BBB": snap(4.0, 2.0),     # +100%
+                "CCC": snap(6.0, 4.0),     # +50%
+                "DDD": snap(5.5, 5.0),     # +10% (below threshold)
+                "LOW": snap(9.0, 3.0),     # +200%, but under the cum-volume floor
+            }
+
+        def get_assets(self):
+            return [{"symbol": s, "exchange": "NASDAQ", "status": "active",
+                     "tradable": True, "name": s} for s in self.snaps]
+
+        def _data(self, method, path, params=None):
+            assert path == "/v2/stocks/snapshots" and params["feed"] == "sip"
+            self.snapshot_calls += 1
+            return {s: self.snaps[s] for s in params["symbols"].split(",")}
+
+        def get_minute_bars(self, symbols, start_iso, end_iso=None, **kw):
+            self.bar_calls.append((tuple(symbols), start_iso))
+            return {s: [{"v": 10_000 if s == "LOW" else 400_000}] for s in symbols}
+
+    class Wrapper:   # TradierClient-style: data delegated to data_client
+        def __init__(self, dc):
+            self.data_client = dc
+
+    dc = FakeData()
+    cfg = SnapshotScanConfig(**{**CFG.__dict__, "batch_size": 2})
+    scanner = SnapshotScanner(Wrapper(dc), cfg, excluded=set())
+    res = scanner.sweep(NOW)
+    assert dc.snapshot_calls == 3                 # 5 symbols / batch 2
+    assert res.requests == 1 + 3 + 1              # assets + snapshots + cum-volume bars
+    assert res.scanned == 5
+    assert [h.symbol for h in res.hits][:2] in (["AAA", "LOW"], ["LOW", "AAA"])
+    assert "DDD" not in {h.symbol for h in res.hits}
+    # LOW dropped by the cum-volume floor; cap of 2 keeps the top two by pct
+    assert [h.symbol for h in res.injected] == ["AAA", "BBB"]
+    assert res.injected[0].cum_volume == 400_000
+    assert dc.bar_calls[0][1] == "2026-09-25T08:00:00Z"   # 04:00 ET
+    # next sweep: already-announced names aren't re-injected; watched ones skipped
+    res2 = scanner.sweep(NOW + timedelta(seconds=45), already_watched={"CCC"})
+    assert res2.injected == []
+    assert res2.requests == 3 + 1                 # universe cached for the day
+
+
+def test_fetch_snapshots_survives_a_failed_batch():
+    class Flaky:
+        settings = type("S", (), {"feed": "sip"})()
+        n = 0
+
+        def _data(self, method, path, params=None):
+            Flaky.n += 1
+            if Flaky.n == 1:
+                raise RuntimeError("503")
+            return {s: {} for s in params["symbols"].split(",")}
+
+    out, reqs, errs = fetch_snapshots(Flaky(), ["A", "B", "C"], batch_size=2)
+    assert reqs == 2 and len(errs) == 1 and set(out) == {"C"}
+
+
+def test_emitted_event_source_and_evidence():
+    class Store:
+        def __init__(self):
+            self.events = []
+
+        def emit(self, ev):
+            self.events.append(ev)
+
+    store = Store()
+    hits = qualify({"KITT": snap(3.50, 0.49)}, CFG, NOW)
+    hits[0].cum_volume = 2_500_000
+    emit_discoveries(store, hits, mode=EventMode.PAPER, correlation_id="s1",
+                     scan_duration_s=3.2)
+    (ev,) = store.events
+    assert ev.event_type == EventType.SYMBOL_DISCOVERED
+    assert ev.symbol == "KITT"
+    assert ev.symbol_data["source"] == SOURCE == "snapshot_scan"
+    assert ev.payload["source"] == "snapshot_scan"
+    assert ev.symbol_data["price"] == 3.5
+    assert ev.symbol_data["prev_close"] == 0.49
+    assert ev.symbol_data["pct_change"] > 600
+    assert ev.symbol_data["minute_volume"] == 50_000
+    assert ev.symbol_data["cum_volume"] == 2_500_000
+    assert "via snapshot_scan" in ev.message
+
+
+def test_scan_window_04_to_16_et_weekdays():
+    assert in_scan_window(datetime(2026, 9, 25, 8, 0, tzinfo=timezone.utc))       # 04:00 ET
+    assert not in_scan_window(datetime(2026, 9, 25, 7, 59, tzinfo=timezone.utc))  # 03:59 ET
+    assert not in_scan_window(datetime(2026, 9, 25, 20, 0, tzinfo=timezone.utc))  # 16:00 ET
+    assert not in_scan_window(datetime(2026, 9, 26, 13, 0, tzinfo=timezone.utc))  # Saturday
+
+
+def test_config_from_env(monkeypatch):
+    monkeypatch.setenv("WATCHER_PRICE_MIN", "2")
+    monkeypatch.setenv("DISCOVER_SNAPSHOT_MIN_PCT", "35")
+    monkeypatch.setenv("DISCOVER_SNAPSHOT_MAX_PER_SWEEP", "3")
+    cfg = SnapshotScanConfig.from_env()
+    assert (cfg.price_min, cfg.min_pct, cfg.max_per_sweep) == (2.0, 35.0, 3)