scripts/run_patterns.py

"""
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()