import os import sys import json import time import subprocess import logging from datetime import datetime from typing import List, Dict, Any logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) OPTIMIZER_DIR = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, OPTIMIZER_DIR) from ea_config_parser import load_all_ea_configs, EAConfig from smart_search import SmartSearch from parameter_constraint import create_constraint_engine from analyze_results import ResultAnalyzer class BatchAutoOptimizer: def __init__(self, base_dir: str = "optimizer", mt5_path: str = None): self.base_dir = base_dir self.mt5_path = mt5_path self.configs_dir = os.path.join(base_dir, "configs") self.set_files_dir = os.path.join(base_dir, "set_files") self.reports_dir = os.path.join(base_dir, "reports") self.results_dir = os.path.join(base_dir, "results") self.scripts_dir = os.path.join(base_dir, "scripts") for d in [self.configs_dir, self.set_files_dir, self.reports_dir, self.results_dir, self.scripts_dir]: os.makedirs(d, exist_ok=True) def generate_set_files(self, ea_config: EAConfig, max_samples: int = 1000) -> List[Dict]: logger.info(f"Generating SET files for {ea_config.ea_name}") logger.info(f" Combinations: {ea_config.get_total_combinations()}") searcher = SmartSearch(ea_config) combinations = searcher.generate_combinations(max_samples) logger.info(f" Generated {len(combinations)} samples") engine = create_constraint_engine(ea_config) valid_combos = engine.filter_valid_combinations(combinations) logger.info(f" Valid combinations: {len(valid_combos)}") ea_set_dir = os.path.join(self.set_files_dir, ea_config.ea_name) os.makedirs(ea_set_dir, exist_ok=True) param_mapping = [] for i, combo in enumerate(valid_combos, 1): set_id = str(i).zfill(8) set_filename = f"params_{set_id}.set" set_path = os.path.join(ea_set_dir, set_filename) self._write_set_file(set_path, combo) param_mapping.append({ 'set_id': set_id, 'set_file': set_filename, 'set_path': set_path, 'report_name': f"{ea_config.ea_name}_{set_id}.htm", 'params': combo }) mapping_path = os.path.join(self.results_dir, f"{ea_config.ea_name}_param_mapping.json") with open(mapping_path, 'w', encoding='utf-8') as f: json.dump(param_mapping, f, ensure_ascii=False, indent=2) logger.info(f" SET files saved to {ea_set_dir}") return param_mapping def _write_set_file(self, set_path: str, params: Dict): lines = ["; MT5 EA Parameters", f"; Generated: {datetime.now()}", ""] for name, value in params.items(): if isinstance(value, bool): lines.append(f"{name} <{'true' if value else 'false'}>") else: lines.append(f"{name} <{value}> <{value}>") with open(set_path, 'w', encoding='utf-8') as f: f.write("\n".join(lines)) def generate_ini(self, ea_config: EAConfig) -> str: test_config = ea_config.test_config ini_path = os.path.join(self.configs_dir, f"{ea_config.ea_name}.ini") date_from = test_config.get("from_date", "2025.01.01") date_to = test_config.get("to_date", "2025.12.31") lines = [ "; MT5 Strategy Tester Configuration", f"; Generated: {datetime.now()}", "", "[Tester]", f"Expert={ea_config.ea_path}", "ExpertParameters=", f"Symbol={test_config.get('symbol', 'EURUSD')}", f"Period={test_config.get('period', 'H1')}", f"Model={test_config.get('model', 1)}", f"FromDate={date_from}", f"ToDate={date_to}", f"Deposit={test_config.get('deposit', 10000)}", f"Leverage={test_config.get('leverage', '1:100')}", "ReplaceReport=true", "ShutdownTerminal=true", ] with open(ini_path, 'w', encoding='utf-8') as f: f.write("\n".join(lines)) logger.info(f"INI saved: {ini_path}") return ini_path def generate_powershell_script(self, ea_config: EAConfig, param_mapping: List[Dict]) -> str: ps_path = os.path.join(self.scripts_dir, f"run_{ea_config.ea_name}.ps1") ea_set_dir = os.path.join(self.set_files_dir, ea_config.ea_name) ea_report_dir = os.path.join(self.reports_dir, ea_config.ea_name) os.makedirs(ea_report_dir, exist_ok=True) ini_path = os.path.join(self.configs_dir, f"{ea_config.ea_name}.ini") lines = [ f"# MT5 Batch Optimization for {ea_config.ea_name}", f"# Generated: {datetime.now()}", "", f'$MT5Path = "{self.mt5_path or "terminal64.exe"}"', f'$IniFile = "{ini_path}"', f'$SetDir = "{ea_set_dir}"', "", f"Write-Host 'Starting batch optimization for {ea_config.ea_name}'", "", "$SetFiles = Get-ChildItem -Path $SetDir -Filter 'params_*.set'", "$Total = $SetFiles.Count", "$Current = 0", "", "foreach ($SetFile in $SetFiles) {", " $Current++", ' Write-Host "[$Current/$Total] $($SetFile.Name)"', "", " $IniContent = Get-Content $IniFile", ' $IniContent = $IniContent -replace "ExpertParameters=.*", "ExpertParameters=$($SetFile.FullName)"', ' $TempIni = Join-Path $env:TEMP "temp_$([guid]::NewGuid().ToString().Substring(0,8)).ini"', " $IniContent | Set-Content $TempIni -Encoding UTF8", "", " Start-Process -FilePath $MT5Path -ArgumentList '/portable',\"/config:$TempIni\" -Wait", " Start-Sleep -Seconds 3", "}", "", 'Write-Host "Batch optimization completed!"', ] with open(ps_path, 'w', encoding='utf-8') as f: f.write("\n".join(lines)) logger.info(f"PowerShell script: {ps_path}") return ps_path def run_full_auto(self, ea_names: List[str] = None, max_samples: int = 1000): logger.info("=" * 60) logger.info("MT5 Batch Auto Optimizer") logger.info("=" * 60) configs = load_all_ea_configs(self.configs_dir) if not configs: logger.error(f"No configs found in {self.configs_dir}") return if ea_names: configs = {k: v for k, v in configs.items() if k in ea_names} for ea_name, ea_config in configs.items(): logger.info(f"\nProcessing: {ea_name}") param_mapping = self.generate_set_files(ea_config, max_samples) self.generate_ini(ea_config) self.generate_powershell_script(ea_config, param_mapping) logger.info(f" SET files: {len(param_mapping)}") logger.info(f" Next: Run the PS1 script in MT5 terminal") def analyze_all(self, ea_names: List[str] = None) -> List[Dict]: logger.info("Analyzing results...") all_results = [] search_dirs = [self.reports_dir] if not ea_names else [os.path.join(self.reports_dir, name) for name in ea_names] for search_dir in search_dirs: if not os.path.exists(search_dir): continue for ea_name in os.listdir(search_dir): ea_report_dir = os.path.join(search_dir, ea_name) if not os.path.isdir(ea_report_dir): continue analyzer = ResultAnalyzer(ea_report_dir) results = analyzer.parse_all_reports() for r in results: r['ea_name'] = ea_name all_results.extend(results) logger.info(f" {ea_name}: {len(results)} reports") if not all_results: logger.warning("No results found") return [] all_results.sort(key=lambda x: x.get('profit_factor', 0), reverse=True) output_path = os.path.join(self.results_dir, "optimization_results.json") with open(output_path, 'w', encoding='utf-8') as f: json.dump(all_results, f, indent=2, ensure_ascii=False) logger.info(f"Results: {len(all_results)} total") logger.info(f"Best PF: {all_results[0].get('profit_factor', 0) if all_results else 'N/A'}") return all_results def main(): import argparse parser = argparse.ArgumentParser(description="MT5 Batch Auto Optimizer") parser.add_argument("--base-dir", "-d", default="optimizer", help="Base directory") parser.add_argument("--ea", "-e", nargs="+", help="EA names") parser.add_argument("--max-samples", "-m", type=int, default=1000, help="Max samples per EA") parser.add_argument("--mt5-path", "-p", help="MT5 terminal path") parser.add_argument("--analyze", "-a", action="store_true", help="Analyze only") args = parser.parse_args() optimizer = BatchAutoOptimizer(args.base_dir, args.mt5_path) if args.analyze: optimizer.analyze_all(args.ea) else: optimizer.run_full_auto(args.ea, args.max_samples) if __name__ == "__main__": main()