""" premarket_scanner.py Runs at ~09:00 ET. Scans the tradable universe ($5-$15, liquid enough) and produces a ranked list of candidates using scorer.py. Called by monitor.py, but also runnable standalone for testing: python premarket_scanner.py It does NOT buy anything — this module only discovers and ranks. """ import sys from datetime import datetime, timedelta, timezone from config_loader import get_config from logger_setup import get_logger from alpaca_client import get_client from scorer import score_premarket_candidate import symbol_filters import data_store log = get_logger("premarket_scanner") def _bars_to_dicts(bars) -> list: out = [] for b in bars: out.append({ "t": b.timestamp, "o": float(b.open), "h": float(b.high), "l": float(b.low), "c": float(b.close), "v": float(b.volume), }) return out def get_universe_symbols(client) -> list: """ Pulls active, tradable US equities and applies: 1. the include/exclude_symbols lists from config 2. ETF/fund/trust exclusion (symbol_filters.is_etf_or_fund) 3. leveraged/inverse ("multiplier") product exclusion 4. company-name syllable-count cap Alpaca's asset_class=US_EQUITY includes ETFs/ETNs, so filters 2-4 are done on the asset's `name` field via symbol_filters.py. """ cfg = get_config()["universe"] if cfg.get("include_only_symbols"): return cfg["include_only_symbols"] assets = client.get_tradable_assets() excluded = set(cfg.get("exclude_symbols", [])) candidates = [a for a in assets if getattr(a, "tradable", False) and a.symbol not in excluded] before = len(candidates) survivors = symbol_filters.filter_symbol_list(candidates, log_rejections=True) log.info(f"[PREMARKET] Symbol filters (ETF/leveraged/name-complexity) removed " f"{before - len(survivors)} of {before} candidates") return survivors def prefilter_by_snapshot(client, symbols: list) -> list: """ Cheap first pass using snapshot data to cut the universe down to a manageable set before pulling minute bars for full scoring. Filters on price band and a basic volume floor. """ cfg = get_config()["universe"] scoring_cfg = get_config()["premarket_scoring"] survivors = [] snapshots = client.get_snapshots(symbols) if not snapshots: return survivors for symbol, snap in snapshots.items(): try: latest_trade = snap.latest_trade daily_bar = snap.daily_bar if latest_trade is None: continue price = float(latest_trade.price) if not (cfg["price_min"] <= price <= cfg["price_max"]): continue prev_daily_vol = float(daily_bar.volume) if daily_bar else 0 if prev_daily_vol and prev_daily_vol < cfg["min_avg_daily_volume"] * 0.05: # extremely thin so far today even accounting for premarket partial data continue survivors.append(symbol) except (AttributeError, TypeError): continue return survivors def scan(candidate_count: int = None, prefiltered: list = None, lookback_hours: float = 6) -> list: """ Full premarket scan. Returns a ranked list of scored candidate dicts, truncated to `candidate_count` (defaults to config's premarket_candidate_count, i.e. 20). [FEATURE 2026-08-17] `prefiltered` lets a caller (monitor.py) supply an already-computed universe+snapshot-filtered symbol list instead of this function pulling and filtering the full tradable-asset list again. monitor.py caches this list once at 09:00 and reuses it for every subsequent 30-minute intraday full rescan, so the expensive get_tradable_assets() + bulk snapshot call only ever happens once per day rather than once per rescan. `lookback_hours` defaults to 6 (covers the overnight/premarket session) but monitor.py's intraday rescans pass a much shorter window (intraday_health. full_rescan_lookback_hours, default 1) since a 6-hour window is unnecessary once the regular session is already underway. """ cfg = get_config() candidate_count = candidate_count or cfg["candidates"]["premarket_candidate_count"] client = get_client() log.info("[PREMARKET] Starting scan") if prefiltered is None: universe = get_universe_symbols(client) log.info(f"[PREMARKET] Universe size after asset filtering: {len(universe)}") prefiltered = prefilter_by_snapshot(client, universe) log.info(f"[PREMARKET] Prefiltered to {len(prefiltered)} symbols in price/volume band") now = datetime.now(timezone.utc) start = now - timedelta(hours=lookback_hours) scored = [] for symbol in prefiltered: bars = _bars_to_dicts(client.get_minute_bars(symbol, start, now)) if len(bars) < 3: log.debug(f"[PREMARKET] {symbol} rejected: insufficient bar data ({len(bars)} bars)") continue pm_high = max(b["h"] for b in bars) pm_low = min(b["l"] for b in bars) quote = client.get_latest_quote(symbol) bid = float(quote.bid_price) if quote and quote.bid_price else None ask = float(quote.ask_price) if quote and quote.ask_price else None # Baseline average volume for RVOL: use sum of daily bar volume # from snapshot if available, else fall back to today's premarket volume * 20 avg_vol_baseline = get_config()["universe"]["min_avg_daily_volume"] result = score_premarket_candidate(symbol, bars, pm_high, pm_low, avg_vol_baseline, bid, ask) result["pm_high"] = pm_high result["pm_low"] = pm_low if not result["flags"].get("meets_min_volume", False): log.debug(f"[PREMARKET] {symbol} rejected: below min premarket volume") continue if not result["flags"].get("spread_ok", True): log.debug(f"[PREMARKET] {symbol} rejected: spread too wide") continue scored.append(result) data_store.append_premarket_snapshot(symbol, result) scored.sort(key=lambda r: r["total_score"], reverse=True) top = scored[:candidate_count] for r in top: log.info(f"[PREMARKET] {r['symbol']} score={r['total_score']}") log.info(f"[PREMARKET] Selected {len(top)} candidates") data_store.write_premarket_candidates(top) return top if __name__ == "__main__": results = scan() for r in results: print(f"{r['symbol']:6s} score={r['total_score']:6.2f} " f"price={r['metrics']['price']:.2f} rvol={r['metrics']['rvol']:.2f}")