使用 Python 和 MetaTrader5 python 软件包及 ONNX 模型文件进行深度学习预测和排序·进阶篇
搭好 Python 端的行情与建模环境
在 MT5 终端之外做行情挖掘和模型训练，第一步是把通信与计算依赖拉进 Python 进程。下面这段导入语句就是给后续拉取外汇、贵金属报价和跑机器学习模型铺路。 import MetaTrader5 as mt5 建立与本地 MT5 终端的通信句柄；from MetaTrader5 import * 把账户、订单、历史报价等接口全量引入。通配符虽省事，但和其他库同名函数撞车时排查成本很高，实盘脚本里更推荐显式引用。 数值层用 import numpy as np，后续处理 OHLC 数组和指标矩阵都靠它。机器学习侧从 sklearn 取了 train_test_split、cross_val_score、StandardScaler 以及回归常用的均方误差、平均绝对误差、R2 三个评估量；深度学习走 TensorFlow 的 Sequential 堆叠 Dense 层，并用 l2 做权重衰减抑制过拟合。 KFold 单独引入，说明后面倾向用交叉验证看模型在 EURUSD 或 XAUUSD 这类高波动品种上的稳定度。外汇与贵金属杠杆高、跳空频繁，任何历史回测给出的预测力都只是概率倾向，实盘前务必在 MT5 用策略测试器交叉核对。 代码里 MetaTrader5 与 sklearn 的导入出现重复，直接删掉第二遍即可，解释器不会报错但会显得环境没整理干净。
class="kw">import MetaTrader5 as mt5 from MetaTrader5 class="kw">import * class="kw">import numpy as np from sklearn.model_selection class="kw">import train_test_split, cross_val_score from sklearn.preprocessing class="kw">import StandardScaler from sklearn.metrics class="kw">import mean_squared_error, mean_absolute_error, r2_score from tensorflow.keras.models class="kw">import Sequential from tensorflow.keras.layers class="kw">import Dense from tensorflow.keras.regularizers class="kw">import l2 from sklearn.model_selection class="kw">import KFold
◍ 从 MT5 拉分时报价进 Pandas
要把 MT5 的分时报价喂给后续的 Python 分析，第一步是带着账号信息把终端拉起来并连上服务器。下面这段代码用 mt5.initialize 读入路径、登录名、密码和服务器，连不上会直接打印 last_error，方便你当场排错。 [CODE] # You will need to update the values for path, login, pass, and server according to your specific case. creds = { "path": "C:/Program Files/XXX MT5/terminal64.exe", "login": account, "pass": clave_secreta, "server": server_account, "timeout": 60000, "portable": False } # We launch the MT5 platform and connect to the server with our username and password. if mt5.initialize(path=creds['path'], login=creds['login'], password=creds['pass'], server=creds['server'], timeout=creds['timeout'], portable=creds['portable']): print("Plataform MT5 launched correctly") else: print(f"There has been a problem with initialization: {mt5.last_error()}") rates = rates.drop(index=rates.index) rates = mt5.copy_ticks_from(symbol, utc_from, 1000000000, mt5.COPY_TICKS_ALL) rates_frame=pd.DataFrame() # Empty DataFrame rates_frame = rates_frame.drop(index=rates_frame.index) rates_frame = pd.DataFrame(rates) rates_frame['time']=pd.to_datetime(rates_frame['time'], unit='s') rates_frame['close']=(rates_frame['ask']+rates_frame['bid'])/2 [/CODE] 代码逐行拆解：creds 字典里 path 填你本机终端 exe 路径，login/pass/server 换成自己账户，timeout 给 60000 毫秒算宽裕。mt5.initialize 成功返回 True 并打印启动正常，失败则抛出 last_error 内容。 rates.drop 先清空旧数据框，copy_ticks_from 从 utc_from 时刻起最多拉 1000000000 条 tick，COPY_TICKS_ALL 表示买卖价全要。rates_frame 先建空 DataFrame 再清一遍，然后用 rates 填充。 time 列由秒级时间戳转成 datetime，close 列取 ask 与 bid 的均值——这一步把双向报价压成单值，后续做深度学习输入时维度更干净。外汇和贵金属杠杆高，tick 级数据噪声大，均值化处理只是降维起点，不代表任何方向判断。
# You will need to update the values for path, login, pass, and server according to your specific case. creds = { "path": "C:/Program Files/XXX MT5/terminal64.exe", "login": account, "pass": clave_secreta, "server": server_account, "timeout": class="num">60000, "portable": False } # We launch the MT5 platform and connect to the server with our username and password. if mt5.initialize(path=creds[&class="macro">#x27;path&class="macro">#x27;], login=creds[&class="macro">#x27;login&class="macro">#x27;], password=creds[&class="macro">#x27;pass&class="macro">#x27;], server=creds[&class="macro">#x27;server&class="macro">#x27;], timeout=creds[&class="macro">#x27;timeout&class="macro">#x27;], portable=creds[&class="macro">#x27;portable&class="macro">#x27;]): print("Plataform MT5 launched correctly") else: print(f"There has been a problem with initialization: {mt5.last_error()}") rates = rates.drop(index=rates.index) rates = mt5.copy_ticks_from(symbol, utc_from, class="num">1000000000, mt5.COPY_TICKS_ALL) rates_frame=pd.DataFrame() # Empty DataFrame rates_frame = rates_frame.drop(index=rates_frame.index) rates_frame = pd.DataFrame(rates) rates_frame[&class="macro">#x27;time&class="macro">#x27;]=pd.to_datetime(rates_frame[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;) rates_frame[&class="macro">#x27;close&class="macro">#x27;]=(rates_frame[&class="macro">#x27;ask&class="macro">#x27;]+rates_frame[&class="macro">#x27;bid&class="macro">#x27;])/class="num">2
「原始行情与加工指标的取舍」
把 OHLC 原始报价直接喂给模型，好处是没动过刀：所有波动细节都在，模型自己从裸数据里找模式，不预设市场该长什么样。代价也实在——tick 级或分钟级原始流里混着大量无意义的杂讯，不先做去均值或标准化，网络大概率在前几层就学歪。 另一路是先用振荡器或技术指标做特征工程，再把加工后的序列送进去。RSI、MACD、ATR 这类量把「超买」「动量」「波动区间」直接结构化成一条线，模型输入维度更干净，收敛往往更快。但这一步你已经替模型做了假设：你选的指标的窗口参数、平滑方式，决定了它能看见什么、看不见什么。 实盘里常见翻车点是：拿 14 周期 RSI 训出来的网络，换到 5 分钟黄金和日线欧美两套品种上，有效性差异极大——因为指标本身对波动尺度敏感。原始数据路线容错高但吃算力，衍生指标路线省事却把偏见焊死在输入端，两条路没有免费午餐。
喂给神经网络的行情数据该怎么收拾
做 MT5 上的行情序列建模，第一步是把不同量纲的特征压到同一区间。最常用的是 min-max 归一化，把每个维度线性映射到 0~1，避免价格绝对值大的列在梯度更新里压过波动率类特征，训练收敛明显更稳。 序列长度直接决定模型能「看多远」。太短漏掉趋势结构，太长既吃算力又容易把噪声当规律。实盘里 30~120 根 bar 的窗口比较常见，具体要在回测里用不同长度跑同一架构，看验证集误差拐点落在哪。 时间依赖别只堆原始 OHLC。延迟值、滑动均值、滚动波动率这类衍生特征，往往比裸价格更能暴露节奏。特征子集也要做 ablation：每次拿掉一组，看哪组对任务精度贡献真大，而不是凭直觉全塞进去。 架构上，RNN / LSTM 对连续行情比普通前馈网更顺手，但高维输入必须上 dropout 类正则化扛过拟合。超参数（隐藏层维度、学习率、dropout 率）没有通用解，网格或随机搜一轮再定。外汇和贵金属波动受事件驱动，市场状态会漂移，隔几周重测一次特征与模型配置才算靠谱。
◍ 给收盘价加时间偏移做预测目标
把行情塞进深度学习之前，得先解决一个关键问题：模型要学的不是「现在」的值，而是「若干秒之后」的值。MQL5 里讲神经网络的文章已经不少，这里只说落地时绕不开的一步——构造带时间偏移的标签列。 在时间序列里，RNN、LSTM 这类结构靠前后顺序抓依赖。把 close 列整体往前挪 seconds 秒，每一行就变成「过去观测 → 未来目标」的一对样本；输入和输出在时间轴上对齐，模型才学得出来历史背景到预测的映射。 代码里 number_of_rows 直接等于秒数，先造一个全 NaN 的空 DataFrame 接在原表后面，保证索引连续不断层；再用 shift(-seconds) 把 close 负向平移，生成 target 列。最后 dropna() 清掉末尾因平移产生的空行，留下的就是能直接喂给 TensorFlow 的干净表。 在 MT5 导出秒级 CSV 后，用这段逻辑试 seconds=300：你会看到原表行数增加了 300 行空记录，target 列前段等于 5 分钟后的收盘价，末段 300 行全空被 drop 掉。外汇与贵金属秒级数据噪声大、滑点高，实盘使用前先在历史段回测偏移窗口的稳定性。
number_of_rows= seconds empty_rows = pd.DataFrame(np.nan, index=range(number_of_rows), columns=df.columns) df = df._append(empty_rows, ignore_index=True) df[&class="macro">#x27;target&class="macro">#x27;] = df[&class="macro">#x27;close&class="macro">#x27;].shift(-seconds) print("df modified",df) # Drop NaN values df=df.dropna()
「用 Keras 搭一个收盘价回归前馈网」
想在 MT5 导出的历史数据上做一步预测，可以用 TensorFlow/Keras 跑一个最朴素的前馈网络。上面这段脚本把收盘价当作唯一特征，目标变量是平移后生成的 target，按 8:2 切分且不打乱时序，避免未来信息泄入训练集。 网络结构是 Sequential 线性堆叠：首层 Dense(128, relu, input_shape=(1,)) 并带 L2 正则（系数由 k_reg 控制），随后 256、128、64 三个 relu 隐藏层，最后一层单个线性神经元做回归输出。若是分类任务，末层才会换成 sigmoid 或 softmax。 特征先用 StandardScaler 做 z-score 标准化，训练用 adam 优化器加均方误差损失，batch_size=256、验证集占 0.2。脚本尾部取 df2 末尾 segundos 根 K 线，过同一 scaler 后送 model.predict，直接打印最近 1 根实际收盘价与接下来若干根预测值——你在本地把 epoch 和 k_reg 调小一点，几分钟就能在 CPU 上复现。 外汇与贵金属杠杆高、滑点跳空频繁，这类单变量神经网络仅给出概率倾向，实盘前务必用样本外数据验证稳定性。
# Split the data into features(X) and target variable(y) X=[] y=[] X = df2[[&class="macro">#x27;close&class="macro">#x27;]] y = df2[&class="macro">#x27;target&class="macro">#x27;] # Split the data into training and testing sets X_train=[] X_test=[] y_train=[] y_test=[] X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=class="num">0.2, shuffle=False) # Standardize the features X_train_scaled=[] X_test_scaled=[] scaler = StandardScaler() X_train_scaled = scaler.fit_transform(X_train) X_test_scaled = scaler.transform(X_test) # Build a neural network model model=None model = Sequential() model.add(Dense(class="num">128, activation=&class="macro">#x27;relu&class="macro">#x27;, input_shape=(X_train.shape[class="num">1],), kernel_regularizer=l2(k_reg))) model.add(Dense(class="num">256, activation=&class="macro">#x27;relu&class="macro">#x27;, kernel_regularizer=l2(k_reg))) model.add(Dense(class="num">128, activation=&class="macro">#x27;relu&class="macro">#x27;, kernel_regularizer=l2(k_reg))) model.add(Dense(class="num">64, activation=&class="macro">#x27;relu&class="macro">#x27;, kernel_regularizer=l2(k_reg))) model.add(Dense(class="num">1, activation=&class="macro">#x27;linear&class="macro">#x27;)) # Compile the model model.compile(optimizer=&class="macro">#x27;adam&class="macro">#x27;, loss=&class="macro">#x27;mean_squared_error&class="macro">#x27;) # Train the model model.fit(X_train_scaled, y_train, epochs=class="type">int(epoch), batch_size=class="num">256, validation_split=class="num">0.2, verbose=class="num">1) # Use the model to predict the next class="num">4 instances X_predict=[] X_predict_scaled=[] predictions = pd.DataFrame() predictions=[] # Empty DataFrame class="macro">#predictions = predictions.drop(index=predictions.index) X_predict = df2.tail(segundos)[[&class="macro">#x27;close&class="macro">#x27;]] X_predict_scaled = scaler.transform(X_predict) predictions = model.predict(X_predict_scaled) # Print actual and predicted values for the next n instances print("Actual Value for the Last Instances:") print(df.tail(class="num">1)[&class="macro">#x27;close&class="macro">#x27;].values) print("\nPredicted Value for the Next Instances:") print(predictions[:, class="num">0]) predictions=pd.DataFrame(predictions)