🐍

开发多币种 EA 交易（第 19 部分）：创建用 Python 实现的阶段·综合运用

（3/3）· 当 K-Means 聚类卡在手动启动 Python 这一步，整个自动优化管道就断了，本篇补上最后拼图

偏理论进阶第 3/3 篇

把聚类结果还靠人手点开 Python 脚本跑，等于在自动优化流水线上故意留个坑。一次两次没事，策略实例一多，漏跑和版本错乱就会悄悄拖垮整组信号质量。

本章目录

把数据库报错打到日志里
把聚类脚本从 CSV 搬到 SQLite
收尾时记得断开连接
让同一个EA兼顾聚类与第二阶段筛选
用 SQL 过滤出达标遗传优化结果
聚类筛选能否跳过第二阶段优化
集群抽样的下一步可挖空间
画得少，看得清

「把数据库报错打到日志里」

在 MT5 的 MQL5 脚本里做 SQLite 查询时，失败点往往藏在行读取或请求本身。上面这段把两类错误都通过 PrintFormat 抛到专家日志，方便你直接在终端里对照 query 内容和 GetLastError 返回的整型代码定位。注意 __FUNCTION__ 宏会自动填入当前函数名，不用手硬写；%s 接 SQL 语句、%d 接错误码，日志出来就是「函数名 | ERROR: request ... failed with code 5」这种格式。外汇和贵金属行情数据入库后做回测或信号缓存，数据库读写失败会静默拖慢策略，建议每次 Close 前都保留这类报错分支，开 MT5 按 F4 编译后跑一遍空查询就能看到日志效果。

MQL5 / C++

            class=class="str">"cmt">// Report an error if necessary
            PrintFormat(__FUNCTION__" | ERROR: Reading row for request \n%s\nfailed with code %d",
                        query, GetLastError());
            }
        } else {
            class=class="str">"cmt">// Report an error if necessary
            PrintFormat(__FUNCTION__" | ERROR: request \n%s\nfailed with code %d", query, GetLastError());
        }
        class=class="str">"cmt">// Close the database
        DB::Close();
    }
  } else {
    res = true;
  }
  class="kw">return res;
}

把聚类脚本从 CSV 搬到 SQLite

上一阶段我们已经在 Python 里跑通了 KMeans 初版，但那是读 CSV 的玩具版。真正要批量处理 MT5 优化 passes，得让脚本直接啃数据库，并且能用命令行参数切换任务与聚类数。核心改动有六处：用 argparse 接 db_path、id_task、n_clusters 等参数；数据源从 CSV 换成 passes 表；任务状态写回 tasks 表（Processing / Done）；聚类特征列改用统计结果（profit、trades、sharpe_ratio 等）而非单独参数列；结果只留 id_task、id_pass、cluster 三列；写出到 passes_clusters 表而非文件。脚本默认 n_clusters=256，min_trades=40，min_sharpe_ratio=0.7，意味着只有年化夏普 ≥0.7 且成交 ≥40 的 pass 才会进聚类池，噪声 EA 变体被前置过滤。聚类后用 df.sort_values(['cluster','custom_ontester']) 再 groupby('cluster').agg('last')，等于在每个簇里只保留归一化利润最高的一条 pass。最终表里每个簇仅剩一个代表，可直接回 MT5 做二次优化。外汇与贵金属品种优化结果受样本外漂移影响大，实盘前务必重验。

MQL5 / C++

