profitable-expert-advisor/backtesting/MT5/run_optimize.py

"""
Walk-forward random-search optimizer for RSI scalping.

Optimizes on in-sample (train), ranks by out-of-sample (validation) score.
"""

from __future__ import annotations

import argparse
import json
import random
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any

import MetaTrader5 as mt5
import pandas as pd

from rsi_scalping_backtest import (
    CostModel,
    RsiScalpParams,
    backtest_rsi_scalping,
    load_rates,
    split_walk_forward,
)
from set_parser import SetParam, parse_set_file

TF_MAP = {
    "M1": mt5.TIMEFRAME_M1,
    "M5": mt5.TIMEFRAME_M5,
    "M10": mt5.TIMEFRAME_M10,
    "M15": mt5.TIMEFRAME_M15,
    "M30": mt5.TIMEFRAME_M30,
    "H1": mt5.TIMEFRAME_H1,
    "H4": mt5.TIMEFRAME_H4,
    "D1": mt5.TIMEFRAME_D1,
}

SET_TO_PARAM = {
    "RSI_Period": "rsi_period",
    "RSI_Overbought": "rsi_overbought",
    "RSI_Oversold": "rsi_oversold",
    "RSI_Target_Buy": "rsi_target_buy",
    "RSI_Target_Sell": "rsi_target_sell",
    "BarsToWait": "bars_to_wait",
    "UseTrailingStop": "use_trailing",
    "TrailingStopDistancePoints": "trail_distance_pts",
    "TrailingActivationPoints": "trail_activation_pts",
}


def _sample_value(p: SetParam, rng: random.Random) -> Any:
    if not p.optimize:
        return p.value
    if isinstance(p.start, bool):
        return rng.choice([p.start, p.stop])
    if isinstance(p.start, int) and isinstance(p.stop, int):
        step = int(p.step) if int(p.step) != 0 else 1
        vals = list(range(int(p.start), int(p.stop) + 1, step))
        return rng.choice(vals) if vals else p.value
    step = float(p.step) if float(p.step) != 0 else 1.0
    start, stop = float(p.start), float(p.stop)
    n = int((stop - start) / step) + 1
    idx = rng.randint(0, max(n - 1, 0))
    return round(start + idx * step, 4)


def sample_params(set_params: dict[str, SetParam], rng: random.Random, defaults: dict, fixed_lot: float) -> RsiScalpParams:
    raw = dict(defaults)
    for set_name, field in SET_TO_PARAM.items():
        if set_name in set_params:
            raw[field] = _sample_value(set_params[set_name], rng)
    raw["lot_size"] = fixed_lot
    return RsiScalpParams.from_dict(raw)


def _result_dict(r, label: str) -> dict:
    return {
        "label": label,
        "net_profit": r.net_profit,
        "total_trades": r.total_trades,
        "win_rate": r.win_rate,
        "profit_factor": r.profit_factor,
        "max_drawdown_pct": r.max_drawdown_pct,
        "total_costs": r.total_costs,
        "score": r.score,
        "params": r.params.__dict__,
    }


