#!/usr/bin/env python3 """ ╔══════════════════════════════════════════════════════════════════════════════╗ ║ IMPACT TRADING ACADEMY — COMPREHENSIVE DAY-TRADING SCANNER SUITE ║ ║ Version 2.0 | Python 3.9+ ║ ╠══════════════════════════════════════════════════════════════════════════════╣ ║ LIST-BASED SCANNERS ║ ║ 1. Top Gappers (>7% gap up/down, sortable) ║ ║ 2. Penny Gappers ($0.10–$5.00) ║ ║ 3. Large Cap Gappers (market cap ≥ $2B) ║ ║ 4. Large Cap + Earnings (gap + earnings within 48 h) ║ ║ 5. Top Relative Volume (highest RVOL movers) ║ ║ 6. Top RSI (5-min overbought / oversold) ║ ║ 7. 5-Min Volume Leaders (biggest 5-min volume spikes) ║ ║ 8. Continuation Scanner (2-week movers, bull flags, flat tops) ║ ║ 9. After-Hours Gainers (post-4 PM, % gain from close) ║ ║ ║ ║ ALERT-STYLE MOMENTUM SCANNERS (small-cap focus) ║ ║ 10. Low Float / High RVOL (<$20, low float, extreme RVOL) ║ ║ 11. Low Float Former Momo (historical spike track record) ║ ║ 12. Squeeze Up 5 % / 5 min (pre-alert) ║ ║ 13. Squeeze Up 10% / 10min (full alert) ║ ║ 14. Medium Float Grinder (<$20, algo-driven) ║ ║ 15. High Float Grinder ($20+, algo-driven) ║ ║ 16. 52-Week Breakout (new highs / Blue Sky ATH) ║ ║ 17. Halt Alert (timestamped, NASDAQ halts feed) ║ ║ ║ ║ LARGE CAP REVERSAL SCANNERS ║ ║ 18. Extreme Reversal (consecutive candles + BB + reversal candle) ║ ╠══════════════════════════════════════════════════════════════════════════════╣ ║ DATA COLUMNS (all scanners): ║ ║ ticker · price · daily_volume · float_shares · rvol · rvol_5min ║ ║ pct_change · atr · short_interest · short_ratio ║ ╠══════════════════════════════════════════════════════════════════════════════╣ ║ INSTALL: pip install yfinance pandas requests tabulate colorama ║ ║ RUN: python scanner_suite.py --scanner gappers ║ ║ python scanner_suite.py --scanner all ║ ║ python scanner_suite.py --scanner halt --watch ║ ╚══════════════════════════════════════════════════════════════════════════════╝ """ import os, sys, time, math, json, argparse, datetime, threading import warnings warnings.filterwarnings("ignore") import requests import pandas as pd from io import StringIO from typing import Optional, List, Dict, Tuple # ── Optional dependency guards ──────────────────────────────────────────────── try: import yfinance as yf HAS_YF = True except ImportError: HAS_YF = False print("[WARN] yfinance not found. Run: pip install yfinance") try: from tabulate import tabulate HAS_TAB = True except ImportError: HAS_TAB = False try: from colorama import Fore, Style, init as colorama_init colorama_init(autoreset=True) HAS_COLOR = True except ImportError: HAS_COLOR = False class _NoColor: GREEN = RED = YELLOW = CYAN = MAGENTA = WHITE = RESET_ALL = "" Fore = _NoColor() Style = _NoColor() # ═══════════════════════════════════════════════════════════════════════════════ # CONFIGURATION # ═══════════════════════════════════════════════════════════════════════════════ CFG = { # Gap thresholds "gap_pct_min": 0.07, # 7% minimum gap "penny_price_max": 5.00, # penny stock upper bound "penny_price_min": 0.10, # penny stock lower bound "large_cap_mcap_min": 2e9, # $2 B market cap minimum # Float thresholds (shares) "low_float_max": 20e6, # ≤ 20 M shares = low float "med_float_max": 100e6, # 20 M–100 M = medium float # (high float = > 100 M) # RVOL thresholds "rvol_alert_min": 3.0, # min RVOL for alert-style scanners "rvol_list_min": 1.5, # min RVOL for list scanners "rvol_5min_min": 2.0, # 5-min RVOL threshold # Squeeze "squeeze_5m_pct": 0.05, # 5% move in 5 min "squeeze_10m_pct": 0.10, # 10% move in 10 min # RSI "rsi_period": 14, "rsi_overbought": 70, "rsi_oversold": 30, # Bollinger Bands "bb_period": 20, "bb_std": 2.0, # Continuation lookback "continuation_days": 14, "continuation_pct": 0.15, # 15% move over 14 days minimum # Reversal "reversal_candles": 7, # consecutive same-color candles trigger # Earnings "earnings_window_h": 48, # ATR period "atr_period": 14, # Small-cap universe defaults (used by fetch_smallcap_universe below) "smallcap_max_price": 20.0, "smallcap_min_cap": 5e7, # $50M "smallcap_max_cap": 2e9, # $2B "smallcap_limit": 150, # Universe — expand as needed "universe": [ # Mega-cap / S&P 500 core "AAPL","MSFT","NVDA","AMZN","GOOG","META","TSLA","AVGO","JPM","V", "MA","UNH","JNJ","PG","XOM","CVX","HD","MRK","ABBV","PEP", "KO","LLY","BAC","WMT","COST","NFLX","CRM","ADBE","AMD","INTC", # High-beta / small-cap movers "COIN","MSTR","PLTR","HOOD","SOFI","SMCI","SOUN","IONQ","RGTI","QBTS", "ACHR","JOBY","BLNK","CHPT","RIVN","LCID","NKLA","WKHS","IDAI","MULN", "FFIE","CLOV","SPCE","GRAB","GRAB","OPEN","OFFERPAD","ATAI","SAVA", # ETFs for broad market context "SPY","QQQ","IWM","DIA","XLK","XLE","XLF","XLV","ARKK","SQQQ","SPXU", ], } HEADERS = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" } # ═══════════════════════════════════════════════════════════════════════════════ # DATA LAYER # ═══════════════════════════════════════════════════════════════════════════════ def _fmt_float(n: Optional[float], decimals: int = 2) -> str: if n is None or (isinstance(n, float) and math.isnan(n)): return "—" return f"{n:,.{decimals}f}" def _fmt_volume(v: Optional[float]) -> str: if v is None or (isinstance(v, float) and math.isnan(v)): return "—" if v >= 1_000_000_000: return f"{v/1e9:.2f}B" if v >= 1_000_000: return f"{v/1e6:.2f}M" if v >= 1_000: return f"{v/1e3:.0f}K" return str(int(v)) def _color_float(shares: Optional[float]) -> str: """Color-code float size: green=low, yellow=medium, white=high.""" if not HAS_COLOR or shares is None: return _fmt_volume(shares) if shares <= CFG["low_float_max"]: return Fore.GREEN + _fmt_volume(shares) + Style.RESET_ALL if shares <= CFG["med_float_max"]: return Fore.YELLOW + _fmt_volume(shares) + Style.RESET_ALL return _fmt_volume(shares) def _color_pct(pct: Optional[float]) -> str: if not HAS_COLOR or pct is None: return f"{_fmt_float(pct)}%" color = Fore.GREEN if pct > 0 else (Fore.RED if pct < 0 else "") return color + f"{pct:+.2f}%" + Style.RESET_ALL def fetch_finviz_screener(filters: str = "geo_usa") -> pd.DataFrame: """ Scrape Finviz screener for fundamental data (float, short interest, etc.) Returns DataFrame with columns: Ticker, Price, Change, Volume, Float, ShortFloat, ShortRatio filters example: "geo_usa,sh_gap_u7" for US stocks with gap > 7% """ url = ( f"https://finviz.com/screener.ashx?v=152&f={filters}" "&o=-volume&c=1,2,3,4,5,6,7,8,65,66,67,68" ) try: r = requests.get(url, headers=HEADERS, timeout=12) r.raise_for_status() tables = pd.read_html(StringIO(r.text)) # Finviz screener table is usually the last large table for tbl in reversed(tables): if "Ticker" in tbl.columns or (tbl.shape[1] > 5 and tbl.shape[0] > 2): # Rename columns if needed tbl.columns = [str(c).strip() for c in tbl.columns] if "No." in tbl.columns: tbl = tbl.drop(columns=["No."], errors="ignore") return tbl except Exception as e: print(f"[WARN] Finviz fetch failed: {e}") return pd.DataFrame() def fetch_smallcap_universe(max_price: float = None, min_cap: float = None, max_cap: float = None, limit: int = None) -> List[str]: """ Live small/micro-cap candidate list, pulled fresh from Finviz's screener every call — NOT the static CFG['universe']. This is what the low-float, former-momo, squeeze, and grinder alerts scan against, since their whole thesis (explosive small-cap moves) is undermined by ranking the same ~60 mega-cap tickers every run. Filters: US stocks, price <= max_price, market cap in [min_cap, max_cap] (roughly micro-to-small-cap band), sorted by relative volume. Falls back to CFG['universe'] if Finviz is unreachable or returns nothing. All params default to CFG['smallcap_*'] values when not supplied, so callers can override per-alert (e.g. former-momo widens price/cap since it screens for *historical* spikes rather than the current price). """ max_price = CFG["smallcap_max_price"] if max_price is None else max_price min_cap = CFG["smallcap_min_cap"] if min_cap is None else min_cap max_cap = CFG["smallcap_max_cap"] if max_cap is None else max_cap limit = CFG["smallcap_limit"] if limit is None else limit price_filter = f",sh_price_u{int(max_price)}" if max_price and max_price < 999 else "" filt = f"geo_usa,cap_smallover,cap_microunder{price_filter}" df = fetch_finviz_screener(filt + ",o=-relativevolume") ticker_col = next((c for c in df.columns if c.lower() in ("ticker", "symbol")), None) if ticker_col and not df.empty: tickers = df[ticker_col].dropna().tolist()[:limit] if tickers: return tickers print("[WARN] fetch_smallcap_universe: Finviz unavailable or empty — falling back to CFG['universe']") return CFG["universe"] def fetch_yf_quotes(symbols: List[str]) -> Dict[str, dict]: """Fetch bulk quotes from Yahoo Finance via yfinance.""" if not HAS_YF: return {} results = {} try: tickers = yf.Tickers(" ".join(symbols)) for sym in symbols: try: info = tickers.tickers[sym].fast_info results[sym] = { "price": getattr(info, "last_price", None), "prev_close": getattr(info, "previous_close", None), "volume": getattr(info, "three_month_average_volume", None), "day_volume": getattr(info, "last_volume", None), "market_cap": getattr(info, "market_cap", None), "52w_high": getattr(info, "year_high", None), "52w_low": getattr(info, "year_low", None), } except Exception: pass except Exception as e: print(f"[WARN] yfinance bulk fetch failed: {e}") return results def fetch_yf_intraday(symbol: str, period: str = "1d", interval: str = "5m") -> pd.DataFrame: """Fetch intraday OHLCV data for one symbol.""" if not HAS_YF: return pd.DataFrame() try: df = yf.download(symbol, period=period, interval=interval, progress=False, auto_adjust=True) df.index = pd.to_datetime(df.index) return df except Exception: return pd.DataFrame() def fetch_yf_history(symbol: str, period: str = "3mo") -> pd.DataFrame: """Fetch daily OHLCV history.""" if not HAS_YF: return pd.DataFrame() try: df = yf.download(symbol, period=period, interval="1d", progress=False, auto_adjust=True) return df except Exception: return pd.DataFrame() def fetch_nasdaq_halts() -> pd.DataFrame: """ Fetch NASDAQ halts from the official public CSV feed. Returns DataFrame with: Symbol, HaltDate, HaltTime, ResumeDate, ResumeTime, Reason, PauseThreshold """ url = "https://www.nasdaqtrader.com/rss.aspx?feed=tradehalts" try: r = requests.get(url, headers=HEADERS, timeout=10) r.raise_for_status() # NASDAQ returns XML/RSS; parse with pandas xml reader import xml.etree.ElementTree as ET root = ET.fromstring(r.text) ns = {"ns": "http://www.nasdaqtrader.com/"} rows = [] for item in root.iter("item"): def g(tag): el = item.find(f"ns:{tag}", ns) return el.text if el is not None else None rows.append({ "Symbol": g("IssueSymbol"), "HaltDate": g("HaltDate"), "HaltTime": g("HaltTime"), "ResumeDate":g("ResumptionDate"), "ResumeTime":g("ResumptionTradeTime"), "Reason": g("ReasonCode"), }) return pd.DataFrame(rows) except Exception: # Fallback: try CSV endpoint try: csv_url = "https://www.nasdaqtrader.com/dynamic/symdir/nasdaqtraded.txt" # This is the full halt list r = requests.get( "https://www.nasdaqtrader.com/rss.aspx?feed=tradehalts", headers=HEADERS, timeout=10 ) return pd.DataFrame(columns=["Symbol","HaltDate","HaltTime","Reason"]) except Exception: return pd.DataFrame(columns=["Symbol","HaltDate","HaltTime","Reason"]) def compute_rsi(closes: pd.Series, period: int = 14) -> pd.Series: """Wilder RSI.""" delta = closes.diff() gain = delta.clip(lower=0) loss = (-delta).clip(lower=0) avg_g = gain.ewm(com=period - 1, min_periods=period).mean() avg_l = loss.ewm(com=period - 1, min_periods=period).mean() rs = avg_g / avg_l.replace(0, float("nan")) return 100 - (100 / (1 + rs)) def compute_atr(df: pd.DataFrame, period: int = 14) -> float: """Average True Range from OHLC DataFrame.""" if df.empty or len(df) < 2: return float("nan") hi, lo, cl = df["High"], df["Low"], df["Close"] prev_cl = cl.shift(1) tr = pd.concat([hi - lo, (hi - prev_cl).abs(), (lo - prev_cl).abs()], axis=1).max(axis=1) return float(tr.ewm(span=period, min_periods=period).mean().iloc[-1]) def compute_bollinger(closes: pd.Series, period: int = 20, std_mult: float = 2.0) -> Tuple[pd.Series, pd.Series, pd.Series]: """Returns (upper, mid, lower) Bollinger Bands.""" mid = closes.rolling(period).mean() std = closes.rolling(period).std() return mid + std_mult * std, mid, mid - std_mult * std def estimate_rvol(symbol: str, current_vol: float, avg_vol_30d: float, minutes_elapsed: int) -> float: """ Estimate intraday RVOL adjusted for time of day. Compares current volume pace to what the 30-day avg would be at this point. """ if avg_vol_30d <= 0 or minutes_elapsed <= 0: return 0.0 trading_minutes = 390 # 6.5 h × 60 expected_vol = avg_vol_30d * (minutes_elapsed / trading_minutes) return current_vol / expected_vol if expected_vol > 0 else 0.0 def build_base_row(symbol: str, info: dict, hist: pd.DataFrame) -> dict: """Build the standard data columns row for any scanner output.""" price = info.get("price") or float("nan") prev = info.get("prev_close") or float("nan") day_vol = info.get("day_volume") or float("nan") avg_vol = info.get("volume") or float("nan") # 3-month avg mcap = info.get("market_cap") or float("nan") pct_chg = ((price - prev) / prev * 100) if (price and prev and prev != 0) else float("nan") atr = compute_atr(hist) if not hist.empty else float("nan") now = datetime.datetime.now() market_open = now.replace(hour=9, minute=30, second=0, microsecond=0) elapsed = max(1, int((now - market_open).total_seconds() / 60)) rvol = estimate_rvol(symbol, day_vol, avg_vol, elapsed) return { "Ticker": symbol, "Price": price, "DailyVol": day_vol, "AvgVol30d": avg_vol, "Float": float("nan"), # populated by Finviz layer "RVOL": rvol, "RVOL5m": float("nan"), # populated by intraday layer "Chg%": pct_chg, "ATR": atr, "ShortInt": float("nan"), # populated by Finviz layer "ShortRatio": float("nan"), # populated by Finviz layer "MarketCap": mcap, "_52wHigh": info.get("52w_high") or float("nan"), "_52wLow": info.get("52w_low") or float("nan"), } def enrich_with_finviz(rows: List[dict]) -> List[dict]: """Overlay Finviz data (float, short interest) onto base rows.""" symbols = [r["Ticker"] for r in rows] try: fv = fetch_finviz_screener( f"geo_usa,idx_sp500,idx_ndx,idx_dji" ) if fv.empty: return rows # Attempt to align columns col_map = {} for col in fv.columns: cl = col.lower() if "float" in cl: col_map[col] = "Float" if "short float" in cl: col_map[col] = "ShortInt" if "short ratio" in cl: col_map[col] = "ShortRatio" fv = fv.rename(columns=col_map) ticker_col = next((c for c in fv.columns if c.lower() in ("ticker","symbol")), None) if ticker_col: fv = fv.set_index(ticker_col) for row in rows: sym = row["Ticker"] if sym in fv.index: def _parse(val): try: return float(str(val).replace("%","").replace(",","").replace("B","e9").replace("M","e6").replace("K","e3")) except Exception: return float("nan") if "Float" in fv.columns: row["Float"] = _parse(fv.at[sym,"Float"]) if "ShortInt" in fv.columns: row["ShortInt"] = _parse(fv.at[sym,"ShortInt"]) if "ShortRatio" in fv.columns: row["ShortRatio"] = _parse(fv.at[sym,"ShortRatio"]) except Exception as e: print(f"[WARN] Finviz enrichment failed: {e}") return rows def display(rows: List[dict], title: str, sort_col: str = "RVOL", ascending: bool = False, top_n: int = 30) -> None: """Render scanner output as a formatted table.""" if not rows: print(f"\n{Fore.YELLOW}[{title}]{Style.RESET_ALL} — No results.\n") return df = pd.DataFrame(rows) if sort_col in df.columns: df = df.sort_values(sort_col, ascending=ascending) df = df.head(top_n) # Build display columns disp = pd.DataFrame() disp["Ticker"] = df["Ticker"] disp["Price"] = df["Price"].apply(lambda x: f"${_fmt_float(x)}") disp["Daily Vol"] = df["DailyVol"].apply(_fmt_volume) disp["Float"] = df["Float"].apply(lambda x: _color_float(x) if not (isinstance(x,float) and math.isnan(x)) else "—") disp["RVOL"] = df["RVOL"].apply(lambda x: f"{_fmt_float(x)}×") disp["RVOL 5m"] = df["RVOL5m"].apply(lambda x: f"{_fmt_float(x)}×" if not (isinstance(x,float) and math.isnan(x)) else "—") disp["Chg %"] = df["Chg%"].apply(lambda x: _color_pct(x) if not (isinstance(x,float) and math.isnan(x)) else "—") disp["ATR"] = df["ATR"].apply(lambda x: f"${_fmt_float(x)}") disp["Short Int"] = df["ShortInt"].apply(lambda x: f"{_fmt_float(x)}%" if not (isinstance(x,float) and math.isnan(x)) else "—") disp["Short Ratio"]= df["ShortRatio"].apply(lambda x: _fmt_float(x) if not (isinstance(x,float) and math.isnan(x)) else "—") ts = datetime.datetime.now().strftime("%H:%M:%S") bar = "═" * 90 print(f"\n{Fore.CYAN}{bar}") print(f" ⚡ {title} [{ts}]") print(f"{bar}{Style.RESET_ALL}") if HAS_TAB: print(tabulate(disp, headers="keys", tablefmt="simple", showindex=False)) else: print(disp.to_string(index=False)) print() # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 1 — TOP GAPPERS # ═══════════════════════════════════════════════════════════════════════════════ def scanner_top_gappers(min_gap: float = 0.07, direction: str = "both", sort_by: str = "Chg%") -> List[dict]: """ Lists stocks gapping > min_gap (default 7%) at the open vs prior close. direction: 'up' | 'down' | 'both' sort_by: 'Chg%' | 'Float' | 'DailyVol' | 'RVOL' Uses Finviz pre-market gap filter + yfinance for full data. """ print("[Scanner 1] Fetching Top Gappers…") # Finviz gap filter filt = "geo_usa,sh_gap_u7" if direction in ("up","both") else "geo_usa,sh_gap_d-7" if direction == "both": up_fv = fetch_finviz_screener("geo_usa,sh_gap_u7") down_fv = fetch_finviz_screener("geo_usa,sh_gap_d-7") fv_df = pd.concat([up_fv, down_fv], ignore_index=True) else: fv_df = fetch_finviz_screener(filt) # Extract tickers from Finviz or fall back to universe ticker_col = next((c for c in fv_df.columns if c.lower() in ("ticker","symbol")), None) if ticker_col and not fv_df.empty: candidates = fv_df[ticker_col].dropna().tolist()[:80] else: candidates = CFG["universe"] quotes = fetch_yf_quotes(candidates) results = [] for sym, info in quotes.items(): price = info.get("price") or 0 prev = info.get("prev_close") or 0 if not prev: continue gap = (price - prev) / prev if abs(gap) < min_gap: continue if direction == "up" and gap < 0: continue if direction == "down" and gap > 0: continue hist = fetch_yf_history(sym, period="1mo") row = build_base_row(sym, info, hist) results.append(row) results = enrich_with_finviz(results) display(results, "TOP GAPPERS (gap ≥ 7%)", sort_col=sort_by) return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 2 — PENNY GAPPERS # ═══════════════════════════════════════════════════════════════════════════════ def scanner_penny_gappers() -> List[dict]: """ Penny stocks ($0.10–$5.00) gapping > 7%. Low price + high gap = explosive potential, but elevated risk. """ print("[Scanner 2] Fetching Penny Gappers…") all_gappers = scanner_top_gappers(min_gap=0.07, direction="up") results = [r for r in all_gappers if CFG["penny_price_min"] <= (r.get("Price") or 0) <= CFG["penny_price_max"]] display(results, f"PENNY GAPPERS (${CFG['penny_price_min']}–${CFG['penny_price_max']})", sort_col="Chg%") return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 3 — LARGE CAP GAPPERS # ═══════════════════════════════════════════════════════════════════════════════ def scanner_large_cap_gappers() -> List[dict]: """ Large cap stocks (market cap ≥ $2 B) gapping > 7%. Institutional-grade moves — lower risk than penny but still powerful. """ print("[Scanner 3] Fetching Large Cap Gappers…") all_gappers = scanner_top_gappers(min_gap=0.07, direction="both") results = [r for r in all_gappers if (r.get("MarketCap") or 0) >= CFG["large_cap_mcap_min"]] display(results, f"LARGE CAP GAPPERS (MCap ≥ $2B)", sort_col="Chg%") return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 4 — LARGE CAP + EARNINGS GAPPERS # ═══════════════════════════════════════════════════════════════════════════════ def scanner_earnings_gappers() -> List[dict]: """ Large cap stocks gapping up that reported earnings within the last 48 hours. Earnings gap-ups are the cleanest momentum setups in large caps. """ print("[Scanner 4] Fetching Large Cap + Earnings Gappers…") if not HAS_YF: print("[SKIP] yfinance required for earnings dates.") return [] large_cap = scanner_large_cap_gappers() results = [] window = datetime.timedelta(hours=CFG["earnings_window_h"]) now = datetime.datetime.utcnow() for row in large_cap: sym = row["Ticker"] try: ticker = yf.Ticker(sym) cal = ticker.earnings_dates if cal is None or cal.empty: continue latest = cal.index[0] # Convert to naive UTC for comparison if hasattr(latest, "tz_localize"): latest = latest.tz_convert("UTC").replace(tzinfo=None) if (now - latest) <= window: row["EarningsDate"] = str(latest.date()) results.append(row) except Exception: pass display(results, "LARGE CAP + EARNINGS GAPPERS (within 48 h)", sort_col="Chg%") return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 5 — TOP RELATIVE VOLUME # ═══════════════════════════════════════════════════════════════════════════════ def scanner_top_rvol(min_rvol: float = 2.0, universe: Optional[List[str]] = None) -> List[dict]: """ Identifies stocks with highest intraday relative volume. RVOL = (current day's volume) / (expected volume at this time of day). universe: optional ticker list override — e.g. a live small-cap pull from fetch_smallcap_universe(). Defaults to CFG['universe'] when not given, so existing callers (breakout/reversal scanners, ad-hoc CLI runs) are unaffected. """ print("[Scanner 5] Fetching Top Relative Volume…") scan_universe = universe if universe else CFG["universe"] quotes = fetch_yf_quotes(scan_universe) results = [] for sym, info in quotes.items(): hist = fetch_yf_history(sym, period="2mo") row = build_base_row(sym, info, hist) if row["RVOL"] >= min_rvol: results.append(row) results = enrich_with_finviz(results) display(results, f"TOP RELATIVE VOLUME (RVOL ≥ {min_rvol}×)", sort_col="RVOL") return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 6 — TOP RSI (5-MINUTE) # ═══════════════════════════════════════════════════════════════════════════════ def scanner_top_rsi(mode: str = "both") -> List[dict]: """ Tracks stocks with highest (overbought) or lowest (oversold) 5-min RSI. mode: 'overbought' | 'oversold' | 'both' These extremes often precede sharp mean-reversion moves. """ print("[Scanner 6] Computing 5-Min RSI…") results_ob, results_os = [], [] for sym in CFG["universe"][:30]: # limit to avoid rate limit df = fetch_yf_intraday(sym, period="2d", interval="5m") if df.empty or len(df) < CFG["rsi_period"] + 5: continue closes = df["Close"].squeeze() rsi_series = compute_rsi(closes, CFG["rsi_period"]) latest_rsi = float(rsi_series.iloc[-1]) if math.isnan(latest_rsi): continue info = fetch_yf_quotes([sym]).get(sym, {}) hist = fetch_yf_history(sym, period="1mo") row = build_base_row(sym, info, hist) row["RSI5m"] = latest_rsi row["RVOL5m"] = _compute_rvol_5m(df) if latest_rsi >= CFG["rsi_overbought"] and mode in ("overbought","both"): results_ob.append(row) elif latest_rsi <= CFG["rsi_oversold"] and mode in ("oversold","both"): results_os.append(row) results_ob = sorted(results_ob, key=lambda x: x.get("RSI5m",0), reverse=True) results_os = sorted(results_os, key=lambda x: x.get("RSI5m",0)) if mode in ("overbought","both"): disp_rows = [{**r, "RSI": f"{r.get('RSI5m',0):.1f} 🔴"} for r in results_ob] display(disp_rows, f"TOP RSI — OVERBOUGHT (RSI ≥ {CFG['rsi_overbought']})", sort_col="RSI5m", ascending=False) if mode in ("oversold","both"): disp_rows = [{**r, "RSI": f"{r.get('RSI5m',0):.1f} 🟢"} for r in results_os] display(disp_rows, f"TOP RSI — OVERSOLD (RSI ≤ {CFG['rsi_oversold']})", sort_col="RSI5m", ascending=True) return results_ob + results_os def _compute_rvol_5m(df_5m: pd.DataFrame) -> float: """5-min RVOL: latest bar volume vs 10-day avg of the same 5-min slot.""" if df_5m.empty or len(df_5m) < 2: return float("nan") try: latest_vol = float(df_5m["Volume"].iloc[-1]) avg_vol_5m = float(df_5m["Volume"].iloc[-50:-1].mean()) # prior bars today return (latest_vol / avg_vol_5m) if avg_vol_5m > 0 else float("nan") except Exception: return float("nan") # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 7 — 5-MINUTE VOLUME LEADERS # ═══════════════════════════════════════════════════════════════════════════════ def scanner_5m_volume_leaders() -> List[dict]: """ Stocks with biggest 5-minute volume spikes right now. Key for spotting institutional entries in real time. """ print("[Scanner 7] Fetching 5-Min Volume Leaders…") results = [] for sym in CFG["universe"][:40]: df = fetch_yf_intraday(sym, period="1d", interval="5m") if df.empty or len(df) < 5: continue vol_5m = float(df["Volume"].iloc[-1]) avg_5m = float(df["Volume"].iloc[:-1].mean()) rvol_5m = (vol_5m / avg_5m) if avg_5m > 0 else 0 if rvol_5m < CFG["rvol_5min_min"]: continue info = fetch_yf_quotes([sym]).get(sym, {}) hist = fetch_yf_history(sym, period="1mo") row = build_base_row(sym, info, hist) row["Vol5m"] = vol_5m row["RVOL5m"] = rvol_5m results.append(row) results = enrich_with_finviz(results) display(results, "5-MIN VOLUME LEADERS", sort_col="RVOL5m") return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 8 — CONTINUATION SCANNER # ═══════════════════════════════════════════════════════════════════════════════ def scanner_continuation() -> List[dict]: """ Stocks that made large moves (≥15%) over the past 2 weeks. Useful for spotting: daily breakouts, flat tops, bull flag setups. Includes flag detection: tight 3-day range after large move. """ print("[Scanner 8] Running Continuation Scanner…") results = [] for sym in CFG["universe"]: hist = fetch_yf_history(sym, period="3mo") if hist.empty or len(hist) < CFG["continuation_days"] + 5: continue closes = hist["Close"].squeeze() price_now = float(closes.iloc[-1]) price_14d_ago= float(closes.iloc[-CFG["continuation_days"]]) if price_14d_ago <= 0: continue move_14d = (price_now - price_14d_ago) / price_14d_ago if abs(move_14d) < CFG["continuation_pct"]: continue # Bull flag detection: last 3 bars tight (< 2% range each) recent = hist.tail(3) hi = recent["High"].squeeze(); lo = recent["Low"].squeeze() flag_tight = all(((h - l) / l) < 0.02 for h, l in zip(hi, lo)) # Flat top: last 3 closes within 1% last3_closes = closes.tail(3) flat_top = (last3_closes.max() - last3_closes.min()) / last3_closes.mean() < 0.01 info = fetch_yf_quotes([sym]).get(sym, {}) row = build_base_row(sym, info, hist) row["Move14d%"] = move_14d * 100 row["BullFlag"] = "✅" if flag_tight else "—" row["FlatTop"] = "✅" if flat_top else "—" results.append(row) results = enrich_with_finviz(results) display(results, "CONTINUATION SCANNER (14-day movers ≥ 15%)", sort_col="Move14d%") return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 9 — AFTER-HOURS GAINERS # ═══════════════════════════════════════════════════════════════════════════════ def scanner_after_hours() -> List[dict]: """ Percentage gainers from the regular-session close. Only active after 4:00 PM ET. Uses Yahoo Finance extended-hours data. """ now = datetime.datetime.now() if now.hour < 16: print("[Scanner 9] After-Hours Gainers — market not yet closed (activates after 4 PM).") return [] print("[Scanner 9] Fetching After-Hours Gainers…") results = [] for sym in CFG["universe"]: if not HAS_YF: break try: ticker = yf.Ticker(sym) info_d = ticker.fast_info ah_price = getattr(info_d, "last_price", None) reg_close= getattr(info_d, "regular_market_previous_close", None) or \ getattr(info_d, "previous_close", None) if not ah_price or not reg_close: continue pct = (ah_price - reg_close) / reg_close * 100 if pct < 1.0: continue # min +1% to appear info = { "price": ah_price, "prev_close": reg_close, "day_volume": getattr(info_d, "three_month_average_volume", None), "volume": getattr(info_d, "three_month_average_volume", None), "market_cap": getattr(info_d, "market_cap", None), "52w_high": getattr(info_d, "year_high", None), "52w_low": getattr(info_d, "year_low", None), } hist = fetch_yf_history(sym, period="1mo") row = build_base_row(sym, info, hist) results.append(row) except Exception: pass results = enrich_with_finviz(results) display(results, "AFTER-HOURS GAINERS (post 4 PM)", sort_col="Chg%") return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 10 — LOW FLOAT / HIGH RVOL ALERT (< $20) # ═══════════════════════════════════════════════════════════════════════════════ def alert_low_float_high_rvol() -> List[dict]: """ ALERT: Stock priced < $20, float ≤ 20 M shares, RVOL ≥ 3×. The holy grail of small-cap day trading setups. Low float + high RVOL = explosive move potential. Scans a LIVE small-cap universe (fetch_smallcap_universe) rather than the static CFG['universe'], which is 90% mega-cap and structurally can't surface fresh low-float names. """ print("[Alert 10] Scanning Low Float / High RVOL (<$20)…") base = scanner_top_rvol(min_rvol=CFG["rvol_alert_min"], universe=fetch_smallcap_universe()) results = [] for row in base: price = row.get("Price") or 0 fl = row.get("Float") or float("nan") rvol = row.get("RVOL") or 0 if (price <= 20 and not math.isnan(fl) and fl <= CFG["low_float_max"] and rvol >= CFG["rvol_alert_min"]): row["ALERT"] = f"🚨 LOW FLOAT + RVOL {rvol:.1f}×" results.append(row) display(results, "🚨 ALERT: LOW FLOAT / HIGH RVOL (<$20, Float ≤ 20M, RVOL ≥ 3×)", sort_col="RVOL") return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 11 — LOW FLOAT FORMER MOMO STOCK # ═══════════════════════════════════════════════════════════════════════════════ def alert_low_float_former_momo() -> List[dict]: """ Highly sensitive scanner for low-float stocks with a history of massive spikes. Criteria: float ≤ 20 M, at least one 100%+ daily move in past 6 months, currently showing elevated RVOL (≥ 2×). Former momo stocks can wake up suddenly — catch them early. Scans a LIVE small/micro-cap universe instead of CFG['universe'] — price cap is widened (max_price=999) and market cap band lowered (0–$500M) since this screens for *historical* spike behavior, not today's price band. """ print("[Alert 11] Scanning Low Float Former Momo…") results = [] candidates = fetch_smallcap_universe(max_price=999, min_cap=0, max_cap=5e8) for sym in candidates: hist = fetch_yf_history(sym, period="6mo") if hist.empty or len(hist) < 10: continue closes = hist["Close"].squeeze() # Check for any single-day 100%+ move daily_chg = closes.pct_change() max_single_day = float(daily_chg.max()) * 100 if max_single_day < 50: # at least 50% single-day spike in 6 months continue info = fetch_yf_quotes([sym]).get(sym, {}) row = build_base_row(sym, info, hist) row["MaxDaySpike%"] = max_single_day results.append(row) results = enrich_with_finviz(results) # Filter: low float only results = [r for r in results if not math.isnan(r.get("Float") or float("nan")) and r.get("Float",float("nan")) <= CFG["low_float_max"]] display(results, "🔥 LOW FLOAT FORMER MOMO (float ≤ 20M, prior ≥50% single-day spike)", sort_col="MaxDaySpike%") return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 12 — SQUEEZE UP ALERT: 5% in 5 MINUTES (pre-alert) # ═══════════════════════════════════════════════════════════════════════════════ def alert_squeeze_5pct_5min() -> List[dict]: """ PRE-ALERT: Stock squeezing up ≥ 5% within the last 5-minute bar. These are the first warning signs of a bigger squeeze developing. Fires early so you can prepare before the main move. Scans a live small-cap universe (fetch_smallcap_universe) instead of the fixed CFG['universe'][:40]. """ print("[Alert 12] Scanning Squeeze 5% / 5min…") results = [] for sym in fetch_smallcap_universe(limit=40): df = fetch_yf_intraday(sym, period="1d", interval="5m") if df.empty or len(df) < 2: continue closes = df["Close"].squeeze() last_close = float(closes.iloc[-1]) prior_close = float(closes.iloc[-2]) if prior_close <= 0: continue move_5m = (last_close - prior_close) / prior_close if move_5m >= CFG["squeeze_5m_pct"]: info = fetch_yf_quotes([sym]).get(sym, {}) hist = fetch_yf_history(sym, period="1mo") row = build_base_row(sym, info, hist) row["Squeeze5m%"] = move_5m * 100 row["RVOL5m"] = _compute_rvol_5m(df) row["ALERT"] = f"⚡ +{move_5m*100:.1f}% in 5 min" results.append(row) display(results, "⚡ PRE-ALERT: SQUEEZE UP 5% IN 5 MIN", sort_col="Squeeze5m%") return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 13 — SQUEEZE UP ALERT: 10% in 10 MINUTES (full alert) # ═══════════════════════════════════════════════════════════════════════════════ def alert_squeeze_10pct_10min() -> List[dict]: """ FULL ALERT: Stock squeezing up ≥ 10% across the last two 5-minute bars. Confirmed momentum — high probability of continuation or halt. Prioritizes low float for most explosive candidates. Scans a live small-cap universe (fetch_smallcap_universe) instead of the fixed CFG['universe'][:40]. """ print("[Alert 13] Scanning Squeeze 10% / 10min…") results = [] for sym in fetch_smallcap_universe(limit=40): df = fetch_yf_intraday(sym, period="1d", interval="5m") if df.empty or len(df) < 3: continue closes = df["Close"].squeeze() last_close = float(closes.iloc[-1]) prior10m = float(closes.iloc[-3]) # 2 bars back = 10 min ago if prior10m <= 0: continue move_10m = (last_close - prior10m) / prior10m if move_10m >= CFG["squeeze_10m_pct"]: info = fetch_yf_quotes([sym]).get(sym, {}) hist = fetch_yf_history(sym, period="1mo") row = build_base_row(sym, info, hist) row["Squeeze10m%"] = move_10m * 100 row["RVOL5m"] = _compute_rvol_5m(df) row["ALERT"] = f"🚀 +{move_10m*100:.1f}% in 10 min" results.append(row) results = enrich_with_finviz(results) display(results, "🚀 FULL ALERT: SQUEEZE UP 10% IN 10 MIN", sort_col="Squeeze10m%") return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 14 — MEDIUM FLOAT GRINDER (< $20, algo-driven) # ═══════════════════════════════════════════════════════════════════════════════ def alert_medium_float_grinder_under20() -> List[dict]: """ Medium float (20M–100M shares) stocks under $20 with high RVOL. These are algorithm-driven slow grinders — less explosive than low float but more predictable. Good for scalping and multi-day holds. Scans a live small-cap universe (fetch_smallcap_universe) rather than the static CFG['universe']. """ print("[Alert 14] Scanning Medium Float Grinder (<$20)…") base = scanner_top_rvol(min_rvol=CFG["rvol_list_min"], universe=fetch_smallcap_universe()) results = [] for row in base: price = row.get("Price") or 0 fl = row.get("Float") or float("nan") if (price < 20 and not math.isnan(fl) and CFG["low_float_max"] < fl <= CFG["med_float_max"]): results.append(row) display(results, "🔵 MEDIUM FLOAT GRINDER (<$20, Float 20M–100M)", sort_col="RVOL") return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 15 — HIGH FLOAT GRINDER ($20+, algo-driven) # ═══════════════════════════════════════════════════════════════════════════════ def alert_high_float_grinder_over20() -> List[dict]: """ High float stocks (> 100M shares) priced $20+ with elevated RVOL. Institutional order flow drives these. Slower, larger moves. Great for options plays and trend trading. """ print("[Alert 15] Scanning High Float Grinder ($20+)…") base = scanner_top_rvol(min_rvol=CFG["rvol_list_min"]) results = [] for row in base: price = row.get("Price") or 0 fl = row.get("Float") or float("nan") if (price >= 20 and not math.isnan(fl) and fl > CFG["med_float_max"]): results.append(row) display(results, "⚪ HIGH FLOAT GRINDER ($20+, Float >100M)", sort_col="RVOL") return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 16 — 52-WEEK BREAKOUT / BLUE SKY ATH # ═══════════════════════════════════════════════════════════════════════════════ def alert_52_week_breakout() -> List[dict]: """ Alerts when a stock hits its 52-week high (breakout) or surpasses it into 'Blue Sky' territory (all-time high — no overhead resistance). 52-week breakouts have statistically significant positive drift. """ print("[Alert 16] Scanning 52-Week Breakouts / Blue Sky ATH…") results = [] quotes = fetch_yf_quotes(CFG["universe"]) for sym, info in quotes.items(): price = info.get("price") or 0 high52 = info.get("52w_high") or 0 if not price or not high52: continue pct_from_52h = (price - high52) / high52 * 100 if pct_from_52h < -0.5: # not at / near 52wk high continue hist = fetch_yf_history(sym, period="2y") row = build_base_row(sym, info, hist) is_ath = (price >= high52 * 0.999) # within 0.1% of 52w high row["52wHigh"] = high52 row["Pct52wHigh%"] = pct_from_52h row["BlueSky"] = "🔵 BLUE SKY" if is_ath else "🟡 NEAR HIGH" row["ALERT"] = "NEW 52W HIGH" if is_ath else f"Within {abs(pct_from_52h):.1f}%" results.append(row) results = enrich_with_finviz(results) display(results, "🔵 52-WEEK BREAKOUT / BLUE SKY ATH", sort_col="Pct52wHigh%", ascending=False) return results # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 17 — HALT ALERT (timestamped, NASDAQ halts feed) # ═══════════════════════════════════════════════════════════════════════════════ def alert_halt(watch: bool = False, poll_seconds: int = 30) -> None: """ Real-time halt monitor using NASDAQ's public trade halt RSS feed. Outputs a timestamped alert the exact moment a halt is detected. watch=True: runs in a continuous polling loop (press Ctrl+C to stop). """ seen_halts: set = set() print(f"\n[Alert 17] {'WATCHING' if watch else 'SNAPSHOT'} — HALT MONITOR") print(f"{'═'*70}") print(f" {'Time':10} {'Symbol':8} {'Halt Time':12} {'Reason':30}") print(f"{'─'*70}") def _check(): df = fetch_nasdaq_halts() if df.empty: return for _, row in df.iterrows(): sym = str(row.get("Symbol","")).strip() ht = str(row.get("HaltTime","")).strip() hd = str(row.get("HaltDate","")).strip() rsn = str(row.get("Reason","")).strip() key = f"{sym}_{hd}_{ht}" if key in seen_halts: continue seen_halts.add(key) ts = datetime.datetime.now().strftime("%H:%M:%S") rsn_friendly = { "LUDP": "Volatility (Up) 🔺", "LUDS": "Volatility (Down)🔻", "T1": "Regulatory", "T2": "Regulatory", "T5": "Single Price", "T6": "Extraordinary Event", "M": "Market Wide Circuit Breaker", }.get(rsn, rsn) color = Fore.GREEN if "Up" in rsn_friendly else (Fore.RED if "Down" in rsn_friendly else Fore.YELLOW) print(f" {color}{ts:10} {sym:8} {ht:12} {rsn_friendly:30}{Style.RESET_ALL}") if watch: print(f"Polling every {poll_seconds}s. Press Ctrl+C to stop.\n") try: while True: _check() time.sleep(poll_seconds) except KeyboardInterrupt: print("\n[Halt Monitor] Stopped.") else: _check() # ═══════════════════════════════════════════════════════════════════════════════ # SCANNER 18 — EXTREME REVERSAL SCANNER # ═══════════════════════════════════════════════════════════════════════════════ def scanner_extreme_reversal() -> List[dict]: """ Large Cap Reversal Scanner. Identifies stocks due for a top or bottom reversal. Criteria (ALL must be met for a high-confidence signal): A. ≥ N consecutive same-color 5-min candles (default N=7) B. Price trading OUTSIDE the 20-period Bollinger Band C. Most recent candle is a reversal candle: - Doji (open ≈ close, large wick) - Hammer / Shooting Star (long lower/upper wick) - Engulfing (current bar fully engulfs prior bar) Outputs: symbol, signal direction, consecutive count, BB deviation, candle type. """ print("[Scanner 18] Running Extreme Reversal Scanner…") results = [] for sym in CFG["universe"][:35]: df = fetch_yf_intraday(sym, period="2d", interval="5m") if df.empty or len(df) < CFG["bb_period"] + CFG["reversal_candles"] + 2: continue op = df["Open"].squeeze() hi = df["High"].squeeze() lo = df["Low"].squeeze() cl = df["Close"].squeeze() # A. Consecutive candles colors = ["green" if cl.iloc[i] > op.iloc[i] else "red" for i in range(len(cl))] consec = 1 for i in range(len(colors) - 2, -1, -1): if colors[i] == colors[-1]: consec += 1 else: break if consec < CFG["reversal_candles"]: continue signal_dir = "SHORT" if colors[-1] == "green" else "LONG" # B. Outside Bollinger Bands bb_up, bb_mid, bb_lo = compute_bollinger(cl, CFG["bb_period"], CFG["bb_std"]) last_close = float(cl.iloc[-1]) last_bb_up = float(bb_up.iloc[-1]) last_bb_lo = float(bb_lo.iloc[-1]) outside_bb = last_close > last_bb_up or last_close < last_bb_lo if not outside_bb: continue bb_dev = ((last_close - last_bb_up) if last_close > last_bb_up else (last_bb_lo - last_close)) # C. Reversal candle last_op = float(op.iloc[-1]) last_hi = float(hi.iloc[-1]) last_lo = float(lo.iloc[-1]) body = abs(last_close - last_op) rng = last_hi - last_lo candle_type = None if rng > 0: upper_wick = last_hi - max(last_close, last_op) lower_wick = min(last_close, last_op) - last_lo # Doji: body < 10% of range, wicks at least 40% if body / rng < 0.10 and (upper_wick + lower_wick) / rng > 0.80: candle_type = "Doji" # Hammer (bullish reversal): lower wick > 2× body, small upper wick elif lower_wick > 2 * body and upper_wick < body and signal_dir == "LONG": candle_type = "Hammer" # Shooting Star (bearish reversal): upper wick > 2× body elif upper_wick > 2 * body and lower_wick < body and signal_dir == "SHORT": candle_type = "Shooting Star" # Engulfing: current range > prior bar range elif len(df) >= 2: prior_body = abs(float(cl.iloc[-2]) - float(op.iloc[-2])) if body > prior_body * 1.2: candle_type = f"Engulfing ({'Bullish' if signal_dir=='LONG' else 'Bearish'})" if candle_type is None: continue # No reversal candle confirmation info = fetch_yf_quotes([sym]).get(sym, {}) hist = fetch_yf_history(sym, period="1mo") row = build_base_row(sym, info, hist) row["Signal"] = ("🟢 LONG (Bottom Rev)" if signal_dir == "LONG" else "🔴 SHORT (Top Rev)") row["Consec"] = consec row["BBDev"] = bb_dev row["CandleType"] = candle_type row["OutsideBB"] = ("Above" if last_close > last_bb_up else "Below") results.append(row) results = enrich_with_finviz(results) display(results, "↩️ EXTREME REVERSAL SCANNER (consec candles + BB + reversal candle)", sort_col="Consec") return results # ═══════════════════════════════════════════════════════════════════════════════ # MASTER RUN — ALL SCANNERS # ═══════════════════════════════════════════════════════════════════════════════ SCANNER_MAP = { "gappers": scanner_top_gappers, "penny": scanner_penny_gappers, "largecap": scanner_large_cap_gappers, "earnings": scanner_earnings_gappers, "rvol": scanner_top_rvol, "rsi": scanner_top_rsi, "vol5m": scanner_5m_volume_leaders, "continuation": scanner_continuation, "afterhours": scanner_after_hours, "lowfloat": alert_low_float_high_rvol, "formermomo": alert_low_float_former_momo, "squeeze5": alert_squeeze_5pct_5min, "squeeze10": alert_squeeze_10pct_10min, "medgrinder": alert_medium_float_grinder_under20, "highgrinder": alert_high_float_grinder_over20, "breakout52": alert_52_week_breakout, "reversal": scanner_extreme_reversal, } def run_all(): print(f"\n{'█'*90}") print(f" IMPACT TRADING ACADEMY — FULL SCANNER SUITE RUN [{datetime.datetime.now():%Y-%m-%d %H:%M:%S}]") print(f"{'█'*90}\n") for name, fn in SCANNER_MAP.items(): try: fn() except Exception as e: print(f"[ERROR] Scanner '{name}' failed: {e}") # Halt is handled separately (it's a watcher, not a list scanner) alert_halt(watch=False) # ═══════════════════════════════════════════════════════════════════════════════ # CLI ENTRY POINT # ═══════════════════════════════════════════════════════════════════════════════ if __name__ == "__main__": parser = argparse.ArgumentParser( description="Impact Trading Academy — Day-Trading Scanner Suite", formatter_class=argparse.RawTextHelpFormatter ) parser.add_argument( "--scanner", "-s", default="all", help=( "Which scanner to run. Options:\n" " all — run every scanner in sequence\n" " gappers — Top Gappers (>7%%)\n" " penny — Penny Gappers ($0.10–$5)\n" " largecap — Large Cap Gappers\n" " earnings — Large Cap + Earnings\n" " rvol — Top Relative Volume\n" " rsi — Top RSI (5-min)\n" " vol5m — 5-Min Volume Leaders\n" " continuation — Continuation Scanner\n" " afterhours — After-Hours Gainers\n" " lowfloat — Low Float / High RVOL Alert\n" " formermomo — Low Float Former Momo\n" " squeeze5 — Squeeze 5%% / 5 min\n" " squeeze10 — Squeeze 10%% / 10 min\n" " medgrinder — Medium Float Grinder\n" " highgrinder — High Float Grinder\n" " breakout52 — 52-Week Breakout\n" " halt — Halt Alert (live NASDAQ feed)\n" " reversal — Extreme Reversal Scanner\n" ) ) parser.add_argument( "--watch", "-w", action="store_true", help="For 'halt' scanner: run in continuous polling loop." ) parser.add_argument( "--sort", default=None, help="Sort column override (e.g. 'Chg%', 'RVOL', 'Float')." ) parser.add_argument( "--direction", default="both", choices=["up","down","both"], help="For gappers: 'up', 'down', or 'both'." ) args = parser.parse_args() if args.scanner == "all": run_all() elif args.scanner == "halt": alert_halt(watch=args.watch) elif args.scanner in SCANNER_MAP: fn = SCANNER_MAP[args.scanner] # Pass direction arg to gappers only if args.scanner == "gappers" and args.direction: scanner_top_gappers(direction=args.direction) else: fn() else: print(f"Unknown scanner: '{args.scanner}'. Use --help for options.") sys.exit(1)