94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
"""Run the MT5 Strategy Tester and wait for the report (doc 07 §3, §7).
|
|
|
|
Topology A (all-Windows): launch ``terminal64.exe /config:tester.ini`` locally.
|
|
``ShutdownTerminal=1`` makes the terminal close itself when the run finishes;
|
|
the bridge waits for the process to exit (or for the report file to appear),
|
|
then parses the report. No copy/poll step — read the HTML straight from the
|
|
terminal's report folder.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
DEFAULT_MT5_INSTALL = r"C:\Program Files\MetaTrader 5 IC Markets Global"
|
|
|
|
# MT5 writes reports into the terminal's installation directory.
|
|
DEFAULT_REPORT_SUBDIR = "Reports"
|
|
|
|
|
|
def run_tester(
|
|
ini_path: str | Path,
|
|
*,
|
|
mt5_install: str = DEFAULT_MT5_INSTALL,
|
|
timeout: int = 1800,
|
|
report_subdir: str = DEFAULT_REPORT_SUBDIR,
|
|
poll_interval: float = 5.0,
|
|
) -> tuple[int, Optional[Path]]:
|
|
"""Launch the tester with the generated ini and wait for the report.
|
|
|
|
Returns ``(exit_code, report_path)``. ``report_path`` is ``None`` if no
|
|
report appeared before ``timeout``. The terminal is launched
|
|
non-interactively; ``ShutdownTerminal=1`` in the ini closes it on finish.
|
|
"""
|
|
ini = Path(ini_path)
|
|
if not ini.exists():
|
|
raise FileNotFoundError(f"tester.ini not found: {ini}")
|
|
terminal = Path(mt5_install) / "terminal64.exe"
|
|
if not terminal.exists():
|
|
raise FileNotFoundError(f"terminal64.exe not found: {terminal}")
|
|
|
|
cmd = [str(terminal), f"/config:{ini}"]
|
|
# Detach so the terminal's own GUI lifecycle controls shutdown.
|
|
proc = subprocess.Popen(cmd)
|
|
report_dir = Path(mt5_install) / report_subdir
|
|
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if proc.poll() is not None:
|
|
break
|
|
# Some runs leave the terminal open if ShutdownTerminal didn't fire;
|
|
# check for the report regardless.
|
|
rep = _find_latest_report(report_dir, since=ini.stat().st_mtime)
|
|
if rep is not None:
|
|
return proc.wait(), rep
|
|
time.sleep(poll_interval)
|
|
|
|
# One last check after exit / timeout.
|
|
rep = _find_latest_report(report_dir, since=ini.stat().st_mtime)
|
|
exit_code = proc.poll() if proc.poll() is not None else -1
|
|
return exit_code, rep
|
|
|
|
|
|
def wait_for_report(
|
|
report_dir: str | Path,
|
|
*,
|
|
since_ts: float,
|
|
timeout: int = 1800,
|
|
poll_interval: float = 5.0,
|
|
) -> Optional[Path]:
|
|
"""Poll ``report_dir`` for a fresh ``report*.htm`` and return its path."""
|
|
deadline = time.time() + timeout
|
|
rdir = Path(report_dir)
|
|
while time.time() < deadline:
|
|
rep = _find_latest_report(rdir, since=since_ts)
|
|
if rep is not None:
|
|
return rep
|
|
time.sleep(poll_interval)
|
|
return None
|
|
|
|
|
|
def _find_latest_report(report_dir: Path, since: float) -> Optional[Path]:
|
|
"""Return the newest ``.htm`` report in ``report_dir`` newer than ``since``."""
|
|
if not report_dir.exists():
|
|
return None
|
|
candidates = [
|
|
f for f in report_dir.glob("*.htm")
|
|
if f.stat().st_mtime >= since and f.stat().st_size > 1024
|
|
]
|
|
if not candidates:
|
|
return None
|
|
return max(candidates, key=lambda f: f.stat().st_mtime)
|