2462 lines
110 KiB
Python
2462 lines
110 KiB
Python
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("<MouseWheel>", 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('<FocusOut>', 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('<FocusOut>', 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('<FocusOut>', 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('<<ComboboxSelected>>', 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('<<ComboboxSelected>>', 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('<<ComboboxSelected>>', 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("<Double-1>", self._on_row_double_click)
|
||
self.report_tree.bind("<Button-3>", 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('<<ComboboxSelected>>', 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('<Double-1>', 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('<<ComboboxSelected>>', lambda e: self._update_resolved_symbol_hint())
|
||
self.opt_symbol.bind('<FocusOut>', 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('<<ComboboxSelected>>', 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('<Return>', lambda e: on_validate())
|
||
entry.bind('<Escape>', 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()
|