""" Stock Screener — Bulk Market Screening Engine ============================================== Downloads stock data in bulk for a selected market, screens every stock against 8 technical criteria, scores and ranks them, and outputs the top N. Supported Markets: sp500 S&P 500 (US large-cap) nasdaq100 Nasdaq 100 (US tech-heavy) dow30 Dow Jones Industrial Average china_all All China A-shares (~5000+ stocks, Shanghai + Shenzhen) china_csi CSI 300 (Shanghai + Shenzhen blue-chips) china_sse50 SSE 50 (Shanghai top 50) hongkong Hang Seng Index (Hong Kong) ftse100 FTSE 100 (UK) dax40 DAX 40 (Germany) nikkei225 Nikkei 225 (Japan) custom Read tickers from a file Scoring (max 170 pts): C1 (10pt) Price > 50-day SMA C2 (10pt) 50-day SMA > 200-day SMA C3 (10pt) Up weeks > down weeks (15 weeks, Mon-Fri) C4 (var) High-volume score (50-day: 2x avg=5pt, 3x avg=10pt per day) C5 (10pt) Adaptive rising trend (ATR-based swing detection) C6 (20pt) Price >= 75% of 52-week intraday high C7 (20pt) Price >= 130% of 52-week intraday low C8 (20pt) Relative strength > 70 (50-day ratio method) C9 (30pt) Price between 1.2x and 1.5x of 52-week low C10 (30pt) 5-day volatility < 5% Usage: python stock_screener.py --market sp500 python stock_screener.py --market sp500 --top 20 --detail python stock_screener.py --market china_csi --export results.csv python stock_screener.py --market custom --file watchlist.txt python stock_screener.py --list-markets Dependencies: pip install yfinance pandas requests tabulate lxml openpyxl Optional (for best china_all results): pip install akshare """ import argparse import csv import math import os import sys import time from dataclasses import dataclass, field from typing import Optional try: import numpy as np except ImportError: sys.exit("ERROR: numpy is required.\n pip install numpy") try: import requests as req except ImportError: sys.exit("ERROR: requests is required.\n pip install requests") try: import yfinance as yf except ImportError: sys.exit("ERROR: yfinance is required.\n pip install yfinance") try: import pandas as pd except ImportError: sys.exit("ERROR: pandas is required.\n pip install pandas") try: from tabulate import tabulate except ImportError: tabulate = None # ═══════════════════════════════════════════════════════════════════════════════ # MARKET DEFINITIONS # ═══════════════════════════════════════════════════════════════════════════════ # Wikipedia blocks bare Python requests — need a browser User-Agent _HEADERS = { "User-Agent": ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36" ) } def _fetch_wiki_table(url: str, symbol_col: str = "Symbol") -> list[str]: """Fetch a stock list from a Wikipedia table, with proper User-Agent.""" try: resp = req.get(url, headers=_HEADERS, timeout=15) resp.raise_for_status() tables = pd.read_html(resp.text) for tbl in tables: for col in tbl.columns: # Match column name case-insensitively if str(col).strip().lower() == symbol_col.lower(): tickers = tbl[col].dropna().astype(str).tolist() tickers = [t.strip().replace("\n", "") for t in tickers if t.strip()] return tickers # If exact match fails, try partial match for tbl in tables: for col in tbl.columns: if symbol_col.lower() in str(col).lower(): tickers = tbl[col].dropna().astype(str).tolist() tickers = [t.strip().replace("\n", "") for t in tickers if t.strip()] return tickers print(f" Warning: '{symbol_col}' column not found at {url}", file=sys.stderr) return [] except Exception as e: print(f" Warning: Could not fetch from {url}: {e}", file=sys.stderr) return [] def get_sp500_tickers() -> list[str]: url = "https://en.wikipedia.org/wiki/List_of_S%26P_500_companies" tickers = _fetch_wiki_table(url, "Symbol") return [t.replace(".", "-") for t in tickers] def get_nasdaq100_tickers() -> list[str]: url = "https://en.wikipedia.org/wiki/Nasdaq-100" tickers = _fetch_wiki_table(url, "Ticker") if not tickers: tickers = _fetch_wiki_table(url, "Symbol") return [t.replace(".", "-") for t in tickers] def get_dow30_tickers() -> list[str]: url = "https://en.wikipedia.org/wiki/Dow_Jones_Industrial_Average" tickers = _fetch_wiki_table(url, "Symbol") return [t.replace(".", "-") for t in tickers] def get_ftse100_tickers() -> list[str]: url = "https://en.wikipedia.org/wiki/FTSE_100_Index" tickers = _fetch_wiki_table(url, "Ticker") if not tickers: tickers = _fetch_wiki_table(url, "EPIC") return [f"{t.strip()}.L" for t in tickers if t.strip()] def get_dax40_tickers() -> list[str]: url = "https://en.wikipedia.org/wiki/DAX" tickers = _fetch_wiki_table(url, "Ticker symbol") if not tickers: tickers = _fetch_wiki_table(url, "Ticker") return [f"{t.strip()}.DE" for t in tickers if t.strip()] def get_nikkei225_tickers() -> list[str]: url = "https://en.wikipedia.org/wiki/Nikkei_225" try: resp = req.get(url, headers=_HEADERS, timeout=15) resp.raise_for_status() tables = pd.read_html(resp.text) tickers = [] for tbl in tables: for col in tbl.columns: if "code" in str(col).lower() or "ticker" in str(col).lower(): codes = tbl[col].dropna().astype(str).tolist() for c in codes: c = c.strip().split(".")[0] if c.isdigit() and len(c) == 4: tickers.append(f"{c}.T") return tickers except Exception as e: print(f" Warning: Nikkei 225 fetch failed: {e}", file=sys.stderr) return [] def get_china_csi300_tickers() -> list[str]: shanghai = [ "600519", "601318", "600036", "600900", "601166", "600276", "600030", "600887", "600309", "601888", "600048", "600585", "601398", "601288", "601328", "601601", "601668", "600690", "601012", "600196", "600809", "601899", "603259", "600406", "601225", "600031", "600050", "600104", "603288", "601088", "600436", "601138", "600150", "601186", "600570", "600000", "601857", "601766", "600660", "600918", ] shenzhen = [ "000858", "000333", "000651", "002714", "000568", "000725", "002475", "000001", "002304", "000002", "002032", "002230", "002415", "000538", "000063", "002352", "002594", "000661", "002371", "002027", "300750", "300015", "300059", "300124", "300122", "300014", "300033", "300136", "300142", "300347", "300413", "300454", "300529", "300760", "300782", ] return [f"{t}.SS" for t in shanghai] + [f"{t}.SZ" for t in shenzhen] def get_china_sse50_tickers() -> list[str]: codes = [ "600519", "601318", "600036", "600900", "601166", "600276", "600030", "600887", "600309", "601888", "600048", "600585", "601398", "601288", "601328", "601601", "601668", "600690", "601012", "600196", "600809", "601899", "603259", "600406", "601225", "600031", "600050", "600104", "603288", "601088", "600436", "601138", "600150", "601186", "600570", "600000", "601857", "601766", "600660", "600918", "600089", "600111", "600176", "600346", "600588", "600763", "601006", "601169", "601211", "601229", ] return [f"{c}.SS" for c in codes] def get_china_all_ashares() -> list[str]: """ Full China A-share universe (~5000+ stocks). Strategy: 1. Try akshare (most reliable, returns live ticker list from EastMoney) 2. Fallback: generate all valid code ranges algorithmically Shanghai (.SS): 600xxx, 601xxx, 603xxx, 605xxx (main board), 688xxx (STAR/科创板) Shenzhen (.SZ): 000xxx, 001xxx, 002xxx, 003xxx (main+SME), 300xxx, 301xxx (ChiNext/创业板) """ # ── Method 1: akshare (if installed) ── try: import akshare as ak print(" Using akshare for live A-share ticker list...", file=sys.stderr) # Temporarily bypass proxy — akshare's eastmoney API often fails through proxies _proxy_vars = ("HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy", "ALL_PROXY") _saved = {k: os.environ.pop(k) for k in _proxy_vars if k in os.environ} try: df = ak.stock_zh_a_spot_em() finally: os.environ.update(_saved) codes = df["代码"].astype(str).tolist() tickers = [] for code in codes: code = code.strip() if not code or len(code) != 6 or not code.isdigit(): continue if code.startswith(("6",)): tickers.append(f"{code}.SS") # Shanghai elif code.startswith(("0", "3")): tickers.append(f"{code}.SZ") # Shenzhen if len(tickers) > 1000: print(f" akshare returned {len(tickers)} A-share tickers.", file=sys.stderr) return tickers print(f" akshare returned only {len(tickers)} tickers, falling back...", file=sys.stderr) except ImportError: print(" akshare not installed — using algorithmic ticker generation.", file=sys.stderr) print(" (For best results: pip install akshare)", file=sys.stderr) except Exception as e: print(f" akshare failed ({e}), using algorithmic fallback.", file=sys.stderr) # ── Method 2: Algorithmic generation of all valid code ranges ── tickers = [] # Shanghai main board: 600000-600999, 601000-601999, 603000-603999, 605000-605999 for prefix in ["600", "601", "603", "605"]: for i in range(1000): tickers.append(f"{prefix}{i:03d}.SS") # Shanghai STAR Market (科创板): 688000-688999 for i in range(1000): tickers.append(f"688{i:03d}.SS") # Shenzhen main board: 000001-000999, 001000-001999 for prefix in ["000", "001"]: for i in range(1000): code = f"{prefix}{i:03d}" if code == "000000": continue tickers.append(f"{code}.SZ") # Shenzhen SME board: 002000-002999, 003000-003999 for prefix in ["002", "003"]: for i in range(1000): tickers.append(f"{prefix}{i:03d}.SZ") # Shenzhen ChiNext (创业板): 300000-300999, 301000-301999 for prefix in ["300", "301"]: for i in range(1000): tickers.append(f"{prefix}{i:03d}.SZ") print(f" Generated {len(tickers)} candidate tickers (will filter by data availability).", file=sys.stderr) return tickers def get_hongkong_tickers() -> list[str]: codes = [ "0700", "9988", "0941", "1299", "2318", "0005", "1398", "3690", "0388", "2020", "0027", "1109", "0011", "0883", "2628", "0066", "0016", "1038", "0003", "0002", "0012", "0006", "0823", "1928", "2388", "0175", "0017", "1997", "9999", "0288", "2269", "0101", "0688", "1177", "2313", "0267", "0669", "1113", "2007", "0960", "0762", "0857", "1810", "9618", "9626", "2382", "0386", "3988", "1211", "0241", ] return [f"{c.zfill(4)}.HK" for c in codes] MARKETS = { "sp500": {"name": "S&P 500 (US)", "fetch": get_sp500_tickers, "benchmark": "SPY"}, "nasdaq100": {"name": "Nasdaq 100 (US)", "fetch": get_nasdaq100_tickers, "benchmark": "SPY"}, "dow30": {"name": "Dow Jones 30 (US)", "fetch": get_dow30_tickers, "benchmark": "SPY"}, "china_all": {"name": "All China A-shares (~5000+)","fetch": get_china_all_ashares, "benchmark": "000001.SS"}, "china_csi": {"name": "CSI 300 (China A-shares)", "fetch": get_china_csi300_tickers,"benchmark": "000001.SS"}, "china_sse50": {"name": "SSE 50 (Shanghai)", "fetch": get_china_sse50_tickers,"benchmark": "000001.SS"}, "hongkong": {"name": "Hang Seng (Hong Kong)", "fetch": get_hongkong_tickers, "benchmark": "^HSI"}, "ftse100": {"name": "FTSE 100 (UK)", "fetch": get_ftse100_tickers, "benchmark": "^FTSE"}, "dax40": {"name": "DAX 40 (Germany)", "fetch": get_dax40_tickers, "benchmark": "^GDAXI"}, "nikkei225": {"name": "Nikkei 225 (Japan)", "fetch": get_nikkei225_tickers, "benchmark": "^N225"}, } # Default benchmark when market is "custom" and no --benchmark is given DEFAULT_BENCHMARK = "SPY" # ═══════════════════════════════════════════════════════════════════════════════ # DATA STRUCTURES # ═══════════════════════════════════════════════════════════════════════════════ @dataclass class FailedTicker: """Record of a ticker that failed to download.""" ticker: str reason: str # e.g. "delisted", "insufficient data", "network error" retryable: bool # True if failure was transient (network) @dataclass class CriterionResult: id: int label: str points: int passed: bool score: int detail: str = "" @dataclass class StockResult: ticker: str company: str current_price: float sma_50: float sma_200: float week52_high: float week52_low: float up_weeks: int down_weeks: int high_vol_score: int trend_rising: bool relative_strength: float price_near_52w_low: bool volatility_5d: float criteria: list = field(default_factory=list) total_score: int = 0 error: Optional[str] = None CRITERIA_DEFS = [ {"id": 1, "label": "Price > 50-Day SMA", "points": 10}, {"id": 2, "label": "50-Day SMA > 200-Day SMA", "points": 10}, {"id": 3, "label": "Up Weeks > Down Weeks (15W)", "points": 10}, {"id": 4, "label": "High-Volume Score", "points": "var"}, {"id": 5, "label": "Adaptive Rising Trend (50D)", "points": 10}, {"id": 6, "label": "Price >= 75% of 52W High", "points": 20}, {"id": 7, "label": "Price >= 130% of 52W Low", "points": 20}, {"id": 8, "label": "Relative Strength > 70", "points": 20}, {"id": 9, "label": "Price 1.2x-1.5x of 52W Low", "points": 30}, {"id": 10, "label": "5-Day Volatility < 5%", "points": 30}, ] # Max score is not fixed due to variable C4, but we define a reference max MAX_SCORE = 160 # 10+10+10+0(var)+10+20+20+20+30+30 # ═══════════════════════════════════════════════════════════════════════════════ # BULK DATA DOWNLOAD — handles yfinance MultiIndex properly # ═══════════════════════════════════════════════════════════════════════════════ def _flatten_single_ticker_df(df: pd.DataFrame) -> pd.DataFrame: """ yfinance >= 0.2.31 returns MultiIndex columns even for single tickers: ('Close', 'AAPL'), ('High', 'AAPL'), ... This flattens them back to simple: Close, High, ... """ if isinstance(df.columns, pd.MultiIndex): # Drop the ticker level, keep just Price level df = df.copy() df.columns = df.columns.get_level_values(0) return df def download_single_ticker(ticker: str, period: str = "1y") -> Optional[pd.DataFrame]: """Download data for a single ticker, handling MultiIndex columns.""" try: df = yf.download(ticker, period=period, interval="1d", progress=False) if df.empty: return None df = _flatten_single_ticker_df(df) df = df[["Open", "High", "Low", "Close", "Volume"]].dropna() if len(df) < 50: return None return df except Exception: return None def _is_network_error(exc: Exception) -> bool: """Heuristic: check if an exception is likely a transient network issue.""" msg = str(exc).lower() network_keywords = [ "timeout", "timed out", "connection", "network", "unreachable", "reset by peer", "broken pipe", "temporary failure", "dns", "proxy", "ssl", "eof", "remote end closed", "max retries", "503", "502", "429", "read timed out", ] return any(kw in msg for kw in network_keywords) def _parse_batch_result( raw: pd.DataFrame, batch: list[str], all_data: dict, failures: list, ): """Parse a yfinance bulk-download result, populating all_data and failures.""" if raw.empty: for t in batch: if t not in all_data: failures.append(FailedTicker(t, "no data returned (possibly delisted)", False)) return if isinstance(raw.columns, pd.MultiIndex): try: available_tickers = raw.columns.get_level_values(1).unique().tolist() except Exception: available_tickers = [] for t in batch: if t in all_data: continue try: if t in available_tickers: df = pd.DataFrame({ "Open": raw[("Open", t)], "High": raw[("High", t)], "Low": raw[("Low", t)], "Close": raw[("Close", t)], "Volume": raw[("Volume", t)], }).dropna() if len(df) >= 50: all_data[t] = df elif len(df) > 0: failures.append(FailedTicker( t, f"insufficient data ({len(df)} days, need >=50)", False)) else: failures.append(FailedTicker(t, "all data NaN (possibly delisted)", False)) else: failures.append(FailedTicker(t, "delisted or invalid ticker", False)) except Exception as e: failures.append(FailedTicker(t, f"parse error: {e}", False)) else: # Non-MultiIndex (single ticker batch) if len(batch) == 1: t = batch[0] if t not in all_data: try: df = raw[["Open", "High", "Low", "Close", "Volume"]].dropna() if len(df) >= 50: all_data[t] = df elif len(df) > 0: failures.append(FailedTicker( t, f"insufficient data ({len(df)} days, need >=50)", False)) else: failures.append(FailedTicker(t, "all data NaN (possibly delisted)", False)) except Exception as e: failures.append(FailedTicker(t, f"parse error: {e}", False)) def download_bulk_data( tickers: list[str], period: str = "1y", verbose: bool = True, max_retries: int = 2, ) -> tuple[dict[str, pd.DataFrame], list[FailedTicker]]: """ Download OHLCV data for all tickers in bulk using yfinance. Handles the MultiIndex column structure that yfinance >= 0.2.31 returns. Classifies failures as permanent (delisted/invalid) or transient (network). Automatically retries transient failures. Returns: (dict mapping ticker -> DataFrame, list of FailedTicker records) """ if verbose: print(f" Downloading data for {len(tickers)} tickers...", file=sys.stderr) all_data = {} failures = [] # will be rebuilt after each pass batch_size = 50 network_failed_batches = [] # list of (batch, exception_msg) # ── Pass 1: initial bulk download ── for batch_start in range(0, len(tickers), batch_size): batch = tickers[batch_start:batch_start + batch_size] batch_num = batch_start // batch_size + 1 total_batches = (len(tickers) + batch_size - 1) // batch_size if verbose: print(f" Batch {batch_num}/{total_batches}: {len(batch)} tickers...", file=sys.stderr) try: raw = yf.download( tickers=batch, period=period, interval="1d", threads=True, progress=False, ) _parse_batch_result(raw, batch, all_data, failures) except Exception as e: if verbose: print(f" Batch {batch_num} failed: {e}", file=sys.stderr) if _is_network_error(e): network_failed_batches.append((batch, str(e))) else: for t in batch: if t not in all_data: failures.append(FailedTicker(t, f"download error: {e}", False)) if batch_start + batch_size < len(tickers): time.sleep(1) # ── Pass 2+: retry network failures ── for retry_round in range(1, max_retries + 1): # Collect tickers that failed due to network issues retryable = [f.ticker for f in failures if f.retryable] # Also add tickers from entire batches that failed due to network for batch, _ in network_failed_batches: for t in batch: if t not in all_data and t not in retryable: retryable.append(t) network_failed_batches.clear() if not retryable: break if verbose: print(f"\n Retry round {retry_round}: {len(retryable)} tickers with network errors...", file=sys.stderr) # Remove retryable items from failures list — they'll be re-evaluated failures = [f for f in failures if f.ticker not in retryable] time.sleep(3 * retry_round) # progressive back-off for batch_start in range(0, len(retryable), batch_size): batch = retryable[batch_start:batch_start + batch_size] batch_num = batch_start // batch_size + 1 total_batches = (len(retryable) + batch_size - 1) // batch_size if verbose: print(f" Retry batch {batch_num}/{total_batches}: {len(batch)} tickers...", file=sys.stderr) try: raw = yf.download( tickers=batch, period=period, interval="1d", threads=True, progress=False, ) _parse_batch_result(raw, batch, all_data, failures) except Exception as e: if verbose: print(f" Retry batch {batch_num} failed again: {e}", file=sys.stderr) if _is_network_error(e) and retry_round < max_retries: network_failed_batches.append((batch, str(e))) else: for t in batch: if t not in all_data: failures.append(FailedTicker(t, f"network error (after retries): {e}", False)) if batch_start + batch_size < len(retryable): time.sleep(2) # ── Mark any remaining network-failed-batch tickers ── for batch, emsg in network_failed_batches: for t in batch: if t not in all_data: already = {f.ticker for f in failures} if t not in already: failures.append(FailedTicker(t, f"network error (exhausted retries): {emsg}", False)) # ── Deduplicate failures (keep first reason) and remove any that succeeded ── seen = set() deduped = [] for f in failures: if f.ticker not in all_data and f.ticker not in seen: seen.add(f.ticker) deduped.append(f) failures = deduped if verbose: n_ok = len(all_data) n_fail = len(failures) print(f" Got data for {n_ok}/{len(tickers)} tickers. " f"({n_fail} failed)", file=sys.stderr) return all_data, failures # ═══════════════════════════════════════════════════════════════════════════════ # BULK COMPANY NAME LOOKUP # ═══════════════════════════════════════════════════════════════════════════════ def _fetch_names_akshare(tickers: list[str], verbose: bool = True) -> dict[str, str]: """ Fetch company names for China A-share tickers via akshare (one API call). Returns dict mapping ticker (e.g. '000001.SZ') -> company name. """ try: import akshare as ak if verbose: print(" Fetching company names via akshare (bulk)...", file=sys.stderr) # Bypass proxy for akshare _proxy_vars = ("HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy", "ALL_PROXY") _saved = {k: os.environ.pop(k) for k in _proxy_vars if k in os.environ} try: df = ak.stock_zh_a_spot_em() finally: os.environ.update(_saved) # Build code -> name mapping from akshare data # akshare columns: 代码 (code), 名称 (name) code_to_name = {} for _, row in df.iterrows(): code = str(row["代码"]).strip() name = str(row["名称"]).strip() if code and name: # Map to yfinance-style tickers if code.startswith("6"): code_to_name[f"{code}.SS"] = name elif code.startswith(("0", "3")): code_to_name[f"{code}.SZ"] = name # Match against requested tickers result = {} for t in tickers: if t in code_to_name: result[t] = code_to_name[t] if verbose: print(f" akshare returned names for {len(result)}/{len(tickers)} tickers.", file=sys.stderr) return result except ImportError: return {} except Exception as e: if verbose: print(f" akshare name lookup failed: {e}", file=sys.stderr) return {} def _fetch_names_yfinance(tickers: list[str], verbose: bool = True) -> dict[str, str]: """ Fetch company names via yfinance individual lookups (threaded). Used for non-China markets or as a fallback. """ from concurrent.futures import ThreadPoolExecutor, as_completed result = {} total = len(tickers) def _get_name(t): try: info = yf.Ticker(t).fast_info # fast_info doesn't always have the name; fall back to .info name = None # Try the lightweight Ticker object attributes first try: tk = yf.Ticker(t) # .info is cached per session; one HTTP call full_info = tk.info name = full_info.get("longName") or full_info.get("shortName") except Exception: pass return (t, name) except Exception: return (t, None) if verbose: print(f" Fetching company names via yfinance ({total} tickers, threaded)...", file=sys.stderr) with ThreadPoolExecutor(max_workers=8) as executor: futures = {executor.submit(_get_name, t): t for t in tickers} done_count = 0 for future in as_completed(futures): done_count += 1 if verbose and done_count % 200 == 0: print(f" ... names fetched: {done_count}/{total}", file=sys.stderr) ticker, name = future.result() if name: result[ticker] = name if verbose: print(f" Got names for {len(result)}/{total} tickers.", file=sys.stderr) return result def fetch_company_names( tickers: list[str], market_key: str = "", verbose: bool = True, ) -> dict[str, str]: """ Fetch company names for all tickers using the best available method. Returns dict mapping ticker -> company name. Tickers not found will be absent from the dict. """ names = {} is_china = market_key in ("china_all", "china_csi", "china_sse50") # Strategy 1: akshare for China A-shares (one call for all ~5000 names) if is_china: names = _fetch_names_akshare(tickers, verbose=verbose) # Strategy 2: yfinance for remaining tickers that have no name yet remaining = [t for t in tickers if t not in names] if remaining: yf_names = _fetch_names_yfinance(remaining, verbose=verbose) names.update(yf_names) return names # ═══════════════════════════════════════════════════════════════════════════════ # TECHNICAL ANALYSIS # ═══════════════════════════════════════════════════════════════════════════════ def compute_sma(closes: pd.Series, period: int) -> Optional[float]: if len(closes) < period: return None return float(closes.iloc[-period:].mean()) def compute_weekly_returns(df: pd.DataFrame, num_weeks: int = 15): """ Compute up/down weeks using Monday-Friday week boundaries. Friday close = weekly close. Current unfinished week uses latest close. Compare consecutive Friday closes to determine up/down. """ closes = df["Close"] if isinstance(closes, pd.DataFrame): closes = closes.iloc[:, 0] # Ensure datetime index try: idx = pd.DatetimeIndex(closes.index) except Exception: # Fallback: simple 5-day chunking if index is not datetime data = closes.values chunks = [] for i in range(0, len(data), 5): chunk = data[i:i + 5] if len(chunk) >= 1: chunks.append(float(chunk[-1])) chunks = chunks[-(num_weeks + 1):] up = sum(1 for i in range(1, len(chunks)) if chunks[i] > chunks[i - 1]) down = sum(1 for i in range(1, len(chunks)) if chunks[i] < chunks[i - 1]) return up, down closes = closes.copy() closes.index = idx # Build (year, week) keys manually to avoid isocalendar() DataFrame issues iso = idx.isocalendar() year_vals = iso.year.values week_vals = iso.week.values keys = list(zip(year_vals, week_vals)) # Group by (year, week) preserving order from collections import OrderedDict groups = OrderedDict() for i, key in enumerate(keys): groups.setdefault(key, []).append(i) weekly_closes = [] for (yr, wk), indices in groups.items(): sub = closes.iloc[indices] # Find Friday (dayofweek == 4) if available, else use last day fridays = sub[sub.index.dayofweek == 4] if len(fridays) > 0: weekly_closes.append(float(fridays.iloc[-1])) else: weekly_closes.append(float(sub.iloc[-1])) # Use only the most recent num_weeks+1 entries (need n+1 to get n comparisons) weekly_closes = weekly_closes[-(num_weeks + 1):] up, down = 0, 0 for i in range(1, len(weekly_closes)): if weekly_closes[i] > weekly_closes[i - 1]: up += 1 elif weekly_closes[i] < weekly_closes[i - 1]: down += 1 return up, down def compute_high_volume_score(volumes: pd.Series, lookback: int = 50) -> int: """ Score based on high-volume days in the most recent `lookback` trading days. Volume > 2x 50-day avg = 5 points per day. Volume > 3x 50-day avg = 10 points per day (replaces the 5). """ if isinstance(volumes, pd.DataFrame): volumes = volumes.iloc[:, 0] n = min(lookback, len(volumes)) v = volumes.iloc[-n:].values avg_vol = v.mean() if avg_vol <= 0: return 0 score = 0 for vol in v: if vol > avg_vol * 3: score += 10 elif vol > avg_vol * 2: score += 5 return score def compute_trend_structure(df: pd.DataFrame, lookback: int = 50) -> bool: """ Adaptive trend detection using ATR-based swing points. A swing high requires the high to be the maximum within an adaptive window sized by ATR. Similarly for swing lows. This captures meaningful swings that may take 10+ bars to form. """ closes = df["Close"] highs_col = df["High"] lows_col = df["Low"] if isinstance(closes, pd.DataFrame): closes = closes.iloc[:, 0] if isinstance(highs_col, pd.DataFrame): highs_col = highs_col.iloc[:, 0] if isinstance(lows_col, pd.DataFrame): lows_col = lows_col.iloc[:, 0] n = min(lookback, len(closes)) h = highs_col.iloc[-n:].values l = lows_col.iloc[-n:].values c = closes.iloc[-n:].values if n < 20: return False # Compute ATR to determine adaptive window size tr = np.maximum(h[1:] - l[1:], np.maximum(np.abs(h[1:] - c[:-1]), np.abs(l[1:] - c[:-1]))) atr = np.mean(tr[-14:]) if len(tr) >= 14 else np.mean(tr) avg_price = np.mean(c[-14:]) atr_pct = atr / avg_price if avg_price > 0 else 0.02 # Adaptive window: higher volatility -> larger window (min 5, max 15) win = int(np.clip(atr_pct * 500, 5, 15)) swing_highs = [] swing_lows = [] for i in range(win, n - win): left_start = max(0, i - win) right_end = min(n, i + win + 1) if h[i] == np.max(h[left_start:right_end]): swing_highs.append((i, h[i])) if l[i] == np.min(l[left_start:right_end]): swing_lows.append((i, l[i])) # Need at least 2 swing highs and 2 swing lows if len(swing_highs) < 2 or len(swing_lows) < 2: return False # Check if the most recent swing highs are rising recent_highs = [v for _, v in swing_highs[-3:]] recent_lows = [v for _, v in swing_lows[-3:]] highs_rising = all(recent_highs[i] > recent_highs[i - 1] for i in range(1, len(recent_highs))) lows_rising = all(recent_lows[i] > recent_lows[i - 1] for i in range(1, len(recent_lows))) return highs_rising and lows_rising def compute_relative_strength(stock_closes: pd.Series, bench_closes: pd.Series, lookback: int = 50) -> float: """ Compute relative strength score (1-99) using the ratio method: RS_ratio(t) = (P_s(t) / P_s(t0)) / (P_b(t) / P_b(t0)) over the most recent `lookback` trading days, with exponential weighting. Score > 70 indicates strong performance. """ n_s = min(lookback, len(stock_closes) - 1) n_b = min(lookback, len(bench_closes) - 1) n = min(n_s, n_b) if n < 5: return 50.0 s = stock_closes.iloc[-(n + 1):].values b = bench_closes.iloc[-(n + 1):].values s0 = s[0] b0 = b[0] if s0 <= 0 or b0 <= 0: return 50.0 # Compute daily RS ratios rs_ratios = np.array([(s[t] / s0) / (b[t] / b0) for t in range(1, n + 1)]) # Exponential weights (more weight on recent days) weights = np.exp(np.linspace(-1, 0, n)) weights /= weights.sum() weighted_rs = np.dot(rs_ratios, weights) # Map to 1-99 scale. RS ratio of 1.0 = 50, scale deviations. # A ratio of 1.10 (~10% outperformance) maps roughly to ~80 score = 50 + (weighted_rs - 1.0) * 300 return float(np.clip(score, 1.0, 99.0)) def compute_5day_volatility(closes: pd.Series) -> float: """Return the annualized volatility over the last 5 trading days as a percentage.""" if isinstance(closes, pd.DataFrame): closes = closes.iloc[:, 0] if len(closes) < 6: return 99.0 # not enough data, return high value recent = closes.iloc[-6:].values # 6 prices -> 5 returns returns = np.diff(recent) / recent[:-1] return float(np.std(returns) * 100) # daily std as percentage # ═══════════════════════════════════════════════════════════════════════════════ # SCORING ENGINE # ═══════════════════════════════════════════════════════════════════════════════ def evaluate_stock(ticker: str, df: pd.DataFrame, bench_closes: pd.Series, company_name: str = "") -> StockResult: """ Score a single stock against all criteria. Args: company_name: Pre-fetched company name. If empty, the ticker is used as a fallback (name resolution should be done externally via fetch_company_names). """ closes = df["Close"] highs_col = df["High"] lows_col = df["Low"] volumes = df["Volume"] # Ensure we're working with plain Series (not DataFrames from MultiIndex) if isinstance(closes, pd.DataFrame): closes = closes.iloc[:, 0] if isinstance(highs_col, pd.DataFrame): highs_col = highs_col.iloc[:, 0] if isinstance(lows_col, pd.DataFrame): lows_col = lows_col.iloc[:, 0] if isinstance(volumes, pd.DataFrame): volumes = volumes.iloc[:, 0] price = float(closes.iloc[-1]) sma50 = compute_sma(closes, 50) sma200 = compute_sma(closes, 200) # 1e/1f: Use intraday high/low for 52-week extremes w52h = float(highs_col.max()) w52l = float(lows_col.min()) # 1a: Stricter weekly returns using Mon-Fri weeks, 15 weeks up_w, dn_w = compute_weekly_returns(df, 15) # 1b: Volume scoring hv_score = compute_high_volume_score(volumes, 50) # 1c: Adaptive trend trend = compute_trend_structure(df, 50) # 1d: Ratio-based relative strength rs = compute_relative_strength(closes, bench_closes, 50) # 2a: Price relative to 52-week low near_low = w52l > 0 and (1.2 * w52l <= price <= 1.5 * w52l) # 2b: 5-day volatility vol_5d = compute_5day_volatility(closes) company = company_name if company_name else ticker cr = [] c1 = sma50 is not None and price > sma50 cr.append(CriterionResult(1, CRITERIA_DEFS[0]["label"], 10, c1, 10 if c1 else 0, f"${price:.2f} {'>' if c1 else '<='} SMA50 ${sma50:.2f}" if sma50 else "N/A")) c2 = sma50 is not None and sma200 is not None and sma50 > sma200 cr.append(CriterionResult(2, CRITERIA_DEFS[1]["label"], 10, c2, 10 if c2 else 0, f"SMA50 ${sma50:.2f} {'>' if c2 else '<='} SMA200 ${sma200:.2f}" if sma50 and sma200 else "N/A")) c3 = up_w > dn_w cr.append(CriterionResult(3, CRITERIA_DEFS[2]["label"], 10, c3, 10 if c3 else 0, f"{up_w} up vs {dn_w} down (15W)")) # C4: variable score based on high-volume days c4_passed = hv_score > 0 cr.append(CriterionResult(4, CRITERIA_DEFS[3]["label"], hv_score, c4_passed, hv_score, f"Volume score: {hv_score} pts")) c5 = trend cr.append(CriterionResult(5, CRITERIA_DEFS[4]["label"], 10, c5, 10 if c5 else 0, "Rising (adaptive)" if c5 else "Not rising")) c6 = w52h > 0 and price >= 0.75 * w52h cr.append(CriterionResult(6, CRITERIA_DEFS[5]["label"], 20, c6, 20 if c6 else 0, f"{price/w52h*100:.1f}% of high ${w52h:.2f} (need >=75%)")) c7 = w52l > 0 and price >= 1.30 * w52l cr.append(CriterionResult(7, CRITERIA_DEFS[6]["label"], 20, c7, 20 if c7 else 0, f"{price/w52l*100:.1f}% of low ${w52l:.2f} (need >=130%)")) c8 = rs > 70 cr.append(CriterionResult(8, CRITERIA_DEFS[7]["label"], 20, c8, 20 if c8 else 0, f"RS = {rs:.1f} (need >70)")) # C9: Price relative to 52-week low (1.2x to 1.5x) c9 = near_low cr.append(CriterionResult(9, CRITERIA_DEFS[8]["label"], 30, c9, 30 if c9 else 0, f"Price/52WL = {price/w52l:.2f}x (need 1.2-1.5x)" if w52l > 0 else "N/A")) # C10: 5-day volatility < 5% c10 = vol_5d < 5.0 cr.append(CriterionResult(10, CRITERIA_DEFS[9]["label"], 30, c10, 30 if c10 else 0, f"5D vol = {vol_5d:.2f}% (need <5%)")) total = sum(c.score for c in cr) return StockResult( ticker=ticker, company=company, current_price=price, sma_50=sma50 or 0, sma_200=sma200 or 0, week52_high=w52h, week52_low=w52l, up_weeks=up_w, down_weeks=dn_w, high_vol_score=hv_score, trend_rising=trend, relative_strength=rs, price_near_52w_low=near_low, volatility_5d=vol_5d, criteria=cr, total_score=total, ) # ═══════════════════════════════════════════════════════════════════════════════ # MAIN SCREENER # ═══════════════════════════════════════════════════════════════════════════════ def screen_market(market_key: str, tickers: list[str] = None, top_n: int = 10, verbose: bool = True, benchmark: str = None) -> tuple[list[StockResult], list[FailedTicker]]: """ Run the screener on the given market. Args: benchmark: Ticker symbol for the benchmark index. If None, the market's default benchmark is used (e.g. SPY for US, 000001.SS for China). Returns (scored_results, download_failures). """ if tickers: ticker_list = tickers elif market_key in MARKETS: market_info = MARKETS[market_key] if verbose: print(f"\n{'='*60}", file=sys.stderr) print(f" Market: {market_info['name']}", file=sys.stderr) print(f"{'='*60}", file=sys.stderr) print(f" Fetching ticker list...", file=sys.stderr) ticker_list = market_info["fetch"]() if verbose: print(f" Found {len(ticker_list)} tickers.", file=sys.stderr) else: print(f"ERROR: Unknown market '{market_key}'.", file=sys.stderr) return [], [] if not ticker_list: print("ERROR: No tickers found.", file=sys.stderr) return [], [] # Determine benchmark ticker if benchmark is None: if market_key in MARKETS: benchmark = MARKETS[market_key]["benchmark"] else: benchmark = DEFAULT_BENCHMARK # Bulk download (with retry for network failures) stock_data, download_failures = download_bulk_data( ticker_list, period="1y", verbose=verbose) # Benchmark download if verbose: print(f" Fetching benchmark ({benchmark})...", file=sys.stderr) bench_df = download_single_ticker(benchmark, period="1y") bench_closes = bench_df["Close"] if bench_df is not None else pd.Series(dtype=float) if bench_df is None and verbose: print(f" WARNING: Could not download benchmark {benchmark}. " f"RS scores will default to 50.", file=sys.stderr) # Fetch company names in bulk (always, regardless of universe size) stock_tickers = list(stock_data.keys()) name_map = fetch_company_names(stock_tickers, market_key=market_key, verbose=verbose) # Evaluate if verbose: print(f" Scoring {len(stock_data)} stocks...", file=sys.stderr) results = [] eval_failed = 0 t_start = time.time() for i, (ticker, df) in enumerate(stock_data.items()): if verbose and (i + 1) % 100 == 0: elapsed = time.time() - t_start rate = (i + 1) / elapsed if elapsed > 0 else 0 remaining = (len(stock_data) - i - 1) / rate if rate > 0 else 0 print(f" ... scored {i+1}/{len(stock_data)} " f"({rate:.0f}/sec, ~{remaining:.0f}s remaining)", file=sys.stderr) try: result = evaluate_stock(ticker, df, bench_closes, company_name=name_map.get(ticker, "")) results.append(result) except Exception as e: eval_failed += 1 results.append(StockResult( ticker=ticker, company=name_map.get(ticker, ticker), current_price=0, sma_50=0, sma_200=0, week52_high=0, week52_low=0, up_weeks=0, down_weeks=0, high_vol_score=0, trend_rising=False, relative_strength=0, price_near_52w_low=False, volatility_5d=99.0, criteria=[], total_score=0, error=str(e), )) results.sort(key=lambda r: (-r.total_score, r.ticker)) if verbose: valid = len([r for r in results if r.error is None]) print(f"\n Done: {valid} scored, {eval_failed} eval errors, " f"{len(download_failures)} download failures.", file=sys.stderr) return results, download_failures # ═══════════════════════════════════════════════════════════════════════════════ # DISPLAY # ═══════════════════════════════════════════════════════════════════════════════ def format_results(results: list[StockResult], top_n: int = 10) -> str: if not results: return "No results to display." lines = ["", "=" * 72, " STOCK SCREENER — RESULTS", "=" * 72] valid = [r for r in results if r.error is None] top = valid[:top_n] if top: lines += ["", f" ★ TOP {len(top)} STOCKS:", ""] for i, r in enumerate(top): bar_len = int(r.total_score / MAX_SCORE * 30) bar = "█" * bar_len + "░" * (30 - bar_len) lines.append( f" {i+1:>3}. {r.ticker:<10} {r.total_score:>3}/{MAX_SCORE} " f"{bar} ${r.current_price:>9.2f} {r.company[:30]}" ) lines += ["", "-" * 72, ""] if tabulate and valid: rows = [] for i, r in enumerate(valid): passed = sum(1 for c in r.criteria if c.passed) rows.append([i+1, r.ticker, r.company[:25], f"${r.current_price:.2f}", r.total_score, f"{passed}/10", f"{r.relative_strength:.0f}"]) lines.append(tabulate(rows, headers=["#", "Ticker", "Company", "Price", "Score", "Pass", "RS"], tablefmt="simple", stralign="left", numalign="right")) elif valid: for i, r in enumerate(valid): lines.append(f" {i+1:>3} {r.ticker:<10} {r.total_score:>3}/{MAX_SCORE}" f" ${r.current_price:>9.2f} {r.company[:30]}") lines.append("") return "\n".join(lines) def format_detail(r: StockResult) -> str: if r.error: return f"\n {r.ticker}: ERROR — {r.error}\n" lines = [ "", f" {'─'*60}", f" {r.ticker} — {r.company}", f" {'─'*60}", f" Price: ${r.current_price:.2f} Score: {r.total_score}/{MAX_SCORE}", f" SMA50: ${r.sma_50:.2f} SMA200: ${r.sma_200:.2f}", f" 52W High: ${r.week52_high:.2f} 52W Low: ${r.week52_low:.2f}", f" Up/Down Weeks: {r.up_weeks}↑ / {r.down_weeks}↓ HiVol Score: {r.high_vol_score}", f" Relative Strength: {r.relative_strength:.1f}", "", ] for c in r.criteria: icon = "✓" if c.passed else "✗" pts = f"+{c.score:>2}" if c.passed else " 0" lines.append(f" {icon} C{c.id} {c.label:<30} {pts}pt {c.detail}") lines.append("") return "\n".join(lines) def export_csv(results: list[StockResult], filepath: str): with open(filepath, "w", newline="") as f: w = csv.writer(f) w.writerow(["Rank","Ticker","Company","Price","Total_Score", "SMA50","SMA200","52W_High","52W_Low", "Up_Wks","Down_Wks","HiVol_Score","Trend","RS", "Vol_5D","Near_52WL", "C1","C2","C3","C4","C5","C6","C7","C8","C9","C10","Error"]) for i, r in enumerate(results): cv = [c.score for c in r.criteria] if r.criteria else [0]*10 w.writerow([i+1, r.ticker, r.company, f"{r.current_price:.2f}", r.total_score, f"{r.sma_50:.2f}", f"{r.sma_200:.2f}", f"{r.week52_high:.2f}", f"{r.week52_low:.2f}", r.up_weeks, r.down_weeks, r.high_vol_score, r.trend_rising, f"{r.relative_strength:.1f}", f"{r.volatility_5d:.2f}", r.price_near_52w_low, *cv, r.error or ""]) print(f" Exported CSV to {filepath}", file=sys.stderr) def export_excel(results: list[StockResult], filepath: str, download_failures: list[FailedTicker] = None): """ Export screening results to an Excel file. Sheet 1 ("Screening Results"): scored stocks with per-criterion detail. Sheet 2 ("Failed Downloads"): tickers that could not be downloaded. """ # ── Sheet 1: Screening Results ── scored_rows = [] for i, r in enumerate(results): row = { "Rank": i + 1, "Ticker": r.ticker, "Company": r.company, "Price": r.current_price, "Total Score": r.total_score, "SMA50": r.sma_50, "SMA200": r.sma_200, "52W High": r.week52_high, "52W Low": r.week52_low, "Up Weeks": r.up_weeks, "Down Weeks": r.down_weeks, "Relative Strength": r.relative_strength, "5D Volatility %": r.volatility_5d, } # Add each criterion's score and detail for c in r.criteria: row[f"C{c.id} Score"] = c.score row[f"C{c.id} Detail"] = c.detail if r.error: row["Error"] = r.error scored_rows.append(row) df_scored = pd.DataFrame(scored_rows) # ── Sheet 2: Failed Downloads ── fail_rows = [] if download_failures: for f in download_failures: fail_rows.append({"Ticker": f.ticker, "Failure Reason": f.reason}) df_fails = pd.DataFrame(fail_rows) if fail_rows else pd.DataFrame( columns=["Ticker", "Failure Reason"]) # ── Write both sheets ── with pd.ExcelWriter(filepath, engine="openpyxl") as writer: df_scored.to_excel(writer, sheet_name="Screening Results", index=False) df_fails.to_excel(writer, sheet_name="Failed Downloads", index=False) print(f" Exported Excel to {filepath} " f"({len(scored_rows)} scored, {len(fail_rows)} failed)", file=sys.stderr) # ═══════════════════════════════════════════════════════════════════════════════ # CLI # ═══════════════════════════════════════════════════════════════════════════════ def main(): parser = argparse.ArgumentParser( description="Stock Screener — Bulk market screening engine", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" examples: python stock_screener.py --market sp500 python stock_screener.py --market sp500 --top 20 --detail python stock_screener.py --market china_csi --export results.csv python stock_screener.py --market custom --file watchlist.txt --benchmark 000001.SS python stock_screener.py --list-markets """, ) parser.add_argument("--market", "-m", type=str) parser.add_argument("--file", "-f", type=str) parser.add_argument("--benchmark", "-b", type=str, default=None, help="Benchmark ticker for relative strength " "(default: auto-selected per market, e.g. SPY for US, 000001.SS for China)") parser.add_argument("--top", "-n", type=int, default=10) parser.add_argument("--detail", "-d", action="store_true") parser.add_argument("--export", "-e", type=str, help="Export to CSV file") parser.add_argument("--excel", "-x", type=str, help="Export to Excel (.xlsx) file") parser.add_argument("--quiet", "-q", action="store_true") parser.add_argument("--list-markets", action="store_true") args = parser.parse_args() if args.list_markets: print("\nAvailable markets:\n") print(f" {'KEY':<15} {'MARKET':<35} {'BENCHMARK'}") print(f" {'-'*15} {'-'*35} {'-'*12}") for key, info in MARKETS.items(): print(f" {key:<15} {info['name']:<35} {info['benchmark']}") print(f"\n {'custom':<15} {'Read tickers from --file':<35} " f"(default: {DEFAULT_BENCHMARK}, override with --benchmark)\n") sys.exit(0) if not args.market: parser.print_help() sys.exit(1) custom_tickers = None if args.market == "custom": if not args.file: sys.exit("ERROR: --market custom requires --file ") try: with open(args.file) as f: custom_tickers = [l.strip() for l in f if l.strip() and not l.startswith("#")] except FileNotFoundError: sys.exit(f"ERROR: File not found: {args.file}") results, download_failures = screen_market( args.market, tickers=custom_tickers, top_n=args.top, verbose=not args.quiet, benchmark=args.benchmark) print(format_results(results, top_n=args.top)) if args.detail: valid = [r for r in results if r.error is None] for r in valid[:args.top]: print(format_detail(r)) if args.export: export_csv(results, args.export) if args.excel: export_excel(results, args.excel, download_failures=download_failures) if __name__ == "__main__": main()