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

#!/usr/bin/env python3
"""SimpleEMA v5 — 20-symbol portfolio backtest (shared params, per-symbol spread)."""

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import asdict, replace
from datetime import datetime
from pathlib import Path

import MetaTrader5 as mt5
import pandas as pd

ROOT = Path(__file__).resolve().parents[3]
LAB = Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT / "backtesting" / "MT5"))
sys.path.insert(0, str(LAB))

from cluster_audit.backtest_core import CostModel, load_bars, resolve_symbol  # noqa: E402
from run_backtest import pip_size  # noqa: E402
from strategy_v5 import V5Params, load_v5_cache, market_from_cache, simulate_v5  # noqa: E402


def load_portfolio_config(path: Path) -> dict:
    return json.loads(path.read_text(encoding="utf-8"))


def load_best_params(path: Path) -> V5Params:
    data = json.loads(path.read_text(encoding="utf-8"))
    return V5Params(**data["params"])


def portfolio_metrics(trades: pd.DataFrame, initial: float) -> dict:
    if trades.empty:
        return {"net_profit": 0, "total_trades": 0, "profit_factor": 0, "win_rate": 0, "max_drawdown_pct": 0}
    wins = trades[trades["profit"] > 0]["profit"]
    losses = trades[trades["profit"] <= 0]["profit"]
    gp = float(wins.sum()) if len(wins) else 0.0
    gl = abs(float(losses.sum())) if len(losses) else 0.0
    eq = initial + trades.sort_values("close_time")["profit"].cumsum()
    dd = abs(float(((eq - eq.cummax()) / eq.cummax() * 100).min())) if len(eq) else 0.0
    return {
        "net_profit": round(float(trades["profit"].sum()), 2),
        "total_trades": len(trades),
        "profit_factor": round(gp / gl, 2) if gl > 0 else 0.0,
        "win_rate": round(100.0 * len(wins) / len(trades), 1),
        "max_drawdown_pct": round(dd, 2),
        "profitable_symbols": int((trades.groupby("symbol")["profit"].sum() > 0).sum()),
        "symbol_count": trades["symbol"].nunique(),
    }


def load_params_map(args) -> tuple[dict[str, V5Params], float, list[dict]]:
    """Return symbol->params, lot, member metadata (may be empty)."""
    if args.params and args.params.name == "portfolio_params.json" and args.params.exists():
        data = json.loads(args.params.read_text(encoding="utf-8"))
        lot = data.get("config", {}).get("lot_per_symbol", 0.05)
        mapping: dict[str, V5Params] = {}
        members = []
        for m in data.get("members", []):
            if not m.get("enabled", True) or "params" not in m:
                continue
            mapping[m["symbol"]] = V5Params(**m["params"])
            members.append(m)
        return mapping, lot, members

    base = load_best_params(args.params)
    lot = base.lot_size
    return {}, lot, []


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", type=Path, default=LAB / "portfolio_symbols.json")
    ap.add_argument("--params", type=Path, default=LAB / "portfolio_params.json")
    ap.add_argument("--shared-params", type=Path, default=LAB / "best_params.json", help="fallback single-param set")
    args = ap.parse_args()

    cfg = load_portfolio_config(args.config)
    per_sym, lot, members = load_params_map(args)
    if not per_sym:
        base = load_best_params(args.shared_params)
        lot = cfg.get("lot_per_symbol", base.lot_size)
    else:
        lot = cfg.get("lot_per_symbol", lot)
    initial = cfg.get("initial_balance", 10000)

    start = datetime.fromisoformat(cfg["period"][0])
    end = datetime.fromisoformat(cfg["period"][1])

    if not mt5.initialize():
        raise SystemExit("MT5 init failed")
    try:
        sym_rows = []
        all_trades: list[dict] = []
        skipped: list[str] = []

        for entry in cfg["symbols"]:
            req = entry["name"]
            try:
                sym = resolve_symbol(req)
                if per_sym and sym not in per_sym:
                    if members:
                        print(f"  SKIP {sym}: disabled in portfolio_params")
                        continue
                spread_cap = entry.get("max_spread_pips", 8.0)
                if per_sym and sym in per_sym:
                    p = per_sym[sym]
                else:
                    base = load_best_params(args.shared_params)
                    p = replace(base, lot_size=lot, max_spread_pips=spread_cap)
                df = load_bars(sym, mt5.TIMEFRAME_M15, start, end)
                pip = pip_size(sym)
                point = float(mt5.symbol_info(sym).point)
                costs = CostModel.for_symbol(sym)
                r = simulate_v5(market_from_cache(load_v5_cache(df), p), sym, p, costs, pip, point)
                for t in r.trades:
                    all_trades.append(
                        {
                            "symbol": sym,
                            "side": t["side"],
                            "open_time": df.index[t["open_i"]],
                            "close_time": df.index[t["close_i"]],
                            "profit": round(t["profit"], 2),
                            "exit_reason": t["exit_reason"],
                        }
                    )
                sym_rows.append(
                    {
                        "symbol": sym,
                        "trades": r.total_trades,
                        "net_profit": round(r.net_profit, 2),
                        "profit_factor": round(r.profit_factor, 2),
                        "win_rate": round(r.win_rate, 1),
                    }
                )
                print(f"  {sym}: t={r.total_trades} net=${r.net_profit:,.0f} PF={r.profit_factor:.2f}")
            except Exception as exc:  # noqa: BLE001
                skipped.append(f"{req}: {exc}")
                print(f"  SKIP {req}: {exc}")

        tdf = pd.DataFrame(all_trades).sort_values(["close_time", "symbol"]) if all_trades else pd.DataFrame()
        metrics = portfolio_metrics(tdf, initial)
        metrics["target_met_2000_trades"] = metrics["total_trades"] >= 2000
        metrics["target_met_profit"] = metrics["net_profit"] > 0

        out = LAB / "best_run"
        out.mkdir(exist_ok=True)
        tdf.to_csv(out / "portfolio_trades.csv", index=False)
        pd.DataFrame(sym_rows).to_csv(out / "portfolio_by_symbol.csv", index=False)

        mode = "per_symbol" if per_sym else "shared"
        payload = {
            "version": 5,
            "mode": mode,
            "symbol_count": len(sym_rows),
            "skipped": skipped,
            "metrics": metrics,
            "per_symbol": sym_rows,
        }
        (LAB / "portfolio_report.json").write_text(json.dumps(payload, indent=2), encoding="utf-8")

        print(
            f"\nPORTFOLIO ({len(sym_rows)} symbols): "
            f"net=${metrics['net_profit']:,.0f} trades={metrics['total_trades']} "
            f"PF={metrics['profit_factor']:.2f} WR={metrics['win_rate']:.1f}% "
            f"DD={metrics['max_drawdown_pct']:.1f}% "
            f"2000+={'YES' if metrics['target_met_2000_trades'] else 'no'} "
            f"profit={'YES' if metrics['target_met_profit'] else 'no'}"
        )
    finally:
        mt5.shutdown()


if __name__ == "__main__":
    main()