profitable-expert-advisor/backtesting/MT5/cluster_audit/sync_united.py

"""
Sync United EA audit results into main.mq5 defaults and generate .set file.

Usage:
  python -m cluster_audit.sync_united
"""

from __future__ import annotations

import json
import re
from datetime import datetime
from pathlib import Path

from cluster_audit.scoring import DEFAULT_TRADES_PER_DAY, acceptance, period_days, trades_per_day
from cluster_audit.united_registry import PERIODS, UNITED_STRATEGIES

REPORTS = Path(__file__).parent / "reports" / "united_sequential"
MAIN_MQ5 = Path(__file__).resolve().parents[3] / "frontline" / "cluster-latest" / "main.mq5"
OUT_SET = Path(__file__).resolve().parents[3] / "frontline" / "cluster-latest" / "UnitedEA_Optimized.set"
OUT_JSON = REPORTS / "united_manifest.json"

# main.mq5 input name -> (strategy_id, param_key in optimized JSON)
PARAM_PATCHES: dict[str, tuple[str, str]] = {
    "DB_BoxPeriod": ("united_darvas", "box_period"),
    "DB_BoxDeviation": ("united_darvas", "box_deviation"),
    "DB_StopLoss": ("united_darvas", "stop_loss_pts"),
    "DB_TakeProfit": ("united_darvas", "take_profit_pts"),
    "DB_MA_Period": ("united_darvas", "ma_period"),
    "DB_TrendThreshold": ("united_darvas", "trend_threshold"),
    "RC_overboughtLevel": ("united_rsi_cross", "overbought_level"),
    "RC_oversoldLevel": ("united_rsi_cross", "oversold_level"),
    "RC_emaSlopeThreshold": ("united_rsi_cross", "ema_slope_threshold"),
    "RC_emaDistanceThreshold": ("united_rsi_cross", "ema_distance_threshold"),
    "RS_APPL_RSI_Period": ("united_rsi_scalp_appl", "rsi_period"),
    "RS_APPL_RSI_Overbought": ("united_rsi_scalp_appl", "rsi_overbought"),
    "RS_APPL_RSI_Oversold": ("united_rsi_scalp_appl", "rsi_oversold"),
    "RS_APPL_RSI_Target_Buy": ("united_rsi_scalp_appl", "rsi_target_buy"),
    "RS_APPL_RSI_Target_Sell": ("united_rsi_scalp_appl", "rsi_target_sell"),
    "RS_APPL_BarsToWait": ("united_rsi_scalp_appl", "bars_to_wait"),
    "RS_APPL_TrailDistancePoints": ("united_rsi_scalp_appl", "trail_distance_pts"),
    "RS_APPL_TrailActivationPoints": ("united_rsi_scalp_appl", "trail_activation_pts"),
    "RS_BTCUSD_RSI_Period": ("united_rsi_scalp_btc", "rsi_period"),
    "RS_BTCUSD_TrailDistancePoints": ("united_rsi_scalp_btc", "trail_distance_pts"),
    "RS_XAUUSD_RSI_Period": ("united_rsi_scalp_xau", "rsi_period"),
    "RS_XAUUSD_TrailDistancePoints": ("united_rsi_scalp_xau", "trail_distance_pts"),
    "RRA_EURUSD_OverboughtLevel": ("united_rsi_asian_eur", "overbought_level"),
    "RRA_EURUSD_OversoldLevel": ("united_rsi_asian_eur", "oversold_level"),
    "RSS_RSIOverbought": ("united_rsi_secret", "rsi_overbought"),
    "RSS_RSIOversold": ("united_rsi_secret", "rsi_oversold"),
    "UB_MinRangePoints": ("united_usdjpy", "min_range_pts"),
    "UB_OrderBufferPoints": ("united_usdjpy", "order_buffer_pts"),
}

ENABLE_PATCHES: dict[str, str] = {s["enable_key"]: s["id"] for s in UNITED_STRATEGIES}


def load_report(sid: str) -> dict | None:
    p = REPORTS / f"{sid}_2021-2026.json"
    if not p.exists():
        # fallback cluster audit darvas
        alt = Path(__file__).parent / "reports" / "sequential" / "darvas_xau_2021-2026.json"
        if sid == "united_darvas" and alt.exists():
            r = json.loads(alt.read_text(encoding="utf-8"))
            r["id"] = sid
            return r
        return None
    return json.loads(p.read_text(encoding="utf-8"))


