""" Run all pattern presets against historical data and verify results. Usage: python3 scripts/run_patterns.py [--daily-only] [--pattern NAME] Without args, runs all patterns. Results saved to results/patterns/. """ import argparse import json import os import sys import time from datetime import datetime sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from core.data_manager import DataManager from core.scanner_service import ScannerService RESULTS_DIR = os.path.join(os.path.dirname(__file__), "..", "results", "patterns") os.makedirs(RESULTS_DIR, exist_ok=True) # === All pattern presets === PATTERNS = [ { "name": "1. Gap&Crap (estándar)", "daily_only": True, "expr": "gap_pct > 15% and run_pct < 0 and volume > avg(volume, 60) * 2 sort gap_pct desc", "verify_cols": ["symbol", "date", "gap_pct", "run_pct", "volume"], }, { "name": "2. Gap&Crap Reversal", "daily_only": False, "expr": "gap_pct > 30% and volume > avg(volume, 60) * 5 and run_pct < 0 sort gap_pct desc" " | time > '09:30' and time < '10:00' and close < vwap and close < open and volume > 500000" " | time > '10:00' and time < '15:50' and close > vwap and close > open and low > session_low", "verify_cols": ["symbol", "date", "gap_pct", "volume", "run_pct"], }, { "name": "3. Gap and Go", "daily_only": False, "expr": "gap_pct > 15% and volume > avg(volume, 60) * 3 and run_pct > 0 sort gap_pct desc" " | time >= '09:30' and time <= '10:30' and close > pm_high and close > open", "verify_cols": ["symbol", "date", "gap_pct", "run_pct", "volume"], }, { "name": "4. Breakout 52-week", "daily_only": True, "expr": "close > max(close, 252) and volume > 1_000_000", "verify_cols": ["symbol", "date", "close", "volume"], }, { "name": "5. Red to Green", "daily_only": False, "expr": "close < max(close, 252) * 0.6 and volume > avg(volume, 60)" " | time >= '09:30' and time <= '10:30' and close < vwap and close < open" " | time >= '10:00' and time <= '15:50' and close > vwap and close > open and close > session_low", "verify_cols": ["symbol", "date", "run_pct", "volume"], }, { "name": "6. VWAP Bounce", "daily_only": False, "expr": "volume > avg(volume, 60) * 5 sort volume desc" " | time >= '09:30' and time <= '11:30' and low < vwap and close > vwap" " and (session_high - vwap) / vwap > 0.10", "verify_cols": ["symbol", "date", "volume", "run_pct"], }, { "name": "7. VWAP Reclaim", "daily_only": False, "expr": "close > open and volume > avg(volume, 60) * 3" " | time >= '09:30' and time <= '11:00' and (session_high - vwap) / vwap > 0.10" " | time >= '10:00' and time <= '12:00' and close < vwap" " | time >= '11:00' and time <= '15:30' and close > vwap and volume > avg(volume, 20) * 2", "verify_cols": ["symbol", "date", "volume", "run_pct"], }, { "name": "8. Dip Buying Panics", "daily_only": False, "expr": "streak_run_pct > 100% and volume > avg(volume, 60) * 3 sort streak_run_pct desc" " | time >= '09:30' and time <= '10:30' and (session_high - close) / session_high > 0.20 and close < vwap" " | time >= '09:45' and time <= '11:00' and close > vwap and volume > avg(volume, 20) * 2", "verify_cols": ["symbol", "date", "streak_run_pct", "volume"], }, { "name": "9. Swing 1er Día Verde", "daily_only": False, "expr": "run_pct > 20% and volume > avg(volume, 60) * 5 sort volume desc" " | close > vwap and close > session_high * 0.95", "verify_cols": ["symbol", "date", "run_pct", "volume"], }, { "name": "10. Rebote 1er Día Verde", "daily_only": True, "expr": "close > open and (max(high, 20) - close) / max(high, 20) > 0.25" " and (max(high, 20) - close) / max(high, 20) < 0.50" " and volume < avg(volume, 20) * 2", "verify_cols": ["symbol", "date", "run_pct", "volume"], }, ] # Target date range START = "2026-05-18" END = "2026-05-22" def run_daily_only(dm, expr, name): """Run a daily-only filter (no pipe stages).""" t0 = time.time() try: df = dm.get_candidates(START, expr, 0, end_date=END, limit=50) elapsed = time.time() - t0 return { "status": "ok", "count": len(df), "samples": df.head(10).to_dict("records") if not df.empty else [], "elapsed_s": round(elapsed, 1), } except Exception as e: return {"status": "error", "error": str(e), "elapsed_s": round(time.time() - t0, 1)} def run_with_intraday(scanner, expr, name): """Run filter with pipe stages (needs minute data).""" t0 = time.time() try: results = scanner.get_scan_results(expr, start=START, end=END, limit=20) elapsed = time.time() - t0 # Simplify results for display samples = [] for r in results[:10]: s = {"symbol": r.get("symbol"), "date": r.get("date")} # Keep key daily fields for k in ["gap_pct", "run_pct", "change_pct", "volume", "streak_run_pct"]: if k in r: s[k] = r[k] samples.append(s) return { "status": "ok", "count": len(results), "samples": samples, "elapsed_s": round(elapsed, 1), } except Exception as e: import traceback return { "status": "error", "error": str(e), "traceback": traceback.format_exc(), "elapsed_s": round(time.time() - t0, 1), } def main(): parser = argparse.ArgumentParser() parser.add_argument("--daily-only", action="store_true", help="Run only daily patterns") parser.add_argument("--pattern", type=str, help="Run only this pattern (name prefix)") args = parser.parse_args() dm = DataManager() scanner = ScannerService(data_manager=dm) patterns = PATTERNS if args.daily_only: patterns = [p for p in patterns if p["daily_only"]] if args.pattern: patterns = [p for p in patterns if p["name"].startswith(args.pattern)] print(f"{'Status':8} {'Pattern':32} {'Count':6} {'Time':6} Notes") print("=" * 80) all_results = {} for p in patterns: name = p["name"] print(f"\n--- {name} ---") print(f" Filter: {p['expr'][:80]}...") result = run_daily_only(dm, p["expr"], name) if p["daily_only"] else run_with_intraday(scanner, p["expr"], name) all_results[name] = result if result["status"] == "ok": print(f" ✅ {result['count']} matches in {result['elapsed_s']}s") if result["samples"]: print(" Samples:") for s in result["samples"]: cols = {k: v for k, v in s.items() if k in p.get("verify_cols", []) or k == "symbol"} print(f" {cols}") else: print(" (no matches)") else: print(f" ❌ Error: {result.get('error', 'unknown')}") if "traceback" in result: print(f" {result['traceback'][:500]}") # Save results timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") report = { "run_at": timestamp, "date_range": f"{START} to {END}", "results": all_results, } report_path = os.path.join(RESULTS_DIR, f"run_{timestamp}.json") with open(report_path, "w") as f: json.dump(report, f, indent=2, default=str) print(f"\n📄 Report saved: {report_path}") if __name__ == "__main__": main()