基于Python和MQL5的特征工程（第一部分）：为长期 AI 模型预测移动平均线·进阶篇
「用网格扫描找澳日回归的最优参数窗」
这段脚本干的事很直接：把预测步长（Forecast Horizon）和差分周期（Differencing Period）铺成二维网格，对每个组合跑一次 evaluate 算拟合误差，最后用 contourf 把误差面画出来。白点标的是两个轴上各自最小误差的交叉位置，也就是肉眼可抄的较优参数。 注意它前后画了两张图，标题分别是「AUDJPY Daily Return」和「AUDJPY Daily Moving Average Return」——同样扫描逻辑，换了个收益口径，白点位置可能偏移，说明参数对标的序列敏感，不能一套通吃。 末尾的 3D surface 用 subplot_kw={"projection":"3d"} 建轴，fig.set_size_inches(8,8) 定正方画布，plot_surface 把 optimal_nn_res 当高度场。你本地跑时若 optimal_nn_res 没先算好会直接报空，建议先确认前面网格结果已落进这个变量。外汇与贵金属这类品种波动跳变多，网格寻优只给历史样本下的概率较优，实盘仍属高风险。
period = y[i]
for j in np.arange(class="num">0,y.shape[class="num">0]):
results[i,j] = evaluate(look_ahead[j],period[j])
class="kw">return(results)
res = objective(x,y)
res = np.abs(res)
plt.contourf(x,y,res,class="num">100,cmap="jet")
plt.plot(x_axis[res.min(axis=class="num">0).argmin()],y_axis[res.min(axis=class="num">1).argmin()],&class="macro">#x27;.&class="macro">#x27;,class="type">class="kw">color=&class="macro">#x27;white&class="macro">#x27;)
plt.ylabel("Differencing Period")
plt.xlabel("Forecast Horizon")
plt.title("Linear Regression Accuracy Forecasting AUDJPY Daily Return")
plt.contourf(x,y,res,class="num">100,cmap="jet")
plt.plot(x_axis[res.min(axis=class="num">0).argmin()],y_axis[res.min(axis=class="num">1).argmin()],&class="macro">#x27;.&class="macro">#x27;,class="type">class="kw">color=&class="macro">#x27;white&class="macro">#x27;)
plt.ylabel("Differencing Period")
plt.xlabel("Forecast Horizon")
plt.title("Linear Regression Accuracy Forecasting AUDJPY Daily Moving Average Return")
class="macro">#Create a surface plot
fig , ax = plt.subplots(subplot_kw={"projection":"3d"})
fig.set_size_inches(class="num">8,class="num">8)
ax.plot_surface(x,y,optimal_nn_res,cmap="jet")◍ 用 wavelet 硬阈值给均线信号挤水分
线性变换只是特征工程的入门，往非线性走往往靠试错，没有公式能预告哪种变换当下最优。小波变换把序列同时投到时间与频率域，靠识别微小频率成分来分离噪声，低于阈值的系数直接压成 0，得到原始数据的稀疏表达。 相比 FFT 把信号拆成正弦余弦并一刀切滤掉高频，小波对高频域信号更友好，能以无监督方式保留真实波动、只剔噪声。想在 MT5 外做这套，先装 scikit-image 与 PyWavelets；纯 MQL5 从零写变换调试成本过高，接 Python 库反而快。 实测只对新浪均线类输入做变换、不动目标列：基准线性回归交叉验证均值 0.9935846835797412，小波硬阈值（BayesShrink + sym5 + 3 层）后降到 0.9082244556297641。准确率下滑说明去噪削掉了部分拟合用的高频成分，外汇与贵金属行情高风险，这种稀疏化是否利于实盘信号，需你自行回测确认。 硬阈值把噪声系数置 0，软阈值仅令其趋近 0；若想少丢细节可换软阈值重跑上面代码比对。
class="macro">#Benchmark Score np.mean(cross_val_score(LinearRegression(),data.loc[:,["MA"]],data["Target"])) class="macro">#Wavelet denoising data["Denoised"] = denoise_wavelet( data["MA"], method=&class="macro">#x27;BayesShrink&class="macro">#x27;, mode=&class="macro">#x27;hard&class="macro">#x27;, rescale_sigma=True, wavelet_levels = class="num">3, wavelet=&class="macro">#x27;sym5&class="macro">#x27; ) np.mean(cross_val_score(LinearRegression(),np.sqrt(np.log(data.loc[:,["Denoised"]])),data["Target"]))
用双指标状态化喂数据训练回归模型
直接从 MT5 终端抓行情，而不是用离线样本，才能保证训练时看到的 MA、RSI 数值和实盘机器人读到的一致。脚本里把移动平均线周期锁死在前面算出的理想值（示例用 40 期 SMA），同时拉取 RSI 读数，靠两个独立指标共同约束模型，行为可能比单指标预测更稳。 回测要有意义，得拿没见过的数据说话。原文做法是从生成的 CSV 里剔掉最近两年（2023–2024），只用此前约 22 年数据训练，这样跑 2023–2024 回测时，结果倾向反映样本外表现。外汇与贵金属杠杆高，样本外失效风险始终存在。 标记数据时把指标转成离散状态：RSI 分三档（>70、<30、中间），命中档置 1 其余 0；MA 只有上升 / 下降两态。Ridge 模型系数显示，RSI 超 70 时对应均值为 -0.1929 的价格变化，读数小于 70 时反而倾向上涨（系数 0.1482、0.0448），和经典「超买必跌」教条不完全一致。 MA 两态系数 array([-0.15572796, 0.15572796]) 说明：40 期 AUDJPY 日线 MA 连涨 40 根后倾向回落，连跌后倾向反弹，模型自己学出了均值回归。把 MA 当前状态喂进去，预测误差降到约 -0.0096，比单看 RSI 的 -0.0256 更紧。
class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| ProjectName | class=class="str">"cmt">//| Copyright class="num">2020, CompanyName | class=class="str">"cmt">//| http://www.companyname.net | class=class="str">"cmt">//+------------------------------------------------------------------+ class="macro">#class="kw">property copyright "Gamuchirai Zororo Ndawana" class="macro">#class="kw">property link "[MQL5官方文档] class="macro">#class="kw">property version "class="num">1.00" class="macro">#class="kw">property script_show_inputs class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Script Inputs | class=class="str">"cmt">//+------------------------------------------------------------------+ input class="type">int size = class="num">100000; class=class="str">"cmt">//How much data should we fetch? class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Global variables | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">int ma_handler,rsi_handler; class="type">class="kw">double ma_reading[],rsi_reading[]; class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| On start function | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">void OnStart() { class=class="str">"cmt">//--- Load indicator ma_handler = iMA(Symbol(),PERIOD_CURRENT,class="num">40,class="num">0,MODE_SMA,PRICE_CLOSE); rsi_handler = iRSI(Symbol(),PERIOD_CURRENT,class="num">30,PRICE_CLOSE); class=class="str">"cmt">//--- Load the indicator values CopyBuffer(ma_handler,class="num">0,class="num">0,size,ma_reading); CopyBuffer(rsi_handler,class="num">0,class="num">0,size,rsi_reading); ArraySetAsSeries(ma_reading,true);
「把 MT5 导出的 MA+RSI 序列喂给回归模型」
MT5 端先按时间序列方向把 RSI 数组排好，再拼出带品种名与指标名的 CSV 文件，用 FileOpen 以 ANSI+CSV 模式写入，表头含 Time/Open/High/Low/Close/MA/RSI，循环从 size 递减到 0 把每根 K 线的 iTime、iOpen 等和预存的 ma_reading、rsi_reading 逐行落盘，最后 FileClose 收尾。 读回时用 pandas 加载 'Market Data AUDJPY MA RSI As Series.csv'，砍掉末尾 365*2=730 根作为预留验证集，仅留 iloc[365:-(365*2),:] 做训练样本。 特征构造上，MA 状态用当前 MA 与 40 根前 MA 比较：大于置 MA1=1、否则 MA2=1；RSI 按 <30、>70、30~70 三档分别打 RSI1/RSI2/RSI3 标签。目标变量取 22 根后的 Close 差与 MA 差，模拟中期偏移。 数据清洗丢弃 NaN 并再截掉前 40 根（避免 shift(40) 引入的空缺），reset_index 后送 Ridge 回归。模型从样本里学到：RSI 读数低于 30 时，后续价格倾向走弱；其余区间则倾向回升。外汇与贵金属波动剧烈，该结论仅基于历史样本，实盘可能失效，务必用小资金验证。
ArraySetAsSeries(rsi_reading,true); class=class="str">"cmt">//--- File name class="type">class="kw">string file_name = "Market Data " + Symbol() +" MA RSI " + " As Series.csv"; class=class="str">"cmt">//--- Write to file class="type">int file_handle=FileOpen(file_name,FILE_WRITE|FILE_ANSI|FILE_CSV,","); for(class="type">int i= size;i>=class="num">0;i--) { if(i == size) { FileWrite(file_handle,"Time","Open","High","Low","Close","MA","RSI"); } else { FileWrite(file_handle,iTime(Symbol(),PERIOD_CURRENT,i), iOpen(Symbol(),PERIOD_CURRENT,i), iHigh(Symbol(),PERIOD_CURRENT,i), iLow(Symbol(),PERIOD_CURRENT,i), iClose(Symbol(),PERIOD_CURRENT,i), ma_reading[i], rsi_reading[i] ); } } class=class="str">"cmt">//--- Close the file FileClose(file_handle); } class=class="str">"cmt">//+------------------------------------------------------------------+ class="macro">#Read in the data data = pd.read_csv("Market Data AUDJPY MA RSI As Series.csv") class="macro">#Let&class="macro">#x27;s drop the last two years of data. We&class="macro">#x27;ll use that to validate our model in the back test data = data.iloc[class="num">365:-(class="num">365 * class="num">2),:] data class="macro">#MA States data["MA class="num">1"] = class="num">0 data["MA class="num">2"] = class="num">0 data.loc[data["MA"] > data["MA"].shift(class="num">40),"MA class="num">1"] = class="num">1 data.loc[data["MA"] <= data["MA"].shift(class="num">40),"MA class="num">2"] = class="num">1 class="macro">#RSI States data["RSI class="num">1"] = class="num">0 data["RSI class="num">2"] = class="num">0 data["RSI class="num">3"] = class="num">0 data.loc[data["RSI"] < class="num">30,"RSI class="num">1"] = class="num">1 data.loc[data["RSI"] > class="num">70,"RSI class="num">2"] = class="num">1 data.loc[(data["RSI"] >= class="num">30) & (data["RSI"] <= class="num">70),"RSI class="num">3"] = class="num">1 class="macro">#Target data["Target"] = data["Close"].shift(-class="num">22) - data["Close"] data["MA Target"] = data["MA"].shift(-class="num">22) - data["MA"] class="macro">#Clean up the data data = data.dropna() data = data.iloc[class="num">40:,:] data = data.reset_index(drop=True) from sklearn.linear_model class="kw">import Ridge from sklearn.model_selection class="kw">import TimeSeriesSplit,cross_val_score class="macro">#Our model can suggest optimal ways of using the RSI indicator class="macro">#Our model has learned that on average price tends to fall the RSI reading is less than class="num">30 and increases otherwises model = Ridge()
◍ 用岭回归给 RSI 和均线称重
把 RSI 三个周期（RSI 1/2/3）当作特征喂给 Ridge 模型去拟合目标列，再打印 coef_ 看每个周期对方向的线性权重。交叉验证用时间序列切分（tscv），RSI 状态的平均 cv 分数能反映这组特征在样本外的稳定度。 模型学到一个倾向：当 RSI 读数低于 30 时，价格后续倾向于走弱；高于该阈值则倾向回升。这不是确定性规律，外汇与贵金属市场高风险，阈值效应会随品种和周期漂移。 换用两条均线（MA 1/2）重复同样流程，fit 后看 coef_ 正负与量级，再跑一次时间序列交叉验证。对比两组 cv 均值，能判断当前样本里哪类因子对短期方向更有解释力，开 MT5 导出对应周期数据即可复算。
model.fit(data.loc[:,[“RSI class="num">1”,“RSI class="num">2”,“RSI class="num">3”]] , data[“Target”]) model.coef_ class="macro">#RSI state np.mean(cross_val_score(Ridge(),data.loc[:,[“RSI class="num">1”,“RSI class="num">2”,“RSI class="num">3”]] , data[“Target”],cv=tscv)) class="macro">#Our model can suggest optimal ways of using the RSI indicator class="macro">#Our model has learned that on average price tends to fall the RSI reading is less than class="num">30 and increases otherwises model = Ridge() model.fit(data.loc[:,[“MA class="num">1”,“MA class="num">2”]] , data[“Target”]) model.coef_ class="macro">#MA state np.mean(cross_val_score(Ridge(),data.loc[:,[“MA class="num">1”,“MA class="num">2”]] , data[“Target”],cv=tscv))
把 Ridge 模型落盘成 ONNX
前面已经用网格搜索敲定了移动平均线预测的最佳输入窗口（MA 1、MA 2 两列滞后特征），下一步是把训练好的模型从 Python 环境里剥出来，换成 ONNX 这种与语言无关的通用格式。这样一来，MetaTrader 5 里的 MQL5 侧就能直接加载 .onnx 文件做推理，不必重跑训练流程。 具体导出时，RSI 模型用了 3 个输入特征（RSI 1/2/3），MA 模型用了 2 个输入特征（MA 1/2），两者都用 Ridge 回归在全量数据上拟合。target_opset=12 是转换时的算子集版本，文件名里带的 AUDJPY D1 标明这是澳日日线周期的产物——外汇与贵金属杠杆品种波动剧烈，实盘加载前务必在 MT5 策略测试器用历史数据回测验证推断一致性。 代码里 initial_types 的 FloatTensorType([1, 3]) 和 [1, 2] 直接对应模型输入张量的形状，写错一个维度，MQL5 端用 OnnxRunModel 调用就会报输入不匹配。导出后硬盘上会多两个文件：AUDJPY D1 RSI AI F22 P40.onnx 与 AUDJPY D1 MA AI F22 P40.onnx，这就是给盯盘 EA 喂信号的实际权重。
class="macro">#Fetch clean data new_data = clean_data(class="num">140,class="num">130) class="kw">import onnx from skl2onnx class="kw">import convert_sklearn from skl2onnx.common.data_types class="kw">import FloatTensorType class="macro">#First we will class="kw">export the RSI model rsi_model = Ridge() rsi_model.fit(data.loc[:,[&class="macro">#x27;RSI class="num">1&class="macro">#x27;,&class="macro">#x27;RSI class="num">2&class="macro">#x27;,&class="macro">#x27;RSI class="num">3&class="macro">#x27;]],data.loc[:,&class="macro">#x27;Target&class="macro">#x27;]) class="macro">#Finally we will class="kw">export the MA model ma_model = Ridge() ma_model.fit(data.loc[:,[&class="macro">#x27;MA class="num">1&class="macro">#x27;,&class="macro">#x27;MA class="num">2&class="macro">#x27;]],data.loc[:,&class="macro">#x27;MA Target&class="macro">#x27;]) initial_types = [(&class="macro">#x27;float_input&class="macro">#x27;, FloatTensorType([class="num">1, class="num">3]))] onnx.save(convert_sklearn(rsi_model,initial_types=initial_types,target_opset=class="num">12),"AUDJPY D1 RSI AI F22 P40.onnx") initial_types = [(&class="macro">#x27;float_input&class="macro">#x27;, FloatTensorType([class="num">1, class="num">2]))] onnx.save(convert_sklearn(ma_model,initial_types=initial_types,target_opset=class="num">12),"AUDJPY D1 MA AI F22 P40.onnx")