profitable-expert-advisor/lab/EAs/SimpleEMA/run_mt5_portfolio.py

#!/usr/bin/env python3
"""
Run MT5 Strategy Tester for each enabled portfolio symbol (source of truth).

Each symbol: main.mq5 + per-symbol .set from portfolio_params.json.
Aggregates HTML report metrics into best_run/mt5_results.json.

Usage:
  python run_mt5_portfolio.py
  python run_mt5_portfolio.py --only EURUSD,XAUUSD
  python run_mt5_portfolio.py --from 2020.01.01 --to 2026.01.01
"""

from __future__ import annotations

import argparse
import json
import shutil
import sys
import time
from datetime import datetime
from pathlib import Path

LAB = Path(__file__).resolve().parent
ROOT = LAB.parents[2]  # repo root: .../profitable-expert-advisor
sys.path.insert(0, str(LAB))

from run_mt5_tester import mt5_context, run_tester  # noqa: E402


def write_member_set(params: dict, path: Path) -> None:
    """Write MT5 .set from portfolio_params member dict (no Python backtest imports)."""
    p = params
    lines = [
        "; SimpleEMA v5 — per-symbol MT5 set",
        "Timeframe=16388",
        f"FastEmaPeriod={p['fast_ema']}",
        f"SlowEmaPeriod={p['slow_ema']}",
        f"TrendLegBars={p['trend_leg_bars']}",
        f"MinEmaGapPips={p['min_ema_gap_pips']}",
        f"CrossCooldown={p['cross_cooldown']}",
        f"PullbackCooldown={p['pullback_cooldown']}",
        f"UsePullback={'true' if p['use_pullback'] else 'false'}",
        f"PullbackTouch={p['pullback_touch']}",
        f"PullbackAdxMin={p['pullback_adx_min']}",
        f"PullbackMinGapPips={p['pullback_min_gap_pips']}",
        f"MaxPullbacksPerLeg={p['max_pullbacks_per_leg']}",
        f"AtrPeriod={p['atr_period']}",
        f"AtrSlMult={p['atr_sl_mult']}",
        f"AtrTpMult={p['atr_tp_mult']}",
        f"MaxBarsInTrade={p['max_bars_in_trade']}",
        f"HtfEmaPeriod={p['htf_ema_period']}",
        f"UseHtfFilter={'true' if p['use_htf_filter'] else 'false'}",
        f"UseAdxFilter={'true' if p['use_adx_filter'] else 'false'}",
        f"AdxPeriod={p['adx_period']}",
        f"AdxMin={p['adx_min']}",
        f"SessionStartHour={p['session_start']}",
        f"SessionEndHour={p['session_end']}",
        f"MaxSpreadPips={p['max_spread_pips']}",
        f"LotSize={p['lot_size']}",
    ]
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")

SETS_DIR = LAB / "mt5_sets"
OUT_DIR = LAB / "best_run" / "mt5_reports"
RESULTS_JSON = LAB / "best_run" / "mt5_results.json"


def load_members(params_path: Path, only: set[str] | None, enabled_only: bool = False) -> tuple[list[dict], dict]:
    data = json.loads(params_path.read_text(encoding="utf-8"))
    members = [m for m in data.get("members", []) if "params" in m]
    if enabled_only:
        members = [m for m in members if m.get("enabled")]
    if only:
        only_up = {s.upper() for s in only}
        members = [m for m in members if m["symbol"].upper() in only_up or m.get("requested", "").upper() in only_up]
    return members, data.get("config", {})


