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()