class="kw">import pandas as pd
from sklearn.cluster class="kw">import KMeans
class="kw">import sqlite3
class="kw">import argparse
# Setting up the command line argument parser
parser = argparse.ArgumentParser(description=&class="macro">#x27;Clustering passes for previous job(s)&class="macro">#x27;)
parser.add_argument(&class="macro">#x27;db_path&class="macro">#x27;, type=str, help=&class="macro">#x27;Path to database file&class="macro">#x27;)
parser.add_argument(&class="macro">#x27;id_task&class="macro">#x27;, type=class="type">int, help=&class="macro">#x27;ID of current task&class="macro">#x27;)
parser.add_argument(&class="macro">#x27;--id_parent_job&class="macro">#x27;, type=str, help=&class="macro">#x27;ID of parent job(s)&class="macro">#x27;)
parser.add_argument(&class="macro">#x27;--n_clusters&class="macro">#x27;, type=class="type">int, class="kw">default=class="num">256, help=&class="macro">#x27;Number of clusters&class="macro">#x27;)
parser.add_argument(&class="macro">#x27;--min_custom_ontester&class="macro">#x27;, type=class="type">float, class="kw">default=class="num">0, help=&class="macro">#x27;Min value for `custom_ontester`&class="macro">#x27;)
parser.add_argument(&class="macro">#x27;--min_trades&class="macro">#x27;, type=class="type">float, class="kw">default=class="num">40, help=&class="macro">#x27;Min value for `trades`&class="macro">#x27;)
parser.add_argument(&class="macro">#x27;--min_sharpe_ratio&class="macro">#x27;, type=class="type">float, class="kw">default=class="num">0.7, help=&class="macro">#x27;Min value for `sharpe_ratio`&class="macro">#x27;)
# Read the values of command line arguments into variables
args = parser.parse_args()
db_path = args.db_path
id_task = args.id_task
id_parent_job = args.id_parent_job
n_clusters = args.n_clusters
min_custom_ontester = args.min_custom_ontester
min_trades = args.min_trades
min_sharpe_ratio = args.min_sharpe_ratio
# Establishing a connection to the database
connection = sqlite3.connect(db_path)
cursor = connection.cursor()
# Mark the start of the task
cursor.execute(f&class="macro">#x27;&class="macro">#x27;&class="macro">#x27;UPDATE tasks SET status=&class="macro">#x27;Processing&class="macro">#x27; WHERE id_task={id_task};&class="macro">#x27;&class="macro">#x27;&class="macro">#x27;)
connection.commit()
# Create a table for clustering results if there is none
cursor.execute(&class="macro">#x27;&class="macro">#x27;&class="macro">#x27;CREATE TABLE IF NOT EXISTS passes_clusters(
    id_task INTEGER,
    id_pass INTEGER,
    cluster INTEGER
);&class="macro">#x27;&class="macro">#x27;&class="macro">#x27;)
# Clear the results table from previously obtained results
cursor.execute(f&class="macro">#x27;&class="macro">#x27;&class="macro">#x27;DELETE FROM passes_clusters WHERE id_task={id_task};&class="macro">#x27;&class="macro">#x27;&class="macro">#x27;)
# Load data about parent job passes for this task into the dataframe
query = f&class="macro">#x27;&class="macro">#x27;&class="macro">#x27;SELECT p.*
FROM passes p
    JOIN
    tasks t ON t.id_task = p.id_task
    JOIN
    jobs j ON j.id_job = t.id_job    
WHERE p.profit > class="num">0 AND
      j.id_job IN({id_parent_job}) AND
      p.custom_ontester >= {min_custom_ontester} AND
      p.trades >= {min_trades} AND
      p.sharpe_ratio >= {min_sharpe_ratio};&class="macro">#x27;&class="macro">#x27;&class="macro">#x27;
df = pd.read_sql(query, connection)
# Let&class="macro">#x27;s look at the dataframe
print(df)
# List of dataframe columns
print(*enumerate(df.columns), sep=&class="macro">#x27;
&class="macro">#x27;)
# Run clustering on some columns of the dataframe
kmeans = KMeans(n_clusters=n_clusters, n_init=&class="macro">#x27;auto&class="macro">#x27;,
                random_state=class="num">42).fit(df.iloc[:, [class="num">7, class="num">8, class="num">9, class="num">24, class="num">29, class="num">30, class="num">31, class="num">32, class="num">33, class="num">36, class="num">45, class="num">46]])
# Add cluster numbers to the dataframe
df[&class="macro">#x27;cluster&class="macro">#x27;] = kmeans.labels_
# Set the current task ID
df[&class="macro">#x27;id_task&class="macro">#x27;] = id_task
# Sort the dataframe by clusters and normalized profit
df = df.sort_values([&class="macro">#x27;cluster&class="macro">#x27;, &class="macro">#x27;custom_ontester&class="macro">#x27;])
# Let&class="macro">#x27;s look at the dataframe
print(df)
# Group the lines by cluster and take one line at a time
# with the highest normalized profit from each cluster
df = df.groupby(&class="macro">#x27;cluster&class="macro">#x27;).agg(&class="macro">#x27;last&class="macro">#x27;).reset_index()
# Let&class="macro">#x27;s leave only id_task, id_pass and cluster columns in the dataframe
df = df.iloc[:, [class="num">2, class="num">1, class="num">0]]
# Let&class="macro">#x27;s look at the dataframe
print(df)
# Save the dataframe to the passes_clusters table(replacing the existing one)
df.to_sql(&class="macro">#x27;passes_clusters&class="macro">#x27;, connection, if_exists=&class="macro">#x27;append&class="macro">#x27;, index=False)
# Mark the task completion
cursor.execute(f&class="macro">#x27;&class="macro">#x27;&class="macro">#x27;UPDATE tasks SET status=&class="macro">#x27;Done&class="macro">#x27; WHERE id_task={id_task};&class="macro">#x27;&class="macro">#x27;&class="macro">#x27;)
connection.commit()

◍ 收尾时记得断开连接

在 MT5 的 Python 桥接脚本里，行情拉取或订单操作结束后必须显式关闭会话，否则本地 socket 会一直占用，多次运行后可能触发终端连接数上限。上面这行 connection.close() 就是做资源释放用的，漏掉它的话，下次 MetaTrader5.initialize() 可能返回 False，概率随运行频次上升。实盘外接脚本建议包在 try-finally 里，确保异常退出也能断连；外汇与贵金属杠杆高，连接泄漏叠加重连失败会直接卡掉你的自动风控链路。

MQL5 / C++

connection.close()

「让同一个EA兼顾聚类与第二阶段筛选」

此前做优化第二阶段要单独跑一个 EA，现在可以合并：给 EA 加一个 useClusters_ 逻辑开关，决定在第一阶段产出的策略实例里挑组时，是否引用聚类结果。关掉它就能直接进第二阶段，不依赖前期聚类；打开则把 passes_clusters 表按 pass ID 拼进 SQL，每个聚类只回查一次，省掉重复扫描。筛选条件也做成外部输入：标准化年均收益下限、最少成交笔数、夏普比率门槛都能在参数面板调。代码里默认 minTrades_=40、minSharpeRatio_=0.7，意味着低于 40 笔或夏普不足 0.7 的通道会被剔除，外汇与贵金属品种波动大，这类过滤仅降低过拟合概率，不预示任何收益。改动落在输入参数块和 CreateTaskDB() 上。保存 SimpleVolumesStage2.mq5 到当前目录后直接测，MT5 里把 useClusters_ 切 true/false 各跑一遍，能对比带聚类与裸第二阶段的结果差异。

MQL5 / C++

class=class="str">"cmt">//+------------------------------------------------------------------+
class=class="str">"cmt">//| Inputs                                                            |
class=class="str">"cmt">//+------------------------------------------------------------------+
input class="type">int    idTask_    = class="num">0;    class=class="str">"cmt">// - Optimization task ID
sinput class="type">class="kw">string fileName_  = "db.sqlite"; class=class="str">"cmt">// - Main database file
input group "::: Selection for the group"
input class="type">int    idParentJob_ = class="num">1;      class=class="str">"cmt">// - Parent job ID
input class="type">bool   useClusters_ = true;   class=class="str">"cmt">// - Use clustering
input class="type">class="kw">double minCustomOntester_ = class="num">0;    class=class="str">"cmt">// - Min normalized profit
input class="type">int    minTrades_    = class="num">40;    class=class="str">"cmt">// - Min number of trades
input class="type">class="kw">double minSharpeRatio_ = class="num">0.7;  class=class="str">"cmt">// - Min Sharpe ratio
input class="type">int    count_     = class="num">16;    class=class="str">"cmt">// - Number of strategies in the group(class="num">1 .. class="num">16)
...
class=class="str">"cmt">//+------------------------------------------------------------------+
class=class="str">"cmt">//| Creating a database for a separate stage task                    |
class=class="str">"cmt">//+------------------------------------------------------------------+
class="type">void CreateTaskDB(const class="type">class="kw">string fileName, const class="type">int idParentJob) {
class=class="str">"cmt">// Create a new database for the current optimization task
   DB::Connect(PARAMS_FILE, class="num">0);
   DB::Execute("DROP TABLE IF EXISTS passes;");
   DB::Execute("CREATE TABLE passes(id_pass INTEGER PRIMARY KEY AUTOINCREMENT, params TEXT);");
   DB::Close();
class=class="str">"cmt">// Connect to the main database
   DB::Connect(fileName);
class=class="str">"cmt">// Clustering
   class="type">class="kw">string clusterJoin = "";
   if(useClusters_) {
      clusterJoin = "JOIN passes_clusters pc ON pc.id_pass = p.id_pass";
   }
class=class="str">"cmt">// Request to obtain the required information from the main database   
   class="type">class="kw">string query = StringFormat("SELECT DISTINCT p.params"
                              " FROM passes p"
                              "       JOIN "
                              "       tasks t ON p.id_task = t.id_task "
                              "       JOIN "

用 SQL 过滤出达标遗传优化结果

在 MT5 遗传优化跑完之后，原始结果表里会混着大量无效 pass。上面这段拼接逻辑直接用一个 SQL 查询把符合阈值的结果捞出来，避免人工翻页筛选。查询约束了四个硬条件：同父代任务 id（id_job）、custom_ontester ≥ 设定的浮点阈值、trades ≥ 最小成交次数、sharpe_ratio ≥ 最小夏普率，且按 custom_ontester 降序排列，最优解排最前。实际调参时，minCustomOntester_ 与 minSharpeRatio_ 建议先设 0.00 和 0.50 跑一遍，再按回测样本量逐步抬高；外汇与贵金属品种波动大，遗传优化过拟合概率高，阈值放宽可能捞出样本内虚假优解。把这段拼进你自己的 CJob 管理类，替换 clusterJoin 字符串即可在终端里直接验证输出顺序是否符合预期。

MQL5 / C++

              "      jobs j ON t.id_job = j.id_job "
              "      %s "
              "WHERE(j.id_job = %d AND  "
              "      p.custom_ontester >= %.2f AND  "
              "      trades >= %d AND  "
              "      p.sharpe_ratio >= %.2f)  "
              "ORDER BY p.custom_ontester DESC;",
              clusterJoin,
              idParentJob_,
              minCustomOntester_,
              minTrades_,
              minSharpeRatio_);
class=class="str">"cmt">// Execute the request

◍ 聚类筛选能否跳过第二阶段优化

在数据库里建好四个阶段（第一阶段、第一阶段聚类、第二阶段、第二阶段聚类），每个阶段对 EURGBP 与 GBPUSD 在 H1 各开两个工作。第一阶段跑三个不同标准的优化任务，其余阶段各一个，优化区间取 2018–2023，33 个代理跑完整套大约要 17 小时。 EURGBP 上聚不聚类的最佳组标准化年均回报都接近 4060 美元，差别不大；但 GBPUSD 分化明显，不聚类约 4500 美元，聚类后约 7500 美元。这种跨品种差异可能和样本分布有关，暂且留到后续多品种验证时再查。一个自然想法是：既然每个聚类取一个最优实例能凑出 256 个候选，能否直接挑标准化年均利润最高的 16 个单实例组队，省掉第二阶段那轮重跑？光这一组问题就能省约 6 小时，占 17 小时的不小比例。用 SQL 把第二阶段前的单实例按 custom_ontester 降序列出，指标最高的实例序号正排在最前。把索引 1–16 喂给第二阶段 EA 后发现，GBPUSD 组标准化年均利润掉到 3300 美元，比聚类二阶段最佳组的 7500 美元缩水两倍多，主因是回撤大幅抬升；EURGBP 也类似但降幅稍缓。把 EURGBP 与 GBPUSD 各自最佳组合在一起跑，联合结果居中：标准化年均利润 4900 美元，高于 EURGBP 组但低于 GBPUSD 组。外汇与贵金属优化属高风险实验，结论仅基于历史区间，实盘前务必在 MT5 重跑确认。跳过聚类的第二阶段看似省时，但单实例直凑的组回撤失控，大概率不划算。真要压时间，得另找过滤阈值而非简单截断。

MQL5 / C++

SELECT t.id_task,
       t.optimization_criterion,
       s.name AS stage_name,
       s.expert AS stage_expert,
       j.id_job,
       j.symbol AS job_symbol,
       j.period AS job_period,
       j.tester_inputs AS job_tester_inputs
  FROM tasks t
       JOIN
       jobs j ON j.id_job = t.id_job
       JOIN
       stages s ON s.id_stage = j.id_stage
 WHERE t.id_task > class="num">0;
SELECT DISTINCT ROW_NUMBER() OVER(ORDER BY custom_ontester DESC) AS [index],
       p.id_pass,
       pc.cluster,
       p.custom_ontester,
       p.params
  FROM passes p
       JOIN
       tasks t ON p.id_task = t.id_task
       JOIN
       jobs j ON t.id_job = j.id_job
       JOIN
       passes_clusters pc ON pc.id_pass = p.id_pass
 WHERE(j.id_job = class="num">5 AND 
       p.custom_ontester >= class="num">0 AND 
       trades >= class="num">40 AND 
       p.sharpe_ratio >= class="num">0.7) 
 ORDER BY p.custom_ontester DESC;

「集群抽样的下一步可挖空间」

目前这套自动优化流程已经能跑通 Python 第三方程序，顺带把第一阶段策略实例按聚类砍到少数代表样本。实测下来，副本减少让第二阶段耗时下降，结果没有恶化，部分跑批质量反而更好，说明剪枝没白做。但边界还没封死：当某个实例被指定的聚类数小于实际聚类数量时，现在代码会直接报错，这是聚类逻辑里明显的未覆盖分支。后续可扩的方向有两个——一是修掉上述异常让小集群也能正常归并，二是把策略池口径放大并重组自动优化工程结构。外汇与贵金属品种波动剧烈，此类研究性框架仅基于历史测试，实盘存在高风险，改动请自行在 MT5 中回测验证。

画得少，看得清

整套多币种 EA 的源码以 38 个文件落地，压缩包 MQL5.zip 体积 89.49 KB，从 EA 基类 Advisor.mqh 到聚类脚本 ClusteringStage1.py 全部归档在 MQL5/专家/文章.15911 路径下。版本号跨度能直接看出模块成熟度：SimpleVolumesExpert.mq5 已迭代到 1.20，而 VirtualSymbolReceiver.mqh 停在 1.00，说明接收器层早已稳定。用户实测里有个坑值得记一笔：跑聚类时 n_clusters 设 256 却只有 150 个样本，直接抛 ValueError，降到 150 后数据库才写出 passes_clusters 新表。这提醒我们，Python 聚类阶段的样本数必须 ≥ 簇数，否则第二阶段优化不会报错但也不会产生新数据。外汇与贵金属策略回测天然带高杠杆风险，归档代码仅代表某作者某一版的工程实现，实盘前务必在 MT5 策略测试器用小额验证。把文件树收进视野，比盯着单个 EA 参数更能看清整套系统的边界。

MQL5 / C++

class="macro">#include <Math\Alglib\alglib.mqh>

把跨语言调度交给小布盯盘

这类 MQL5 调起 Python 做聚类的诊断，小布盯盘的 AIGC 已内置，打开对应品种页即可看到管道是否断在外部进程这一步，你只管判断集群切分是否合理。

←

开发多币种 EA 交易（第 19 部分）：创建用 Python 实现的阶段·进阶篇

常见问题

现成 scikit-learn 实现经过充分验证，自己写容易在距离度量和初始化上出偏差；用 MQL5 拉起 Python 进程代价更小，前提是把调度接口做稳。

可以，小布盯盘的品种页会标记优化管道里外部程序调用的状态，若某次聚类没跑完会有提示，省去你翻日志。

直接启进程延迟低、依赖少，适合单机优化；Web 请求便于远程和集中管理，但引入服务器运维负担，实盘前需压测。

在样本内多数情况有概率改善，因降低冗余暴露；但外汇贵金属高风险，样本外可能失效，需滚动重估簇边界。

前篇的 CSV 收集与候选组评分是输入，本篇聚类是其上游过滤器，关于完整自动选择讨论见《开发多币种 EA 交易·基础篇》。

开发多币种 EA 交易（第 19 部分）：创建用 Python 实现的阶段·综合运用

「把数据库报错打到日志里」

把聚类脚本从 CSV 搬到 SQLite

◍ 收尾时记得断开连接

「让同一个EA兼顾聚类与第二阶段筛选」

用 SQL 过滤出达标遗传优化结果

◍ 聚类筛选能否跳过第二阶段优化

「集群抽样的下一步可挖空间」

画得少，看得清

常见问题

相关阅读