diff --git a/scripts/mt5_auto_runner.py b/scripts/mt5_auto_runner.py index fa17f11..3f90cd8 100644 --- a/scripts/mt5_auto_runner.py +++ b/scripts/mt5_auto_runner.py @@ -1,362 +1,380 @@ -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() +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, config_path): + if isinstance(config_path, dict): + self.config = config_path + else: + self.config = self._load_config(config_path) + self.mt5_path = self.config["mt5_settings"]["terminal_path"] + self.data_dir = self.config["mt5_settings"].get("data_dir", "") + self.ini_dir = self.config["mt5_settings"]["ini_dir"] + self.reports_dir = os.path.abspath(self.config["mt5_settings"].get("reports_dir", "reports")) + execution_cfg = self.config.get("execution", {}) or {} + self.timeout_per_test = int(execution_cfg.get("timeout_per_test", 30)) + self.kill_between = bool(execution_cfg.get("kill_between", True)) + self.skip_existing = bool(execution_cfg.get("skip_existing", True)) + self.execution_log = [] + os.makedirs("logs", exist_ok=True) + os.makedirs(self.reports_dir, exist_ok=True) + + 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): + try: + with open(log_file, 'r', encoding='utf-8') as f: + content = f.read().strip() + if not content: + return [] + return json.loads(content) + except (json.JSONDecodeError, OSError) as e: + logger.warning(f"Corrupt execution_log.json, ignoring: {e}") + return [] + 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] + + project_report_path = os.path.join(self.reports_dir, f"{base_name}.htm") + if os.path.exists(project_report_path): + return project_report_path + + if self.data_dir and os.path.isdir(self.data_dir): + target_names = {f"{base_name}.htm", f"{base_name}.xml", base_name} + for root, dirs, files in os.walk(self.data_dir): + for f in files: + if f in target_names: + found = os.path.join(root, f) + logger.info(f"Report found at: {found}") + return found + + logger.warning(f"Report not found for {ini_name}, base={base_name}, data_dir={self.data_dir}") + return project_report_path + + def _get_report_siblings(self, src_path: str) -> list: + parent = os.path.dirname(src_path) + base = os.path.splitext(os.path.basename(src_path))[0] + siblings = [] + try: + for f in os.listdir(parent): + name, ext = os.path.splitext(f) + if name == base and ext.lower() in (".htm", ".html", ".xml", ".png", ".gif", ".csv"): + siblings.append(os.path.join(parent, f)) + except OSError: + pass + return siblings + + 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") + if os.path.abspath(src_path) == os.path.abspath(dest_path): + return dest_path + + 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 + + actual_report = self._get_report_path(ini_name) + if os.path.exists(actual_report): + result["status"] = "completed" + result["report_found"] = True + copied_path = self._copy_report_to_project(ini_name) + result["report_path"] = copied_path or actual_report + logger.info(f"Success: {ini_name} ({result['duration_sec']:.0f}s)") + else: + result["status"] = "failed" + logger.warning(f"Failed: {ini_name}, looked at: {actual_report}") + + 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}") + + 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()