fithia2/scripts/v12_experiments.py

#!/usr/bin/env python3
"""V12 experiment runner: structural improvements and new engine features.

Tests breakout volume confirmation, time-decay trailing, SPY trend filter,
VWAP confirmation, wider ORB, compounding, and combinations.

Usage:
    python -u scripts/v12_experiments.py [--days N]
"""
import subprocess
import sys
import tempfile
import yaml
from pathlib import Path

BASE_CONFIG = "configs/intraday/strategies/orb_gainers_v10.yaml"
DAYS = 400


def load_base():
    with open(BASE_CONFIG) as f:
        return yaml.safe_load(f)


def run_experiment(name: str, overrides: dict, days: int = DAYS) -> dict | None:
    """Run a single backtest with config overrides and return parsed metrics."""
    cfg = load_base()
    for key, val in overrides.items():
        cfg["orb_strategy"][key] = val

    with tempfile.NamedTemporaryFile(
        mode="w", suffix=".yaml", prefix="v12_", delete=False, dir="/tmp"
    ) as f:
        yaml.dump(cfg, f, default_flow_style=False)
        tmp_path = f.name

    cmd = [
        sys.executable, "-u", "-m", "apps.intraday_bt.run",
        "--config", tmp_path,
        "--days", str(days),
    ]

    print(f"\n{'='*70}")
    print(f"  {name}")
    print(f"  Overrides: {overrides or '(baseline)'}")
    print(f"{'='*70}", flush=True)

    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
    except subprocess.TimeoutExpired:
        print(f"  TIMEOUT after 600s")
        return None
    finally:
        Path(tmp_path).unlink(missing_ok=True)

    if result.returncode != 0:
        print(f"  FAILED (rc={result.returncode})")
        if result.stderr:
            print(f"  stderr: {result.stderr[-500:]}")
        return None

    output = result.stdout
    metrics = {}
    for line in output.split("\n"):
        line = line.strip()
        if "Total return" in line and "%" in line:
            try:
                val = line.split("│")[-2].strip().replace("%", "").replace("+", "")
                metrics["total_return"] = float(val)
            except (ValueError, IndexError):
                pass
        elif "Max drawdown" in line and "%" in line:
            try:
                val = line.split("│")[-2].strip().replace("%", "").replace("+", "")
                metrics["max_drawdown"] = float(val)
            except (ValueError, IndexError):
                pass
        elif "Sharpe ratio" in line:
            try:
                val = line.split("│")[-2].strip()
                metrics["sharpe"] = float(val)
            except (ValueError, IndexError):
                pass
        elif "Total trades" in line:
            try:
                val = line.split("│")[-2].strip()
                metrics["trades"] = int(val)
            except (ValueError, IndexError):
                pass
        elif "Win rate" in line and "%" in line:
            try:
                val = line.split("│")[-2].strip().replace("%", "").replace("+", "")
                metrics["win_rate"] = float(val)
            except (ValueError, IndexError):
                pass
        elif "Profit factor" in line:
            try:
                val = line.split("│")[-2].strip()
                metrics["profit_factor"] = float(val)
            except (ValueError, IndexError):
                pass

    if not metrics:
        print("  WARNING: Could not parse metrics from output")
        for line in output.split("\n")[-30:]:
            print(f"    {line}")
        return None

    print(f"  => return={metrics.get('total_return', '?'):+.2f}%, "
          f"DD={metrics.get('max_drawdown', '?'):.2f}%, "
          f"Sharpe={metrics.get('sharpe', '?'):.2f}, "
          f"trades={metrics.get('trades', '?')}, "
          f"WR={metrics.get('win_rate', '?'):.1f}%, "
          f"PF={metrics.get('profit_factor', '?'):.3f}", flush=True)
    return metrics


