from __future__ import annotations import gc import json import logging import time from dataclasses import dataclass, field from datetime import date, datetime, timedelta, timezone from pathlib import Path from typing import Any, Dict, List, Optional from trading.market_registry import get_scan_results_path, to_provider_symbol from trading.universe_provider import get_market_universe def _default_backtest_end() -> str: """Return the most recent completed trading weekday as YYYY-MM-DD.""" d = date.today() - timedelta(days=1) while d.weekday() >= 5: # Saturday=5, Sunday=6 d -= timedelta(days=1) return d.isoformat() @dataclass(frozen=True) class ScanConfig: market_id: str universe_name: str title: str symbols: Optional[List[str]] = None max_symbols: Optional[int] = None min_data_days: int = 252 min_avg_volume: int = 500_000 min_price: float = 1.0 backtest_start: str = "2023-01-01" backtest_end: str = field(default_factory=_default_backtest_end) train_window: int = 252 days_ahead: int = 7 # --- Eligibility gates (tightened to resist overfit / selection bias) --- # Rationale: backtest dir_acc ~57% historically collapsed to 37% live, the # signature of an edge barely above coin-flip being amplified by picking the # best N of ~100 stocks. Stricter full-period gates + a recent out-of-sample # (OOS) holdout gate require the edge to PERSIST on unseen recent data. min_dir_acc: float = 0.55 # was 0.53 — must clear coin-flip noise band min_sharpe: float = 0.30 # was 0.0 — demand real risk-adjusted edge min_hit_rate: float = 45.0 # was 40.0 min_trades: int = 8 # was 3 — kills 4-trade "100% hit" noise # OOS holdout gates (recent 30% of eval window the selection never optimised on) min_oos_dir_acc: float = 0.52 # recent edge must stay above coin-flip require_oos: bool = True # if no holdout could be computed → exclude min_oos_return_pct: float = -2.0 # recent strategy return not deeply negative # Chunked / resumable scan support: if set, run_scan() processes at most # N additional Stage 2 stocks per call and then returns without marking # `completed=True`. Subsequent calls resume where it left off (work file # holds progress). Used by HF free-tier external cron pinger. max_stage2_per_run: Optional[int] = None @property def final_results_file(self) -> Path: return get_scan_results_path(self.market_id, completed=True) @property def work_results_file(self) -> Path: return get_scan_results_path(self.market_id, completed=False) @property def universe_key(self) -> str: parts = [self.market_id, self.universe_name] if self.symbols: parts.append("symbols=" + ",".join(self.symbols)) if self.max_symbols: parts.append(f"max={self.max_symbols}") return "|".join(parts) def _load_results(config: ScanConfig) -> Dict[str, Any]: src = None if config.work_results_file.exists(): src = config.work_results_file elif config.final_results_file.exists(): src = config.final_results_file if src is not None and src.exists(): try: return json.loads(src.read_text()) except Exception: pass return { "market_id": config.market_id, "scan_started": None, "universe": None, "universe_key": None, "stage1": {}, "stage2": {}, "completed": False, } def _save_results(config: ScanConfig, data: Dict[str, Any]) -> None: config.work_results_file.parent.mkdir(parents=True, exist_ok=True) data["market_id"] = config.market_id data["updated_at"] = datetime.now(timezone.utc).isoformat() tmp = config.work_results_file.with_suffix(".tmp") tmp.write_text(json.dumps(data, indent=2, default=str)) tmp.rename(config.work_results_file) def _publish_final_if_complete(config: ScanConfig, data: Dict[str, Any]) -> None: if not data.get("completed", False): return config.final_results_file.parent.mkdir(parents=True, exist_ok=True) tmp = config.final_results_file.with_suffix(".tmp") tmp.write_text(json.dumps(data, indent=2, default=str)) tmp.rename(config.final_results_file) def stage1_prefilter(symbol: str, config: ScanConfig) -> Dict[str, Any]: import yfinance as yf ticker = to_provider_symbol(symbol, market_id=config.market_id) result: Dict[str, Any] = { "market_id": config.market_id, "symbol": symbol, "provider_symbol": ticker, "passed": False, "reason": "", "avg_volume": 0.0, "data_days": 0, "last_price": 0.0, "checked_at": datetime.now(timezone.utc).isoformat(), } try: df = yf.download(ticker, period="2y", progress=False, auto_adjust=True) if df is None or df.empty: result["reason"] = "no_data" return result data_days = len(df) result["data_days"] = data_days if data_days < config.min_data_days: result["reason"] = f"insufficient_data ({data_days} < {config.min_data_days} days)" return result avg_vol = 0.0 if "Volume" in df.columns: avg_vol_raw = df["Volume"].mean() avg_vol = float(avg_vol_raw.iloc[0]) if hasattr(avg_vol_raw, "iloc") else float(avg_vol_raw) result["avg_volume"] = round(avg_vol, 0) if avg_vol < config.min_avg_volume: result["reason"] = f"low_volume ({avg_vol:,.0f} < {config.min_avg_volume:,.0f})" return result last_price = float(df["Close"].iloc[-1].iloc[0]) if hasattr(df["Close"].iloc[-1], "iloc") else float(df["Close"].iloc[-1]) result["last_price"] = round(last_price, 2) if last_price < config.min_price: result["reason"] = f"below_min_price (price={last_price:.2f} < {config.min_price:.2f})" return result result["passed"] = True result["reason"] = "OK" return result except Exception as exc: result["reason"] = f"error: {exc}" return result def stage2_backtest(symbol: str, config: ScanConfig, logger: logging.Logger) -> Dict[str, Any]: from analysis.walk_forward_backtest import walk_forward_backtest result: Dict[str, Any] = { "market_id": config.market_id, "symbol": symbol, "eligible": False, "reason": "", "evaluated_at": datetime.now(timezone.utc).isoformat(), } try: _, metrics = walk_forward_backtest( symbol=symbol, start_date=config.backtest_start, end_date=config.backtest_end, market_id=config.market_id, days_ahead=config.days_ahead, train_window=config.train_window, model_type="rf", use_technical_gate=True, initial_capital=100_000.0, commission_bps=10.0, slippage_bps=10.0, exit_rule="signal_or_fixed", max_hold_days=config.days_ahead, stop_loss_pct=0.05, take_profit_pct=0.10, trailing_stop_pct=0.07, max_position_pct=0.50, max_risk_per_trade_pct=0.02, ) dir_acc = metrics["direction_accuracy"] sharpe = metrics["sharpe"] hit_rate = metrics.get("hit_rate_pct", 0.0) total_ret = metrics["total_return_pct"] trades = metrics["trades_count"] max_dd = metrics.get("max_drawdown_pct", 0.0) oos_dir_acc = metrics.get("oos_direction_accuracy") oos_samples = metrics.get("oos_samples", 0) oos_ret = metrics.get("oos_return_pct") result.update({ "dir_acc": round(dir_acc, 4), "sharpe": round(sharpe, 3), "hit_rate": round(hit_rate, 1), "total_return_pct": round(total_ret, 1), "trades": trades, "max_drawdown_pct": round(max_dd, 1), "oos_dir_acc": round(oos_dir_acc, 4) if oos_dir_acc is not None else None, "oos_samples": oos_samples, "oos_return_pct": round(oos_ret, 1) if oos_ret is not None else None, }) reasons: List[str] = [] # --- Full-period gates --- if dir_acc < config.min_dir_acc: reasons.append(f"dir_acc={dir_acc:.1%}<{config.min_dir_acc:.0%}") if sharpe < config.min_sharpe: reasons.append(f"sharpe={sharpe:.3f}<{config.min_sharpe}") if hit_rate < config.min_hit_rate: reasons.append(f"hit_rate={hit_rate:.1f}%<{config.min_hit_rate}%") if trades < config.min_trades: reasons.append(f"trades={trades}<{config.min_trades}") # --- Out-of-sample holdout gates (the real overfit defense) --- if oos_dir_acc is None: if config.require_oos: reasons.append("no_oos_holdout(insufficient_recent_samples)") else: if oos_dir_acc < config.min_oos_dir_acc: reasons.append(f"oos_dir_acc={oos_dir_acc:.1%}<{config.min_oos_dir_acc:.0%}") if oos_ret is not None and oos_ret < config.min_oos_return_pct: reasons.append(f"oos_ret={oos_ret:.1f}%<{config.min_oos_return_pct}%") result["eligible"] = len(reasons) == 0 result["reason"] = "; ".join(reasons) if reasons else "OK" return result except Exception as exc: result["reason"] = f"backtest_error: {exc}" logger.warning("Stage 2 failed for %s: %s", symbol, exc) return result def get_universe(config: ScanConfig, logger: logging.Logger) -> List[str]: if config.symbols: symbols = list(dict.fromkeys(config.symbols)) if config.max_symbols: symbols = symbols[: config.max_symbols] logger.info("Using explicit symbol subset for %s/%s: %s", config.market_id, config.universe_name, ", ".join(symbols)) return symbols try: result = get_market_universe(config.market_id, config.universe_name) symbols = result.symbols if config.max_symbols: symbols = symbols[: config.max_symbols] logger.info( "Fetched %d stocks from %s/%s, limited to first %d symbols", len(result.symbols), result.market_id, result.name, len(symbols), ) else: logger.info("Fetched %d stocks from %s/%s", len(result.symbols), result.market_id, result.name) return symbols except Exception as exc: logger.error("Failed to fetch universe %s/%s: %s", config.market_id, config.universe_name, exc) if config.market_id == "bist": return [ "THYAO", "AKBNK", "GARAN", "EREGL", "SISE", "TUPRS", "KCHOL", "ASELS", "BIMAS", "SAHOL", "YKBNK", "HALKB", "VAKBN", "TCELL", "ARCLK", "PETKM", "TOASO", "KOZAA", "KOZAL", "SASA", "TAVHL", "TTKOM", "ENKAI", "FROTO", "EKGYO", "PGSUS", "SOKM", "DOHOL", "GUBRF", "ISCTR", ] if config.market_id == "us": return [ "AAPL", "MSFT", "AMZN", "GOOGL", "META", "NVDA", "TSLA", "BRK-B", "JPM", "JNJ", "V", "UNH", "PG", "HD", "MA", "DIS", "BAC", "XOM", "PFE", "CSCO", "ABBV", "AVGO", "KO", "PEP", "TMO", "COST", "MRK", "WMT", "ABT", "CRM", ] return [] def run_scan(config: ScanConfig, logger: logging.Logger, *, force: bool = False, stage1_only: bool = False) -> Dict[str, Any]: symbols = get_universe(config, logger) if not symbols: logger.error("No symbols found for universe: %s", config.universe_name) return {} data = _load_results(config) existing_key = data.get("universe_key") or data.get("universe") is_same_universe = existing_key == config.universe_key if not is_same_universe or force: logger.info("Starting fresh scan for %s/%s (%d stocks)", config.market_id, config.universe_name, len(symbols)) data = { "market_id": config.market_id, "scan_started": datetime.now(timezone.utc).isoformat(), "universe": config.universe_name, "universe_key": config.universe_key, "total_stocks": len(symbols), "stage1": {}, "stage2": {}, "completed": False, } _save_results(config, data) else: s1_done = len(data.get("stage1", {})) s2_done = len(data.get("stage2", {})) logger.info("Resuming scan: %d/%d Stage1, %d Stage2 done", s1_done, len(symbols), s2_done) if is_same_universe and not force and data.get("completed"): s1_done_map = data.get("stage1", {}) or {} s2_done_map = data.get("stage2", {}) or {} s1_passed_done = [s for s, v in s1_done_map.items() if isinstance(v, dict) and v.get("passed")] if s1_passed_done and len(s2_done_map) >= len(s1_passed_done): logger.info( "Scan already completed for %s/%s; returning existing results without updating timestamps", config.market_id, config.universe_name, ) return data logger.info("=" * 60) logger.info("STAGE 1: Liquidity & Data Quality Pre-Filter") logger.info("=" * 60) s1_todo = [s for s in symbols if s not in data.get("stage1", {})] total_s1 = len(symbols) done_s1 = total_s1 - len(s1_todo) for i, sym in enumerate(s1_todo, start=done_s1 + 1): t0 = time.time() result = stage1_prefilter(sym, config) elapsed = time.time() - t0 status = "PASS" if result["passed"] else f"FAIL ({result['reason']})" logger.info("[Stage1 %d/%d] %s: %s (%.1fs)", i, total_s1, sym, status, elapsed) data.setdefault("stage1", {})[sym] = result _save_results(config, data) gc.collect() s1_passed = [s for s, v in data["stage1"].items() if v.get("passed")] s1_failed = [s for s, v in data["stage1"].items() if not v.get("passed")] logger.info("") logger.info("Stage 1 Results: %d PASS / %d FAIL out of %d", len(s1_passed), len(s1_failed), total_s1) logger.info("Passed: %s", ", ".join(sorted(s1_passed))) logger.info("") if stage1_only: data["completed"] = True _save_results(config, data) _publish_final_if_complete(config, data) return data logger.info("=" * 60) logger.info("STAGE 2: ML Walk-Forward Backtest") logger.info("=" * 60) s2_todo = [s for s in s1_passed if s not in data.get("stage2", {})] total_s2 = len(s1_passed) done_s2 = total_s2 - len(s2_todo) times_s2: List[float] = [] chunk_limit = config.max_stage2_per_run processed_this_run = 0 for i, sym in enumerate(s2_todo, start=done_s2 + 1): t0 = time.time() result = stage2_backtest(sym, config, logger) elapsed = time.time() - t0 times_s2.append(elapsed) if result["eligible"]: status = ( f"ELIGIBLE (ret={result.get('total_return_pct', 0):.1f}%, " f"sharpe={result.get('sharpe', 0):.3f}, " f"hit={result.get('hit_rate', 0):.0f}%)" ) else: status = f"EXCLUDED ({result['reason']})" avg_time = sum(times_s2) / len(times_s2) remaining = total_s2 - i eta_min = (remaining * avg_time) / 60 logger.info("[Stage2 %d/%d] %s: %s (%.1fs, ETA: %.0f min)", i, total_s2, sym, status, elapsed, eta_min) data.setdefault("stage2", {})[sym] = result _save_results(config, data) gc.collect() processed_this_run += 1 if chunk_limit and processed_this_run >= chunk_limit and i < total_s2: data["chunk_paused_at"] = datetime.now(timezone.utc).isoformat() data["chunk_last_index"] = i _save_results(config, data) logger.info( "Chunk limit reached: processed %d Stage 2 stocks this run " "(%d/%d total). Pausing — resume on next call.", processed_this_run, i, total_s2, ) return data # NOT completed=True; next call resumes data["completed"] = True data["scan_finished"] = datetime.now(timezone.utc).isoformat() _save_results(config, data) _publish_final_if_complete(config, data) print_report(config, logger, data) return data def print_report(config: ScanConfig, logger: logging.Logger, data: Optional[Dict[str, Any]] = None) -> None: if data is None: data = _load_results(config) s2 = data.get("stage2", {}) if not s2: logger.info("No Stage 2 results found. Run scan first.") return eligible = {s: v for s, v in s2.items() if v.get("eligible")} excluded = {s: v for s, v in s2.items() if not v.get("eligible")} print("\n" + "=" * 70) print(f" {config.title} SCAN RESULTS — {data.get('universe', '?').upper()}") print("=" * 70) s1 = data.get("stage1", {}) s1_pass = sum(1 for v in s1.values() if v.get("passed")) s1_fail = sum(1 for v in s1.values() if not v.get("passed")) print(f"\nStage 1 (Liquidity Filter): {s1_pass} pass / {s1_fail} fail / {len(s1)} total") print(f"Stage 2 (ML Backtest): {len(eligible)} eligible / {len(excluded)} excluded / {len(s2)} tested") if eligible: print(f"\n{'─' * 70}") print(f" ELIGIBLE STOCKS ({len(eligible)})") print(f"{'─' * 70}") print(f" {'Symbol':<10} {'Return%':>9} {'Sharpe':>8} {'DirAcc':>7} {'OOSAcc':>7} {'OOSRet%':>8} {'Trades':>7}") print(f" {'─'*10} {'─'*9} {'─'*8} {'─'*7} {'─'*7} {'─'*8} {'─'*7}") sorted_eligible = sorted(eligible.items(), key=lambda kv: kv[1].get("sharpe", -999), reverse=True) for sym, metrics in sorted_eligible: _oa = metrics.get("oos_dir_acc") _or = metrics.get("oos_return_pct") print( f" {sym:<10} {metrics.get('total_return_pct', 0):>+8.1f}% " f"{metrics.get('sharpe', 0):>8.3f} " f"{metrics.get('dir_acc', 0)*100:>6.1f}% " f"{(_oa*100 if _oa is not None else 0):>6.1f}% " f"{(_or if _or is not None else 0):>+7.1f}% " f"{metrics.get('trades', 0):>7d}" ) if excluded: print(f"\n{'─' * 70}") print(f" EXCLUDED STOCKS ({len(excluded)})") print(f"{'─' * 70}") for sym, metrics in sorted(excluded.items()): print(f" {sym:<10} {metrics.get('reason', 'unknown')}")