基于Python和MQL5的特征工程（第一部分）：为长期 AI 模型预测移动平均线（基础篇）
「用均线拉长 AI 模型的预测视野」
把 AI 接进 MT5 做行情预测，第一步不是堆模型，而是先把原始报价变成模型能消化的特征。这个过程叫特征工程：对市场价格做变换与重组，让输入里携带更多关于现实状态的信息，从而降低模型误差。 本篇先只做一件事——用移动平均线（MA）扩展 AI 模型的预测跨度。直接拿收盘价喂模型，往往只能看到眼前一两根 K 线；叠加不同周期的 MA 后，模型能同时读到趋势平滑值与价格偏离，预测窗口可能从短期延伸到更长的持有周期。 关键是你要自己掌控特征构造，而不是黑盒代劳。在 MT5 里用 Python 通过 MetaTrader5 包拉数据、算 MA，再喂给模型，你才清楚每一路输入在策略里起了什么作用，也才好评估整体有效性。外汇与贵金属杠杆高、跳空频繁，这类特征方案仅降低误差概率，不保证方向判断正确。
为什么先让模型猜均线而不是猜价格
把同一套 AI 模型在 200 多个交易品种上各跑一遍，一组预测未来价格、一组预测移动平均线，对比下来预测价格比预测均线平均准确率掉 34%。具体数：猜均线平均能到 70%，猜价格只有 52%。 均线并不总贴着价格走。价格可能在 20 根蜡烛里下跌，同区间均线却在上扬，这种背离会让你看对均线方向却吃价格的亏。跨所有市场统计，背离率基本钉在 31% 附近，而模型预判背离的准确率平均有 68%。 模型对背离的判断方差仅 0.000041，背离本身发生的方差是 0.000386，说明它纠错不是瞎蒙，而是有稳定节律。当前对比都锁在 M1 周期，原因是 297 个市场在这个周期才能凑齐足量样本做公平对照。 从线性角度看，均线本就是过去价格的求和，天然契合线性回归假设；价格却是多重现实变量纠缠的结果，硬猜它违背模型底层假设。外汇与贵金属属高风险品种，这种统计优势不承诺实盘收益，仅表明建模切入点可优先选均线。
◍ 先把 MT5 终端和品种池拉起来
做跨市场扫描的第一步，是用 Python 把 MetaTrader 5 终端初始化，并确认当前账户环境下到底能交易多少标的。实测一次标准环境，symbols_total() 返回 297，也就是终端里挂了 297 个交易品种——外汇、贵金属、指数、差价合约都在内，外汇与贵金属杠杆高、滑点跳空频繁，后续建模必须带着这个风险底色。 拿到全量品种名后，建一个 DataFrame 作为存储骨架，索引是品种名，列预留 OHLC Error、MAR Error、Noise Levels、Divergence Error 四类准确率字段。这个结构相当于给每个市场留好位置，后面循环跑模型时直接往里填交叉验证分数。 时间序列切分不能用随机 K 折，得用 TimeSeriesSplit(n_splits=5, gap=10)，避免用未来信息污染训练。代码里对全部 297 个品种逐个取 50000 根 M1 数据，算经典收盘价目标、移动平均目标与背离目标，再用逻辑回归做交叉验证。跑完进度打印到 99.66% complete，说明全市场扫描基本通了，剩下的就是看哪些品种噪声比高、哪些模型准确率有区分度。
class="macro">#Load the libraries we need class="kw">import pandas as pd class="kw">import numpy as np class="kw">import MetaTrader5 as mt5 from sklearn.model_selection class="kw">import TimeSeriesSplit,cross_val_score from sklearn.linear_model class="kw">import LogisticRegression,LinearRegression class="kw">import matplotlib.pyplot as plt class="macro">#Initialize the terminal mt5.initialize() class="macro">#The total number of symbols we have print(f"Total Symbols Available: ",mt5.symbols_total()) class="macro">#Get the names of all pairs symbols = mt5.symbols_get() idx = [s.name for s in symbols] global_params = pd.DataFrame(index=idx,columns=["OHLC Error","MAR Error","Noise Levels","Divergence Error"]) global_params class="macro">#Define the time series split object tscv = TimeSeriesSplit(n_splits=class="num">5,gap=class="num">10) class="macro">#Iterate over all symbols for i in np.arange(global_params.dropna().shape[class="num">0],len(idx)): class="macro">#Fetch M1 Data data = pd.DataFrame(mt5.copy_rates_from_pos(cols[i],mt5.TIMEFRAME_M1,class="num">0,class="num">50000)) data.rename(columns={"open":"Open","high":"High","low":"Low","close":"Close"},inplace=True) class="macro">#Define our period period = class="num">10 class="macro">#Add the classical target data.loc[data["Close"].shift(-period) > data["Close"],"OHLC Target"] = class="num">1 class="macro">#Calculate the returns data.loc[:,["Open","High","Low","Close"]] = data.loc[:,["Open","High","Low","Close"]].diff(period) data["RMA"] = data["Close"].rolling(period).mean() class="macro">#Calculate our new target data.dropna(inplace=True) data.reset_index(inplace=True,drop=True) data.loc[data["RMA"].shift(-period) > data["RMA"],"New Target"] = class="num">1 data = data.iloc[class="num">0:-period,:] class="macro">#Calculate the divergence target data.loc[data["OHLC Target"] != data["New Target"],"Divergence Target"] = class="num">1 class="macro">#Noise ratio global_params.iloc[i,class="num">2] = data.loc[data["New Target"] != data["OHLC Target"]].shape[class="num">0] / data.shape[class="num">0] class="macro">#Test our accuracy predicting the future close price score = cross_val_score(LogisticRegression(),data.loc[:,["Open","High","Low","Close"]],data["OHLC Target"],cv=tscv) global_params.iloc[i,class="num">0] = score.mean() class="macro">#Test our accuracy predicting the moving average of future returns score = cross_val_score(LogisticRegression(),data.loc[:,["Open","Close","RMA"]],data["New Target"],cv=tscv) global_params.iloc[i,class="num">1] = score.mean() class="macro">#Test our accuracy predicting the future divergence between price and its moving average score = cross_val_score(LogisticRegression(),data.loc[:,["Open","Close","RMA"]],data["Divergence Target"],cv=tscv) global_params.iloc[i,class="num">3] = score.mean() print(f"{((i/len(idx)) * class="num">100)}% complete") class="macro">#We are done
「一行打印背后的执行终点」
MQL5 里用 print 向专家日志输出字符串，是最直接的运行探针。上面这行把 "Done" 推到 MT5 终端的「专家」标签页，说明某段逻辑已跑到末尾。 在实盘或回测里，若你写的 EA 没按预期打印 Done，大概率前面某步订单发送或数组越界被静默中断——开 MT5 按 F4 编译后挂到图表，看日志比猜更快。外汇与贵金属杠杆高，日志只帮定位，不替你兜底风险。
print("Done")跨市场回测里模型到底猜中了什么
把 5 折交叉验证铺到全部品种后，先盯未来收盘价变化方向。蓝线是所有市场的平均命中率，红线是 50% 基准，两者几乎贴在一起——平均准确率谈不上比抛硬币强，这个结论对外汇和贵金属交易者而言意味着直接拿价格变动做单信号大概率会亏。 不过拆开看个别标的，有特定市场命中率冲过 65%，这类 outlier 值得单独拉出来验，究竟是样本偶然还是该品种波动结构更规则，得进一步跑子集回测才能定性。 换到移动平均线变化预测，黄线（价格变化命中）和蓝线（均线变化命中）同框对比，模型对均线的捕捉明显更稳。表格里的数字更直接：价格误差 0.525353，均线误差 0.705468，噪声水平 0.317187，背离误差 0.682069——哪怕均线天然滞后，猜它反转的胜率仍显著高于猜价格本身。 背离维度上，全市场噪声压在 30%–35% 区间，说明价格与均线反向的频率不算失控；而模型预测二者是否背离的平均准确率接近 70%，这部分是少数能拿去辅助过滤假突破的产出。 下方代码把四张诊断图一口气画出来，想在 MT5 外接 Python 验证的同好可直接复用：先逐列 plot 准确率与噪声，再标均值线和 0.5 红线，最后按 MAR Error 降序排，找出表现最好的市场。
global_params.iloc[:,class="num">0].plot() plt.title("OHLC Accuracy") plt.xlabel("Market") plt.ylabel("class="num">5-fold Accuracy %") plt.axhline(global_params.iloc[:,class="num">0].mean(),linestyle=&class="macro">#x27;--&class="macro">#x27;) plt.axhline(class="num">0.5,linestyle=&class="macro">#x27;--&class="macro">#x27;,class="type">class="kw">color=&class="macro">#x27;red&class="macro">#x27;) global_params.iloc[:,class="num">1].plot() plt.title("Moving Average Returns Accuracy") plt.xlabel("Market") plt.ylabel("class="num">5-fold Accuracy %") plt.axhline(global_params.iloc[:,class="num">1].mean(),linestyle=&class="macro">#x27;--&class="macro">#x27;) plt.axhline(global_params.iloc[:,class="num">0].mean(),linestyle=&class="macro">#x27;--&class="macro">#x27;,class="type">class="kw">color=&class="macro">#x27;orange&class="macro">#x27;) plt.axhline(class="num">0.5,linestyle=&class="macro">#x27;--&class="macro">#x27;,class="type">class="kw">color=&class="macro">#x27;red&class="macro">#x27;) global_params.iloc[:,class="num">2].plot() plt.title("Noise Level") plt.xlabel("Market") plt.ylabel("Percentage of Divergence:Price And Moving Average") plt.axhline(global_params.iloc[:,class="num">2].mean(),linestyle=&class="macro">#x27;--&class="macro">#x27;) global_params.iloc[:,class="num">3].plot() plt.title("Divergence Accuracy") plt.xlabel("Market") plt.ylabel("class="num">5-fold Accuracy %") plt.axhline(global_params.iloc[:,class="num">3].mean(),linestyle=&class="macro">#x27;--&class="macro">#x27;) global_params.sort_values("MAR Error",ascending=False)
◍ AUDJPY 上预测均线变化比预测价格走得更远
把优化锁定在 AUDJPY 日线这一表现较优的市场上，直接预测收盘价变化与预测移动平均线变化，结果差别很大。回测显示：直接预测价格未来变化时，模型最多只能向前看 1 步，相当于被锁死在下一根蜡烛；而预测均线变化时，最优预测范围右移到 22 步，且最优点上两者误差水平相同——即预测 40 步后的均线变化，难度与预测 1 步价格变化相当。 外汇与贵金属属高风险品种，AUDJPY 交叉盘流动性与波动结构特殊，上述结论仅在该品种日线样本（约 5000 根、2 年）下成立，换市场需重跑。 三维误差面把差异画得很直白：价格变化目标随预测步数拉长单调变差，模型「短视」超过 20 步就崩；均线变化目标则随周期与前瞻步数增加先平稳降至最低点再回升，说明以均线为监督信号更容易训出泛化好的结构。 下面这段代码就是跑这套 5 折时序交叉验证的骨架，接 MT5 终端拉 AUDJPY 日线，用 meshgrid 扫 look_ahead 与 period，白点即最低误差组合。
symbol = "AUDJPY" class="macro">#Reach the terminal mt5.initialize() data = pd.DataFrame(mt5.copy_rates_from_pos(symbol,mt5.TIMEFRAME_D1,class="num">365*class="num">2,class="num">5000)) class="macro">#Standard libraries class="kw">import seaborn as sns from mpl_toolkits.mplot3d class="kw">import Axes3D from sklearn.linear_model class="kw">import LinearRegression from sklearn.neural_network class="kw">import MLPRegressor from sklearn.metrics class="kw">import mean_squared_error from sklearn.model_selection class="kw">import cross_val_score,TimeSeriesSplit class="macro">#Define the input range x_min , x_max = class="num">2,class="num">100 class="macro">#Look ahead y_min , y_max = class="num">2,class="num">100 class="macro">#Period class="macro">#Sample input range uniformly x_axis = np.arange(x_min,x_max,class="num">2) class="macro">#Look ahead y_axis = np.arange(y_min,y_max,class="num">2) class="macro">#Period class="macro">#Create a meshgrid x , y = np.meshgrid(x_axis,y_axis) def clean_data(look_ahead,period): class="macro">#Fetch the data from our terminal and clean it up data = pd.DataFrame(mt5.copy_rates_from_pos(&class="macro">#x27;AUDJPY&class="macro">#x27;,mt5.TIMEFRAME_D1,class="num">365*class="num">2,class="num">5000)) data[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(data[&class="macro">#x27;time&class="macro">#x27;],unit=&class="macro">#x27;s&class="macro">#x27;) data[&class="macro">#x27;MA&class="macro">#x27;] = data[&class="macro">#x27;close&class="macro">#x27;].rolling(period).mean() class="macro">#Transform the data class="macro">#Target data[&class="macro">#x27;Target&class="macro">#x27;] = data[&class="macro">#x27;MA&class="macro">#x27;].shift(-look_ahead) - data[&class="macro">#x27;MA&class="macro">#x27;] class="macro">#Change in price data[&class="macro">#x27;close&class="macro">#x27;] = data[&class="macro">#x27;close&class="macro">#x27;] - data[&class="macro">#x27;close&class="macro">#x27;].shift(period) class="macro">#Change in MA data[&class="macro">#x27;MA&class="macro">#x27;] = data[&class="macro">#x27;MA&class="macro">#x27;] - data[&class="macro">#x27;MA&class="macro">#x27;].shift(period) data.dropna(inplace=True) data.reset_index(drop=True,inplace=True) class="kw">return(data) class="macro">#Evaluate the objective function def evaluate(look_ahead,period): class="macro">#Define the model model = LinearRegression() class="macro">#Define our time series split tscv = TimeSeriesSplit(n_splits=class="num">5,gap=look_ahead) temp = clean_data(look_ahead,period) score = np.mean(cross_val_score(model,temp.loc[:,["Open","High","Low","Close"]],temp["Target"],cv=tscv)) class="kw">return(score) class="macro">#Define the objective def objective(x,y): class="macro">#Define the output matrix results = np.zeros([x.shape[class="num">0],y.shape[class="num">0]]) class="macro">#Fill in the output matrix for i in np.arange(class="num">0,x.shape[class="num">0]): class="macro">#Select the rows look_ahead = x[i]