diff --git a/.env.example b/.env.example index 757b8d0..b71534f 100644 --- a/.env.example +++ b/.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 diff --git a/research/ingestion/snapshot_scan.py b/research/ingestion/snapshot_scan.py new file mode 100644 index 0000000..193d541 --- /dev/null +++ b/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}, + )) diff --git a/run_live_paper.py b/run_live_paper.py index 8ede6b3..d2b2f2e 100644 --- a/run_live_paper.py +++ b/run_live_paper.py @@ -1558,6 +1558,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() diff --git a/tests/unit/research/test_snapshot_scan.py b/tests/unit/research/test_snapshot_scan.py new file mode 100644 index 0000000..bd08969 --- /dev/null +++ b/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)