import os import sys import json import time import shutil import threading import datetime as _dt_mod from datetime import datetime import tkinter as tk from tkinter import ttk, messagebox, filedialog import yaml import subprocess import ctypes import ctypes.wintypes sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), 'scripts')) _IMPORT_ERRORS: list = [] try: from mt5_paths import resolve_mt5_settings from ini_generator import INIGenerator from mt5_auto_runner import MT5AutoRunner from result_parser import ResultParser from report_generator import ReportGenerator from symbol_resolver import resolve_symbol except Exception as _e: _IMPORT_ERRORS.append(repr(_e)) sys.stderr.write(f"[IMPORT FAIL] {_e}\n") resolve_mt5_settings = None # type: ignore INIGenerator = None # type: ignore MT5AutoRunner = None # type: ignore ResultParser = None # type: ignore ReportGenerator = None # type: ignore resolve_symbol = None # type: ignore def _gui_log(msg: str) -> None: """输出到 stdout 同时也追加到 logs/gui_run.log。""" print(f"[gui] {msg}", flush=True) try: os.makedirs("logs", exist_ok=True) with open("logs/gui_run.log", "a", encoding="utf-8") as f: f.write(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}\n") except Exception: pass def _now() -> str: return datetime.now().strftime('%H:%M:%S') _user32 = ctypes.windll.user32 _kernel32 = ctypes.windll.kernel32 SW_MINIMIZE = 6 SWP_NOMOVE = 0x0002 SWP_NOSIZE = 0x0001 SWP_NOACTIVATE = 0x0010 SWP_NOZORDER = 0x0004 SWP_FRAMECHANGED = 0x0020 _EnumWindowsProc = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.wintypes.HWND, ctypes.wintypes.LPARAM) def _mt5_startupinfo(): si = subprocess.STARTUPINFO() si.dwFlags |= subprocess.STARTF_USESHOWWINDOW si.wShowWindow = 6 return si # MT5 策略测试器回测模式（与 MT5 官方界面"每个报价/1分钟OHLC/仅开盘价/数学计算/基于每个真实报价"一致） # 值是写入 INI 的 Model= 编号。批量页和优化页共用本列表，切换显示文字时只改这一处。 MT5_MODELS = [ '每个报价 (0)', '1分钟OHLC (1)', '仅开盘价 (2)', '数学计算 (3)', '基于每个真实报价 (4)', ] MT5_MODEL_MAP = {int(s.split('(')[1].rstrip(')')): s for s in MT5_MODELS} def _find_mt5_hwnds(pid, max_seconds=3.0): hwnds = [] deadline = time.time() + max_seconds while time.time() < deadline and not hwnds: def cb(hwnd, _lParam): owner_pid = ctypes.wintypes.DWORD() _user32.GetWindowThreadProcessId(hwnd, ctypes.byref(owner_pid)) if owner_pid.value != pid or not _user32.IsWindowVisible(hwnd): return True length = _user32.GetWindowTextLengthW(hwnd) if length <= 0: hwnds.append(hwnd) return True buf = ctypes.create_unicode_buffer(length + 1) _user32.GetWindowTextW(hwnd, buf, length + 1) title = buf.value if "MetaTrader" in title or "Strategy" in title or "Tester" in title: hwnds.append(hwnd) return True _user32.EnumWindows(_EnumWindowsProc(cb), 0) if not hwnds: time.sleep(0.2) return hwnds def _minimize_mt5_windows(pid, max_seconds=20.0): deadline = time.time() + max_seconds while time.time() < deadline: for hwnd in _find_mt5_hwnds(pid, max_seconds=0.5): _user32.ShowWindow(hwnd, SW_MINIMIZE) _user32.SetWindowPos( hwnd, 0, 0, 0, 0, 0, SWP_NOMOVE | SWP_NOSIZE | SWP_NOACTIVATE | SWP_NOZORDER | SWP_FRAMECHANGED, ) time.sleep(0.3) def _clear_mt5_chart_workspace(data_dir): if not data_dir: return charts_root = os.path.join(data_dir, "MQL5", "Profiles", "Charts") if not os.path.isdir(charts_root): return for entry in os.listdir(charts_root): profile_dir = os.path.join(charts_root, entry) if not os.path.isdir(profile_dir): continue for fname in os.listdir(profile_dir): if fname.startswith("chart") and fname.endswith(".chr"): try: os.remove(os.path.join(profile_dir, fname)) except OSError: pass def _launch_mt5(mt5_path, ini_path, data_dir, log_lines: list = None): """启动 MT5 终端跑测试。同步等 5 秒检查进程还活着，活着就让它跑。""" if log_lines is None: log_lines = [] _clear_mt5_chart_workspace(data_dir) cmd = [mt5_path, "/config:" + os.path.abspath(ini_path)] log_lines.append(f" 命令: {' '.join(cmd)!s}") if not os.path.isfile(mt5_path): log_lines.append(f" [ERR] terminal64.exe 不存在: {mt5_path!r}") return None try: process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, startupinfo=_mt5_startupinfo(), ) except OSError as e: log_lines.append(f" [ERR] 启动失败: {e!r}") return None except Exception as e: log_lines.append(f" [ERR] 异常: {e!r}") return None log_lines.append(f" pid={process.pid}") # 同步短等待（5s），检查是否启动后立即挂掉 try: process.wait(timeout=5.0) log_lines.append(f" [WARN] MT5 进程在 5s 内退出, returncode={process.returncode}") try: err = (process.stderr.read() or b"").decode('utf-8', 'replace') if err.strip(): log_lines.append(f" stderr: {err[:500]}") except Exception: pass return process except subprocess.TimeoutExpired: log_lines.append(" MT5 正常运行中（5s 后仍在跑）") threading.Thread(target=_minimize_mt5_windows, args=(process.pid,), daemon=True).start() return process def _find_latest_optimizer_xml(search_dirs, ea_name): """在多个候选目录里找最新的优化 XML 结果。""" patterns = [ f'{ea_name}_optimization.xml', f'{ea_name}_optimization[1].xml', f'{ea_name}_optimization[2].xml', ] latest = None latest_mtime = 0 for base in search_dirs: if not base or not os.path.isdir(base): continue for pattern in patterns: p = os.path.join(base, pattern) if os.path.exists(p): try: mt = os.path.getmtime(p) except OSError: continue if mt > latest_mtime: latest_mtime = mt latest = p return latest class MT5BacktestGUI: _CONFIG_FILE = "config/ea_configs.yaml" # 仅给"回测配置"标签页用 _MT5_STATE_FILE = "optimizer/mt5_state.json" # 优化器用的 MT5 路径，独立小文件 def __init__(self, root): self.root = root self.root.title("MT5 EA 批量回测工具") self.root.geometry("900x2000") self.config = self._load_or_default_config() self.scanned_eas = [] self.opt_ea_paths = {} self._scan_eas() self._create_widgets() self._load_config_to_gui() def _load_or_default_config(self): """回测配置（yaml） + MT5 路径（optimizer/mt5_state.json 优先，缺失字段从 yaml 兜底）。""" default = self._load_default_config() # 1) 优先读 mt5_state.json（路径专用小文件） if os.path.isfile(self._MT5_STATE_FILE): try: with open(self._MT5_STATE_FILE, "r", encoding="utf-8") as f: state = json.load(f) for k in ("terminal_path", "data_dir", "reports_dir", "ini_dir"): if k in state and state[k]: default["mt5_settings"][k] = state[k] _gui_log(f"已加载 MT5 路径: {self._MT5_STATE_FILE}") except Exception as e: _gui_log(f"读 mt5_state 失败 ({e})") # 2) 用 yaml 兜底（老数据 / 用户编辑过的字段；mt5_state 已有的路径字段不覆盖） if os.path.isfile(self._CONFIG_FILE): try: with open(self._CONFIG_FILE, "r", encoding="utf-8") as f: saved = yaml.safe_load(f) or {} for k in ("backtest_settings", "mt5_settings", "execution"): if k in saved and isinstance(saved.get(k), dict): for kk, vv in saved[k].items(): if vv in (None, "", []): continue # mt5_state 已经填了相同字段，跳过 yaml 的（避免覆盖最新值） if k == "mt5_settings" and kk in ("terminal_path", "data_dir", "reports_dir", "ini_dir"): if default["mt5_settings"].get(kk): continue default[k][kk] = vv if "eas" in saved and saved["eas"]: default["eas"] = saved["eas"] _gui_log(f"已加载配置: {self._CONFIG_FILE}") except Exception as e: _gui_log(f"读 yaml 失败 ({e})，使用默认") return default def _load_default_config(self): return { "eas": [], "backtest_settings": { "symbols": [ "EURUSD", "GBPUSD", "USDJPY", "USDCHF", "AUDUSD", "USDCAD", "NZDUSD", "GBPJPY", "EURJPY", "AUDJPY", "EURGBP", "EURAUD", "GBPAUD", "XAUUSD", "XAGUSD", "US30", "US500", "USTEC", "BTCUSD", "ETHUSD" ], "timeframes": ["M1", "H1"], "date_range": {"from": "2026.01.01", "to": "2026.01.04"}, "model": 0, "execution_delay": 0, "deposit": 10000, "leverage": "1:100", "currency": "USD", "visual": 0, "replace_report": True, "shutdown_terminal": True, "optimization": 0, "forward_mode": 0, "forward_date": "" }, "mt5_settings": (resolve_mt5_settings({ "terminal_path": "", "data_dir": "", "reports_dir": "{PROJECT_ROOT}/reports", "ini_dir": "{PROJECT_ROOT}/config/generated", }) if resolve_mt5_settings else { "terminal_path": "", "data_dir": "", "reports_dir": "reports", "ini_dir": os.path.join("config", "generated"), }), "execution": { "timeout_per_test": 30, "skip_existing": True, "kill_between": True } } def _scan_eas(self): data_dir = self.config["mt5_settings"]["data_dir"] experts_dir = os.path.join(data_dir, "MQL5", "Experts") if data_dir else "" self.scanned_eas = [] if experts_dir and os.path.exists(experts_dir): for root, dirs, files in os.walk(experts_dir): for f in files: if f.endswith('.ex5'): rel_path = os.path.relpath(os.path.join(root, f), experts_dir) self.scanned_eas.append(rel_path) else: print(f"[scan] MT5 Experts dir not found: {experts_dir!r} (data_dir={data_dir!r})") def _create_widgets(self): notebook = ttk.Notebook(self.root) notebook.pack(fill='both', expand=True, padx=5, pady=5) tab_basic = ttk.Frame(notebook) tab_execution = ttk.Frame(notebook) tab_reports = ttk.Frame(notebook) tab_optimizer = ttk.Frame(notebook) notebook.add(tab_basic, text="回测配置") notebook.add(tab_execution, text="执行选项") notebook.add(tab_reports, text="报告查看") notebook.add(tab_optimizer, text="参数优化") self._create_basic_tab(tab_basic) self._create_execution_tab(tab_execution) self._create_reports_tab(tab_reports) self._create_optimizer_tab(tab_optimizer) self.notebook = notebook btn_frame = ttk.Frame(self.root) btn_frame.pack(fill='x', padx=5, pady=5) ttk.Button(btn_frame, text="开始回测", command=self._on_start_backtest).pack(side='left', padx=5) ttk.Button(btn_frame, text="保存配置", command=self._on_save_config).pack(side='left', padx=5) ttk.Button(btn_frame, text="查看报告", command=self._on_view_reports).pack(side='left', padx=5) ttk.Button(btn_frame, text="参数优化", command=self._on_start_optimizer).pack(side='left', padx=5) ttk.Button(btn_frame, text="重新扫描EA", command=self._on_rescan_eas).pack(side='right', padx=5) def _create_basic_tab(self, parent): main_frame = ttk.Frame(parent) main_frame.pack(fill='both', expand=True) ea_frame = ttk.LabelFrame(main_frame, text="EA 选择 (勾选多选)") ea_frame.pack(fill='x', padx=5, pady=5) ttk.Label(ea_frame, text="已扫描到 %d 个EA，勾选需要回测的EA" % len(self.scanned_eas)).pack(anchor='w', padx=5, pady=2) search_frame = ttk.Frame(ea_frame) search_frame.pack(fill='x', padx=5, pady=2) ttk.Label(search_frame, text="搜索:").pack(side='left', padx=2) self.ea_search_var = tk.StringVar() self.ea_search_var.trace('w', lambda *args: self._filter_ea_list()) ea_search_entry = ttk.Entry(search_frame, textvariable=self.ea_search_var, width=30) ea_search_entry.pack(side='left', padx=2) ttk.Button(search_frame, text="清除", command=lambda: self.ea_search_var.set('')).pack(side='left', padx=2) container = ttk.Frame(ea_frame) container.pack(fill='both', expand=True, padx=5, pady=5) canvas = tk.Canvas(container, height=120) canvas.pack(side='left', fill='both', expand=True) scrollbar = ttk.Scrollbar(container, orient='vertical', command=canvas.yview) scrollbar.pack(side='right', fill='y') canvas['yscrollcommand'] = scrollbar.set cb_window = tk.Frame(canvas) canvas.create_window((0, 0), window=cb_window, anchor='nw') self.ea_vars = {} self.ea_cbs = {} self._rebuild_ea_checkboxes(cb_window, canvas) cb_window.update_idletasks() canvas.configure(scrollregion=canvas.bbox("all")) def on_mousewheel(event): canvas.yview_scroll(int(-1 * (event.delta / 120)), "units") canvas.bind("", on_mousewheel) btn_frame = ttk.Frame(ea_frame) btn_frame.pack(anchor='w', padx=5, pady=2) ttk.Button(btn_frame, text="全选", command=self._select_all_eas).pack(side='left', padx=2) ttk.Button(btn_frame, text="取消", command=self._deselect_all_eas).pack(side='left', padx=2) ttk.Button(btn_frame, text="反选", command=self._invert_eas).pack(side='left', padx=2) self._ea_canvas = canvas self._ea_cb_window = cb_window sym_frame = ttk.LabelFrame(main_frame, text="交易品种") sym_frame.pack(fill='x', padx=5, pady=5) self.symbol_vars = {} for i, sym in enumerate([ "EURUSD", "GBPUSD", "USDJPY", "USDCAD", "USDCHF", "AUDUSD", "NZDUSD", "EURGBP", "EURJPY", "GBPJPY", "EURCHF", "AUDJPY", "EURAUD", "GBPAUD", "EURCAD", "GBPCAD", "NZDJPY", "AUDCAD", "CADJPY", "CHFJPY", "AUDCHF", "CADCHF", "NZDCHF", "XAUUSD", "XAGUSD", "XBRUSD", "XTIUSD", "XNGUSD", "BTCUSD", "ETHUSD", "LTCUSD", "USDCNH", "USDHKD", "USDCZK", "USDDKK", "USDHUF", "USDMXN", "USDPLN", "USDTHB", "NOKJPY", "SEKJPY", "XRPUSD", "XLMUSD", "AUS200", "CHINA50", "ES35", "STOXX50", "F40", "HK50", "IT40", "JP225", "UK100", "US2000", "US30", "US500", "USTEC", "CA60", "NETH25", "SE30", "SWI20", "CHINAH", "NOR25", "TecDE30" ]): var = tk.BooleanVar(value=sym in self.config["backtest_settings"]["symbols"]) cb = ttk.Checkbutton(sym_frame, text=sym, variable=var) cb.grid(row=i//10, column=i%10, sticky='w', padx=2, pady=1) self.symbol_vars[sym] = var var.trace_add('write', lambda *_: self._update_batch_symbol_hint()) self.batch_sym_hint_var = tk.StringVar(value="") ttk.Label(sym_frame, textvariable=self.batch_sym_hint_var, foreground='#0066cc', wraplength=900, justify='left' ).grid(row=99, column=0, columnspan=10, sticky='w', padx=2, pady=(4, 2)) self._update_batch_symbol_hint() tf_frame = ttk.LabelFrame(main_frame, text="时间周期") tf_frame.pack(fill='x', padx=5, pady=5) self.tf_vars = {} for i, tf in enumerate(["M1", "M5", "M15", "M30", "H1", "H4", "D1", "W1"]): var = tk.BooleanVar(value=tf in self.config["backtest_settings"]["timeframes"]) cb = ttk.Checkbutton(tf_frame, text=tf, variable=var) cb.grid(row=0, column=i, padx=3, pady=2) self.tf_vars[tf] = var date_model_frame = ttk.Frame(main_frame) date_model_frame.pack(fill='x', padx=5, pady=5) date_frame = ttk.LabelFrame(date_model_frame, text="回测日期") date_frame.pack(side='left', fill='x', padx=5) ttk.Label(date_frame, text="开始:").grid(row=0, column=0, sticky='w', padx=5) self.from_date = ttk.Entry(date_frame, width=12) self.from_date.insert(0, self.config["backtest_settings"]["date_range"]["from"]) self.from_date.grid(row=0, column=1, padx=5) ttk.Label(date_frame, text="结束:").grid(row=0, column=2, sticky='w', padx=5) self.to_date = ttk.Entry(date_frame, width=12) self.to_date.insert(0, self.config["backtest_settings"]["date_range"]["to"]) self.to_date.grid(row=0, column=3, padx=5) forward_frame = ttk.LabelFrame(date_model_frame, text="Forward 前向测试") forward_frame.pack(side='left', fill='x', padx=5) self.forward_var = tk.StringVar(value="No") forward_values = ["No", "1/2", "1/3", "1/4", "Custom"] ttk.Radiobutton(forward_frame, text="无 (No)", variable=self.forward_var, value="No").pack(anchor='w', padx=2) ttk.Radiobutton(forward_frame, text="1/2", variable=self.forward_var, value="1/2").pack(anchor='w', padx=2) ttk.Radiobutton(forward_frame, text="1/3", variable=self.forward_var, value="1/3").pack(anchor='w', padx=2) ttk.Radiobutton(forward_frame, text="1/4", variable=self.forward_var, value="1/4").pack(anchor='w', padx=2) custom_row = ttk.Frame(forward_frame) custom_row.pack(fill='x', padx=2, pady=2) ttk.Radiobutton(custom_row, text="自定义:", variable=self.forward_var, value="Custom").pack(side='left', padx=2) self.forward_custom_date = ttk.Entry(custom_row, width=12) self.forward_custom_date.insert(0, self.config["backtest_settings"].get("forward_date", "")) self.forward_custom_date.pack(side='left', padx=2) ttk.Label(custom_row, text="(日期)").pack(side='left') model_frame = ttk.LabelFrame(date_model_frame, text="回测模式") model_frame.pack(side='left', fill='x', padx=5) self.model_var = tk.IntVar(value=self.config["backtest_settings"]["model"]) for m in MT5_MODELS: val = int(m.split('(')[1].rstrip(')')) ttk.Radiobutton(model_frame, text=m, variable=self.model_var, value=val).pack(anchor='w', padx=2) delay_frame = ttk.LabelFrame(date_model_frame, text="执行延迟") delay_frame.pack(side='left', fill='x', padx=5) saved_delay = int((self.config.get("backtest_settings") or {}).get("execution_delay", 0) or 0) if saved_delay == 0: delay_type_value = 0 delay_value_text = "100" elif saved_delay == -1: delay_type_value = 1 delay_value_text = "100" else: delay_type_value = 2 delay_value_text = str(saved_delay) self.delay_type = tk.IntVar(value=delay_type_value) ttk.Radiobutton(delay_frame, text="无延迟 (0)", variable=self.delay_type, value=0).pack(anchor='w', padx=2) ttk.Radiobutton(delay_frame, text="随机延迟 (-1)", variable=self.delay_type, value=1).pack(anchor='w', padx=2) delay_row = ttk.Frame(delay_frame) delay_row.pack() ttk.Radiobutton(delay_row, text="固定延迟:", variable=self.delay_type, value=2).pack(side='left', padx=2) self.delay_value = ttk.Entry(delay_row, width=6) self.delay_value.insert(0, delay_value_text) self.delay_value.pack(side='left') ttk.Label(delay_row, text="ms").pack(side='left') acct_frame = ttk.LabelFrame(date_model_frame, text="账户") acct_frame.pack(side='left', fill='x', padx=5) ttk.Label(acct_frame, text="保证金:").pack(side='left', padx=2) self.deposit = ttk.Entry(acct_frame, width=10) self.deposit.insert(0, str(self.config["backtest_settings"]["deposit"])) self.deposit.pack(side='left', padx=2) ttk.Label(acct_frame, text="杠杆:").pack(side='left', padx=2) self.leverage = ttk.Combobox(acct_frame, width=8, values=["1:10", "1:50", "1:100", "1:200", "1:500"]) self.leverage.set(self.config["backtest_settings"]["leverage"]) self.leverage.pack(side='left', padx=2) ttk.Button(acct_frame, text="开始回测", command=self._on_start_backtest).pack(side='left', padx=5) def _create_execution_tab(self, parent): exec_frame = ttk.LabelFrame(parent, text="执行设置") exec_frame.pack(fill='x', padx=5, pady=10) ttk.Label(exec_frame, text="超时(分钟):").grid(row=0, column=0, sticky='w', padx=5, pady=3) self.timeout = ttk.Entry(exec_frame, width=8) self.timeout.insert(0, str(self.config["execution"]["timeout_per_test"])) self.timeout.grid(row=0, column=1, sticky='w', padx=5, pady=3) self.skip_var = tk.BooleanVar(value=self.config["execution"]["skip_existing"]) ttk.Checkbutton(exec_frame, text="跳过已完成的测试", variable=self.skip_var).grid(row=1, column=0, columnspan=2, sticky='w', padx=5, pady=3) self.kill_var = tk.BooleanVar(value=self.config["execution"]["kill_between"]) ttk.Checkbutton(exec_frame, text="测试间关闭MT5进程", variable=self.kill_var).grid(row=2, column=0, columnspan=2, sticky='w', padx=5, pady=3) self.visual_var = tk.BooleanVar(value=self.config["backtest_settings"]["visual"] == 1) ttk.Checkbutton(exec_frame, text="可视化模式(慢)", variable=self.visual_var).grid(row=3, column=0, columnspan=2, sticky='w', padx=5, pady=3) path_frame = ttk.LabelFrame(parent, text="路径设置") path_frame.pack(fill='x', padx=5, pady=10) ttk.Label(path_frame, text="MT5终端 (terminal64.exe):").grid(row=0, column=0, sticky='w', padx=5, pady=3) self.mt5_path = ttk.Entry(path_frame, width=50) self.mt5_path.insert(0, self.config["mt5_settings"].get("terminal_path", "")) self.mt5_path.grid(row=0, column=1, padx=5, pady=3, sticky='ew') self.mt5_path.bind('', lambda e: self._persist_path_settings()) ttk.Button(path_frame, text="浏览...", command=self._browse_mt5).grid(row=0, column=2, padx=5) ttk.Label(path_frame, text="MT5数据目录 (含 MQL5/Experts 的那一层):").grid(row=1, column=0, sticky='w', padx=5, pady=3) self.data_dir = ttk.Entry(path_frame, width=50) self.data_dir.insert(0, self.config["mt5_settings"].get("data_dir", "")) self.data_dir.grid(row=1, column=1, padx=5, pady=3, sticky='ew') self.data_dir.bind('', lambda e: self._persist_path_settings()) ttk.Button(path_frame, text="浏览...", command=self._browse_data).grid(row=1, column=2, padx=5) ttk.Label(path_frame, text="报告输出目录:").grid(row=2, column=0, sticky='w', padx=5, pady=3) self.reports_dir = ttk.Entry(path_frame, width=50) self.reports_dir.insert(0, self.config["mt5_settings"].get("reports_dir", "")) self.reports_dir.grid(row=2, column=1, padx=5, pady=3, sticky='ew') self.reports_dir.bind('', lambda e: self._persist_path_settings()) ttk.Button(path_frame, text="浏览...", command=self._browse_reports).grid(row=2, column=2, padx=5) path_hint = ttk.Label(path_frame, text="提示: 数据目录 = File → 打开数据文件夹 (MT5 菜单里)，里面要能看见 MQL5\\Experts 文件夹", foreground='gray') path_hint.grid(row=3, column=0, columnspan=3, sticky='w', padx=5, pady=(5, 0)) def _browse_mt5(self): path = filedialog.askopenfilename( title="选择 terminal64.exe", filetypes=[("MT5 终端", "terminal64.exe"), ("所有文件", "*.*")], initialdir=os.path.dirname(self.mt5_path.get()) if self.mt5_path.get() else None, ) if path: self.mt5_path.delete(0, tk.END) self.mt5_path.insert(0, path) def _browse_reports(self): path = filedialog.askdirectory(initialdir=self.reports_dir.get()) if path: self.reports_dir.delete(0, tk.END) self.reports_dir.insert(0, path) def _browse_data(self): path = filedialog.askdirectory(initialdir=self.data_dir.get()) if path: self.data_dir.delete(0, tk.END) self.data_dir.insert(0, path) def _select_all_eas(self): for var in self.ea_vars.values(): var.set(True) def _deselect_all_eas(self): for var in self.ea_vars.values(): var.set(False) def _invert_eas(self): for var in self.ea_vars.values(): var.set(not var.get()) def _rebuild_ea_checkboxes(self, cb_window, canvas): for widget in cb_window.winfo_children(): widget.destroy() search_text = self.ea_search_var.get().lower() filtered_eas = [ea for ea in self.scanned_eas if search_text in os.path.basename(ea).lower()] self.ea_vars = {} for i, ea in enumerate(filtered_eas): var = tk.BooleanVar(value=False) if ea in getattr(self, '_previous_ea_states', {}): var.set(self._previous_ea_states[ea]) cb = ttk.Checkbutton(cb_window, text=os.path.basename(ea), variable=var) cb.grid(row=i // 4, column=i % 4, sticky='w', padx=5, pady=1) self.ea_vars[ea] = var cb_window.update_idletasks() if hasattr(self, '_ea_canvas'): self._ea_canvas.configure(scrollregion=self._ea_canvas.bbox("all")) def _filter_ea_list(self): self._previous_ea_states = {ea: var.get() for ea, var in self.ea_vars.items()} if hasattr(self, '_ea_cb_window') and hasattr(self, '_ea_canvas'): self._rebuild_ea_checkboxes(self._ea_cb_window, self._ea_canvas) def _get_selected_eas(self): selected = [] for ea, var in self.ea_vars.items(): if var.get(): ea_name = os.path.splitext(os.path.basename(ea))[0] selected.append({"name": ea_name, "filename": ea}) return selected def _on_rescan_eas(self): self._scan_eas() for ea, cb in list(self.ea_vars.items())[:]: if ea not in self.scanned_eas: del self.ea_vars[ea] for i, ea in enumerate(self.scanned_eas): if ea not in self.ea_vars: var = tk.BooleanVar(value=False) self.ea_vars[ea] = var messagebox.showinfo("扫描完成", "扫描到 %d 个EA文件" % len(self.scanned_eas)) def _load_config_to_gui(self): pass def _get_config(self): symbols = [s for s, v in self.symbol_vars.items() if v.get()] timeframes = [t for t, v in self.tf_vars.items() if v.get()] delay = self.delay_type.get() if delay == 0: execution_delay = 0 elif delay == 1: execution_delay = -1 else: try: execution_delay = int(self.delay_value.get()) except: execution_delay = 0 forward = self.forward_var.get() forward_mode_map = {"No": 0, "1/2": 1, "1/3": 2, "1/4": 3, "Custom": 4} forward_mode = forward_mode_map.get(forward, 0) forward_date = self.forward_custom_date.get() if forward == "Custom" else "" return { "eas": self._get_selected_eas(), "backtest_settings": { "symbols": symbols, "timeframes": timeframes, "date_range": {"from": self.from_date.get(), "to": self.to_date.get()}, "model": self.model_var.get(), "execution_delay": execution_delay, "deposit": float(self.deposit.get()), "leverage": self.leverage.get(), "currency": "USD", "visual": 1 if self.visual_var.get() else 0, "replace_report": True, "shutdown_terminal": True, "optimization": 0, "forward_mode": forward_mode, "forward_date": forward_date, # symbol_aliases 在 GUI 里没有编辑控件，直接从内存配置透传（避免保存时把它清掉） "symbol_aliases": (self.config.get("backtest_settings") or {}).get("symbol_aliases") or {}, }, "mt5_settings": { "terminal_path": self.mt5_path.get(), "data_dir": self.data_dir.get(), "reports_dir": self.reports_dir.get(), "ini_dir": os.path.join(os.path.dirname(self.reports_dir.get()), "config", "generated"), }, "execution": { "timeout_per_test": int(self.timeout.get()), "skip_existing": self.skip_var.get(), "kill_between": self.kill_var.get() } } def _on_save_config(self): config = self._get_config() os.makedirs("config", exist_ok=True) with open("config/ea_configs.yaml", 'w', encoding='utf-8') as f: yaml.dump(config, f, allow_unicode=True) messagebox.showinfo("保存成功", "配置已保存到 config/ea_configs.yaml") def _persist_path_settings(self): """三个路径输入框失焦时调用：写到 optimizer/mt5_state.json（不写 yaml）。""" try: self.config.setdefault("mt5_settings", {}) self.config["mt5_settings"]["terminal_path"] = self.mt5_path.get() self.config["mt5_settings"]["data_dir"] = self.data_dir.get() self.config["mt5_settings"]["reports_dir"] = self.reports_dir.get() reports_parent = os.path.dirname(self.reports_dir.get() or ".") or "." self.config["mt5_settings"]["ini_dir"] = os.path.join(reports_parent, "config", "generated") os.makedirs(os.path.dirname(self._MT5_STATE_FILE) or ".", exist_ok=True) state = {k: self.config["mt5_settings"][k] for k in ("terminal_path", "data_dir", "reports_dir", "ini_dir")} with open(self._MT5_STATE_FILE, "w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False, indent=2) _gui_log(f"已保存 MT5 路径: {self._MT5_STATE_FILE}") except Exception as e: _gui_log(f"自动保存失败: {e}") def _on_start_backtest(self): _gui_log("开始回测 button clicked") if _IMPORT_ERRORS: _gui_log(f"导入错误: {_IMPORT_ERRORS}") messagebox.showerror("依赖缺失", f"导入失败: {_IMPORT_ERRORS[0]}\n请运行 pip install -r requirements.txt") return if not all([INIGenerator, MT5AutoRunner, ResultParser, ReportGenerator]): messagebox.showerror("依赖缺失", "模块为 None（导入被吞）") return config = self._get_config() _gui_log(f"config keys: {list(config.keys())}, eas={len(config.get('eas',[]))}, sym={len(config['backtest_settings']['symbols'])}, tf={len(config['backtest_settings']['timeframes'])}") if not config["eas"]: messagebox.showerror("错误", "请选择至少一个EA文件"); return if not config["backtest_settings"]["symbols"]: messagebox.showerror("错误", "请选择至少一个交易品种"); return if not config["backtest_settings"]["timeframes"]: messagebox.showerror("错误", "请选择至少一个时间周期"); return os.makedirs("logs", exist_ok=True) def _log(msg): self.opt_log.insert('end', msg + "\n") self.opt_log.see('end') _gui_log(msg) def _run(): try: _log("=== 开始回测 ===") generator = INIGenerator(config) _log(f" INIGenerator OK, output_dir={generator.output_dir}") ini_files = generator.generate_ini_files() _log(f" 生成 INI: {len(ini_files)} 个") if not ini_files: _log(" [ERR] 未生成任何 INI — 检查 EA 路径与 MT5 Experts 目录") return runner = MT5AutoRunner(config) reports_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "reports") os.makedirs(reports_dir, exist_ok=True) runner.reports_dir = reports_dir _log(f" mt5_path={runner.mt5_path}") _log(f" data_dir={runner.data_dir}") _log(f" reports_dir={runner.reports_dir}") _log(" 开始调用 MT5...（每个 EA/Symbol/TF 组合会启动一次 MT5 子进程，耐心等待）") results = runner.run_full_auto( ini_files, skip_if_exists=config["execution"].get("skip_existing", False), ) completed = sum(1 for r in results if r["status"] == "completed") failed = [r for r in results if r["status"] not in ("completed", "already_completed")] _log(f" 完成: {completed}/{len(results)}") if failed: _log(f" 失败: {len(failed)} 个") for r in failed[:3]: _log(f" - {r.get('ini_file','?')}: {r.get('error', r.get('status'))}") if completed > 0: parser = ResultParser(reports_dir) parsed = parser.parse_all_reports() _log(f" 解析: {len(parsed)} 个报告") if parsed: rg = ReportGenerator(parsed) rg.generate_excel(os.path.join(reports_dir, "batch_summary.xlsx")) rg.generate_csv(os.path.join(reports_dir, "batch_summary.csv")) rg.generate_markdown(os.path.join(reports_dir, "batch_summary.md")) _log(f" 报告: {reports_dir}") _log("=== 全部完成 ===") except Exception as e: import traceback tb = traceback.format_exc() _log(f"[ERR] {e}") _log(tb) _gui_log(f"thread crashed: {e}\n{tb}") threading.Thread(target=_run, daemon=True).start() def _on_start_backtest_from_opt(self): if not all([INIGenerator, MT5AutoRunner]): messagebox.showerror("错误", "依赖未就绪，请检查 Python 环境") return ea_name = self.opt_ea_var.get() if not ea_name: messagebox.showwarning('警告', '请先选择一个EA') return symbol = self.opt_symbol.get() period = self.opt_period.get() from_date = self.opt_from_date.get() to_date = self.opt_to_date.get() model_str = self.opt_model.get() model = int(model_str.split('(')[1].split(')')[0]) if '(' in model_str else 1 config = { "eas": [{"name": ea_name, "filename": f"Experts\\{ea_name}.ex5"}], "backtest_settings": { "symbols": [symbol], "timeframes": [period], "date_range": {"from": from_date, "to": to_date}, "model": model, "execution_mode": 0, "execution_delay": 0, "deposit": 10000, "leverage": "1:100", "currency": "USD", "visual": 0, "replace_report": True, "shutdown_terminal": True, "optimization": 0, "forward_mode": 0, "forward_date": "", }, "mt5_settings": self.config["mt5_settings"], "execution": self.config["execution"] } os.makedirs("logs", exist_ok=True) def _run(): try: self.opt_log.insert('end', f"=== 单EA回测: {ea_name} ===\n") self.opt_log.see('end') generator = INIGenerator(config) ini_files = generator.generate_ini_files() self.opt_log.insert('end', f" 生成 INI: {len(ini_files)}\n") runner = MT5AutoRunner(config) runner.reports_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "reports") os.makedirs(runner.reports_dir, exist_ok=True) results = runner.run_full_auto(ini_files, skip_if_exists=False) completed = sum(1 for r in results if r["status"] == "completed") self.opt_log.insert('end', f" 完成: {completed}/{len(results)}\n") self.opt_log.insert('end', "=== 完成 ===\n") self.opt_log.see('end') except Exception as e: import traceback self.opt_log.insert('end', f"[ERR] {e}\n{traceback.format_exc()}\n") self.opt_log.see('end') threading.Thread(target=_run, daemon=True).start() def _on_view_reports(self): path = self.reports_dir.get() if os.path.exists(path): subprocess.Popen(['explorer', path]) else: messagebox.showwarning("警告", "报告目录不存在") def _on_start_optimizer(self): self.notebook.select(3) def _create_reports_tab(self, parent): main_frame = ttk.Frame(parent) main_frame.pack(fill='both', expand=True) control_frame = ttk.Frame(main_frame) control_frame.pack(fill='x', padx=5, pady=5) ttk.Label(control_frame, text="报告目录:").pack(side='left', padx=5) self.report_dir = ttk.Entry(control_frame, width=40) self.report_dir.insert(0, self.config["mt5_settings"]["reports_dir"]) self.report_dir.pack(side='left', padx=5) ttk.Button(control_frame, text="浏览", command=self._browse_report_dir).pack(side='left', padx=5) ttk.Button(control_frame, text="刷新", command=self._refresh_reports).pack(side='left', padx=5) search_frame = ttk.LabelFrame(main_frame, text="搜索和筛选") search_frame.pack(fill='x', padx=5, pady=5) search_row = ttk.Frame(search_frame) search_row.pack(fill='x', padx=5, pady=3) ttk.Label(search_row, text="搜索:").pack(side='left', padx=5) self.search_var = tk.StringVar() self.search_var.trace('w', lambda *args: self._apply_filter()) search_entry = ttk.Entry(search_row, textvariable=self.search_var, width=20) search_entry.pack(side='left', padx=5) ttk.Button(search_row, text="清除", command=lambda: self.search_var.set('')).pack(side='left', padx=2) filter_row = ttk.Frame(search_frame) filter_row.pack(fill='x', padx=5, pady=3) ttk.Label(filter_row, text="EA筛选:").pack(side='left', padx=5) self.filter_ea_var = tk.StringVar(value='全部') self.filter_ea_combo = ttk.Combobox(filter_row, textvariable=self.filter_ea_var, width=20, state='readonly') self.filter_ea_combo.pack(side='left', padx=5) self.filter_ea_combo.bind('<>', lambda e: self._apply_filter()) ttk.Label(filter_row, text="品种:").pack(side='left', padx=10) self.filter_symbol_var = tk.StringVar(value='全部') self.filter_symbol_combo = ttk.Combobox(filter_row, textvariable=self.filter_symbol_var, width=10, state='readonly') self.filter_symbol_combo.pack(side='left', padx=5) self.filter_symbol_combo.bind('<>', lambda e: self._apply_filter()) ttk.Label(filter_row, text="周期:").pack(side='left', padx=10) self.filter_period_var = tk.StringVar(value='全部') self.filter_period_combo = ttk.Combobox(filter_row, textvariable=self.filter_period_var, width=8, state='readonly') self.filter_period_combo.pack(side='left', padx=5) self.filter_period_combo.bind('<>', lambda e: self._apply_filter()) ttk.Button(filter_row, text="重置筛选", command=self._reset_filter).pack(side='left', padx=10) options_frame = ttk.LabelFrame(main_frame, text="排序选项") options_frame.pack(fill='x', padx=5, pady=5) sort_row = ttk.Frame(options_frame) sort_row.pack(anchor='w', padx=5, pady=3) ttk.Label(sort_row, text="排序字段:").pack(side='left', padx=5) self.sort_field = ttk.Combobox(sort_row, width=15, values=[ "profit_factor", "net_profit", "gross_profit", "win_rate", "total_trades", "max_drawdown", "sharpe_ratio", "expert", "symbol", "period" ]) self.sort_field.set("profit_factor") self.sort_field.pack(side='left', padx=5) self.sort_asc = tk.BooleanVar(value=False) ttk.Radiobutton(sort_row, text="降序", variable=self.sort_asc, value=False).pack(side='left', padx=5) ttk.Radiobutton(sort_row, text="升序", variable=self.sort_asc, value=True).pack(side='left', padx=5) ttk.Button(sort_row, text="应用排序", command=self._apply_filter).pack(side='left', padx=10) export_row = ttk.Frame(options_frame) export_row.pack(fill='x', padx=5, pady=3) ttk.Label(export_row, text="导出CSV:").pack(side='left', padx=5) self.csv_path_var = tk.StringVar() self.csv_path = ttk.Entry(export_row, textvariable=self.csv_path_var, width=50, state='readonly') self.csv_path.pack(side='left', padx=5, fill='x', expand=True) ttk.Button(export_row, text="另存为...", command=self._export_csv_as).pack(side='left', padx=2) ttk.Button(export_row, text="导出", command=self._export_csv).pack(side='left', padx=2) self.csv_export_dir = os.path.join(self.config["mt5_settings"]["reports_dir"], "exports") self._update_csv_suggested_name() tree_frame = ttk.Frame(main_frame) tree_frame.pack(fill='both', expand=True, padx=5, pady=5) columns = [ "expert", "symbol", "period", "net_profit", "gross_profit", "gross_loss", "profit_factor", "win_rate", "total_trades", "max_drawdown", "sharpe_ratio" ] self.report_tree = ttk.Treeview(tree_frame, columns=columns, show='headings', height=15, selectmode='extended') for col in columns: self.report_tree.heading(col, text=col) self.report_tree.column(col, width=100) scrollbar = ttk.Scrollbar(tree_frame, orient='vertical', command=self.report_tree.yview) self.report_tree.configure(yscrollcommand=scrollbar.set) self.report_tree.pack(side='left', fill='both', expand=True) scrollbar.pack(side='right', fill='y') self.report_tree.bind("", self._on_row_double_click) self.report_tree.bind("", self._on_right_click) self.report_data = [] self.tree_item_to_path = {} self._create_context_menu() self._refresh_reports() def _browse_report_dir(self): path = filedialog.askdirectory(initialdir=self.report_dir.get()) if path: self.report_dir.delete(0, tk.END) self.report_dir.insert(0, path) self._refresh_reports() def _refresh_reports(self): for item in self.report_tree.get_children(): self.report_tree.delete(item) sys.path.insert(0, os.path.join(os.getcwd(), 'scripts')) from result_parser import ResultParser reports_dir = self.report_dir.get() if not os.path.exists(reports_dir): return parser = ResultParser(reports_dir) self.report_data = parser.parse_all_reports() self._update_filter_options() self._apply_filter() def _get_filtered_sorted_data(self): if not self.report_data: return [] search_text = self.search_var.get().lower() filter_ea = self.filter_ea_var.get() filter_symbol = self.filter_symbol_var.get() filter_period = self.filter_period_var.get() filtered_data = [] for r in self.report_data: info = r.get("test_info", {}) expert = info.get("expert", "") symbol = info.get("symbol", "") period = info.get("period", "") if search_text and search_text not in expert.lower() and search_text not in symbol.lower(): continue if filter_ea != '全部' and expert != filter_ea: continue if filter_symbol != '全部' and symbol != filter_symbol: continue if filter_period != '全部' and period != filter_period: continue filtered_data.append(r) sort_by = self.sort_field.get() reverse = not self.sort_asc.get() def _coerce_numeric(val): if isinstance(val, bool): return float(val) if isinstance(val, (int, float)): return float(val) if val is None or val == "": return 0.0 try: return float(str(val).replace(",", "")) except (ValueError, TypeError): return None def _to_str(val): if val is None: return "" if isinstance(val, bool): return "1" if val else "0" if isinstance(val, (int, float)): return str(val) return str(val).lower() sort_as_number = False for r in filtered_data: metrics = r.get("metrics", {}) if sort_by in metrics: v = metrics[sort_by] if isinstance(v, (int, float)) and not isinstance(v, bool): sort_as_number = True break if isinstance(v, str) and v.strip(): try: float(v.replace(",", "")) sort_as_number = True break except ValueError: pass if not sort_as_number: for r in filtered_data: test_info = r.get("test_info", {}) if sort_by in test_info: v = test_info[sort_by] if isinstance(v, (int, float)) and not isinstance(v, bool): sort_as_number = True break if isinstance(v, str) and v.strip(): try: float(v.replace(",", "")) sort_as_number = True break except ValueError: pass def get_sort_key(r): metrics = r.get("metrics", {}) test_info = r.get("test_info", {}) if sort_as_number: if sort_by in metrics: n = _coerce_numeric(metrics[sort_by]) if n is not None: return n if sort_by in test_info: n = _coerce_numeric(test_info[sort_by]) if n is not None: return n return 0.0 else: if sort_by in metrics: return _to_str(metrics[sort_by]) if sort_by in test_info: return _to_str(test_info[sort_by]) return "" return sorted(filtered_data, key=get_sort_key, reverse=reverse) def _apply_filter(self): for item in self.report_tree.get_children(): self.report_tree.delete(item) self.tree_item_to_path = {} if not self.report_data: return sorted_data = self._get_filtered_sorted_data() if hasattr(self, 'csv_path_var'): self._update_csv_suggested_name() for r in sorted_data: info = r.get("test_info", {}) metrics = r.get("metrics", {}) values = [ info.get("expert", ""), info.get("symbol", ""), info.get("period", ""), metrics.get("net_profit", ""), metrics.get("gross_profit", ""), metrics.get("gross_loss", ""), metrics.get("profit_factor", ""), metrics.get("win_rate", ""), metrics.get("total_trades", ""), metrics.get("max_drawdown", ""), metrics.get("sharpe_ratio", "") ] iid = self.report_tree.insert("", "end", values=values) self.tree_item_to_path[iid] = r.get("file_path", "") def _reset_filter(self): self.search_var.set('') self.filter_ea_var.set('全部') self.filter_symbol_var.set('全部') self.filter_period_var.set('全部') self._apply_filter() def _update_filter_options(self): ea_list = ['全部'] symbol_list = ['全部'] period_list = ['全部'] for r in self.report_data: info = r.get("test_info", {}) ea = info.get("expert", "") symbol = info.get("symbol", "") period = info.get("period", "") if ea and ea not in ea_list: ea_list.append(ea) if symbol and symbol not in symbol_list: symbol_list.append(symbol) if period and period not in period_list: period_list.append(period) self.filter_ea_combo['values'] = ea_list self.filter_symbol_combo['values'] = symbol_list self.filter_period_combo['values'] = period_list def _suggest_csv_filename(self): import re ts = time.strftime("%Y%m%d_%H%M%S") parts = [] ea = self.filter_ea_var.get() if ea and ea != '全部': parts.append(ea) sym = self.filter_symbol_var.get() if sym and sym != '全部': parts.append(sym) per = self.filter_period_var.get() if per and per != '全部': parts.append(per) search = self.search_var.get().strip() if search: parts.append("q-" + search[:10]) cleaned = [] for p in parts: s = re.sub(r'[^\w\u4e00-\u9fff\-]', '_', p).strip('_') if s: cleaned.append(s) sig = "_".join(cleaned) if cleaned else "all" if len(sig) > 60: sig = sig[:60].rstrip('_') return f"results_{ts}_{sig}.csv" def _get_csv_export_dir(self): base = self.csv_export_dir try: os.makedirs(base, exist_ok=True) except Exception: base = self.report_dir.get() os.makedirs(base, exist_ok=True) return base def _update_csv_suggested_name(self): name = self._suggest_csv_filename() self.csv_path_var.set(os.path.join(self._get_csv_export_dir(), name)) def _export_csv(self): sys.path.insert(0, os.path.join(os.getcwd(), 'scripts')) from result_parser import ResultParser reports_dir = self.report_dir.get() filtered_results = self._get_filtered_sorted_data() if not filtered_results: messagebox.showwarning("警告", "当前筛选条件下没有数据可导出") return self._update_csv_suggested_name() csv_path = self.csv_path_var.get() target_dir = os.path.dirname(csv_path) or self._get_csv_export_dir() os.makedirs(target_dir, exist_ok=True) if os.path.exists(csv_path): if not messagebox.askyesno("文件已存在", f"目标文件已存在，是否覆盖？\n{csv_path}"): return sort_by = self.sort_field.get() reverse = not self.sort_asc.get() parser = ResultParser(reports_dir) parser.export_to_csv(filtered_results, csv_path, sort_by=sort_by, reverse=reverse) messagebox.showinfo("导出成功", f"已按当前筛选条件导出 {len(filtered_results)} 条记录到:\n{csv_path}") self._update_csv_suggested_name() def _export_csv_as(self): self._update_csv_suggested_name() default_name = os.path.basename(self.csv_path_var.get()) default_dir = os.path.dirname(self.csv_path_var.get()) or self._get_csv_export_dir() os.makedirs(default_dir, exist_ok=True) path = filedialog.asksaveasfilename( title="导出 CSV", defaultextension=".csv", initialfile=default_name, initialdir=default_dir, filetypes=[("CSV 文件", "*.csv"), ("所有文件", "*.*")] ) if not path: return self.csv_path_var.set(path) def _on_row_double_click(self, event): selection = self.report_tree.selection() if not selection: return path = self.tree_item_to_path.get(selection[0], "") if path and os.path.exists(path): subprocess.Popen(['notepad', path]) return values = self.report_tree.item(selection[0])["values"] label = f"{values[0]} {values[1]} {values[2]}" if len(values) >= 3 else "" messagebox.showwarning("警告", f"未找到对应的报告文件: {label}") def _create_context_menu(self): self.context_menu = tk.Menu(self.root, tearoff=0) self.context_menu.add_command(label="打开报告", command=self._open_selected_report) self.context_menu.add_command(label="打开所在目录", command=self._open_report_directory) self.context_menu.add_separator() self.context_menu.add_command(label="复制报告文字", command=self._copy_selected_reports_text) self.context_menu.add_command(label="复制报告到...", command=self._copy_reports_to) self.context_menu.add_command(label="删除选中报告", command=self._delete_selected_reports) self.context_menu.add_separator() self.context_menu.add_command(label="全选", command=lambda: self.report_tree.selection_set(self.report_tree.get_children())) self.context_menu.add_command(label="取消选择", command=lambda: self.report_tree.selection_remove(self.report_tree.selection())) def _on_right_click(self, event): self.context_menu.post(event.x_root, event.y_root) def _get_selected_report_paths(self): selection = self.report_tree.selection() if not selection: return [] paths = [] for item in selection: p = self.tree_item_to_path.get(item, "") if p and os.path.exists(p): paths.append(p) return paths def _open_selected_report(self): paths = self._get_selected_report_paths() if not paths: messagebox.showwarning("警告", "未找到报告文件") return for path in paths[:1]: subprocess.Popen(['notepad', path]) def _open_report_directory(self): selection = self.report_tree.selection() if not selection: return path = self.tree_item_to_path.get(selection[0], "") if path and os.path.exists(path): os.startfile(os.path.dirname(path)) return values = self.report_tree.item(selection[0])["values"] label = f"{values[0]} {values[1]} {values[2]}" if len(values) >= 3 else "" messagebox.showwarning("警告", f"未找到对应的报告文件: {label}") def _copy_selected_reports_text(self): selection = self.report_tree.selection() if not selection: messagebox.showwarning("警告", "请先选择要复制文字的报告") return columns = list(self.report_tree["columns"]) headers = [self.report_tree.heading(col, "text") or col for col in columns] rows = [] for item in selection: values = self.report_tree.item(item, "values") rows.append([str(v) if v is not None else "" for v in values]) col_widths = [len(h) for h in headers] for row in rows: for i, cell in enumerate(row): if i < len(col_widths): col_widths[i] = max(col_widths[i], len(cell)) def _format_row(cells): return " ".join(cells[i].ljust(col_widths[i]) for i in range(len(cells))) lines = [_format_row(headers), _format_row(["-" * w for w in col_widths])] for row in rows: lines.append(_format_row(row)) text = "\n".join(lines) if len(selection) > 1: text += f"\n\n(共 {len(selection)} 条报告)" try: self.root.clipboard_clear() self.root.clipboard_append(text) self.root.update() except Exception as e: messagebox.showerror("错误", f"复制到剪贴板失败: {str(e)}") return self.root.after(100, lambda: self._show_copy_toast(len(selection), len(text))) def _show_copy_toast(self, count, length): try: self.context_menu.unpost() except Exception: pass if hasattr(self, "_copy_toast") and self._copy_toast is not None: try: self._copy_toast.destroy() except Exception: pass toast = tk.Toplevel(self.root) toast.overrideredirect(True) toast.attributes("-topmost", True) x = self.root.winfo_rootx() + self.root.winfo_width() // 2 - 100 y = self.root.winfo_rooty() + self.root.winfo_height() // 2 - 25 toast.geometry(f"200x50+{x}+{y}") ttk.Label(toast, text=f"已复制 {count} 条报告文字\n(共 {length} 字符)", anchor="center", justify="center").pack(expand=True, fill='both') self._copy_toast = toast self.root.after(1500, lambda: self._destroy_copy_toast(toast)) def _destroy_copy_toast(self, toast): try: toast.destroy() except Exception: pass if getattr(self, "_copy_toast", None) is toast: self._copy_toast = None def _copy_reports_to(self): paths = self._get_selected_report_paths() if not paths: messagebox.showwarning("警告", "请先选择要复制的报告") return target_dir = filedialog.askdirectory(title="选择目标目录") if not target_dir: return import shutil copied = 0 for src in paths: try: filename = os.path.basename(src) target = os.path.join(target_dir, filename) shutil.copy2(src, target) copied += 1 except Exception as e: messagebox.showerror("错误", f"复制失败: {str(e)}") return messagebox.showinfo("成功", f"已复制 {copied} 个报告到: {target_dir}") def _delete_selected_reports(self): paths = self._get_selected_report_paths() if not paths: messagebox.showwarning("警告", "请先选择要删除的报告") return count = len(paths) if messagebox.askyesno("确认删除", f"确定要删除选中的 {count} 个报告文件吗？\n此操作不可撤销。"): deleted = 0 for path in paths: try: os.remove(path) deleted += 1 except Exception as e: messagebox.showerror("错误", f"删除失败: {str(e)}") return messagebox.showinfo("成功", f"已删除 {deleted} 个报告") self._refresh_reports() def _create_optimizer_tab(self, parent): main_frame = ttk.Frame(parent) main_frame.pack(fill='both', expand=True) opt_frame = ttk.LabelFrame(main_frame, text="EA 参数优化") opt_frame.pack(fill='both', expand=True, padx=5, pady=5) top_row = ttk.Frame(opt_frame) top_row.pack(fill='x', padx=5, pady=5) ttk.Label(top_row, text="优化目录:").pack(side='left', padx=5) self.opt_dir = ttk.Entry(top_row, width=35) self.opt_dir.insert(0, "optimizer") self.opt_dir.pack(side='left', padx=5) ttk.Button(top_row, text="浏览", command=self._browse_opt_dir).pack(side='left', padx=5) ea_row = ttk.Frame(opt_frame) ea_row.pack(fill='x', padx=5, pady=5) ttk.Label(ea_row, text="选择EA:").pack(side='left', padx=5) self.opt_ea_var = tk.StringVar() self.opt_ea_combo = ttk.Combobox(ea_row, textvariable=self.opt_ea_var, width=35, state='readonly') self.opt_ea_combo.pack(side='left', padx=5) self.opt_ea_combo.bind('<>', self._on_ea_selected) ttk.Button(ea_row, text="刷新EA列表", command=self._refresh_opt_ea_list).pack(side='left', padx=5) ttk.Button(ea_row, text="自动扫描MT5", command=self._auto_scan_eas).pack(side='left', padx=5) ttk.Button(ea_row, text="批量执行优化", command=self._run_batch_optimize).pack(side='left', padx=5) param_frame = ttk.LabelFrame(opt_frame, text="参数配置 (点击编辑)") param_frame.pack(fill='both', expand=True, padx=5, pady=5) tree_scroll_y = ttk.Scrollbar(param_frame, orient='vertical') tree_scroll_y.pack(side='right', fill='y') tree_scroll_x = ttk.Scrollbar(param_frame, orient='horizontal') tree_scroll_x.pack(side='bottom', fill='x') cols = ('param_name', 'param_type', 'default', 'min', 'max', 'step', 'optimize') self.param_tree = ttk.Treeview(param_frame, columns=cols, show='headings', yscrollcommand=tree_scroll_y.set, xscrollcommand=tree_scroll_x.set, height=8) tree_scroll_y.config(command=self.param_tree.yview) tree_scroll_x.config(command=self.param_tree.xview) self.param_tree.heading('param_name', text='参数名') self.param_tree.heading('param_type', text='类型') self.param_tree.heading('default', text='默认值') self.param_tree.heading('min', text='最小值') self.param_tree.heading('max', text='最大值') self.param_tree.heading('step', text='步进') self.param_tree.heading('optimize', text='优化') self.param_tree.column('param_name', width=150, anchor='w') self.param_tree.column('param_type', width=80, anchor='center') self.param_tree.column('default', width=80, anchor='center') self.param_tree.column('min', width=80, anchor='center') self.param_tree.column('max', width=80, anchor='center') self.param_tree.column('step', width=60, anchor='center') self.param_tree.column('optimize', width=50, anchor='center') self.param_tree.pack(fill='both', expand=True) self.param_tree.bind('', self._on_param_double_click) btn_frame = ttk.Frame(opt_frame) btn_frame.pack(fill='x', padx=5, pady=5) ttk.Button(btn_frame, text="添加参数", command=self._add_param).pack(side='left', padx=3) ttk.Button(btn_frame, text="删除选中", command=self._del_param).pack(side='left', padx=3) ttk.Button(btn_frame, text="保存配置", command=self._save_ea_config).pack(side='left', padx=3) ttk.Button(btn_frame, text="加载SET文件", command=self._load_set_file).pack(side='left', padx=3) ttk.Button(btn_frame, text="切换优化", command=self._toggle_param_optimize).pack(side='left', padx=3) test_frame = ttk.LabelFrame(opt_frame, text="回测配置") test_frame.pack(fill='x', padx=5, pady=5) action_row = ttk.Frame(test_frame) action_row.pack(fill='x', padx=5, pady=5) ttk.Button(action_row, text="开始回测", command=self._on_start_backtest_from_opt).pack(side='left', padx=5) sym_row = ttk.Frame(test_frame) sym_row.pack(fill='x', padx=5, pady=3) ttk.Label(sym_row, text="逻辑品种:").pack(side='left', padx=5) self.opt_symbol = ttk.Combobox(sym_row, width=12, values=['EURUSD', 'GBPUSD', 'USDJPY', 'USDCAD', 'USDCHF', 'AUDUSD', 'NZDUSD', 'EURGBP', 'EURJPY', 'GBPJPY', 'EURCHF', 'AUDJPY', 'EURAUD', 'GBPAUD', 'EURCAD', 'GBPCAD', 'XAUUSD', 'XAGUSD', 'XBRUSD', 'XTIUSD', 'BTCUSD', 'ETHUSD', 'LTCUSD', 'USDCNH', 'USDHKD', 'USDCZK', 'USDDKK', 'USDHUF', 'USDMXN', 'USDPLN', 'USDTHB', 'NOKJPY', 'SEKJPY', 'XRPUSD', 'XLMUSD', 'AUS200', 'CHINA50', 'ES35', 'STOXX50', 'F40', 'HK50', 'IT40', 'JP225', 'UK100', 'US2000', 'US30', 'US500', 'USTEC', 'CA60', 'NETH25', 'SE30', 'SWI20', 'CHINAH', 'NOR25', 'TecDE30']) self.opt_symbol.set('XAUUSD') self.opt_symbol.pack(side='left', padx=5) self.opt_symbol_resolved = ttk.Label(sym_row, text="→ XAUUSD", foreground='#666666') self.opt_symbol_resolved.pack(side='left', padx=(10, 2)) ttk.Label(sym_row, text="周期:").pack(side='left', padx=10) self.opt_period = ttk.Combobox(sym_row, width=8, values=['M1', 'M5', 'M15', 'M30', 'H1', 'H4', 'D1', 'W1']) self.opt_period.set('M1') self.opt_period.pack(side='left', padx=5) ttk.Label(sym_row, text="模式:").pack(side='left', padx=10) self.opt_model = ttk.Combobox(sym_row, width=20, values=MT5_MODELS) self.opt_model.set('1分钟OHLC (1)') self.opt_model.pack(side='left', padx=5) delay_row = ttk.Frame(test_frame) delay_row.pack(fill='x', padx=5, pady=3) ttk.Label(delay_row, text="延迟:").pack(side='left', padx=5) self.opt_delay_type = ttk.Combobox(delay_row, width=18, values=['无延迟 (0)', '随机延迟 (-1)', '固定延迟']) self.opt_delay_type.set('无延迟 (0)') self.opt_delay_type.pack(side='left', padx=5) ttk.Label(delay_row, text="ms:").pack(side='left', padx=2) self.opt_delay_value = ttk.Entry(delay_row, width=8) self.opt_delay_value.insert(0, "200") self.opt_delay_value.pack(side='left', padx=2) ttk.Label(delay_row, text="优化指标:").pack(side='left', padx=10) self.opt_criterion = ttk.Combobox(delay_row, width=18, values=[ '余额最大值 (0)', '盈利因子最大值 (1)', '期望收益 (2)', '回撤最小值 (3)', '恢复因子 (4)', '夏普比率 (5)', '自定义 (6)', '复合指标 (7)']) self.opt_criterion.set('盈利因子最大值 (1)') self.opt_criterion.pack(side='left', padx=5) mode_row = ttk.Frame(test_frame) mode_row.pack(fill='x', padx=5, pady=3) ttk.Label(mode_row, text="优化模式:").pack(side='left', padx=5) self.opt_optimization_mode = ttk.Combobox(mode_row, width=20, values=[ '禁用 (0)', '慢速完整算法 (1)', '快速遗传算法 (2)', 'MarketWatch所有符号 (3)']) self.opt_optimization_mode.set('快速遗传算法 (2)') self.opt_optimization_mode.pack(side='left', padx=5) date_row = ttk.Frame(test_frame) date_row.pack(fill='x', padx=5, pady=3) ttk.Label(date_row, text="开始:").pack(side='left', padx=5) self.opt_from_date = ttk.Entry(date_row, width=12) self.opt_from_date.insert(0, "2026.01.01") self.opt_from_date.pack(side='left', padx=5) ttk.Label(date_row, text="结束:").pack(side='left', padx=10) self.opt_to_date = ttk.Entry(date_row, width=12) self.opt_to_date.insert(0, "2026.01.04") self.opt_to_date.pack(side='left', padx=5) self.opt_symbol.bind('<>', lambda e: self._update_resolved_symbol_hint()) self.opt_symbol.bind('', lambda e: self._update_resolved_symbol_hint()) acct_row = ttk.Frame(test_frame) acct_row.pack(fill='x', padx=5, pady=3) ttk.Label(acct_row, text="保证金:").pack(side='left', padx=5) self.opt_deposit = ttk.Entry(acct_row, width=10) self.opt_deposit.insert(0, "10000") self.opt_deposit.pack(side='left', padx=5) ttk.Label(acct_row, text="杠杆:").pack(side='left', padx=10) self.opt_leverage = ttk.Combobox(acct_row, width=8, values=["1:10", "1:50", "1:100", "1:200", "1:500"]) self.opt_leverage.set("1:100") self.opt_leverage.pack(side='left', padx=5) ttk.Label(acct_row, text="等待超时(分钟):").pack(side='left', padx=10) self.opt_timeout_min = ttk.Entry(acct_row, width=8) self.opt_timeout_min.insert(0, "60") self.opt_timeout_min.pack(side='left', padx=5) wf_frame = ttk.LabelFrame(test_frame, text="向前优化 (Forward)") wf_frame.pack(fill='x', padx=5, pady=5) wf_row1 = ttk.Frame(wf_frame) wf_row1.pack(fill='x', padx=5, pady=3) self.wf_enabled = tk.BooleanVar(value=False) ttk.Checkbutton(wf_row1, text="启用向前优化", variable=self.wf_enabled, command=self._update_wf_ui_state).pack(side='left', padx=5) ttk.Label(wf_row1, text="模式:").pack(side='left', padx=5) self.wf_mode = ttk.Combobox(wf_row1, width=18, values=['1/2 (50%)', '1/3 (33%)', '1/4 (25%)', '自定义日期 (4)']) self.wf_mode.set('1/3 (33%)') self.wf_mode.pack(side='left', padx=5) self.wf_mode.bind('<>', lambda e: self._update_wf_ui_state()) ttk.Label(wf_row1, text="ForwardDate:").pack(side='left', padx=(10, 5)) self.wf_custom_date = ttk.Entry(wf_row1, width=12) self.wf_custom_date.insert(0, "") self.wf_custom_date.pack(side='left', padx=5) self.wf_mode_map = {'1/2 (50%)': 1, '1/3 (33%)': 2, '1/4 (25%)': 3, '自定义日期 (4)': 4} self.wf_mode_reverse_map = {v: k for k, v in self.wf_mode_map.items()} self._update_wf_ui_state() action_frame = ttk.Frame(opt_frame) action_frame.pack(fill='x', padx=5, pady=5) ttk.Button(action_frame, text="运行优化", command=self._run_mt5_optimize).pack(side='left', padx=5) ttk.Button(action_frame, text="分析结果", command=self._analyze_opt_results).pack(side='left', padx=5) self.opt_progress = ttk.Progressbar(action_frame, mode='determinate') self.opt_progress.pack(side='left', fill='x', expand=True, padx=10) log_frame = ttk.LabelFrame(main_frame, text="执行日志") log_frame.pack(fill='both', expand=True, padx=5, pady=5) self.opt_log = tk.Text(log_frame, height=10, width=80) self.opt_log.pack(fill='both', expand=True, padx=5, pady=5) self._refresh_opt_ea_list() self._update_resolved_symbol_hint() def _browse_opt_dir(self): path = filedialog.askdirectory(initialdir=self.opt_dir.get()) if path: self.opt_dir.delete(0, tk.END) self.opt_dir.insert(0, path) def _refresh_opt_ea_list(self): configs_dir = os.path.join(self.opt_dir.get(), 'configs') if not os.path.exists(configs_dir): self.opt_log.insert('end', f" [ERR] 配置目录不存在: {configs_dir}\n") self.opt_log.insert('end', f" 创建方法: mkdir {configs_dir}\n") return sys.path.insert(0, os.path.join(os.getcwd(), 'scripts', 'optimizer')) from ea_config_parser import load_all_ea_configs configs = load_all_ea_configs(configs_dir) if not configs: self.opt_log.insert('end', f" [WARN] {configs_dir} 下没有 JSON 配置\n") self.opt_log.insert('end', f" 请把 EA 的参数配置 .json 放进该目录\n") self.opt_ea_combo['values'] = [] return installed_names = None mt5_data = self.data_dir.get() if not mt5_data: self.opt_log.insert('end', f" [WARN] '回测配置' 里 MT5 数据目录为空，跳过 EA 存在性校验\n") else: try: from ea_auto_scanner import EAAutoScanner scanner = EAAutoScanner(mt5_data) experts = scanner.scan_experts() if experts: installed_names = {e['name'] for e in experts} for e in experts: self.opt_ea_paths[e['name']] = e['path'] else: self.opt_log.insert('end', f" [WARN] {mt5_data}\\MQL5\\Experts 下没有 .ex5 文件\n") except Exception as e: import traceback self.opt_log.insert('end', f" [ERR] 扫描MT5失败: {e}\n") self.opt_log.insert('end', f" 请确认 '回测配置' 里的 MT5 数据目录正确\n") self.opt_log.insert('end', f" 目录应包含 MQL5\\Experts 子目录\n") filtered_names = [] removed_names = [] for name in configs.keys(): if installed_names is not None and name not in installed_names: removed_names.append(name) else: filtered_names.append(name) if removed_names: self.opt_log.insert('end', f"已过滤 {len(removed_names)} 个MT5中不存在的EA: {', '.join(removed_names)}\n") for name in removed_names: self.opt_log.insert('end', f" - {name}.json\n") configs = {k: configs[k] for k in filtered_names} ea_names = list(configs.keys()) self.opt_ea_combo['values'] = ea_names if ea_names: self.opt_ea_var.set(ea_names[0]) self._load_ea_params(ea_names[0]) self.opt_log.insert('end', f"已加载 {len(configs)} 个EA配置\n") def _auto_scan_eas(self): mt5_data = self.data_dir.get() if not mt5_data: messagebox.showwarning('警告', '请先在回测配置中设置MT5数据目录') return sys.path.insert(0, os.path.join(os.getcwd(), 'scripts', 'optimizer')) from ea_auto_scanner import EAAutoScanner self.opt_log.insert('end', f"扫描MT5目录: {mt5_data}\n") self.opt_log.update() scanner = EAAutoScanner(mt5_data) experts = scanner.scan_experts() if not experts: self.opt_log.insert('end', "未找到EA文件\n") return self.opt_log.insert('end', f"找到 {len(experts)} 个EA:\n") self.opt_log.update() ea_list = [] mt5_tester_dir = os.path.join(mt5_data, 'MQL5', 'Profiles', 'Tester') for ea_info in experts: ea_name = ea_info['name'] ea_list.append(ea_name) self.opt_ea_paths[ea_name] = ea_info['path'] set_filename = ea_name + '_optimization.set' set_path = os.path.join(mt5_tester_dir, set_filename) if os.path.exists(set_path): self.opt_log.insert('end', f" ✓ {ea_name} (有SET)\n") else: set_files = [f for f in os.listdir(mt5_tester_dir) if f.endswith('.set') and ea_name.lower() in f.lower().replace('.set', '')] if set_files: self.opt_log.insert('end', f" ~ {ea_name} (匹配: {set_files[0]})\n") else: self.opt_log.insert('end', f" - {ea_name}\n") self.opt_log.insert('end', f"\n共 {len(ea_list)} 个EA\n") self.opt_log.update() self.opt_ea_combo['values'] = ea_list if ea_list: self.opt_ea_var.set(ea_list[0]) self._auto_load_ea_with_set(mt5_data, ea_list[0]) def _auto_load_ea_with_set(self, mt5_data, ea_name=None): if ea_name is None: ea_name = self.opt_ea_var.get() for item in self.param_tree.get_children(): self.param_tree.delete(item) mt5_tester_dir = os.path.join(mt5_data, 'MQL5', 'Profiles', 'Tester') set_filename = ea_name + '_optimization.set' set_path = os.path.join(mt5_tester_dir, set_filename) if not os.path.exists(set_path): set_files = [f for f in os.listdir(mt5_tester_dir) if f.endswith('.set') and ea_name.lower() in f.lower().replace('.set', '')] if set_files: set_path = os.path.join(mt5_tester_dir, set_files[0]) self.opt_log.insert('end', f"加载SET: {set_files[0]}\n") if os.path.exists(set_path): params = self._parse_mt5_set_file(set_path) for pname, pinfo in params.items(): self.param_tree.insert('', 'end', values=( pname, pinfo.get('type', 'int'), pinfo.get('default', 0), pinfo.get('min', 0), pinfo.get('max', 100), pinfo.get('step', 1), 'Y' if pinfo.get('optimize', True) else 'N' )) self.opt_log.insert('end', f"已加载 {len(params)} 个参数\n") else: self.opt_log.insert('end', f"无SET文件，请手动添加参数\n") def _on_ea_selected(self, event=None): ea_name = self.opt_ea_var.get() if ea_name: mt5_data = self.data_dir.get() self._auto_load_ea_with_set(mt5_data, ea_name) def _load_ea_params(self, ea_name): for item in self.param_tree.get_children(): self.param_tree.delete(item) configs_dir = os.path.join(self.opt_dir.get(), 'configs') sys.path.insert(0, os.path.join(os.getcwd(), 'scripts', 'optimizer')) from ea_config_parser import load_ea_config mt5_data = self.data_dir.get() config_path = os.path.join(configs_dir, ea_name + '.json') if os.path.exists(config_path): config = load_ea_config(config_path) test_cfg = config.test_config or {} wf_cfg = config.walk_forward or {} if test_cfg.get('symbol'): self.opt_symbol.set(str(test_cfg.get('symbol'))) self._update_resolved_symbol_hint() if test_cfg.get('period'): self.opt_period.set(str(test_cfg.get('period'))) if 'model' in test_cfg: model_val = int(test_cfg.get('model', 1)) self.opt_model.set(MT5_MODEL_MAP.get(model_val, '1分钟OHLC (1)')) if test_cfg.get('from_date'): self.opt_from_date.delete(0, tk.END) self.opt_from_date.insert(0, str(test_cfg.get('from_date'))) if test_cfg.get('to_date'): self.opt_to_date.delete(0, tk.END) self.opt_to_date.insert(0, str(test_cfg.get('to_date'))) if 'deposit' in test_cfg: self.opt_deposit.delete(0, tk.END) self.opt_deposit.insert(0, str(test_cfg.get('deposit'))) if test_cfg.get('leverage'): self.opt_leverage.set(str(test_cfg.get('leverage'))) if 'timeout_min' in test_cfg: self.opt_timeout_min.delete(0, tk.END) self.opt_timeout_min.insert(0, str(test_cfg.get('timeout_min'))) exec_delay = int(test_cfg.get('execution_delay', 0) or 0) if exec_delay == 0: self.opt_delay_type.set('无延迟 (0)') elif exec_delay == -1: self.opt_delay_type.set('随机延迟 (-1)') else: self.opt_delay_type.set('固定延迟') self.opt_delay_value.delete(0, tk.END) self.opt_delay_value.insert(0, str(exec_delay)) opt_mode = int(test_cfg.get('optimization_mode', 2) or 2) opt_mode_map = { 0: '禁用 (0)', 1: '慢速完整算法 (1)', 2: '快速遗传算法 (2)', 3: 'MarketWatch所有符号 (3)', } self.opt_optimization_mode.set(opt_mode_map.get(opt_mode, '快速遗传算法 (2)')) opt_criterion = int(test_cfg.get('optimization_criterion', 1) or 1) criterion_map = { 0: '余额最大值 (0)', 1: '盈利因子最大值 (1)', 2: '期望收益 (2)', 3: '回撤最小值 (3)', 4: '恢复因子 (4)', 5: '夏普比率 (5)', 6: '自定义 (6)', 7: '复合指标 (7)', } self.opt_criterion.set(criterion_map.get(opt_criterion, '盈利因子最大值 (1)')) wf_enabled = bool(wf_cfg.get('enabled', False)) self.wf_enabled.set(wf_enabled) wf_mode = int(wf_cfg.get('forward_mode', 2) or 2) self.wf_mode.set(self.wf_mode_reverse_map.get(wf_mode, '1/3 (33%)')) self.wf_custom_date.delete(0, tk.END) self.wf_custom_date.insert(0, str(wf_cfg.get('forward_date', '') or '')) self._update_wf_ui_state() if config.parameters: for pname, param in config.parameters.items(): optimize = 'Y' if param.optimize else 'N' self.param_tree.insert('', 'end', values=( pname, param.param_type, param.default, param.min_value, param.max_value, param.step, optimize )) self.opt_log.insert('end', f"从配置加载 {len(config.parameters)} 个参数\n") return self._auto_load_ea_with_set(mt5_data, ea_name) def _add_param(self): self.param_tree.insert('', 'end', values=('NewParam', 'int', 0, 0, 100, 1, 'Y')) def _del_param(self): sel = self.param_tree.selection() if sel: self.param_tree.delete(sel) def _toggle_param_optimize(self): sel = self.param_tree.selection() if sel: cur = self.param_tree.item(sel[0], 'values') new_opt = 'N' if cur[6] == 'Y' else 'Y' self.param_tree.item(sel[0], values=(cur[0], cur[1], cur[2], cur[3], cur[4], cur[5], new_opt)) def _on_param_double_click(self, event): item = self.param_tree.selection() if not item: return col = self.param_tree.identify_column(event.x) col_idx = int(col.replace('#', '')) - 1 cols = ('param_name', 'param_type', 'default', 'min', 'max', 'step', 'optimize') if col_idx < 2: return cur_vals = self.param_tree.item(item[0], 'values') entry = ttk.Entry(self.param_tree) entry.insert(0, cur_vals[col_idx]) entry.select_clear() entry.focus() def on_validate(): new_val = entry.get() new_vals = list(cur_vals) new_vals[col_idx] = new_val self.param_tree.item(item[0], values=new_vals) entry.destroy() entry.bind('', lambda e: on_validate()) entry.bind('', lambda e: entry.destroy()) entry.place(x=event.x, y=event.y, width=100) def _update_wf_ui_state(self): """ForwardDate 仅在启用向前优化且选择自定义日期(4)时可编辑。""" enabled = self.wf_enabled.get() self.wf_mode.configure(state='readonly' if enabled else 'disabled') mode_val = self.wf_mode_map.get(self.wf_mode.get(), 2) if enabled else 0 self.wf_custom_date.configure(state='normal' if (enabled and mode_val == 4) else 'disabled') def _format_resolved_hint(self, symbols): """Format a hint string showing logical -> resolved symbol mapping. Shared by batch backtest tab (multi-select) and optimizer tab (single). Returns "" when nothing is selected or no aliases are configured. """ if not symbols: return "" aliases = (self.config.get('backtest_settings') or {}).get('symbol_aliases') or {} if not aliases: return "" parts = [] for sym in symbols: if not sym: continue resolved = resolve_symbol(sym, {'symbol_aliases': aliases}) if resolve_symbol else sym if resolved == sym: parts.append(f"{sym}(直通)") else: parts.append(f"{sym}→{resolved}") return "实际品种：" + ", ".join(parts) if parts else "" def _update_batch_symbol_hint(self): if not hasattr(self, 'batch_sym_hint_var'): return checked = [s for s, v in self.symbol_vars.items() if v.get()] self.batch_sym_hint_var.set(self._format_resolved_hint(checked)) def _resolve_test_config(self): """Build the minimal dict the resolver needs (only symbol_aliases).""" return { 'symbol_aliases': (self.config.get('backtest_settings') or {}).get('symbol_aliases') or {}, } def _update_resolved_symbol_hint(self): if not hasattr(self, 'opt_symbol_resolved') or resolve_symbol is None: return try: sym = self.opt_symbol.get().split()[0] if ' ' in self.opt_symbol.get() else self.opt_symbol.get() resolved = resolve_symbol(sym, self._resolve_test_config()) except Exception: resolved = self.opt_symbol.get() if resolved == sym: self.opt_symbol_resolved.config(text=f"→ {resolved}", foreground='#666666') else: self.opt_symbol_resolved.config(text=f"→ {resolved} (别名)", foreground='#0066cc') def _save_ea_config(self, quiet=False): ea_name = self.opt_ea_var.get() if not ea_name: if not quiet: messagebox.showwarning('警告', '请先选择一个EA') return False wf_mode_val = self.wf_mode_map.get(self.wf_mode.get(), 2) if self.wf_enabled.get() else 0 wf_forward_date = self.wf_custom_date.get().strip() if self.wf_enabled.get() and wf_mode_val == 4 and not wf_forward_date: if not quiet: messagebox.showwarning('警告', 'ForwardMode=自定义日期 (4) 时，必须填写 ForwardDate (YYYY.MM.DD)') return False configs_dir = os.path.join(self.opt_dir.get(), 'configs') os.makedirs(configs_dir, exist_ok=True) params = {} for item in self.param_tree.get_children(): vals = self.param_tree.item(item, 'values') pname, ptype, pdefault, pmin, pmax, pstep, poptimize = vals if pname and pname != 'NewParam': try: default_val = float(pdefault) if pdefault else 0 min_val = float(pmin) if pmin else 0 max_val = float(pmax) if pmax else 100 step_val = float(pstep) if pstep else 1 except: default_val, min_val, max_val, step_val = 0, 0, 100, 1 params[pname] = { 'type': ptype if ptype in ('int', 'double', 'bool') else 'int', 'default': default_val, 'min': min_val, 'max': max_val, 'step': step_val, 'optimize': poptimize == 'Y' if poptimize else True } import json config_path = os.path.join(configs_dir, ea_name + '.json') config_data = { 'ea_name': ea_name, 'ea_path': self.opt_ea_paths.get(ea_name, f"MY-EA\\{ea_name}.ex5"), 'description': 'Edited via GUI', 'search_strategy': 'auto', 'parameters': params, 'test_config': { 'symbol': self.opt_symbol.get().split()[0] if ' ' in self.opt_symbol.get() else self.opt_symbol.get(), 'period': self.opt_period.get(), 'from_date': self.opt_from_date.get(), 'to_date': self.opt_to_date.get(), 'model': int(self.opt_model.get().split('(')[1].split(')')[0]) if '(' in self.opt_model.get() else 1, 'execution_delay': self._get_opt_execution_delay(), 'optimization_mode': self._get_opt_optimization_mode(), 'optimization_criterion': self._get_opt_optimization_criterion(), 'deposit': int(self.opt_deposit.get()) if self.opt_deposit.get().isdigit() else 10000, 'leverage': self.opt_leverage.get(), 'timeout_min': int(self.opt_timeout_min.get()) if self.opt_timeout_min.get().isdigit() else 60 }, 'walk_forward': { 'enabled': self.wf_enabled.get(), 'forward_mode': wf_mode_val, 'forward_date': wf_forward_date if (self.wf_enabled.get() and wf_mode_val == 4) else '' } } with open(config_path, 'w', encoding='utf-8') as f: json.dump(config_data, f, indent=4, ensure_ascii=False) self.opt_log.insert('end', f"配置已保存: {config_path}\n") if not quiet: self.opt_log.update() return True def _get_opt_execution_delay(self): delay_type = self.opt_delay_type.get() if '无延迟' in delay_type: return 0 elif '随机延迟' in delay_type: return -1 else: try: return int(self.opt_delay_value.get()) except ValueError: return 0 def _get_opt_optimization_mode(self): mode_str = self.opt_optimization_mode.get() if '(' in mode_str: return int(mode_str.split('(')[1].split(')')[0]) return 2 def _get_opt_optimization_criterion(self): criterion_map = { '余额最大值': 0, '盈利因子最大值': 1, '期望收益': 2, '回撤最小值': 3, '恢复因子': 4, '夏普比率': 5, '自定义': 6, '复合指标': 7, } criterion_str = self.opt_criterion.get() for key, val in criterion_map.items(): if key in criterion_str: return val return 1 def _load_set_file(self): ea_name = self.opt_ea_var.get() if not ea_name: messagebox.showwarning('警告', '请先选择一个EA') return file_path = filedialog.askopenfilename( title='选择SET文件', initialdir=os.path.join(self.opt_dir.get(), 'set_files'), filetypes=[('SET files', '*.set'), ('All files', '*.*')] ) if not file_path or not os.path.exists(file_path): return self.opt_log.insert('end', f"加载SET文件: {file_path}\n") for item in self.param_tree.get_children(): self.param_tree.delete(item) params = self._parse_set_file(file_path) if params: for pname, ptype, pval, min_val, max_val, step, optimize_flag in params: self.param_tree.insert('', 'end', values=(pname, ptype, pval, 0, max_val, step, optimize_flag)) self.opt_log.insert('end', f"成功导入 {len(params)} 个参数\n") else: self.opt_log.insert('end', f"解析SET文件失败或文件为空\n") messagebox.showwarning('警告', '无法解析SET文件内容') def _parse_set_file(self, set_path): params = [] try: with open(set_path, 'rb') as f: raw = f.read() if raw[:2] in (b'\xff\xfe', b'\xfe\xff'): content = raw[2:].decode('utf-16-le', errors='ignore') else: content = raw.decode('utf-8', errors='ignore') for line in content.split('\n'): line = line.strip() if not line or line.startswith(';') or line.startswith('#'): continue if '=' in line: parts = line.split('=') pname = parts[0].strip() try: values_str = parts[1].strip() if len(parts) > 1 else '' values = [v.strip() for v in values_str.split('||')] if values and values[0]: first_val = values[0].lower() if first_val == 'true': pval = 1.0 ptype = 'bool' elif first_val == 'false': pval = 0.0 ptype = 'bool' else: pval = float(values[0]) ptype = 'double' if pval != int(pval) or '.' in values[0] else 'int' max_val_str = values[3] if len(values) > 3 else '' if max_val_str.lower() == 'true': max_val = 1.0 elif max_val_str.lower() == 'false': max_val = 0.0 elif max_val_str: max_val = float(max_val_str) else: max_val = max(abs(pval) * 10, 100) if pval != 0 else 100 min_val = 0 step = 0.01 if ptype == 'double' else 1 optimize_flag = 'Y' if len(values) > 4 and values[4] == 'Y' else 'N' params.append((pname, ptype, pval, min_val, max_val, step, optimize_flag)) except: pass except Exception as e: pass return params def _parse_mt5_set_file(self, set_path): params = {} try: with open(set_path, 'rb') as f: raw = f.read() if raw[:2] in (b'\xff\xfe', b'\xfe\xff'): content = raw[2:].decode('utf-16-le', errors='ignore') else: content = raw.decode('utf-8', errors='ignore') for line in content.split('\n'): line = line.strip() if not line or line.startswith(';') or line.startswith('#'): continue if '=' in line: parts = line.split('=') pname = parts[0].strip() try: values_str = parts[1].strip() if len(parts) > 1 else '' values = [v.strip() for v in values_str.split('||')] if values and values[0]: first_val = values[0].lower() if first_val == 'true': pval = 1.0 ptype = 'bool' elif first_val == 'false': pval = 0.0 ptype = 'bool' else: pval = float(values[0]) ptype = 'double' if pval != int(pval) or '.' in values[0] else 'int' max_val_str = values[3] if len(values) > 3 else '' if max_val_str.lower() == 'true': max_val = 1.0 elif max_val_str.lower() == 'false': max_val = 0.0 elif max_val_str: max_val = float(max_val_str) else: max_val = max(abs(pval) * 10, 100) if pval != 0 else 100 min_val_str = values[2] if len(values) > 2 else '' if min_val_str.lower() == 'true': min_val = 1.0 elif min_val_str.lower() == 'false': min_val = 0.0 elif min_val_str: min_val = float(min_val_str) else: min_val = 0 step = 0.01 if ptype == 'double' else 1 optimize_flag = values[4] == 'Y' if len(values) > 4 else True params[pname] = { 'type': ptype, 'default': pval, 'min': min_val, 'max': max_val, 'step': step, 'optimize': optimize_flag } except: pass except Exception as e: pass return params def _run_mt5_optimize(self): ea_name = self.opt_ea_var.get() if not ea_name: messagebox.showwarning('警告', '请先选择一个EA') return if not self._save_ea_config(quiet=True): messagebox.showwarning('警告', '当前优化配置无效，请检查 ForwardDate 或先选择 EA') return wf_mode_val = self.wf_mode_map.get(self.wf_mode.get(), 2) if self.wf_enabled.get() else 0 if self.wf_enabled.get() and wf_mode_val == 4 and not self.wf_custom_date.get().strip(): messagebox.showwarning('警告', '当前选择了自定义向前优化日期，请先填写 ForwardDate 并保存配置') return mt5_path = self.mt5_path.get() if not mt5_path or not os.path.isfile(mt5_path): messagebox.showerror('错误', f'MT5终端路径无效: {mt5_path!r}\n请在"回测配置"标签页用"浏览..."按钮设置 terminal64.exe') return sys.path.insert(0, os.path.join(os.getcwd(), 'scripts', 'optimizer')) from ea_batch_optimizer import EABatchOptimizer from ea_config_parser import load_ea_config configs_dir = os.path.join(self.opt_dir.get(), 'configs') config_path = os.path.join(configs_dir, ea_name + '.json') if not os.path.exists(config_path): messagebox.showerror('错误', '请先保存EA配置') return try: config = load_ea_config(config_path) optimizer = EABatchOptimizer(self.opt_dir.get(), mt5_data_dir=self.data_dir.get()) self.opt_log.insert('end', f"正在生成优化配置...\n") self.opt_log.update() ini_path = optimizer.generate_ini_for_ea(config, None) self.opt_log.insert('end', f"INI: {ini_path}\n") self.opt_log.update() abs_ini_path = os.path.abspath(ini_path) self.opt_log.insert('end', "正在启动MT5优化器...\n") self.opt_log.update() mt5_data_dir = self.data_dir.get() log_lines: list = [] _launch_mt5(mt5_path, abs_ini_path, mt5_data_dir, log_lines=log_lines) for line in log_lines: self.opt_log.insert('end', line + "\n") self.opt_log.update() mode = self.opt_optimization_mode.get().split('(')[0] if '(' in self.opt_optimization_mode.get() else '遗传算法' self.opt_log.insert('end', f"MT5优化器运行中({mode})\n") self.opt_log.insert('end', "优化完成后点击'分析结果'查看\n") except Exception as e: self.opt_log.insert('end', f"错误: {str(e)}\n") def _run_batch_optimize(self): _gui_log("点击: 批量执行优化") if not self._save_ea_config(quiet=True): messagebox.showwarning('警告', '当前优化配置无效，请检查 ForwardDate 或先选择 EA') return wf_mode_val = self.wf_mode_map.get(self.wf_mode.get(), 2) if self.wf_enabled.get() else 0 if self.wf_enabled.get() and wf_mode_val == 4 and not self.wf_custom_date.get().strip(): messagebox.showwarning('警告', '当前选择了自定义向前优化日期，请先填写 ForwardDate 并保存配置') return mt5_path = self.mt5_path.get() if not mt5_path or not os.path.isfile(mt5_path): messagebox.showerror('错误', f'MT5终端路径无效: {mt5_path!r}\n请在"回测配置"标签页用"浏览..."按钮设置 terminal64.exe') return mt5_data_dir = self.data_dir.get() if not mt5_data_dir or not os.path.isdir(mt5_data_dir): messagebox.showerror('错误', f'MT5数据目录无效: {mt5_data_dir!r}\n请在"回测配置"标签页设置 MT5 数据目录') return sys.path.insert(0, os.path.join(os.getcwd(), 'scripts', 'optimizer')) from ea_batch_optimizer import EABatchOptimizer from ea_config_parser import load_all_ea_configs configs_dir = os.path.join(self.opt_dir.get(), 'configs') if not os.path.exists(configs_dir): messagebox.showwarning('警告', '配置目录不存在') return configs = load_all_ea_configs(configs_dir) if not configs: messagebox.showwarning('警告', '没有找到EA配置') return self.opt_log.insert('end', f"开始批量优化，共 {len(configs)} 个EA\n") self.opt_log.insert('end', "每个EA优化完成后会自动关闭MT5并启动下一个\n") self.opt_log.update() _gui_log(f"开始批量优化，共 {len(configs)} 个EA") optimizer = EABatchOptimizer(self.opt_dir.get(), mt5_data_dir=self.data_dir.get()) reports_dir = os.path.join(self.opt_dir.get(), 'reports') os.makedirs(reports_dir, exist_ok=True) for i, (ea_name, ea_config) in enumerate(configs.items(), 1): self.opt_log.insert('end', f"\n[{i}/{len(configs)}] 优化: {ea_name}\n") self.opt_log.update() try: ini_path = optimizer.generate_ini_for_ea(ea_config, None) abs_ini_path = os.path.abspath(ini_path) self.opt_log.insert('end', f"启动MT5: {ea_name}\n") self.opt_log.update() _gui_log(f"启动MT5: {ea_name}, ini={abs_ini_path}") log_lines: list = [] process = _launch_mt5(mt5_path, abs_ini_path, mt5_data_dir, log_lines=log_lines) for line in log_lines: self.opt_log.insert('end', line + "\n") _gui_log(line) self.opt_log.update() if process is None: self.opt_log.insert('end', "启动失败，跳过该EA\n") continue mt5_install_dir = os.path.dirname(mt5_path) xml_path = _find_latest_optimizer_xml([mt5_data_dir, mt5_install_dir], ea_name) self.opt_log.insert('end', f"等待XML结果，候选目录: {mt5_data_dir} | {mt5_install_dir}\n") self.opt_log.update() try: wait_timeout = int(self.opt_timeout_min.get()) * 60 except ValueError: wait_timeout = 60 * 60 self.opt_log.insert('end', f"等待超时: {wait_timeout // 60} 分钟\n") self.opt_log.update() start_time = time.time() while time.time() - start_time < wait_timeout: xml_path = _find_latest_optimizer_xml([mt5_data_dir, mt5_install_dir], ea_name) if xml_path and os.path.exists(xml_path): time.sleep(2) try: with open(xml_path, 'r', encoding='utf-8') as f: if len(f.read()) > 100: self.opt_log.insert('end', "结果XML已生成\n") self.opt_log.update() break except: pass time.sleep(5) if process.poll() is not None: xml_path = _find_latest_optimizer_xml([mt5_data_dir, mt5_install_dir], ea_name) if xml_path and os.path.exists(xml_path): self.opt_log.insert('end', "MT5已退出，结果文件存在\n") self.opt_log.update() break self.opt_log.insert('end', "MT5已退出，等待结果文件...\n") self.opt_log.update() time.sleep(3) xml_path = _find_latest_optimizer_xml([mt5_data_dir, mt5_install_dir], ea_name) if xml_path and os.path.exists(xml_path): break if process.poll() is None: self.opt_log.insert('end', "正在关闭MT5...\n") self.opt_log.update() try: process.terminate() process.wait(timeout=5) except: pass dst_xml = os.path.join(reports_dir, ea_name, ea_name + '_optimization.xml') if xml_path and os.path.exists(xml_path): os.makedirs(os.path.dirname(dst_xml), exist_ok=True) shutil.copy2(xml_path, dst_xml) self.opt_log.insert('end', f"已保存结果: {dst_xml}\n") else: self.opt_log.insert('end', f"未找到优化 XML，已检查: {mt5_data_dir} 与 {mt5_install_dir}\n") self.opt_log.insert('end', f"{ea_name} 优化完成\n") self.opt_log.update() except Exception as e: self.opt_log.insert('end', f"错误: {str(e)}\n") self.opt_log.insert('end', f"\n所有EA优化完成!\n") self.opt_log.insert('end', "点击'分析结果'查看优化效果\n") def _analyze_opt_results(self): ea_name = self.opt_ea_var.get() sys.path.insert(0, os.path.join(os.getcwd(), 'scripts', 'optimizer')) from ea_batch_optimizer import EABatchOptimizer from analyze_results import ResultAnalyzer optimizer = EABatchOptimizer(self.opt_dir.get(), mt5_data_dir=self.data_dir.get()) mt5_search_dirs = [self.data_dir.get(), os.path.dirname(self.mt5_path.get())] self.opt_log.insert('end', "正在分析结果...\n") self.opt_log.update() results = [] if any(mt5_search_dirs): reports_dir = os.path.join(self.opt_dir.get(), 'reports', ea_name) analyzer = ResultAnalyzer(reports_dir) xml_path = analyzer.copy_xml_results_to_reports(mt5_search_dirs, ea_name, reports_dir) if xml_path: self.opt_log.insert('end', f"从MT5复制结果: {xml_path}\n") xml_results = analyzer.parse_mt5_xml_optimization(xml_path) results.extend(xml_results) self.opt_log.insert('end', f"解析到 {len(xml_results)} 条优化结果\n") if not results: results = optimizer.analyze_results() if not results: self.opt_log.insert('end', "未找到结果文件\n") return self.opt_log.insert('end', f"共 {len(results)} 个结果\n") criterion = self.opt_criterion.get() criterion_key_map = { '余额最大值': 'net_profit', '盈利因子最大值': 'profit_factor', '期望收益': 'expected_payoff', '回撤最小值': 'max_drawdown', '恢复因子': 'recovery_factor', '夏普比率': 'sharpe_ratio', '自定义': 'custom', '复合指标': 'result', } criterion_key = 'profit_factor' for key, val in criterion_key_map.items(): if key in criterion: criterion_key = val break filtered = [] for r in results: trades = r.get('total_trades', 0) pf = r.get('profit_factor', 0) if trades >= 1 and pf > 0: filtered.append(r) if filtered: filtered.sort(key=lambda x: x.get(criterion_key, 0), reverse=True) best = filtered[0] self.opt_log.insert('end', f"\n最优配置 ({criterion}):\n") self.opt_log.insert('end', f" PF: {best.get('profit_factor', 0)}\n") self.opt_log.insert('end', f" 净利润: {best.get('net_profit', 0)}\n") self.opt_log.insert('end', f" 交易数: {best.get('total_trades', 0)}\n") self.opt_log.insert('end', f" 最大回撤: {best.get('max_drawdown', 0)}\n") self.opt_log.insert('end', f" 夏普比率: {best.get('sharpe_ratio', 0)}\n") report_path = os.path.join(self.opt_dir.get(), 'results', f'{ea_name}_report.txt') analyzer.generate_report(filtered[:10], report_path) self.opt_log.insert('end', f"\n报告已保存: {report_path}\n") else: self.opt_log.insert('end', "没有符合条件的优化结果\n") self.opt_log.insert('end', "\n--- Walk-Forward 分析 ---\n") configs_dir = os.path.join(self.opt_dir.get(), 'configs') config_path = os.path.join(configs_dir, ea_name + '.json') if os.path.exists(config_path): with open(config_path, 'r', encoding='utf-8') as f: import json ea_cfg = json.load(f) wf = ea_cfg.get('walk_forward', {}) if wf.get('enabled'): wf_result = analyzer.analyze_walk_forward(mt5_search_dirs, ea_name, reports_dir) is_pf = wf_result.get('in_sample', {}).get('profit_factor', 0) oos_pf = wf_result.get('out_of_sample', {}).get('profit_factor', 0) self.opt_log.insert('end', f"In-Sample PF: {is_pf}\n") self.opt_log.insert('end', f"Out-of-Sample PF: {oos_pf}\n") self.opt_log.insert('end', f"PF衰减: {wf_result.get('pf_decay', 'N/A')}%\n") self.opt_log.insert('end', f"回撤增长: {wf_result.get('dd_increase', 'N/A')}%\n") self.opt_log.insert('end', f"判定: {wf_result.get('verdict', 'Unknown')}\n") else: self.opt_log.insert('end', "未启用Walk-Forward，跳过分析\n") else: self.opt_log.insert('end', "无配置文件，跳过Walk-Forward分析\n") def main(): root = tk.Tk() app = MT5BacktestGUI(root) root.mainloop() if __name__ == "__main__": main()