import os import subprocess import time import logging import json import shutil from datetime import datetime from typing import List, Dict logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('logs/batch_executor.log', encoding='utf-8'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class MT5AutoRunner: def __init__(self, source): """ source: YAML 文件路径 str 或 已加载的 config dict（GUI 内存直接传入用）。 """ if isinstance(source, dict): self.config = source else: self.config = self._load_config(source) # 占位符解析（不做自动探测） from mt5_paths import resolve_mt5_settings self.config["mt5_settings"] = resolve_mt5_settings(self.config.get("mt5_settings", {})) self.mt5_path = self.config["mt5_settings"]["terminal_path"] self.data_dir = self.config["mt5_settings"].get("data_dir", "") self.reports_dir = os.path.abspath(self.config["mt5_settings"].get("reports_dir", "reports")) self.ini_dir = os.path.abspath(self.config["mt5_settings"].get("ini_dir", os.path.join(os.path.dirname(self.reports_dir), "config", "generated"))) os.makedirs(self.reports_dir, exist_ok=True) os.makedirs(self.ini_dir, exist_ok=True) exec_cfg = self.config.get("execution", {}) or {} self.kill_between = bool(exec_cfg.get("kill_between", True)) self.skip_existing = bool(exec_cfg.get("skip_existing", True)) self.timeout_per_test = int(exec_cfg.get("timeout_per_test", 30)) if not self.mt5_path or not os.path.isfile(self.mt5_path): raise FileNotFoundError( f"MT5 终端路径未配置或不存在: {self.mt5_path!r}\n" "请在 GUI '回测配置' 标签页设置 terminal64.exe 的完整路径后点 '保存配置'。" ) def _load_config(self, config_path: str) -> Dict: import yaml with open(config_path, 'r', encoding='utf-8') as f: return yaml.safe_load(f) def _load_execution_log(self) -> List[Dict]: log_file = os.path.join(self.ini_dir, "..", "results", "execution_log.json") log_file = os.path.normpath(log_file) if os.path.exists(log_file): with open(log_file, 'r', encoding='utf-8') as f: return json.load(f) return [] def _save_execution_log(self, log: List[Dict]): log_file = os.path.join(self.ini_dir, "..", "results", "execution_log.json") log_file = os.path.normpath(log_file) os.makedirs(os.path.dirname(log_file), exist_ok=True) with open(log_file, 'w', encoding='utf-8') as f: json.dump(log, f, indent=2, ensure_ascii=False) def _get_report_path(self, ini_name: str) -> str: base_name = os.path.splitext(ini_name)[0] return os.path.join(self.data_dir, f"{base_name}.htm") if self.data_dir else f"{base_name}.htm" def _copy_report_to_project(self, ini_name: str) -> str: base_name = os.path.splitext(ini_name)[0] src_path = self._get_report_path(ini_name) if not os.path.exists(src_path): logger.warning(f"Source report not found: {src_path}") return None dest_path = os.path.join(self.reports_dir, f"{base_name}.htm") try: shutil.copy2(src_path, dest_path) logger.info(f"Report copied to: {dest_path}") return dest_path except Exception as e: logger.error(f"Failed to copy report: {e}") return None def _kill_mt5(self): try: subprocess.run(['taskkill', '/F', '/IM', 'terminal64.exe'], capture_output=True, text=True) time.sleep(2) except: pass def _read_ini_expert(self, ini_path: str) -> str: try: with open(ini_path, 'r', encoding='utf-8') as f: for line in f: if line.startswith('Expert='): return line.split('=', 1)[1].strip() except: pass return None def _execute_single_ini(self, ini_path: str, timeout_min: int = 30) -> Dict: result = { "ini_file": os.path.basename(ini_path), "status": "pending", "start_time": datetime.now().isoformat(), "end_time": None, "duration_sec": 0, "report_found": False, "report_path": None, "error": None } ini_name = os.path.basename(ini_path) expected_report = self._get_report_path(ini_name) if os.path.exists(expected_report): result["status"] = "already_completed" result["report_found"] = True result["report_path"] = expected_report logger.info(f"Already completed: {ini_name}") return result logger.info(f"Executing: {ini_name}") try: start_time = time.time() proc = subprocess.Popen( [self.mt5_path, f"/config:{ini_path}"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) logger.info(f"MT5 started PID: {proc.pid}") while True: if proc.poll() is not None: logger.info("MT5 process ended") break if os.path.exists(expected_report): try: with open(expected_report, 'r', encoding='utf-8') as f: content = f.read() if len(content) > 1000: break except: pass elapsed = time.time() - start_time if elapsed > timeout_min * 60: logger.warning(f"Timeout: {ini_name}") break time.sleep(2) end_time = time.time() result["duration_sec"] = end_time - start_time if os.path.exists(expected_report): result["status"] = "completed" result["report_found"] = True copied_path = self._copy_report_to_project(ini_name) result["report_path"] = copied_path or expected_report logger.info(f"Success: {ini_name} ({result['duration_sec']:.0f}s)") else: result["status"] = "failed" logger.warning(f"Failed: {ini_name}") result["end_time"] = datetime.now().isoformat() if proc.poll() is None: proc.terminate() try: proc.wait(timeout=5) except: pass except Exception as e: result["status"] = "error" result["error"] = str(e) logger.error(f"Error: {e}") return result def run_full_auto(self, ini_files: List[str] = None, skip_if_exists: bool = None, kill_between_tests: bool = None, timeout_min: int = None) -> List[Dict]: logger.info("="*60) logger.info("MT5 Full Auto Batch Runner") logger.info("="*60) if skip_if_exists is None: skip_if_exists = self.skip_existing if kill_between_tests is None: kill_between_tests = self.kill_between if timeout_min is None: timeout_min = self.timeout_per_test logger.info(f"Timeout per test: {timeout_min} minutes, Kill between tests: {kill_between_tests}, Skip existing: {skip_if_exists}") self.execution_log = self._load_execution_log() if ini_files is None: ini_files = [os.path.join(self.ini_dir, f) for f in os.listdir(self.ini_dir) if f.endswith('.ini')] total = len(ini_files) completed = 0 failed = 0 logger.info(f"Total INI files: {total}") # First run must also start from a clean tester state. Otherwise the first test # can inherit an already-open terminal's previous deposit/delay/symbol settings, # while later tests look correct only because kill_between_tests runs after them. if kill_between_tests: self._kill_mt5() time.sleep(2) for idx, ini_path in enumerate(ini_files, 1): logger.info(f"[{idx}/{total}] {os.path.basename(ini_path)}") result = self._execute_single_ini(ini_path, timeout_min=timeout_min) self.execution_log.append(result) self._save_execution_log(self.execution_log) if result["status"] == "completed": completed += 1 elif result["status"] == "already_completed": completed += 1 else: failed += 1 logger.info(f" Status: {result['status']}, Duration: {result['duration_sec']:.0f}s") if kill_between_tests and result["status"] != "already_completed": self._kill_mt5() time.sleep(2) logger.info("="*60) logger.info(f"DONE: {completed}/{total} completed, {failed} failed") logger.info("="*60) return self.execution_log def run_daemon(self, check_interval: int = 60): logger.info("="*60) logger.info("MT5 Auto Runner - DAEMON MODE") logger.info(f"Monitoring: {self.ini_dir}") logger.info(f"Reports will be copied to: {self.reports_dir}") logger.info(f"Timeout per test: {self.timeout_per_test} minutes") logger.info("Press Ctrl+C to stop") logger.info("="*60) self.execution_log = self._load_execution_log() try: while True: pending_inis = [] for f in os.listdir(self.ini_dir): if f.endswith('.ini'): is_completed = any( log.get("ini_file") == f and log.get("status") == "completed" for log in self.execution_log ) if not is_completed: pending_inis.append(f) if pending_inis: logger.info(f"Found {len(pending_inis)} pending INI files") for ini_name in pending_inis: ini_path = os.path.join(self.ini_dir, ini_name) result = self._execute_single_ini(ini_path, timeout_min=self.timeout_per_test) self.execution_log.append(result) self._save_execution_log(self.execution_log) self._kill_mt5() time.sleep(3) else: logger.info("No pending INI files, waiting...") time.sleep(check_interval) except KeyboardInterrupt: logger.info("Daemon stopped") def generate_report(self) -> str: logger.info("Generating summary report...") from scripts.result_parser import ResultParser parser = ResultParser(self.reports_dir) results = parser.parse_all_reports(pattern="*.ht*") if results: from scripts.report_generator import ReportGenerator generator = ReportGenerator(results) os.makedirs("reports", exist_ok=True) output = generator.generate_excel("reports/batch_summary.xlsx") logger.info(f"Report saved: {output}") return output else: logger.warning("No results to generate report") return None def main(): import argparse parser = argparse.ArgumentParser(description="MT5 Auto Runner") parser.add_argument("--config", "-c", default="config/ea_configs.yaml") parser.add_argument("--mode", "-m", choices=["full", "daemon", "report", "init", "execute"], default="full") parser.add_argument("--interval", "-i", type=int, default=60) parser.add_argument("--no-skip", action="store_true") args = parser.parse_args() runner = MT5AutoRunner(args.config) if args.mode == "daemon": runner.run_daemon(check_interval=args.interval) elif args.mode == "report": runner.generate_report() elif args.mode == "init": from scripts.ini_generator import INIGenerator generator = INIGenerator(args.config) ini_files = generator.generate_ini_files() print(f"Generated {len(ini_files)} INI files") elif args.mode == "execute": from scripts.ini_generator import INIGenerator generator = INIGenerator(args.config) ini_files = generator.generate_ini_files() results = runner.run_full_auto(ini_files, skip_if_exists=not args.no_skip) completed = sum(1 for r in results if r["status"] == "completed") print(f"Completed: {completed}/{len(results)}") else: from scripts.ini_generator import INIGenerator generator = INIGenerator(args.config) ini_files = generator.generate_ini_files() logger.info(f"Generated {len(ini_files)} INI files") results = runner.run_full_auto(ini_files, skip_if_exists=not args.no_skip) runner.generate_report() if __name__ == "__main__": main()