222 lines
7.9 KiB
Python
222 lines
7.9 KiB
Python
import os
|
|
import subprocess
|
|
import time
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import List, Dict, Optional
|
|
from pathlib import Path
|
|
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s',
|
|
handlers=[
|
|
logging.FileHandler('logs/batch_executor.log'),
|
|
logging.StreamHandler()
|
|
]
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class BatchExecutor:
|
|
def __init__(self, config_path: str):
|
|
self.config = self._load_config(config_path)
|
|
self.mt5_path = self.config["mt5_settings"]["terminal_path"]
|
|
self.reports_dir = self.config["mt5_settings"]["reports_dir"]
|
|
self.results_dir = self.config["mt5_settings"].get("results_dir", "results")
|
|
os.makedirs(self.results_dir, exist_ok=True)
|
|
os.makedirs("logs", 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 _wait_for_report(self, report_path: str, timeout: int = 600) -> bool:
|
|
start_time = time.time()
|
|
while time.time() - start_time < timeout:
|
|
if os.path.exists(report_path):
|
|
time.sleep(2)
|
|
try:
|
|
with open(report_path, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
if len(content) > 100:
|
|
return True
|
|
except:
|
|
pass
|
|
time.sleep(5)
|
|
return False
|
|
|
|
def _check_mt5_process(self) -> bool:
|
|
try:
|
|
result = subprocess.run(
|
|
['tasklist', '/FI', 'IMAGENAME eq terminal64.exe'],
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
return 'terminal64.exe' in result.stdout
|
|
except:
|
|
return False
|
|
|
|
def execute_single(self, ini_path: str, wait_time: int = 120) -> Dict:
|
|
result = {
|
|
"ini_file": os.path.basename(ini_path),
|
|
"status": "pending",
|
|
"start_time": None,
|
|
"end_time": None,
|
|
"duration": 0,
|
|
"report_path": None,
|
|
"error": None
|
|
}
|
|
|
|
logger.info(f"Starting backtest: {ini_path}")
|
|
|
|
if not os.path.exists(self.mt5_path):
|
|
result["status"] = "error"
|
|
result["error"] = f"MT5 terminal not found: {self.mt5_path}"
|
|
logger.error(result["error"])
|
|
return result
|
|
|
|
result["start_time"] = datetime.now()
|
|
|
|
try:
|
|
process = subprocess.Popen(
|
|
[self.mt5_path, "/portable", f"/config:{ini_path}"],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE
|
|
)
|
|
|
|
logger.info(f"MT5 process started with PID: {process.pid}")
|
|
|
|
ini_basename = os.path.splitext(os.path.basename(ini_path))[0]
|
|
report_xml = os.path.join(self.results_dir, f"{ini_basename}_report.xml")
|
|
|
|
report_found = self._wait_for_report(report_xml, timeout=wait_time * 60)
|
|
|
|
if report_found:
|
|
result["status"] = "completed"
|
|
result["report_path"] = report_xml
|
|
logger.info(f"Report generated: {report_xml}")
|
|
else:
|
|
if process.poll() is not None:
|
|
result["status"] = "mt5_closed"
|
|
logger.warning("MT5 closed before report was generated")
|
|
else:
|
|
process.terminate()
|
|
process.wait(timeout=10)
|
|
result["status"] = "timeout"
|
|
logger.warning(f"Timeout waiting for report: {ini_path}")
|
|
|
|
if process.poll() is None:
|
|
try:
|
|
process.terminate()
|
|
process.wait(timeout=5)
|
|
except:
|
|
pass
|
|
|
|
except Exception as e:
|
|
result["status"] = "error"
|
|
result["error"] = str(e)
|
|
logger.error(f"Error executing backtest: {e}")
|
|
|
|
result["end_time"] = datetime.now()
|
|
if result["start_time"] and result["end_time"]:
|
|
result["duration"] = (result["end_time"] - result["start_time"]).total_seconds()
|
|
|
|
return result
|
|
|
|
def execute_batch(self, ini_files: List[str],
|
|
max_parallel: int = 1,
|
|
wait_time_per_test: int = 120) -> List[Dict]:
|
|
results = []
|
|
total = len(ini_files)
|
|
|
|
logger.info(f"Starting batch execution: {total} tests")
|
|
logger.info(f"Parallel execution: {max_parallel}")
|
|
|
|
for idx, ini_path in enumerate(ini_files, 1):
|
|
logger.info(f"[{idx}/{total}] Executing: {os.path.basename(ini_path)}")
|
|
result = self.execute_single(ini_path, wait_time_per_test)
|
|
results.append(result)
|
|
|
|
status = result["status"]
|
|
duration = result["duration"]
|
|
logger.info(f" Status: {status}, Duration: {duration:.1f}s")
|
|
|
|
if result["error"]:
|
|
logger.error(f" Error: {result['error']}")
|
|
|
|
success_count = sum(1 for r in results if r["status"] == "completed")
|
|
logger.info(f"Batch completed: {success_count}/{total} successful")
|
|
|
|
return results
|
|
|
|
def execute_validate_tasks(self, ini_files: List[str]) -> str:
|
|
validate_dir = os.path.join(os.path.dirname(self.mt5_path),
|
|
"MQL5", "Files", "ValidateTasks")
|
|
os.makedirs(validate_dir, exist_ok=True)
|
|
|
|
for ini_file in ini_files:
|
|
dest_path = os.path.join(validate_dir, os.path.basename(ini_file))
|
|
with open(ini_file, 'r', encoding='utf-8') as src:
|
|
content = src.read()
|
|
with open(dest_path, 'w', encoding='utf-8') as dst:
|
|
dst.write(content)
|
|
logger.info(f"Copied to ValidateTasks: {os.path.basename(ini_file)}")
|
|
|
|
logger.info(f"All {len(ini_files)} tasks queued in ValidateTasks folder")
|
|
return validate_dir
|
|
|
|
|
|
def main():
|
|
import argparse
|
|
parser = argparse.ArgumentParser(description="MT5 Batch Backtest Executor")
|
|
parser.add_argument("--config", "-c", default="config/ea_configs.yaml",
|
|
help="Path to config file")
|
|
parser.add_argument("--ini-dir", "-i",
|
|
help="Directory containing INI files to execute")
|
|
parser.add_argument("--wait", "-w", type=int, default=120,
|
|
help="Wait time per test in minutes")
|
|
parser.add_argument("--validate", "-v", action="store_true",
|
|
help="Use ValidateTasks method")
|
|
|
|
args = parser.parse_args()
|
|
|
|
executor = BatchExecutor(args.config)
|
|
|
|
if args.ini_dir:
|
|
ini_files = [os.path.join(args.ini_dir, f)
|
|
for f in os.listdir(args.ini_dir)
|
|
if f.endswith('.ini')]
|
|
else:
|
|
generator = __import__('ini_generator', fromlist=['']).INIGenerator(args.config)
|
|
ini_files = generator.generate_ini_files()
|
|
|
|
logger.info(f"Found {len(ini_files)} INI files to execute")
|
|
|
|
if args.validate:
|
|
validate_dir = executor.execute_validate_tasks(ini_files)
|
|
logger.info(f"ValidateTasks method: files copied to {validate_dir}")
|
|
logger.info("Run the Validate EA in MT5 terminal to execute")
|
|
else:
|
|
results = executor.execute_batch(ini_files, wait_time_per_test=args.wait)
|
|
|
|
success = [r for r in results if r["status"] == "completed"]
|
|
failed = [r for r in results if r["status"] != "completed"]
|
|
|
|
print("\n" + "="*60)
|
|
print("BATCH EXECUTION SUMMARY")
|
|
print("="*60)
|
|
print(f"Total tests: {len(results)}")
|
|
print(f"Completed: {len(success)}")
|
|
print(f"Failed: {len(failed)}")
|
|
print("="*60)
|
|
|
|
if failed:
|
|
print("\nFailed tests:")
|
|
for r in failed:
|
|
print(f" - {r['ini_file']}: {r['status']} - {r['error']}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |