scripts/deploy_live_mt5_integration.py

from __future__ import annotations

import argparse
import re
import shutil
import subprocess
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_TERMINAL_DATA = Path(
    r"C:\Users\aksha\AppData\Roaming\MetaQuotes\Terminal\D0E8209F77C8CF37AD8BF550E51FF075"
)
DEFAULT_METAEDITOR = Path(r"C:\Program Files\MetaTrader 5\MetaEditor64.exe")

EA_SOURCE = ROOT / "mql5" / "Experts" / "AFML" / "AFMLLiveONNXEA.mq5"
PRODUCTION_ROOT = ROOT / "production_models" / "live_mt5"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Deploy AFML ONNX models and live EA into the local MT5 terminal."
    )
    parser.add_argument("--terminal-data", default=str(DEFAULT_TERMINAL_DATA))
    parser.add_argument("--metaeditor", default=str(DEFAULT_METAEDITOR))
    parser.add_argument("--timeframe", default="M15")
    return parser.parse_args()


def latest_model(symbol: str, timeframe: str) -> Path:
    model_dir = PRODUCTION_ROOT / symbol / timeframe
    candidates = sorted(model_dir.glob("*.onnx"))
    if not candidates:
        raise FileNotFoundError(f"No ONNX models found in {model_dir}")
    return candidates[-1]


def compile_ea(metaeditor: Path, ea_path: Path, log_path: Path) -> tuple[bool, int, str]:
    proc = subprocess.run(
        [str(metaeditor), f"/compile:{ea_path}", f"/log:{log_path}"],
        capture_output=True,
        text=True,
        timeout=120,
    )
    log_text = log_path.read_text(encoding="utf-16", errors="replace") if log_path.exists() else ""
    # MetaEditor may return non-zero even when compile finishes with warnings only.
    # Prefer the compiler summary line when available.
    match = re.search(r"Result:\s*(\d+)\s+errors", log_text)
    if match:
        return int(match.group(1)) == 0, proc.returncode, log_text
    return proc.returncode == 0, proc.returncode, log_text


def write_set_file(path: Path, model_rel_path: str, timeframe: str, magic_number: int) -> None:
    path.write_text(
        "\n".join(
            [
                f"InpOnnxModelFile={model_rel_path}",
                f"InpSignalTimeframe={timeframe}",
                "InpEnableTrading=false",
                "InpCloseOpposite=true",
                "InpFixedLot=0.01",
                "InpBuyThreshold=0.55",
                "InpSellThreshold=0.45",
                "InpStopAtrMultiplier=1.5",
                "InpTakeAtrMultiplier=2.0",
                "InpMaxSpreadPoints=500",
                "InpSlippagePoints=30",
                f"InpMagicNumber={magic_number}",
            ]
        ),
        encoding="utf-8",
    )


def main() -> int:
    args = parse_args()
    terminal_data = Path(args.terminal_data)
    metaeditor = Path(args.metaeditor)

    experts_dir = terminal_data / "MQL5" / "Experts" / "AFML"
    files_dir = terminal_data / "MQL5" / "Files" / "AFML" / "Models"
    presets_dir = terminal_data / "MQL5" / "Profiles" / "Tester"
    experts_dir.mkdir(parents=True, exist_ok=True)
    files_dir.mkdir(parents=True, exist_ok=True)
    presets_dir.mkdir(parents=True, exist_ok=True)

    deployed = []
    for symbol, stable_name in [("BTCUSD", "btcusd_m15_live.onnx"), ("XAUUSD", "xauusd_m15_live.onnx")]:
        source = latest_model(symbol, args.timeframe)
        target = files_dir / stable_name
        shutil.copy2(source, target)
        deployed.append((symbol, source, target))

    ea_target = experts_dir / EA_SOURCE.name
    shutil.copy2(EA_SOURCE, ea_target)

    write_set_file(
        presets_dir / "AFML_BTCUSD_M15.set",
        "AFML\\Models\\btcusd_m15_live.onnx",
        args.timeframe,
        202605031,
    )
    write_set_file(
        presets_dir / "AFML_XAUUSD_M15.set",
        "AFML\\Models\\xauusd_m15_live.onnx",
        args.timeframe,
        202605032,
    )

    log_path = ROOT / "diagnostics" / "metaeditor_compile.log"
    log_path.parent.mkdir(parents=True, exist_ok=True)
    compile_ok = None
    compile_code = None
    compile_log = ""
    if metaeditor.exists():
        compile_ok, compile_code, compile_log = compile_ea(metaeditor, ea_target, log_path)

    print(f"EA deployed to: {ea_target}")
    for symbol, source, target in deployed:
        print(f"{symbol}: {source.name} -> {target}")
    print(f"Preset files: {presets_dir / 'AFML_BTCUSD_M15.set'}")
    print(f"Preset files: {presets_dir / 'AFML_XAUUSD_M15.set'}")

    if compile_code is not None:
        print(f"MetaEditor exit code: {compile_code}")
        if compile_log:
            print("\nCompile log tail:\n")
            print(compile_log[-4000:])

    if compile_ok is None:
        return 0
    return 0 if compile_ok else 1


if __name__ == "__main__":
    raise SystemExit(main())