momentum-system/research/ingestion/snapshot_scan.py

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