def tester_symbol(sym: str) -> str:
    """Use broker symbol as stored in portfolio_params."""
    return sym.split(".")[0] if "." not in sym else sym


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--params", type=Path, default=LAB / "portfolio_params.json")
    ap.add_argument("--only", default="", help="comma-separated symbols, e.g. EURUSD,XAUUSD")
    ap.add_argument("--enabled-only", action="store_true", help="test only MT5-enabled symbols")
    ap.add_argument("--from", dest="from_date", default="2020.01.01")
    ap.add_argument("--to", dest="to_date", default="2026.01.01")
    ap.add_argument("--period", default="M15", choices=["M15", "M30", "H1", "H4"])
    ap.add_argument("--deposit", type=float, default=10000)
    ap.add_argument("--leverage", type=int, default=100)
    args = ap.parse_args()

    only = {s.strip() for s in args.only.split(",") if s.strip()} or None
    members, cfg = load_members(args.params, only, enabled_only=args.enabled_only)
    if not members:
        raise SystemExit("No enabled members in portfolio_params.json")

    SETS_DIR.mkdir(exist_ok=True)
    OUT_DIR.mkdir(parents=True, exist_ok=True)
    (LAB / "best_run").mkdir(exist_ok=True)

    ctx = mt5_context()
    rows: list[dict] = []
    t0 = time.time()

    print(f"MT5 portfolio backtest: {len(members)} symbols  {args.from_date} -> {args.to_date}  {args.period}")
    print("(Each symbol = separate MT5 Strategy Tester run with its own .set)\n")

    for i, m in enumerate(members, 1):
        sym = m["symbol"]
        test_sym = tester_symbol(sym)
        set_name = f"SimpleEMA_{test_sym}.set"
        set_path = SETS_DIR / set_name
        write_member_set(m["params"], set_path)

        report = f"SimpleEMA_pf_{test_sym}"
        print(f"[{i}/{len(members)}] {test_sym} ...")
        try:
            metrics = run_tester(
                ctx,
                mode="backtest",
                set_path=set_path,
                set_name=set_name,
                report=report,
                symbol=test_sym,
                period=args.period,
                from_date=args.from_date,
                to_date=args.to_date,
                deposit=args.deposit,
                leverage=args.leverage,
                visual=False,
                timeout_sec=7200,
            )
        except Exception as exc:  # noqa: BLE001
            print(f"  FAIL {test_sym}: {exc}")
            rows.append({"symbol": sym, "ready": False, "error": str(exc)})
            continue

        if metrics.get("ready") and metrics.get("report"):
            src = Path(metrics["report"])
            dst = OUT_DIR / src.name
            shutil.copy2(src, dst)
            metrics["report_local"] = str(dst)

        row = {
            "symbol": sym,
            "test_symbol": test_sym,
            "ready": metrics.get("ready", False),
            "net_profit": metrics.get("net_profit"),
            "total_trades": metrics.get("total_trades"),
            "profit_factor": metrics.get("profit_factor"),
            "sharpe": metrics.get("sharpe"),
            "max_drawdown": metrics.get("max_drawdown"),
            "elapsed_sec": metrics.get("elapsed_sec"),
            "report": metrics.get("report_local") or metrics.get("report"),
            "set_file": str(set_path),
        }
        rows.append(row)
        if row["ready"]:
            print(
                f"  OK net={row['net_profit']} trades={row['total_trades']} "
                f"PF={row['profit_factor']}  ({row['elapsed_sec']}s)"
            )
        else:
            print(f"  NO REPORT for {test_sym}")

    ok = [r for r in rows if r.get("ready")]
    total_trades = sum(r.get("total_trades") or 0 for r in ok)
    total_net = sum(r.get("net_profit") or 0 for r in ok)
    gp = sum(r.get("net_profit") or 0 for r in ok if (r.get("net_profit") or 0) > 0)
    gl = abs(sum(r.get("net_profit") or 0 for r in ok if (r.get("net_profit") or 0) < 0))

    payload = {
        "source": "mt5_strategy_tester",
        "generated_at": datetime.now().isoformat(timespec="seconds"),
        "period": {"from": args.from_date, "to": args.to_date, "timeframe": args.period},
        "deposit_per_symbol": args.deposit,
        "note": "Sum of independent MT5 single-symbol runs. NOT Python simulation.",
        "portfolio": {
            "symbols_tested": len(ok),
            "symbols_failed": len(rows) - len(ok),
            "total_trades": total_trades,
            "net_profit_sum": round(total_net, 2),
            "profit_factor_approx": round(gp / gl, 2) if gl > 0 else None,
        },
        "per_symbol": rows,
    }
    RESULTS_JSON.write_text(json.dumps(payload, indent=2), encoding="utf-8")

    print(f"\n=== MT5 Portfolio (aggregated) ===")
    print(f"  symbols OK: {len(ok)}/{len(rows)}")
    print(f"  total_trades: {total_trades}")
    print(f"  net_profit (sum): ${total_net:,.2f}")
    print(f"  saved: {RESULTS_JSON}")
    print(f"  HTML reports: {OUT_DIR}")
    print(f"  elapsed: {time.time() - t0:.0f}s")
    print("\nNext: python generate_mt5_portfolio_report.py")


if __name__ == "__main__":
    main()