开发交易机器人：Python与MQL5结合（第二部分）：模型选择、创建与训练，以及Python自定义测试器·进阶篇
回测里多空平仓的盈亏切分逻辑
这段回测脚本把预测标签映射到具体下单动作：标签为 1 时开多，标签为 0 时开空，平仓点统一取入场后第 48 根 K 线的收盘价。以多单为例，当 exit_price 高于 entry_price 减去 markup 才计为盈利，否则按亏损处理，盈利额用 (exit_price - entry_price - markup) / point_cost 折算成点数单位。 空单镜像对称：entry_price 取自 X_test 的 close，exit_price 取 i+48 的 close，仅当 exit_price < entry_price - markup 时落袋利润。注意 markup 在代码里写死为 0.00001，相当于 EURUSD 这类品种 0.1 点差成本，回测未动态加载真实点值。 初始余额设 10000.0 美元，每笔成交 trades 加 1，profit 序列追加正负值，最后 total_profit = balance - initial_balance 得出累计盈亏。绘图部分用 plt.axvline 标出 FORWARD 分割日期，x 轴按总长度十分之一步长显示 %Y-%m-%d，方便肉眼核对样本外区间。 开 MT5 把 markup 改成你经纪商黄金或欧美实时点差，重跑这段逻辑，看 FORWARD 竖线右侧曲线是否还向上——外汇与贵金属杠杆高，样本外崩坏概率不低。
profit = (exit_price - entry_price - markup) / point_cost balance += profit trades += class="num">1 profits.append(profit) else: # Close the class="type">long position with loss loss = (entry_price - exit_price + markup) / point_cost balance -= loss trades += class="num">1 profits.append(-loss) elif predicted_labels[i] == class="num">0: # Open a class="type">class="kw">short position entry_price = X_test.iloc[i][&class="macro">#x27;close&class="macro">#x27;] exit_price = X_test.iloc[i+class="num">48][&class="macro">#x27;close&class="macro">#x27;] if exit_price < entry_price - markup: # Close the class="type">class="kw">short position with profit profit = (entry_price - exit_price - markup) / point_cost balance += profit trades += class="num">1 profits.append(profit) else: # Close the class="type">class="kw">short position with loss loss = (exit_price - entry_price + markup) / point_cost balance -= loss trades += class="num">1 profits.append(-loss) # Calculate the cumulative profit or loss total_profit = balance - initial_balance # Plot the balance change over the number of trades plt.plot(range(trades), [balance + sum(profits[:i]) for i in range(trades)]) plt.title(&class="macro">#x27;Balance Change&class="macro">#x27;) plt.xlabel(&class="macro">#x27;Trades&class="macro">#x27;) plt.ylabel(&class="macro">#x27;Balance($)&class="macro">#x27;) plt.xticks(range(class="num">0, len(X_test), class="type">int(len(X_test)/class="num">10)), X_test.index[::class="type">int(len(X_test)/class="num">10)].strftime(&class="macro">#x27;%Y-%m-%d&class="macro">#x27;)) # Add dates to the x-axis plt.axvline(FORWARD, class="type">color=&class="macro">#x27;r&class="macro">#x27;, linestyle=&class="macro">#x27;--&class="macro">#x27;) # Add a vertical line for the FORWARD date plt.show() # Print the results print(f"Cumulative profit or loss: {total_profit:.2f} $") print(f"Number of trades: {trades}") # Get test data test_data = raw_data[raw_data.index >= FORWARD] X_test = test_data.drop([&class="macro">#x27;label&class="macro">#x27;, &class="macro">#x27;labels&class="macro">#x27;], axis=class="num">1) y_test = test_data[&class="macro">#x27;labels&class="macro">#x27;] # Test the model with markup and target labels initial_balance = class="num">10000.0 markup = class="num">0.00001 test_model(xgb_clf, X_test, y_test, markup, initial_balance)
「用5折交叉验证给模型质量打分」
做价格方向的机器学习分类，最怕模型在训练集上漂亮、放到新行情就崩。把交叉验证嵌进训练流程，能在多个数据子集上反复测，过拟合的苗头更容易被揪出来。 这里用 sklearn 的 cross_val_score 跑 5 折，对 XGBoost 做评估。注意一个分水岭：交叉验证只负责打分，不参加训练；真正 fit 模型时，仍用 FORWARD 日期前的全量样本，且不走 CV。 回测区间取 1990–2024 年，2010 年之后的样本外测试准确率约 56%。相比不交叉验证的初版，这一步让模型对新价格数据表现出可见的鲁棒性提升，但外汇与贵金属属高风险品种，56% 仅是历史样本概率，实盘仍可能漂移。 下面这段函数改完就能在 MT5 导出的特征 CSV 上直接跑。cross_val_score 那行输出平均准确率，clf.fit 才是落地的训练权重。
def train_xgboost_classifier(data, num_boost_rounds=class="num">500): # 检查数据不为空，空则抛异常 if data.empty: raise ValueError("Data should not be empty") # 检查必需列 label 和 labels 是否存在 required_columns = [&class="macro">#x27;label&class="macro">#x27;, &class="macro">#x27;labels&class="macro">#x27;] if not all(column in data.columns for column in required_columns): raise ValueError(f"Data is missing required columns: {required_columns}") # 丢弃 label 列，因为它不是特征 X = data.drop([&class="macro">#x27;label&class="macro">#x27;, &class="macro">#x27;labels&class="macro">#x27;], axis=class="num">1) y = data[&class="macro">#x27;labels&class="macro">#x27;] # 检查所有特征是否为数值类型 if not all(pd.api.types.is_numeric_dtype(X[column]) for column in X.columns): raise ValueError("All features should have numeric data type") # 创建 XGBoost 二分类模型 clf = xgb.XGBClassifier(objective=&class="macro">#x27;binary:logistic&class="macro">#x27;, max_depth=class="num">10, learning_rate=class="num">0.3, n_estimators=num_boost_rounds, random_state=class="num">1) # 用 class="num">5 折交叉验证评估模型 scores = cross_val_score(clf, X, y, cv=class="num">5) # 计算平均预测准确率 accuracy = scores.mean() print(f"Mean prediction accuracy on cross-validation: {accuracy:.2f}") # 不用交叉验证，在全量数据上训练 clf.fit(X, y) # 返回训练好的模型 class="kw">return clf labeled_data_engineered = feature_engineering(labeled_data_clustered, n_features_to_select=class="num">20) # 获取全部数据 raw_data = labeled_data_engineered # 训练数据取 FORWARD 之前 train_data = raw_data[raw_data.index <= FORWARD] # 用过滤后的数据训练 XGBoost xgb_clf = train_xgboost_classifier(train_data, num_boost_rounds=class="num">100) # 测试数据取 FORWARD 之后 test_data = raw_data[raw_data.index >= FORWARD] X_test = test_data.drop([&class="macro">#x27;label&class="macro">#x27;, &class="macro">#x27;labels&class="macro">#x27;], axis=class="num">1) y_test = test_data[&class="macro">#x27;labels&class="macro">#x27;] predicted_labels = xgb_clf.predict(X_test) # 计算预测准确率 accuracy = (predicted_labels == y_test).mean() print(f"Accuracy: {accuracy:.2f}")
◍ 网格搜索给XGBoost挑参数
把机器学习模型调参类比MT5里给EA做参数优化就很直观：你不是在穷举交易开仓条件，而是在穷举模型超参数。用Scikit-learn的GridSearchCV，可以对网格里每一组参数做交叉验证，挑出平均准确率最高那一组，比人工拍脑袋靠谱。 下面这段代码先定义超参数网格，再挂上XGBoost分类器跑网格搜索。max_depth、learning_rate、n_estimators各取若干离散值，cv=5表示五折交叉验证，scoring用accuracy，最后打印最优参数与交叉验证均分。 from sklearn.model_selection import GridSearchCV # 定义超参数网格 param_grid = { 'max_depth': [3, 5, 7, 10], 'learning_rate': [0.1, 0.3, 0.5], 'n_estimators': [100, 500, 1000] } # 创建XGBoost模型 clf = xgb.XGBClassifier(objective='binary:logistic', random_state=1) # 网格搜索最优超参数 grid_search = GridSearchCV(clf, param_grid, cv=5, scoring='accuracy') grid_search.fit(X_train, y_train) # 打印最优超参数 print("Best hyperparameters:", grid_search.best_params_) # 打印最优参数下交叉验证平均准确率 print("Mean prediction accuracy on cross-validation:", grid_search.best_score_) 逐行拆解：第1行引入网格搜索工具；param_grid里max_depth控制单棵树深度，learning_rate是每棵树的步长，n_estimators是树的数量；cv=5把训练集切五份轮着验证，降低过拟合误判；best_params_给出胜出组合，best_score_是该组合的平均准度。 后面封装的train_xgboost_classifier函数做了数据非空、列名、数值型校验，再跑一遍更宽的网格（learning_rate细化到0.05~0.5，n_estimators上限拉到2000），返回best_estimator_。 实盘外接模型若用装袋+提升的XGBoost集成，在风险收益比1:8的标的上分类准确率测到过73%，较早期单模型版本有明显抬升。外汇与贵金属波动大、杠杆高，这类信号只作概率参考，别直接当下单依据。
from sklearn.model_selection class="kw">import GridSearchCV # Define the grid of hyperparameters param_grid = { &class="macro">#x27;max_depth&class="macro">#x27;: [class="num">3, class="num">5, class="num">7, class="num">10], &class="macro">#x27;learning_rate&class="macro">#x27;: [class="num">0.1, class="num">0.3, class="num">0.5], &class="macro">#x27;n_estimators&class="macro">#x27;: [class="num">100, class="num">500, class="num">1000] } # Create XGBoost model clf = xgb.XGBClassifier(objective=&class="macro">#x27;binary:logistic&class="macro">#x27;, random_state=class="num">1) # Perform grid search to find the best hyperparameters grid_search = GridSearchCV(clf, param_grid, cv=class="num">5, scoring=&class="macro">#x27;accuracy&class="macro">#x27;) grid_search.fit(X_train, y_train) # Print the best hyperparameters print("Best hyperparameters:", grid_search.best_params_) # Print the mean accuracy of the predictions on cross-validation for the best hyperparameters print("Mean prediction accuracy on cross-validation:", grid_search.best_score_) def train_xgboost_classifier(data, num_boost_rounds=class="num">1000): # Check that data is not empty if data.empty: raise ValueError("Data should not be empty") # Check that all required columns are present in the data required_columns = [&class="macro">#x27;label&class="macro">#x27;, &class="macro">#x27;labels&class="macro">#x27;] if not all(column in data.columns for column in required_columns): raise ValueError(f"Data is missing required columns: {required_columns}") # Drop the &class="macro">#x27;label&class="macro">#x27; column since it is not a feature X = data.drop([&class="macro">#x27;label&class="macro">#x27;, &class="macro">#x27;labels&class="macro">#x27;], axis=class="num">1) y = data[&class="macro">#x27;labels&class="macro">#x27;] # Check that all features have numeric data type if not all(pd.api.types.is_numeric_dtype(X[column]) for column in X.columns): raise ValueError("All features should have numeric data type") # Create XGBoostClassifier model clf = xgb.XGBClassifier(objective=&class="macro">#x27;binary:logistic&class="macro">#x27;, random_state=class="num">1) # Define hyperparameters for grid search param_grid = { &class="macro">#x27;max_depth&class="macro">#x27;: [class="num">3, class="num">5, class="num">7, class="num">10], &class="macro">#x27;learning_rate&class="macro">#x27;: [class="num">0.05, class="num">0.1, class="num">0.2, class="num">0.3, class="num">0.5], &class="macro">#x27;n_estimators&class="macro">#x27;: [class="num">50, class="num">100, class="num">600, class="num">1200, class="num">2000] } # Train the model on the data using cross-validation and grid search grid_search = GridSearchCV(clf, param_grid, cv=class="num">5, scoring=&class="macro">#x27;accuracy&class="macro">#x27;) grid_search.fit(X, y) # Calculate the mean accuracy of the predictions accuracy = grid_search.best_score_ print(f"Mean prediction accuracy on cross-validation: {accuracy:.2f}") # Return the trained model with the best hyperparameters class="kw">return grid_search.best_estimator_
XGBoost训练前的特征与标签校验
这段 Python 片段在做一件事：把聚类后的行情特征表喂给 XGBoost 之前，先卡死数据质量。空表直接抛异常，缺少 'label' 或 'labels' 列也立刻中断，避免后续模型在脏数据上静默出错。 特征矩阵 X 会丢掉 'label' 和 'labels' 两列，后者作为 y 单独抽出；同时逐列检查是否为数值型，非数值直接 ValueError。外汇与贵金属行情特征若混进字符串（如未编码的 session 名），在这里就会被拦下，这类品种波动受宏观事件驱动，数据口径不一致会让模型倾向过拟合。 模型用的是 binary:logistic 目标的 XGBClassifier，random_state=1 固定随机种子。网格搜索覆盖 max_depth[3,7,12]、learning_rate[0.1,0.3,0.5]、n_estimators[100,600,1200] 共 27 组组合，5 折交叉验证以 accuracy 选优。 实盘前你可在 MT5 导出的 CSV 上跑这段，把 FORWARD / EXAMWARD 换成你自己的样本切分点；交叉验证得分 best_score_ 若低于 0.55，说明当前 20 个特征对涨跌二分类区分度偏弱，可能要考虑换特征或缩短样本周期。
raise ValueError("Data should not be empty") # Check that all required columns are present in the data required_columns = [&class="macro">#x27;label&class="macro">#x27;, &class="macro">#x27;labels&class="macro">#x27;] if not all(column in data.columns for column in required_columns): raise ValueError(f"Missing required columns in data: {required_columns}") # Remove the &class="macro">#x27;label&class="macro">#x27; column as it is not a feature X = data.drop([&class="macro">#x27;label&class="macro">#x27;, &class="macro">#x27;labels&class="macro">#x27;], axis=class="num">1) y = data[&class="macro">#x27;labels&class="macro">#x27;] # Check that all features have numeric data type if not all(pd.api.types.is_numeric_dtype(X[column]) for column in X.columns): raise ValueError("All features should have numeric data type") # Create XGBoostClassifier model clf = xgb.XGBClassifier(objective=&class="macro">#x27;binary:logistic&class="macro">#x27;, random_state=class="num">1) # Define hyperparameters for grid search param_grid = { &class="macro">#x27;max_depth&class="macro">#x27;: [class="num">3, class="num">7, class="num">12], &class="macro">#x27;learning_rate&class="macro">#x27;: [class="num">0.1, class="num">0.3, class="num">0.5], &class="macro">#x27;n_estimators&class="macro">#x27;: [class="num">100, class="num">600, class="num">1200] } # Train the model on the data using cross-validation and hyperparameter tuning grid_search = GridSearchCV(clf, param_grid, cv=class="num">5, scoring=&class="macro">#x27;accuracy&class="macro">#x27;) grid_search.fit(X, y) # Calculate the mean accuracy of the predictions accuracy = grid_search.best_score_ print(f"Mean accuracy on cross-validation: {accuracy:.2f}") # Return the trained model class="kw">return grid_search.best_estimator_ labeled_data_engineered = feature_engineering(labeled_data_clustered, n_features_to_select=class="num">20) # Get all data raw_data = labeled_data_engineered # Train data train_data = raw_data[raw_data.index <= FORWARD] # Test data test_data = raw_data[raw_data.index <= EXAMWARD] # Train the XGBoost model on the filtered data xgb_clf = train_xgboost_classifier(train_data, num_boost_rounds=class="num">100) # Test the model on all data test_data = raw_data[raw_data.index >= FORWARD] X_test = test_data.drop([&class="macro">#x27;label&class="macro">#x27;, &class="macro">#x27;labels&class="macro">#x27;], axis=class="num">1) y_test = test_data[&class="macro">#x27;labels&class="macro">#x27;] predicted_labels = xgb_clf.predict(X_test) # Calculate the accuracy of the predictions accuracy = (predicted_labels == y_test).mean() print(f"Prediction accuracy: {accuracy:.2f}")
「用 EXAMWARD 之后的数据压模型底牌」
把 EXAMWARD 日期之后的行情单独拎出来跑模型，是最直接的压力测试——这部分数据既没进训练集也没进测试集，等于让模型面对一段它从没见过的未来。只有在这种完全隔离的样本上还能站着赚钱，才算摸到了实盘门槛。 我的切分很硬：2000–2010 年训练，2010–2019 年测试，2019 年之后全部留给验证。验证集模拟的是未知未来的交易环境，样本量必须够代表近期波动结构，否则得出的稳健性结论会塌。 结果不算惊艳但够用：验证准确率掉到 60%，不过模型是盈利的，回撤没放大，整体偏稳健。更关键的是它似乎真学到了风险/收益框架——在 1:8 的风险收益比设定下，它倾向挑低风险、高潜在赔率的位置出手，而不是乱撒网。外汇与贵金属杠杆高，这种样本外表现只能说明概率倾向，不能直接当实盘保票。