Files
2026-06-26 20:50:07 +08:00

263 lines
10 KiB
Python

"""Phase 7 — MT5 bridge: verify a finalist against the MT5 Strategy Tester.
Pipeline (doc 07):
1. Deploy the EA (.ex5 + .mq5) to the terminal's MQL5\\Experts directory.
2. Generate a .set from the finalist's params (UTF-16-LE).
3. Generate a tester.ini (symbol, period, dates, model, shutdown).
4. Launch terminal64.exe /config:tester.ini — the tester runs headless and
closes itself when done (ShutdownTerminal=1).
5. Parse the HTML report (UTF-16-LE) the tester writes.
6. Build a Python-vs-MT5 comparison table + write auto-verification.md.
Usage:
python verify_mt5.py --trials 30 --top-n 2
Runs the optimizer first (to get finalists), then verifies each in MT5.
"""
from __future__ import annotations
import argparse
import shutil
import subprocess
import sys
import time
from pathlib import Path
PROJECT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT))
from shared.config import get_secret, load_env
from shared.data.mt5_report import parse_mt5_report
from shared.mt5_pipeline.compare import build_comparison_table
from shared.mt5_pipeline.ini_gen import (
MODEL_OHLC,
TesterConfig,
write_tester_ini,
)
from shared.mt5_pipeline.runner import run_tester
from shared.mt5_pipeline.set_gen import write_set_file
load_env(PROJECT)
# Reuse the optimizer's assembly.
from run import find_bars_file # noqa: E402
from shared.core.engine import SizingInputs # noqa: E402
from shared.core.metrics import compute_metrics # noqa: E402
from shared.data.loaders import load_bars # noqa: E402
from shared.optimizer.objective import ( # noqa: E402
Constraints,
ObjectiveConfig,
build_objective,
)
from shared.optimizer.selector import select_diverse_topn # noqa: E402
from strategies.gold_scalper_pro.instruments import XAUUSD_REAL # noqa: E402
from strategies.gold_scalper_pro.scalper_engine import ( # noqa: E402
ScalperEngine,
engine_kwargs_from_params,
)
from strategies.gold_scalper_pro.search_space import ( # noqa: E402
FROZEN_BASELINE,
INT_PARAMS,
SEARCH_SPACE,
)
from strategies.gold_scalper_pro.set_mappings import GOLD_SCALPER_MAPPINGS # noqa: E402
from strategies.gold_scalper_pro.signals import build_signals # noqa: E402
import optuna # noqa: E402
def deploy_ea(mt5_data_path: Path) -> None:
"""Copy GoldScalperPro.ex5 + .mq5 into MQL5\\Experts so the tester finds it."""
experts_dir = mt5_data_path / "MQL5" / "Experts"
experts_dir.mkdir(parents=True, exist_ok=True)
for src_name in ("GoldScalperPro.ex5", "GoldScalperPro.mq5"):
src = PROJECT / src_name
dst = experts_dir / src_name
if src.exists():
shutil.copy2(src, dst)
print(f" deployed {src_name}{dst.relative_to(mt5_data_path)}")
else:
raise FileNotFoundError(f"EA source missing: {src}")
def verify_finalist(
params: dict,
label: str,
*,
bars: "pd.DataFrame",
deposit: float,
mt5_install: str,
mt5_data_path: Path,
tester_profiles_dir: Path,
date_from: str,
date_to: str,
) -> int:
"""Verify one finalist in MT5 and print the comparison table."""
print(f"\n=== verifying {label} ===")
# ── 1. Python re-run (fresh snapshot — doc 04 Rule 5) ──────────────────
pack = build_signals(params, bars, XAUUSD_REAL)
engine = ScalperEngine()
result = engine.run(
bars, pack.signals_long, pack.signals_short,
pack.sl_prices, pack.tp_prices,
XAUUSD_REAL, SizingInputs(), deposit,
**engine_kwargs_from_params(params),
)
py_metrics = compute_metrics(result)
print(f" python: net={py_metrics.net_profit:+,.2f} PF={py_metrics.profit_factor:.2f} "
f"trades={py_metrics.total_trades} DD={py_metrics.max_equity_dd:.2f}")
# ── 2. Write the .set (UTF-16-LE) into the tester profiles dir ────────
set_name = f"GoldScalperPro_{label}"
set_path = tester_profiles_dir / f"{set_name}.set"
write_set_file(params, GOLD_SCALPER_MAPPINGS, set_path)
print(f" wrote .set → {set_path.name}")
# ── 3. Write tester.ini ───────────────────────────────────────────────
ini_path = tester_profiles_dir / f"{set_name}.ini"
login = int(get_secret("MT5_DEMO_LOGIN") or 0)
password = get_secret("MT5_DEMO_PASSWORD")
server = get_secret("MT5_DEMO_SERVER")
tcfg = TesterConfig(
expert=r"Experts\GoldScalperPro.ex5",
symbol="XAUUSD",
period="M5",
model=MODEL_OHLC, # 1-min OHLC for routine verification
from_date=date_from,
to_date=date_to,
deposit=deposit,
leverage=100,
report=f"report_{label}",
shutdown_terminal=True,
set_file=str(set_path),
login=login,
password=password,
server=server,
)
write_tester_ini(tcfg, ini_path)
print(f" wrote ini → {ini_path.name}")
# ── 4. Run the tester (headless; closes itself when done) ─────────────
print(f" launching MT5 tester (headless, model=OHLC) ...")
t0 = time.time()
exit_code, report_path = run_tester(
ini_path, mt5_install=mt5_install, timeout=900, poll_interval=5.0,
)
elapsed = time.time() - t0
print(f" tester finished in {elapsed:.0f}s exit={exit_code}")
if report_path is None:
print(" ✗ no report found — tester may have failed to start")
return 1
print(f" report → {report_path}")
# ── 5. Parse the report + build comparison ────────────────────────────
mt5_metrics = parse_mt5_report(report_path)
# Map MT5 report keys to our Metrics field names for the table.
mt5_mapped = {
"net_profit": mt5_metrics.get("Total Net Profit"),
"profit_factor": mt5_metrics.get("Profit Factor"),
"total_trades": mt5_metrics.get("Total Trades"),
"max_equity_dd": mt5_metrics.get("Equity Drawdown Maximal"),
"win_rate": None, # MT5 report doesn't surface this directly
"sharpe": mt5_metrics.get("Sharpe Ratio"),
}
table = build_comparison_table(py_metrics, mt5_mapped)
print("\n " + table.replace("\n", "\n "))
# ── 6. Write auto-verification.md ─────────────────────────────────────
out_md = PROJECT / "registry" / f"auto-verification_{label}.md"
out_md.parent.mkdir(parents=True, exist_ok=True)
body = (
f"# Auto-verification: {label}\n\n"
f"## Parameters\n\n```\n"
)
for k, v in params.items():
body += f" {k} = {v}\n"
body += "```\n\n## Python vs MT5\n\n" + table
body += (
"\n## Decision rule (doc 03 §7)\n"
"If the MT5 number still clears the bar after the expected fidelity "
"gap, the finalist is real. If the edge only existed in the optimistic "
"Python figure, discard it.\n"
)
out_md.write_text(body, encoding="utf-8")
print(f" wrote {out_md.relative_to(PROJECT)}")
return 0
def main() -> int:
ap = argparse.ArgumentParser(description="Verify GoldScalperPro finalists in MT5.")
ap.add_argument("--trials", type=int, default=30)
ap.add_argument("--top-n", type=int, default=2)
ap.add_argument("--deposit", type=float, default=10000.0)
args = ap.parse_args()
mt5_install = get_secret("MT5_TERMINAL_PATH") or r"C:\Program Files\MetaTrader 5 IC Markets Global"
mt5_install_dir = str(Path(mt5_install).parent)
mt5_data_path = Path(get_secret("MT5_DATA_PATH")
or r"C:\Users\Administrator\AppData\Roaming\MetaQuotes\Terminal"
r"\010E047102812FC0C18890992854220E")
tester_profiles_dir = mt5_data_path / "MQL5" / "Profiles" / "Tester"
tester_profiles_dir.mkdir(parents=True, exist_ok=True)
bars_path = find_bars_file()
bars = load_bars(bars_path)
date_from = bars["timestamp"].iloc[0].strftime("%Y.%m.%d")
date_to = bars["timestamp"].iloc[-1].strftime("%Y.%m.%d")
print(f"=== MT5 verification ===")
print(f"bars : {bars_path.name} ({len(bars):,} bars)")
print(f"window : {date_from}{date_to}")
print(f"deposit: {args.deposit:,.0f} USD")
# ── Deploy EA ─────────────────────────────────────────────────────────
print(f"\ndeploying EA to {mt5_data_path.name}/MQL5/Experts/ ...")
deploy_ea(mt5_data_path)
# ── Run optimizer to get finalists ────────────────────────────────────
print(f"\noptimizing ({args.trials} trials) ...")
constraints = Constraints(min_trades=25, min_profit_factor=1.2,
max_equity_dd_pct=0.40)
obj_cfg = ObjectiveConfig(
engine=ScalperEngine(),
bars=bars,
instrument=XAUUSD_REAL,
sizing=SizingInputs(),
initial_deposit=args.deposit,
search_space=SEARCH_SPACE,
int_params=INT_PARAMS,
frozen_baseline=FROZEN_BASELINE,
constraints=constraints,
dd_weight=1.0,
build_signals=build_signals,
build_engine_kwargs=engine_kwargs_from_params,
)
objective = build_objective(obj_cfg)
optuna.logging.set_verbosity(optuna.logging.WARNING)
study = optuna.create_study(direction="maximize",
sampler=optuna.samplers.TPESampler(seed=42))
study.optimize(objective, n_trials=args.trials)
finalists = select_diverse_topn(study, args.top_n, SEARCH_SPACE)
print(f"selected {len(finalists)} finalists")
# ── Verify each finalist ───────────────────────────────────────────────
for i, t in enumerate(finalists, 1):
label = f"f{i}"
verify_finalist(
t.user_attrs["params"], label,
bars=bars, deposit=args.deposit,
mt5_install=mt5_install_dir,
mt5_data_path=mt5_data_path,
tester_profiles_dir=tester_profiles_dir,
date_from=date_from, date_to=date_to,
)
print("\n=== done ===")
print(f"verification reports in registry/auto-verification_*.md")
return 0
if __name__ == "__main__":
raise SystemExit(main())