开发多币种 EA 交易（第 19 部分）：创建用 Python 实现的阶段·进阶篇
（2/3）· 当 K-Means 聚类卡在手动启动 Python 的环节，自动化优化流水线就断了，本篇补齐这道缝
◍ 把优化器拆出去，主文件只留定时器
要让 EA 能顺带拉起 Python 任务，第一步不是急着写调用逻辑，而是先给 Optimization.mq5 瘦身。原文件已经不小，若再把数据库名、任务 ID、解释器路径一股脑塞进去，后续维护会非常难受。做法是把执行主体挪进一个新类 COptimizer，主文件只负责建定时器、在 OnInit/OnDeinit 里管对象生命周期。 具体改动里，两个输入参数很关键：fileName_ 指向主库 sqlite 文件（示例名 database911.sqlite），pythonPath_ 写死本机解释器路径，示例给的是 C:\Python\Python312\python.exe。构造函数只收这一个路径参数，后期若想自动探测解释器可再扩，但现在先用最简方案跑通。 定时器设了 20 秒间隔，OnTimer 里只调 optimizer.Process()，所有脏活都在类里。这样你开 MT5 挂 EA 时，只要 pythonPath_ 填对，就能在不变动调度框架的前提下，把 .py 任务接进原有 stages 表。外汇与贵金属品种波动剧烈，自动化优化只是辅助，参数错配可能放大回测偏差，实盘前务必小样本验证。
sinput class="type">class="kw">string fileName_ = "database911.sqlite"; class=class="str">"cmt">// - File with the main database sinput class="type">class="kw">string pythonPath_ = "C:\Python\Python312\python.exe"; class=class="str">"cmt">// - Path to Python interpreter COptimizer *optimizer; class=class="str">"cmt">// Pointer to the optimizer object class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Expert initialization function | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">int OnInit() { class=class="str">"cmt">// Connect to the main database DB::Test(fileName_); class=class="str">"cmt">// Create an optimizer optimizer = new COptimizer(pythonPath_); class=class="str">"cmt">// Create the timer and start its handler EventSetTimer(class="num">20); OnTimer(); class="kw">return(INIT_SUCCEEDED); } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Expert timer function | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">void OnTimer() { class=class="str">"cmt">// Start the optimizer handling optimizer.Process(); } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Expert deinitialization function | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">void OnDeinit(const class="type">int reason) { EventKillTimer(); class=class="str">"cmt">// Remove the optimizer if(!!optimizer) { class="kw">delete optimizer; } }
「把优化调度抽成独立类」
做自动优化管理器时，核心思路是把调度逻辑和具体任务执行拆开。COptimizer 只暴露构造函数和 Process() 主处理方法，私有侧保留 TotalTasks() 查队列任务数、GetNextTaskId() 取下一个任务 ID，真正的跑测工作下放到 COptimizerTask 对象里。 从旧版 Optimization.mq5 的 OnTimer() 搬过来的 Process() 逻辑，因为引入了任务类反而更干净：这一层不再关心下次是跑 MT5 测试器优化还是调 Python 脚本，流程统一成“队列有活就取参数、启动、等完成、再取下一个”。 看 Process() 的实际代码，它会先打印当前任务 ID，遇到 IsStopped() 就杀定时器并卸 EA；任务跑完且 ID 非空就 Finish()，然后统计 Processing+Queued 状态总数，有剩就 Load+Start 下一个，没剩就 ExpertRemove() 退场。 你在 MT5 里建 COptimizer.mqh 后，直接把下面代码贴进去，编译完用 Comment() 输出的“Total tasks in queue”数字就能实时盯队列余量，判断调度是否卡死。
class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Class for the project auto optimization manager | class=class="str">"cmt">//+------------------------------------------------------------------+ class COptimizer { class=class="str">"cmt">// Current optimization task COptimizerTask m_task; class=class="str">"cmt">// Get the number of tasks with a given status in the queue class="type">int TotalTasks(class="type">class="kw">string status = "Queued"); class=class="str">"cmt">// Get the ID of the next optimization task from the queue class="type">class="kw">ulong GetNextTaskId(); class="kw">public: COptimizer(class="type">class="kw">string p_pythonPath = NULL); class=class="str">"cmt">// Constructor class="type">void Process(); class=class="str">"cmt">// Main processing method }; class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Main handling method | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">void COptimizer::Process() { PrintFormat(__FUNCTION__" | Current Task ID = %d", m_task.Id()); class=class="str">"cmt">// If the EA is stopped, remove the timer and the EA itself from the chart if (IsStopped()) { EventKillTimer(); ExpertRemove(); class="kw">return; } class=class="str">"cmt">// If the current task is completed, if (m_task.IsDone()) { class=class="str">"cmt">// If the current task is not empty, if(m_task.Id()) { class=class="str">"cmt">// Complete the current task m_task.Finish(); } class=class="str">"cmt">// Get the number of tasks in the queue class="type">int totalTasks = TotalTasks("Processing") + TotalTasks("Queued"); class=class="str">"cmt">// If there are tasks, then if(totalTasks) { class=class="str">"cmt">// Get the ID of the next current task class="type">class="kw">ulong taskId = GetNextTaskId(); class=class="str">"cmt">// Load the optimization task parameters from the database m_task.Load(taskId); class=class="str">"cmt">// Launch the current task m_task.Start(); class=class="str">"cmt">// Display the number of remaining tasks and the current task on the chart Comment(StringFormat( "Total tasks in queue: %d\n" "Current Task ID: %d", totalTasks, m_task.Id())); } else { class=class="str">"cmt">// If there are no tasks, remove the EA from the chart PrintFormat(__FUNCTION__" | Finish.", class="num">0); ExpertRemove(); } } }
让任务类自己认得该跑 EA 还是 Python
COptimizerTask 是整个调度器里最该盯紧的一环：它直接拉起 Python 解释器，把写好的 py 程序喂进去执行；遇到 EX5 任务则走 MT5 策略测试器。类头部先 import 了 shell32.dll 的 ShellExecuteW，这是后续跨进程启动 Python 的底层入口。 任务参数分两层存：m_params 结构从数据库 Load() 直接灌入，包含 expert、symbol、period、optimization_criterion 等 11 个字段；任务类型则由 ParseType() 查文件名后缀判定，TASK_TYPE_EX5 与 TASK_TYPE_PY 二选一，未知归为 TASK_TYPE_UNKNOWN。 Parse() 负责拼初始化串——EA 优化走 tester_inputs 参数拼接，Python 任务则拼成解释器命令行。Start() 再按 m_type 分流：测试器优化或 ShellExecuteW 拉起 py。 状态检查也分两条路：EA 看测试器是否停止，py 则回数据库按当前 ID 查任务状态。把这套改完存回 COptimizerTask.mqh，开 MT5 挂上自己库就能验证分流是否生效。
class=class="str">"cmt">// Function to launch an executable file in the operating system class="macro">#class="kw">import "shell32.dll" class="type">int ShellExecuteW(class="type">int hwnd, class="type">class="kw">string lpOperation, class="type">class="kw">string lpFile, class="type">class="kw">string lpParameters, class="type">class="kw">string lpDirectory, class="type">int nShowCmd); class="macro">#class="kw">import class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Optimization task class | class=class="str">"cmt">//+------------------------------------------------------------------+ class COptimizerTask { enum { TASK_TYPE_UNKNOWN, TASK_TYPE_EX5, TASK_TYPE_PY } m_type; class=class="str">"cmt">// Task type(MQL5 or Python) class="type">class="kw">ulong m_id; class=class="str">"cmt">// Task ID class="type">class="kw">string m_setting; class=class="str">"cmt">// String for initializing the EA parameters for the current task class="type">class="kw">string m_pythonPath; class=class="str">"cmt">// Full path to the Python interpreter class=class="str">"cmt">// Data structure for reading a single class="type">class="kw">string of a query result class="kw">struct params { class="type">class="kw">string expert; class="type">int optimization; class="type">class="kw">string from_date; class="type">class="kw">string to_date; class="type">int forward_mode; class="type">class="kw">string forward_date; class="type">class="kw">string symbol; class="type">class="kw">string period; class="type">class="kw">string tester_inputs; class="type">class="kw">ulong id_task; class="type">int optimization_criterion; } m_params; class=class="str">"cmt">// Get the full or relative path to a given file in the current folder class="type">class="kw">string GetProgramPath(class="type">class="kw">string name, class="type">bool rel = true); class=class="str">"cmt">// Get initialization class="type">class="kw">string from task parameters class="type">void Parse(); class=class="str">"cmt">// Get task type from task parameters class="type">void ParseType(); class="kw">public: class=class="str">"cmt">// Constructor COptimizerTask() : m_id(class="num">0) {} class=class="str">"cmt">// Task ID class="type">class="kw">ulong Id() { class="kw">return m_id; } class=class="str">"cmt">// Set the full path to the Python interpreter class="type">void PythonPath(class="type">class="kw">string p_pythonPath) { m_pythonPath = p_pythonPath; }
◍ 从文件后缀拆出任务类型的实现细节
在 MT5 里做分布式优化调度，第一步是判断拿到的任务到底是跑 EA 还是跑 Python 脚本。下面这段类方法直接从任务参数里的 expert 字段尾巴取 3 个字符，比对 .py 和 ex5 来赋值任务类型，未知后缀统一归到 TASK_TYPE_UNKNOWN。
代码里 StringSubstr(m_params.expert, StringLen(m_params.expert)-3) 这个 -3 是硬锚点：它假设传入的 expert 字符串至少带 3 字符后缀。若数据库里填了裸名没带扩展名，截断会吃到正常字符，类型判定会直接掉进 UNKNOWN 分支，任务不会被启动。
真正生成测试配置的是 Parse() 里的 StringFormat 拼装。注意它写死了 Deposit=10000、Leverage=200、Currency=USD 和 Model=1（即每笔 tick 按收盘价撮合）。如果你实盘账户杠杆不是 200 或入金不是 1 万美金，回测的保证金占用曲线会和真实环境偏离，优化出来的参数过拟合概率会上升。
外汇与贵金属杠杆交易本身高风险，用这套固定参数做历史优化时，建议先把 Deposit / Leverage 改成和你账户一致再跑，否则前向测试（ForwardMode 那段）的结论参考价值有限。
class="type">void COptimizerTask::ParseType() { class="type">class="kw">string ext = StringSubstr(m_params.expert, StringLen(m_params.expert) - class="num">3); if(ext == ".py") { m_type = TASK_TYPE_PY; } else if (ext == "ex5") { m_type = TASK_TYPE_EX5; } else { m_type = TASK_TYPE_UNKNOWN; } } class="type">void COptimizerTask::Parse() { ParseType(); if(m_type == TASK_TYPE_EX5) { m_setting = StringFormat( "[Tester]\r\n" "Expert=%s\r\n" "Symbol=%s\r\n" "Period=%s\r\n" "Optimization=%d\r\n" "Model=class="num">1\r\n" "FromDate=%s\r\n" "ToDate=%s\r\n" "ForwardMode=%d\r\n" "ForwardDate=%s\r\n" "Deposit=class="num">10000\r\n" "Currency=USD\r\n" "ProfitInPips=class="num">0\r\n" "Leverage=class="num">200\r\n" "ExecutionMode=class="num">0\r\n" "OptimizationCriterion=%d\r\n"
「任务字符串拼装与启动分支」
分布式优化框架里，每类任务在派发前都要先拼出一段可被终端或解释器直接消费的启动参数字符串。下面这段逻辑区分了 EA 回测任务与 Python 任务两种形态，拼错一个字段就会导致 MT5 测试器静默拒跑。 EA 任务走 StringFormat 把品种、周期、优化起止、前向模式、优化准则、任务 ID、数据库文件名以及 tester_inputs 全部塞进模板。注意 idTask_=%d 与 fileName_=%s 是框架自定义的键值，不是 MT5 原生 ini 字段，需要在被调 EA 侧自行解析。 Python 任务则简单得多：解释器路径 + 数据库路径 + 任务 ID（%I64u 无符号 64 位）+ 启动参数，四段用空格隔开。GetProgramPath 第二个参数传 false 表示不按 EX5 规则拼路径，DB::FileName(true) 才返回带库的绝对路径。 Start() 里先 PrintFormat 把整串打进日志，EA 分支再顺序调用 MTTESTER::CloseNotChart、SetSettings2、ClickStart 触发测试器，并立刻 DB::Connect 回写状态。外汇与贵金属品种在此类批量优化中杠杆与滑点敏感，回测结果向外推演时须清醒其高风险的真实落地概率。
"[TesterInputs]\r\n"
"idTask_=%d\r\n"
"fileName_=%s\r\n"
"%s\r\n",
GetProgramPath(m_params.expert),
m_params.symbol,
m_params.period,
m_params.optimization,
m_params.from_date,
m_params.to_date,
m_params.forward_mode,
m_params.forward_date,
m_params.optimization_criterion,
m_params.id_task,
DB::FileName(),
m_params.tester_inputs
);
class=class="str">"cmt">// If this is a task to launch a Python program
} else if (m_type == TASK_TYPE_PY) {
class=class="str">"cmt">// Form a program launch class="type">class="kw">string on Python with parameters
m_setting = StringFormat("\"%s\" \"%s\" %I64u %s",
GetProgramPath(m_params.expert, false), class=class="str">"cmt">// Python program file
DB::FileName(true), class=class="str">"cmt">// Path to the database file
m_id, class=class="str">"cmt">// Task ID
m_params.tester_inputs class=class="str">"cmt">// Launch parameters
);
}
class=class="str">"cmt">//+------------------------------------------------------------------+
class=class="str">"cmt">//| Start task |
class=class="str">"cmt">//+------------------------------------------------------------------+
class="type">void COptimizerTask::Start() {
PrintFormat(__FUNCTION__" | Task ID = %d\n%s", m_id, m_setting);
class=class="str">"cmt">// If this is the EA optimization task
if(m_type == TASK_TYPE_EX5) {
class=class="str">"cmt">// Launch a new optimization task in the tester
MTTESTER::CloseNotChart();
MTTESTER::SetSettings2(m_setting);
MTTESTER::ClickStart();
class=class="str">"cmt">// Update the task status in the database
DB::Connect();任务状态回写与完成判定的落地细节
在 MT5 里把分布式优化任务交给外部进程后，第一步得把数据库里的任务标记成 Processing，否则调度器会重复派发。下面这段把 id_task 对应的行状态改写，并立刻关闭连接，避免长连接占住 SQLite 句柄。 string query = StringFormat( "UPDATE tasks SET " " status='Processing' " " WHERE id_task=%d", m_id); DB::Execute(query); DB::Close(); // If this is a task to launch a Python program } else if (m_type == TASK_TYPE_PY) { PrintFormat(__FUNCTION__" | SHELL EXEC: %s", m_pythonPath); // Call the system function to launch the program with parameters ShellExecuteW(NULL, NULL, m_pythonPath, m_setting, NULL, 1); } } //+------------------------------------------------------------------+
| // | Task completed? |
|---|
//+------------------------------------------------------------------+ bool COptimizerTask::IsDone() { // If there is no current task, then everything is done if(m_id == 0) { return true; } // Result bool res = false; // If this is the EA optimization task if(m_type == TASK_TYPE_EX5) { // Check if the strategy tester has finished its work res = MTTESTER::IsReady(); // If this is a task to run a Python program, then } else if(m_type == TASK_TYPE_PY) { // Request to get the status of the current task string query = StringFormat("SELECT status " " FROM tasks" " WHERE id_task=%I64u;", m_id); // Open the database if(DB::Connect()) { // Execute the request int request = DatabasePrepare(DB::Id(), query); // If there is no error if(request != INVALID_HANDLE) { // Data structure for reading a single string of a query result struct Row { string status; } row; // Read data from the first result string if(DatabaseReadBind(request, row)) { // Check if the status is Done res = (row.status == "Done"); } else { 判定完成的逻辑分两条线：EA 优化任务直接问 MTTESTER::IsReady()，而 Python 子任务必须回数据库轮询 status 字段。实测中若把 %d 误写成 %I64u 或反过来，StringFormat 在 m_id 为 uint64 时会输出乱码，导致 WHERE 命中 0 行，任务永远卡在 Processing。 DatabaseReadBind 只绑了单行结构体 Row，意味着一旦任务表同 id_task 出现重复行，只读第一条，可能在并发写入时漏掉最新状态。开 MT5 跑这套时，建议先在 tasks 表给 id_task 加唯一索引，再用 PrintFormat 把拼接后的 query 打到日志确认。
class="type">class="kw">string query = StringFormat( "UPDATE tasks SET " " status=&class="macro">#x27;Processing&class="macro">#x27; " " WHERE id_task=%d", m_id); DB::Execute(query); DB::Close(); class=class="str">"cmt">// If this is a task to launch a Python program } else if (m_type == TASK_TYPE_PY) { PrintFormat(__FUNCTION__" | SHELL EXEC: %s", m_pythonPath); class=class="str">"cmt">// Call the system function to launch the program with parameters ShellExecuteW(NULL, NULL, m_pythonPath, m_setting, NULL, class="num">1); } } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Task completed? | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">bool COptimizerTask::IsDone() { class=class="str">"cmt">// If there is no current task, then everything is done if(m_id == class="num">0) { class="kw">return true; } class=class="str">"cmt">// Result class="type">bool res = false; class=class="str">"cmt">// If this is the EA optimization task if(m_type == TASK_TYPE_EX5) { class=class="str">"cmt">// Check if the strategy tester has finished its work res = MTTESTER::IsReady(); class=class="str">"cmt">// If this is a task to run a Python program, then } else if(m_type == TASK_TYPE_PY) { class=class="str">"cmt">// Request to get the status of the current task class="type">class="kw">string query = StringFormat("SELECT status " " FROM tasks" " WHERE id_task=%I64u;", m_id); class=class="str">"cmt">// Open the database if(DB::Connect()) { class=class="str">"cmt">// Execute the request class="type">int request = DatabasePrepare(DB::Id(), query); class=class="str">"cmt">// If there is no error if(request != INVALID_HANDLE) { class=class="str">"cmt">// Data structure for reading a single class="type">class="kw">string of a query result class="kw">struct Row { class="type">class="kw">string status; } row; class=class="str">"cmt">// Read data from the first result class="type">class="kw">string if(DatabaseReadBind(request, row)) { class=class="str">"cmt">// Check if the status is Done res = (row.status == "Done"); } else {