def main() -> None:
    parser = argparse.ArgumentParser(description="Walk-forward RSI scalping optimizer")
    parser.add_argument("--symbol", default="XAUUSD")
    parser.add_argument("--timeframe", default="H1", choices=TF_MAP.keys())
    parser.add_argument("--set", required=True)
    parser.add_argument("--trials", type=int, default=800)
    parser.add_argument("--days", type=int, default=730)
    parser.add_argument("--balance", type=float, default=10000.0)
    parser.add_argument("--lot", type=float, default=0.1)
    parser.add_argument("--train-ratio", type=float, default=0.6)
    parser.add_argument("--slippage", type=float, default=3.0)
    parser.add_argument("--commission", type=float, default=0.0)
    parser.add_argument("--seed", type=int, default=42)
    parser.add_argument("--out", default="optimization_results")
    args = parser.parse_args()

    if not mt5.initialize():
        raise SystemExit(f"MT5 init failed: {mt5.last_error()}")

    try:
        end = datetime.now()
        start = end - timedelta(days=args.days)
        tf = TF_MAP[args.timeframe]
        df_all = load_rates(args.symbol, tf, start, end)
        train_df, test_df = split_walk_forward(df_all, args.train_ratio)
        costs = CostModel.from_symbol(args.symbol, slippage_points=args.slippage, commission_per_lot=args.commission)

        info = mt5.symbol_info(args.symbol)
        spread = info.spread if info else 0
        print(f"Symbol {args.symbol} spread={spread} pts  slippage={args.slippage}  commission/lot={args.commission}")
        print(f"All: {len(df_all)} bars  train: {len(train_df)} ({train_df.index[0]} -> {train_df.index[-1]})")
        print(f"Test:  {len(test_df)} bars ({test_df.index[0]} -> {test_df.index[-1]})")

        set_params = parse_set_file(args.set)
        defaults = {
            "rsi_period": 14,
            "rsi_overbought": 71.0,
            "rsi_oversold": 57.0,
            "rsi_target_buy": 80.0,
            "rsi_target_sell": 57.0,
            "bars_to_wait": 1,
            "use_trailing": True,
            "trail_distance_pts": 71.0,
            "trail_activation_pts": 41.0,
        }

        baseline_params = RsiScalpParams.from_dict({**defaults, "lot_size": args.lot})
        baseline_train = backtest_rsi_scalping(train_df, args.symbol, baseline_params, args.balance, costs=costs)
        baseline_test = backtest_rsi_scalping(test_df, args.symbol, baseline_params, args.balance, costs=costs)
        baseline_full = backtest_rsi_scalping(df_all, args.symbol, baseline_params, args.balance, costs=costs)

        print("\n--- BASELINE (current SuperEA XAUUSD trailing defaults) ---")
        print(f"  train net=${baseline_train.net_profit:.2f}  trades={baseline_train.total_trades}  dd={baseline_train.max_drawdown_pct:.1f}%")
        print(f"  test  net=${baseline_test.net_profit:.2f}  trades={baseline_test.total_trades}  dd={baseline_test.max_drawdown_pct:.1f}%")
        print(f"  full  net=${baseline_full.net_profit:.2f}  trades={baseline_full.total_trades}  dd={baseline_full.max_drawdown_pct:.1f}%")

        rng = random.Random(args.seed)
        rows = []
        best_oos = None
        best_oos_score = float("-inf")

        for n in range(1, args.trials + 1):
            params = sample_params(set_params, rng, defaults, args.lot)
            train_r = backtest_rsi_scalping(train_df, args.symbol, params, args.balance, costs=costs)
            test_r = backtest_rsi_scalping(test_df, args.symbol, params, args.balance, costs=costs)
            full_r = backtest_rsi_scalping(df_all, args.symbol, params, args.balance, costs=costs)

            row = {
                "trial": n,
                "oos_score": test_r.score,
                "train_net": train_r.net_profit,
                "test_net": test_r.net_profit,
                "full_net": full_r.net_profit,
                "train_trades": train_r.total_trades,
                "test_trades": test_r.total_trades,
                "test_pf": test_r.profit_factor,
                "test_dd_pct": test_r.max_drawdown_pct,
                "test_win_rate": test_r.win_rate,
                **params.__dict__,
            }
            rows.append(row)

            if test_r.total_trades >= 15 and test_r.score > best_oos_score:
                best_oos_score = test_r.score
                best_oos = (params, train_r, test_r, full_r)

            if n % 200 == 0 and best_oos:
                _, _, br_test, _ = best_oos
                print(f"  trial {n}/{args.trials} best OOS net=${br_test.net_profit:.2f} score={best_oos_score:.2f}")

        if best_oos is None:
            raise SystemExit("No valid OOS candidate (need >=15 test trades)")

        best_params, best_train, best_test, best_full = best_oos

        out_dir = Path(args.out)
        out_dir.mkdir(parents=True, exist_ok=True)
        results_df = pd.DataFrame(rows).sort_values("oos_score", ascending=False)
        tag = f"{args.symbol}_{args.timeframe}_v2"
        csv_path = out_dir / f"{tag}_rsi_scalp_opt.csv"
        results_df.to_csv(csv_path, index=False)

        report = {
            "version": "v2-conservative-walkforward",
            "symbol": args.symbol,
            "timeframe": args.timeframe,
            "costs": {"spread_pts": spread, "slippage_pts": args.slippage, "commission_per_lot": args.commission},
            "bars": {"all": len(df_all), "train": len(train_df), "test": len(test_df)},
            "periods": {
                "all": [str(df_all.index[0]), str(df_all.index[-1])],
                "train": [str(train_df.index[0]), str(train_df.index[-1])],
                "test": [str(test_df.index[0]), str(test_df.index[-1])],
            },
            "trials": args.trials,
            "baseline": {
                "train": _result_dict(baseline_train, "train"),
                "test": _result_dict(baseline_test, "test"),
                "full": _result_dict(baseline_full, "full"),
            },
            "best_by_oos": {
                "train": _result_dict(best_train, "train"),
                "test": _result_dict(best_test, "test"),
                "full": _result_dict(best_full, "full"),
            },
            "top10_oos": results_df.head(10).to_dict(orient="records"),
        }
        json_path = out_dir / f"{tag}_rsi_scalp_best.json"
        json_path.write_text(json.dumps(report, indent=2), encoding="utf-8")

        print("\n=== BEST BY OUT-OF-SAMPLE (validation) ===")
        for k, v in best_params.__dict__.items():
            print(f"  {k}: {v}")
        print(f"  TRAIN net=${best_train.net_profit:.2f}  trades={best_train.total_trades}  dd={best_train.max_drawdown_pct:.1f}%")
        print(f"  TEST  net=${best_test.net_profit:.2f}  trades={best_test.total_trades}  pf={best_test.profit_factor:.2f}  dd={best_test.max_drawdown_pct:.1f}%")
        print(f"  FULL  net=${best_full.net_profit:.2f}  trades={best_full.total_trades}  dd={best_full.max_drawdown_pct:.1f}%")
        print(f"\nSaved: {csv_path}")
        print(f"Saved: {json_path}")
    finally:
        mt5.shutdown()


if __name__ == "__main__":
    main()