def main():
    days = DAYS
    if "--days" in sys.argv:
        idx = sys.argv.index("--days")
        days = int(sys.argv[idx + 1])

    results = {}

    # V10 baseline (no overrides)
    results["V10_baseline"] = run_experiment("V10 Baseline (control)", {}, days)

    # === NEW ENGINE: Breakout Volume Confirmation ===
    results["V12a_brkout_vol_1.5"] = run_experiment(
        "V12a: Breakout bar vol >= 1.5x avg (filter thin breakouts)",
        {"min_breakout_rel_vol": 1.5}, days,
    )
    results["V12a2_brkout_vol_2.0"] = run_experiment(
        "V12a2: Breakout bar vol >= 2.0x avg (stricter)",
        {"min_breakout_rel_vol": 2.0}, days,
    )
    results["V12a3_brkout_vol_1.2"] = run_experiment(
        "V12a3: Breakout bar vol >= 1.2x avg (mild)",
        {"min_breakout_rel_vol": 1.2}, days,
    )

    # === NEW ENGINE: Time-Decay Trailing ===
    results["V12b_decay_180_0.5"] = run_experiment(
        "V12b: Time-decay trailing start=12:30pm, factor=0.5 (halve trail by close)",
        {"time_decay_start_minutes": 180, "time_decay_factor": 0.5}, days,
    )
    results["V12b2_decay_120_0.5"] = run_experiment(
        "V12b2: Time-decay trailing start=11:30am, factor=0.5",
        {"time_decay_start_minutes": 120, "time_decay_factor": 0.5}, days,
    )
    results["V12b3_decay_180_0.3"] = run_experiment(
        "V12b3: Time-decay trailing start=12:30pm, factor=0.3 (aggressive tighten)",
        {"time_decay_start_minutes": 180, "time_decay_factor": 0.3}, days,
    )

    # === EXISTING UNUSED: SPY Trend Filter ===
    results["V12c_spy_trend_5d"] = run_experiment(
        "V12c: SPY 5-day trend filter (skip if SPY down >3%)",
        {"market_regime_spy_trend_days": 5, "market_regime_spy_trend_threshold": -0.03},
        days,
    )
    results["V12c2_spy_trend_3d"] = run_experiment(
        "V12c2: SPY 3-day trend filter (skip if SPY down >2%)",
        {"market_regime_spy_trend_days": 3, "market_regime_spy_trend_threshold": -0.02},
        days,
    )

    # === EXISTING UNUSED: VWAP Confirmation ===
    results["V12d_vwap_confirm"] = run_experiment(
        "V12d: Require ORB candle close vs VWAP confirmation",
        {"require_vwap_confirmation": True}, days,
    )

    # === WIDER ORB ===
    results["V12e_orb_10min"] = run_experiment(
        "V12e: 10-minute ORB window (wider base, fewer false breakouts)",
        {"orb_minutes": 10}, days,
    )
    results["V12e2_orb_15min"] = run_experiment(
        "V12e2: 15-minute ORB window",
        {"orb_minutes": 15}, days,
    )

    # === COMPOUNDING + SETTLEMENT ===
    results["V12f_compound"] = run_experiment(
        "V12f: compound_returns=true (size with current equity)",
        {"compound_returns": True}, days,
    )
    results["V12f2_settle0"] = run_experiment(
        "V12f2: settlement_days=0 (no settlement delay)",
        {"settlement_days": 0}, days,
    )
    results["V12f3_compound_settle0"] = run_experiment(
        "V12f3: compound + no settlement",
        {"compound_returns": True, "settlement_days": 0}, days,
    )

    # === COMBINATIONS of winners (conditional — using likely best) ===
    results["V12g_vol_decay"] = run_experiment(
        "V12g: Breakout vol 1.5x + time-decay 180/0.5",
        {"min_breakout_rel_vol": 1.5, "time_decay_start_minutes": 180, "time_decay_factor": 0.5},
        days,
    )
    results["V12h_vol_vwap"] = run_experiment(
        "V12h: Breakout vol 1.5x + VWAP confirmation",
        {"min_breakout_rel_vol": 1.5, "require_vwap_confirmation": True},
        days,
    )
    results["V12i_decay_vwap"] = run_experiment(
        "V12i: Time-decay 180/0.5 + VWAP confirmation",
        {"time_decay_start_minutes": 180, "time_decay_factor": 0.5, "require_vwap_confirmation": True},
        days,
    )
    results["V12j_kitchen_sink"] = run_experiment(
        "V12j: Vol 1.5x + decay 180/0.5 + VWAP + SPY 5d trend + compound + settle0",
        {
            "min_breakout_rel_vol": 1.5,
            "time_decay_start_minutes": 180,
            "time_decay_factor": 0.5,
            "require_vwap_confirmation": True,
            "market_regime_spy_trend_days": 5,
            "market_regime_spy_trend_threshold": -0.03,
            "compound_returns": True,
            "settlement_days": 0,
        },
        days,
    )

    # === Summary table ===
    print(f"\n\n{'='*90}")
    print("  V12 EXPERIMENT RESULTS SUMMARY")
    print(f"{'='*90}")
    print(f"{'Experiment':<30} {'Return':>9} {'DD':>9} {'Sharpe':>7} {'Trades':>7} {'WR':>7} {'PF':>7}")
    print(f"{'-'*30} {'-'*9} {'-'*9} {'-'*7} {'-'*7} {'-'*7} {'-'*7}")

    baseline_ret = (results.get("V10_baseline") or {}).get("total_return", 0)
    for name, m in results.items():
        if m is None:
            print(f"{name:<30} {'FAILED':>9}")
            continue
        ret = m.get("total_return", 0)
        delta = ret - baseline_ret
        ret_str = f"{ret:+.2f}%"
        dd = f"{m.get('max_drawdown', 0):.2f}%"
        sh = f"{m.get('sharpe', 0):.2f}"
        tr = f"{m.get('trades', 0)}"
        wr = f"{m.get('win_rate', 0):.1f}%"
        pf = f"{m.get('profit_factor', 0):.3f}"
        marker = " <== BASE" if name == "V10_baseline" else (f" ({delta:+.2f}pp)" if delta != 0 else "")
        print(f"{name:<30} {ret_str:>9} {dd:>9} {sh:>7} {tr:>7} {wr:>7} {pf:>7}{marker}")


if __name__ == "__main__":
    main()