📘

使用MQL5和Python构建自优化的EA（第四部分）：模型堆叠·进阶篇

第 2/3 篇

本章目录

NZDJPY的M1数据先看散点还是先拆趋势
先把价格序列压进稳健尺度
横向比模型：谁在新西兰元兑日元预测上更稳
互信息给特征排座次
用随机搜索榨干回归器参数
残差里的自相关会出卖你的模型

「NZDJPY的M1数据先看散点还是先拆趋势」

拿新西兰元兑日元（NZDJPY）的 M1 周期数据做探索，第一步画开盘价对收盘价的散点图，按涨跌二分类着色。结果点云基本随机铺开，涨和跌两类样本在图上没有可分边界，肉眼看不出线性或明显非线性关系。接着补了箱线图，分别看目标为 0（跌）和 1（涨）时的收盘价分布。两类箱体的中位数、四分位距几乎重叠，说明单看收盘价这一维，多空样本的统计分布差异极小，靠简单阈值切分大概率无效。再做时间序列分解，周期设成 1440（即一个自然日的 M1 根数）。原始收盘价本身就有很强的向下斜率；用 statsmodels 做加法分解后，趋势项把长期下行抽出来，残差里能清楚看到以「日」为单位的季节性摆动——这是直接在原始序列里容易被趋势盖住的部分。三维散点换成 Low/High/Close 作图，想挖更高维的隐藏结构。实际仍然是一团混，没有可辨的聚类或流形，二维里分不开的在这里也一样分不开。外汇与贵金属杠杆高、滑点跳空频繁，这类弱信号不能直接当入场依据，只能作为特征工程的负向参考。

MQL5 / C++