def evaluate(r: dict, days: int) -> tuple[bool, list[str]]:
    from cluster_audit.backtest_core import BacktestReport

    o = r.get("optimized", {})
    rep = BacktestReport(
        strategy_id=r["id"],
        symbol=r.get("symbol", ""),
        timeframe=r.get("timeframe", "H1"),
        period_label="2021-2026",
        net_profit=float(o.get("net_profit", 0)),
        total_trades=int(o.get("total_trades", 0)),
        win_rate=float(o.get("win_rate", 0)),
        profit_factor=float(o.get("profit_factor", 0)),
        sharpe=float(o.get("sharpe", 0)),
        max_drawdown_pct=float(o.get("max_drawdown_pct", 0)),
        avg_win=float(o.get("avg_win", 0)),
        avg_loss=float(o.get("avg_loss", 0)),
        worst_trades=o.get("worst_trades", []),
        losing_trades=o.get("losing_trades", []),
        exit_reason_breakdown=o.get("exit_reason_breakdown", {}),
        monthly_returns=o.get("monthly_returns", {}),
        params=o.get("params", {}),
    )
    return acceptance(rep, days, DEFAULT_TRADES_PER_DAY)


def patch_main_mqh(text: str, manifest: dict) -> str:
    params_by_sid = {k: v.get("params", {}) for k, v in manifest["strategies"].items()}

    for input_name, (sid, pkey) in PARAM_PATCHES.items():
        params = params_by_sid.get(sid, {})
        if pkey not in params:
            continue
        val = params[pkey]
        if isinstance(val, bool):
            lit = "true" if val else "false"
        elif isinstance(val, float):
            lit = str(val) if "." in str(val) else f"{val}.0"
        else:
            lit = str(val)
        text, n = re.subn(
            rf"(^input\s+\w+\s+{re.escape(input_name)}\s*=\s*)[^;]+;",
            rf"\g<1>{lit};",
            text,
            count=1,
            flags=re.MULTILINE,
        )
        if n:
            print(f"  patched {input_name}={lit}")

    for enable_key, sid in ENABLE_PATCHES.items():
        info = manifest["strategies"].get(sid, {})
        if info.get("status") == "no_report" or "passed" not in info:
            continue
        if not info.get("passed"):
            continue  # keep main.mq5 enable flags; only auto-enable winners
        lit = "true"
        text, n = re.subn(
            rf"(^input bool {re.escape(enable_key)}\s*=\s*)[^;]+;",
            rf"\g<1>{lit};",
            text,
            count=1,
            flags=re.MULTILINE,
        )
        if n:
            print(f"  enable {enable_key}={lit}")

    return text


def write_set_file(manifest: dict) -> None:
    lines = [
        "; UnitedEA_Optimized.set — generated from united sequential audit",
        f"; {datetime.now().isoformat()}",
        "",
    ]
    for spec in UNITED_STRATEGIES:
        sid = spec["id"]
        info = manifest["strategies"].get(sid, {})
        passed = info.get("passed", False)
        lines.append(f"; {sid}: {'PASS' if passed else 'DISABLED'}")
        lines.append(f"{spec['enable_key']}={'true' if passed else 'false'}")
        lines.append(f"{spec['lot_key']}={spec['lot']}")
        for k, v in info.get("params", {}).items():
            lines.append(f"; {k}={v}")
        lines.append("")
    OUT_SET.write_text("\n".join(lines), encoding="utf-8")


def main() -> None:
    start, end = PERIODS["2021-2026"]
    days = period_days(start, end)
    manifest: dict = {
        "generated": datetime.now().isoformat(),
        "period_days": days,
        "strategies": {},
    }

    for spec in UNITED_STRATEGIES:
        sid = spec["id"]
        r = load_report(sid)
        if not r:
            manifest["strategies"][sid] = {"passed": False, "status": "no_report"}
            continue
        params = r.get("optimized_params") or r.get("optimized", {}).get("params", {})
        ok, issues = evaluate(r, days)
        o = r.get("optimized", {})
        manifest["strategies"][sid] = {
            "passed": ok,
            "enable_key": spec["enable_key"],
            "lot_key": spec["lot_key"],
            "lot": spec["lot"],
            "trades": o.get("total_trades"),
            "trades_per_day": round(trades_per_day(
                type("R", (), {"total_trades": int(o.get("total_trades", 0))})(), days), 3),
            "net_profit": o.get("net_profit"),
            "profit_factor": o.get("profit_factor"),
            "sharpe": o.get("sharpe"),
            "issues": issues,
            "params": params,
        }

    if MAIN_MQ5.exists():
        text = MAIN_MQ5.read_text(encoding="utf-8")
        print(f"Patching {MAIN_MQ5}")
        text = patch_main_mqh(text, manifest)
        MAIN_MQ5.write_text(text, encoding="utf-8")

    REPORTS.mkdir(parents=True, exist_ok=True)
    OUT_JSON.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
    write_set_file(manifest)
    passed = [k for k, v in manifest["strategies"].items() if v.get("passed")]
    print(f"Wrote {OUT_JSON}")
    print(f"Wrote {OUT_SET}")
    print(f"Passed {len(passed)}/{len(UNITED_STRATEGIES)}: {', '.join(passed) if passed else '(none)'}")


if __name__ == "__main__":
    main()