class="macro">#Lets perform scatter plots
sns.scatterplot(data=nzd_jpy,x=nzd_jpy[&class="macro">#x27;Open&class="macro">#x27;], y=nzd_jpy[&class="macro">#x27;Close&class="macro">#x27;],hue=&class="macro">#x27;Binary Target&class="macro">#x27;)
class="macro">#Let&class="macro">#x27;s create categorical box plots
sns.catplot(data=nzd_jpy,x=&class="macro">#x27;Binary Target&class="macro">#x27;,y=&class="macro">#x27;Close&class="macro">#x27;,kind=&class="macro">#x27;box&class="macro">#x27;)
class="macro">#Time series decomposition
class="kw">import statsmodels.api as sm
nzd_jpy_decomposition = sm.tsa.seasonal_decompose(nzd_jpy[&class="macro">#x27;Close&class="macro">#x27;],period=class="num">1440,model=&class="macro">#x27;additive&class="macro">#x27;)
fig = nzd_jpy_decomposition.plot()
class="macro">#Let&class="macro">#x27;s also perform 3D plots
class="macro">#Visualizing our data in 3D
class="kw">import matplotlib.pyplot as plt
fig = plt.figure(figsize=(class="num">7,class="num">7))
ax = fig.add_subplot(class="num">111,projection=&class="macro">#x27;3d&class="macro">#x27;)
colors = [&class="macro">#x27;blue&class="macro">#x27; if movement == class="num">0 else &class="macro">#x27;red&class="macro">#x27; for movement in nzd_jpy.loc[:,"Binary Target"]]
ax.scatter(nzd_jpy.loc[:,"Low"],nzd_jpy.loc[:,"High"],nzd_jpy.loc[:,"Close"],c=colors)
ax.set_xlabel(&class="macro">#x27;NZDJPY Low&class="macro">#x27;)
ax.set_ylabel(&class="macro">#x27;NZDJPY High&class="macro">#x27;)
ax.set_zlabel(&class="macro">#x27;NZDJPY Close&class="macro">#x27;)

先把价格序列压进稳健尺度

做 NZD/JPY 的序列建模前，不能直接拿原始收盘价喂模型。不同波动阶段的价格量纲差异会放大异常样本权重，先缩放再切训练集才稳。这里用 RobustScaler 而不是 MinMaxScaler，核心原因是它用中位数和四分位距，对汇率跳空、非农瞬刺这类离群点不敏感。外汇与贵金属本身高杠杆、高波动，鲁棒缩放能降低极端行情污染训练分布的概率。代码里把 Close 单列缩放成 X，Open/High/Low 三列缩放进 residuals_X，原 Target 列不动直接赋给 y。你可以在 MT5 导出的 NZD/JPY 1H csv 上跑这段，验证缩放后 Close 的 std 是否落在 1.0 附近。

MQL5 / C++

class="macro">#Let&class="macro">#x27;s prepare the data for modelling
from sklearn.preprocessing class="kw">import RobustScaler
class="macro">#Scale the data
X = pd.DataFrame(RobustScaler().fit_transform(nzd_jpy.loc[:,[&class="macro">#x27;Close&class="macro">#x27;]]),columns=[&class="macro">#x27;Close&class="macro">#x27;])
residuals_X = pd.DataFrame(RobustScaler().fit_transform(nzd_jpy.loc[:,[&class="macro">#x27;Open&class="macro">#x27;,&class="macro">#x27;High&class="macro">#x27;,&class="macro">#x27;Low&class="macro">#x27;]]),columns=[&class="macro">#x27;Open&class="macro">#x27;,&class="macro">#x27;High&class="macro">#x27;,&class="macro">#x27;Low&class="macro">#x27;])
y = nzd_jpy.loc[:,&class="macro">#x27;Target&class="macro">#x27;]

◍ 横向比模型：谁在新西兰元兑日元预测上更稳

把 NZD/JPY 的收盘价序列按 50% 不洗牌切分，训练集与测试集严格按时间顺序分离，避免未来信息泄漏。我们同时把 Open/High/Low 作为残差特征单独切分，供后续建模使用。一次性塞进 15 个回归模型：从 Lasso、Ridge 到随机森林、梯度提升，再到 MLP 神经网络。用 5 折交叉验证跑一遍，负均方误差（neg_MSE）记进一个 5 行×15 列的 DataFrame，每列对应一个模型的逐折表现。箱线图把差异摊开了：MLPRegressor 作为最右侧的 Deep Neural Network，箱体和须都明显更长，波动和离群幅度比其他模型大得多，说明它在这项任务上方差失控、单折成绩很不稳定。但别急着丢掉神经网络——它的交叉验证分数差，可能更多来自默认参数而非结构本身，调参后有可能大幅收敛。最终挑模型看 cv_error.mean()，均值最低的那个才是当前数据上泛化误差倾向最优的选择。下面这段代码可直接在装了 sklearn 的 Python 环境复跑，把 cv_error 打印出来就能比对每个模型的 5 折分数。

MQL5 / C++

class="macro">#Cross validating the models
from sklearn.model_selection class="kw">import cross_val_score,train_test_split
from sklearn.linear_model class="kw">import Lasso,LinearRegression,Ridge,ElasticNet,SGDRegressor,HuberRegressor
from sklearn.ensemble class="kw">import RandomForestRegressor,GradientBoostingRegressor,AdaBoostRegressor,ExtraTreesRegressor,BaggingRegressor
from sklearn.svm class="kw">import LinearSVR
from sklearn.neighbors class="kw">import KNeighborsRegressor
from sklearn.tree class="kw">import DecisionTreeRegressor
from sklearn.neural_network class="kw">import MLPRegressor
class="macro">#Create train-test splits
train_X,test_X,train_y,test_y = train_test_split(nzd_jpy.loc[:,[&class="macro">#x27;Close&class="macro">#x27;]],y,test_size=class="num">0.5,shuffle=False)
residuals_train_X,residuals_test_X,residuals_train_y,residuals_test_y = train_test_split(nzd_jpy.loc[:,[&class="macro">#x27;Open&class="macro">#x27;,&class="macro">#x27;High&class="macro">#x27;,&class="macro">#x27;Low&class="macro">#x27;]],y,test_size=class="num">0.5,shuffle=False)
class="macro">#Store the models
models = [
    Lasso(),
    LinearRegression(),
    Ridge(),
    ElasticNet(),
    SGDRegressor(),
    HuberRegressor(),
    RandomForestRegressor(),
    GradientBoostingRegressor(),
    AdaBoostRegressor(),
    ExtraTreesRegressor(),
    BaggingRegressor(),
    LinearSVR(),
    KNeighborsRegressor(),
    DecisionTreeRegressor(),
    MLPRegressor(),
]
class="macro">#Store the names of the models
model_names = [
    &class="macro">#x27;Lasso&class="macro">#x27;,
    &class="macro">#x27;Linear Regression&class="macro">#x27;,
    &class="macro">#x27;Ridge&class="macro">#x27;,
    &class="macro">#x27;Elastic Net&class="macro">#x27;,
    &class="macro">#x27;SGD Regressor&class="macro">#x27;,
    &class="macro">#x27;Huber Regressor&class="macro">#x27;,
    &class="macro">#x27;Random Forest Regressor&class="macro">#x27;,
    &class="macro">#x27;Gradient Boosting Regressor&class="macro">#x27;,
    &class="macro">#x27;Ada Boost Regressor&class="macro">#x27;,
    &class="macro">#x27;Extra Trees Regressor&class="macro">#x27;,
    &class="macro">#x27;Bagging Regressor&class="macro">#x27;,
    &class="macro">#x27;Linear SVR&class="macro">#x27;,
    &class="macro">#x27;K Neighbors Regressor&class="macro">#x27;,
    &class="macro">#x27;Decision Tree Regressor&class="macro">#x27;,
    &class="macro">#x27;MLP Regressor&class="macro">#x27;,
]
class="macro">#Create a dataframe to store our cv error
cv_error = pd.DataFrame(columns=model_names,index=np.arange(class="num">0,class="num">5))
class="macro">#Cross validate each model
for model in models:
  cv_score = cross_val_score(model,X,y,cv=class="num">5,n_jobs=-class="num">1,scoring=&class="macro">#x27;neg_mean_squared_error&class="macro">#x27;)
  for i in np.arange(class="num">0,class="num">5):
    index = models.index(model)
    cv_error.iloc[i,index] = cv_score[i]
cv_error
cv_error.plot()
sns.boxplot(cv_error)
class="macro">#Our mean validation error
cv_error.mean()

「互信息给特征排座次」

想看清模型到底抓没抓到真关系，特征重要性算法是最直接的探针。先算互信息（MI）得分：它衡量每个预测变量单独预测目标变量的潜在能力，而且以对数尺度计，实战里 MI 高于 3 极少见，这本身就是一个可校验的阈值锚点。把 MI 得分画出来，图里最高价（High）在新西兰元兑日元（NZDJPY）未来收盘价的预测上潜力最大。注意外汇与贵金属杠杆高、跳空频繁，这种单变量潜力不等于实盘胜率，只能说明信息含量偏厚。递归特征消除（RFE）用起来和 scikit-learn 里别的估计器一样：建实例、fit、看 ranking_。本例跑出来是 array([1, 1, 1, 1])，算法认为四个预测变量重要性等同——和 MI 的结论冲突，说明线性包装器与信息论视角给出的特征画像并不总一致。

MQL5 / C++

<span class="preprocessor">class="macro">#Feature </span>importance
from sklearn.feature_selection class="kw">import mutual_info_regression,RFE
mi_score = pd.DataFrame(mutual_info_regression(X,y),columns=[&class="macro">#x27;MI Score&class="macro">#x27;],index=X.columns)
mi_score.plot()
<span class="preprocessor">class="macro">#Select </span>the best features
rfe = RFE(model, n_features_to_select=class="num">5, step=class="num">1)
rfe = rfe.fit(X, y)
rfe.ranking_

代码逐行拆解：第 2 行从 sklearn 导入互信息回归与 RFE 两个工具；第 3 行用 mutual_info_regression 算 X 对 y 的 MI，转成带列名『MI Score』、索引为 X 列名的 DataFrame；第 4 行直接 plot() 出柱状/折线图看排序。第 6 行建 RFE 实例，指定留 5 个特征、每次删 1 个；第 7 行 fit 到数据；第 8 行 ranking_ 输出每个特征的排名，全 1 即未被淘汰、视为同等重要。开 MT5 导出的 NZDJPY 小时 bar 自己跑一遍，能立刻复现这套分歧。

用随机搜索榨干回归器参数

想把 SGD 回归器在价格行为建模里的性能再往上推，直接手调参数效率太低。更实在的做法是在参数空间里做随机采样，让交叉验证替你筛出相对优的组合。下面这段 Python 用了 RandomizedSearchCV，对 loss、penalty、alpha 等 8 个维度各给了一组候选值，n_iter=100 表示抽 100 组随机组合，cv=5 是 5 折交叉验证，n_jobs=-1 吃满本地线程。评分用 neg_mean_squared_error，越接近 0 拟合误差越小。

MQL5 / C++

class="macro">#Parameter tuning
from sklearn.model_selection class="kw">import RandomizedSearchCV
class="macro">#Initialize the model
model = SGDRegressor()
class="macro">#Define the tuner
tuner = RandomizedSearchCV(
				model,
				{
				"loss" : [&class="macro">#x27;squared_error&class="macro">#x27;, &class="macro">#x27;huber&class="macro">#x27;, &class="macro">#x27;epsilon_insensitive&class="macro">#x27;,&class="macro">#x27;squared_epsilon_insensitive&class="macro">#x27;],
				"penalty":[&class="macro">#x27;l2&class="macro">#x27;,&class="macro">#x27;l1&class="macro">#x27;, &class="macro">#x27;elasticnet&class="macro">#x27;, None],
				"alpha":[class="num">0.1,class="num">0.01,class="num">0.001,class="num">0.0001,class="num">0.00001,class="num">0.00001,class="num">0.0000001,class="num">10,class="num">100,class="num">1000,class="num">10000,class="num">100000],
				"tol":[class="num">0.1,class="num">0.01,class="num">0.001,class="num">0.0001,class="num">0.00001,class="num">0.000001,class="num">0.0000001],
				"fit_intercept": [True,False],
				"early_stopping": [True,False],
				"learning_rate":[&class="macro">#x27;constant&class="macro">#x27;,&class="macro">#x27;optimal&class="macro">#x27;,&class="macro">#x27;adaptive&class="macro">#x27;,&class="macro">#x27;invscaling&class="macro">#x27;],
				"shuffle": [True,False]
				},
				n_iter=class="num">100,
				cv=class="num">5,
				n_jobs=-class="num">1,
				scoring="neg_mean_squared_error"
)
class="macro">#Fit the tuner
tuner.fit(train_X,train_y)
class="macro">#Our best parameters
tuner.best_params_

代码逐行拆一下：第 1–2 行导入随机搜索工具；第 4–5 行先建一个默认 SGDRegressor 实例。第 7 行起定义 tuner，候选 loss 含 huber 等 4 种、penalty 含 elasticnet 等 4 种、alpha 跨度从 1e-7 到 1e5 共 12 档、tol 从 0.1 到 1e-7 共 7 档，布尔类参数如 fit_intercept 和 early_stopping 各给 True/False。第 24–25 行设定搜索 100 次、5 折、满线程、以负均方误差打分。第 27–28 行拿训练集跑搜索，第 30 行取出最优参数组。一次跑完，搜索返回的最优组合是：tol=0.001、shuffle=False、penalty='elasticnet'、loss='huber'、learning_rate='adaptive'、fit_intercept=True、early_stopping=True、alpha=1e-05。这组在样本内交叉验证里误差倾向更低，但实际用于外汇或贵金属行情预测时仍只是概率性改善，这类品种高杠杆、高波动，参数优不代表实盘能稳过。

MQL5 / C++

class="macro">#Parameter tuning
from sklearn.model_selection class="kw">import RandomizedSearchCV
class="macro">#Initialize the model
model = SGDRegressor()
class="macro">#Define the tuner
tuner = RandomizedSearchCV(
				model,
				{
				"loss" : [&class="macro">#x27;squared_error&class="macro">#x27;, &class="macro">#x27;huber&class="macro">#x27;, &class="macro">#x27;epsilon_insensitive&class="macro">#x27;,&class="macro">#x27;squared_epsilon_insensitive&class="macro">#x27;],
				"penalty":[&class="macro">#x27;l2&class="macro">#x27;,&class="macro">#x27;l1&class="macro">#x27;, &class="macro">#x27;elasticnet&class="macro">#x27;, None],
				"alpha":[class="num">0.1,class="num">0.01,class="num">0.001,class="num">0.0001,class="num">0.00001,class="num">0.00001,class="num">0.0000001,class="num">10,class="num">100,class="num">1000,class="num">10000,class="num">100000],
				"tol":[class="num">0.1,class="num">0.01,class="num">0.001,class="num">0.0001,class="num">0.00001,class="num">0.000001,class="num">0.0000001],
				"fit_intercept": [True,False],
				"early_stopping": [True,False],
				"learning_rate":[&class="macro">#x27;constant&class="macro">#x27;,&class="macro">#x27;optimal&class="macro">#x27;,&class="macro">#x27;adaptive&class="macro">#x27;,&class="macro">#x27;invscaling&class="macro">#x27;],
				"shuffle": [True,False]
				},
				n_iter=class="num">100,
				cv=class="num">5,
				n_jobs=-class="num">1,
				scoring="neg_mean_squared_error"
)
class="macro">#Fit the tuner
tuner.fit(train_X,train_y)
class="macro">#Our best parameters
tuner.best_params_

◍ 残差里的自相关会出卖你的模型

判断一个回归模型有没有过拟合，最快的办法是盯着它的残差看。理想状态下，模型学干净了，残差就是随机白噪声，前后之间没有可预测关系；一旦发现残差有自相关——比如跌了还接着跌、涨了还接着涨——就说明模型没真正学到规律，只是表面上把参数调顺了。把残差直接画在图表上，往往肉眼就能看出趋势性聚集，这已经是个坏信号。更硬核的做法是拉一张自相关函数（ACF）图：横轴是每个滞后阶数，纵轴尖峰高度代表该滞后下与自身的相关强度，背景蓝色锥体是置信区间，任何戳出锥体的尖峰都算统计显著。原文图14里残差ACF明显有超界尖峰，证明即便做完参数调整，第一版SGD回归器依旧没训透。光看图表不够，还得交叉验证把误差量化。用5折cv在训练残差上跑每个候选模型，neg_mean_squared_error做评分，再按平均误差降序排，Lasso在那组实验里平均验证误差最低，是最优选择；而SGD回归器预测自身误差的箱线图散得很开，说明它连自己错在哪都估不准。外汇与贵金属行情噪声大、过拟合风险高，这类检验在实盘前最好亲自跑一遍。下面这段是原文里做验证的核心脚本，可直接丢进MT5关联的Python环境改路径复用：

MQL5 / C++

class="macro">#Model validation
model = SGDRegressor(
    tol           = tuner.best_params_[&class="macro">#x27;tol&class="macro">#x27;],
    shuffle       = tuner.best_params_[&class="macro">#x27;shuffle&class="macro">#x27;],
    penalty       = tuner.best_params_[&class="macro">#x27;penalty&class="macro">#x27;],
    loss          = tuner.best_params_[&class="macro">#x27;loss&class="macro">#x27;],
    learning_rate = tuner.best_params_[&class="macro">#x27;learning_rate&class="macro">#x27;],
    alpha         = tuner.best_params_[&class="macro">#x27;alpha&class="macro">#x27;],
    fit_intercept = tuner.best_params_[&class="macro">#x27;fit_intercept&class="macro">#x27;],
    early_stopping= tuner.best_params_[&class="macro">#x27;early_stopping&class="macro">#x27;]
)
model.fit(train_X,train_y)
residuals = test_y - model.predict(test_X)
class="macro">#Plot the residuals
residuals.plot()
class="macro">#The residuals appear to have autocorrelation
from statsmodels.graphics.tsaplots class="kw">import plot_acf
fig = plot_acf(residuals)
class="macro">#Prepare the residuals for our second model
model = SGDRegressor(tol= class="num">0.001,
 shuffle=False,
 penalty= &class="macro">#x27;elasticnet&class="macro">#x27;,
 loss= &class="macro">#x27;huber&class="macro">#x27;,
 learning_rate=&class="macro">#x27;adaptive&class="macro">#x27;,
 fit_intercept= True,
 early_stopping= True,
 alpha= class="num">1e-05)
class="macro">#Store the model residuals
model.fit(train_X,train_y)
residuals_train_y = train_y - model.predict(train_X)
residuals_test_y = test_y - model.predict(test_X)
class="macro">#Cross validate each model
for model in models:
  cv_score = cross_val_score(model,residuals_train_X,residuals_train_y,cv=class="num">5,n_jobs=-class="num">1,scoring=&class="macro">#x27;neg_mean_squared_error&class="macro">#x27;)
  for i in np.arange(class="num">0,class="num">5):
    index = models.index(model)
    cv_error.iloc[i,index] = cv_score[i]
class="macro">#Cross validaton error levels
cv_error
class="macro">#Store the model&class="macro">#x27;s performance
cv_error.mean().sort_values(ascending=False)
sns.boxplot(cv_error)
cv_error.plot()

逐行拆解几个关键点：SGDRegressor 用 tuner 搜出的最佳超参重建，fit 后拿 test_y 减 predict 得到残差；residuals.plot() 看趋势，plot_acf 出自相关图；第二版模型换 elasticnet+huber 试图压自相关；cross_val_score 用5折、neg_mean_squared_error 把每个模型在残差上的误差存进 cv_error，最后 mean 排序和 boxplot 一眼辨优劣。

MQL5 / C++

class="macro">#Model validation
model = SGDRegressor(
    tol           = tuner.best_params_[&class="macro">#x27;tol&class="macro">#x27;],
    shuffle       = tuner.best_params_[&class="macro">#x27;shuffle&class="macro">#x27;],
    penalty       = tuner.best_params_[&class="macro">#x27;penalty&class="macro">#x27;],
    loss          = tuner.best_params_[&class="macro">#x27;loss&class="macro">#x27;],
    learning_rate = tuner.best_params_[&class="macro">#x27;learning_rate&class="macro">#x27;],
    alpha         = tuner.best_params_[&class="macro">#x27;alpha&class="macro">#x27;],
    fit_intercept = tuner.best_params_[&class="macro">#x27;fit_intercept&class="macro">#x27;],
    early_stopping= tuner.best_params_[&class="macro">#x27;early_stopping&class="macro">#x27;]
)
model.fit(train_X,train_y)
residuals = test_y - model.predict(test_X)
class="macro">#Plot the residuals
residuals.plot()
class="macro">#The residuals appear to have autocorrelation
from statsmodels.graphics.tsaplots class="kw">import plot_acf
fig = plot_acf(residuals)
class="macro">#Prepare the residuals for our second model
model = SGDRegressor(tol= class="num">0.001,
 shuffle=False,
 penalty= &class="macro">#x27;elasticnet&class="macro">#x27;,
 loss= &class="macro">#x27;huber&class="macro">#x27;,
 learning_rate=&class="macro">#x27;adaptive&class="macro">#x27;,
 fit_intercept= True,
 early_stopping= True,
 alpha= class="num">1e-05)
class="macro">#Store the model residuals
model.fit(train_X,train_y)
residuals_train_y = train_y - model.predict(train_X)
residuals_test_y = test_y - model.predict(test_X)
class="macro">#Cross validate each model
for model in models:
  cv_score = cross_val_score(model,residuals_train_X,residuals_train_y,cv=class="num">5,n_jobs=-class="num">1,scoring=&class="macro">#x27;neg_mean_squared_error&class="macro">#x27;)
  for i in np.arange(class="num">0,class="num">5):
    index = models.index(model)
    cv_error.iloc[i,index] = cv_score[i]
class="macro">#Cross validaton error levels
cv_error
class="macro">#Store the model&class="macro">#x27;s performance
cv_error.mean().sort_values(ascending=False)
sns.boxplot(cv_error)
cv_error.plot()