使用Python和MQL5开发机器人（第一部分）：数据预处理（基础篇）
「先把行情数据洗成能喂模型的样子」
在 MT5 里直接拿到的 tick 或柱数据，往往带着跳空、重复时间戳和时区偏移，直接丢进 Python 做训练会把噪声当特征。第一步必须做对齐与重采样：把不同品种的同周期 K 线按服务器时间统一索引，缺失段用前收或 NaN 标记，别盲目插值。 一个常见坑是点差与库存费没剥离。若用收盘价做回归目标，得先减去对应点差均值，否则模型学到的是「报价幻觉」而非真实可成交价格。外汇与贵金属杠杆高、滑点大，这类预处理疏忽会显著放大实盘偏差。 下面这段 MQL5 把当前图表最近 500 根日线收拢成数组，供 Python 侧通过 ZeroMQ 或文件管道读取，是数据预处理的起点。复制进 EA 的 OnStart 即可在 MT5 终端跑通验证。
class="type">void OnStart() { class="type">class="kw">double close_array[class="num">500]; class=class="str">"cmt">// 声明存放收盘价的数组，长度500 CopyClose(_Symbol, PERIOD_D1, class="num">0, class="num">500, close_array); class=class="str">"cmt">// 从当前品种日线，偏移0取500根收盘价 for(class="type">int i=class="num">0; i<class="num">500; i++) { Print(close_array[i]); class=class="str">"cmt">// 逐根打印，便于确认数据连续性 } }
算法盘口下机器学习交易系统的开发链路
当前市场已不是人工盯盘能覆盖的场域，公开数据表明超过 95% 的成交额由交易机器人贡献，盘口实质是一场算法博弈。在这种背景下，单纯线性策略容易失效，而机器学习模型虽非强人工智能，却能在非线性噪声中抽出可交易规律。 把神经网络接进交易机器人后，系统可批量扫描大数据、识别隐藏节律并给出走势倾向性判断。需要明确：外汇与贵金属属高杠杆品种，模型输出只是概率优势，不是确定性信号。 本系列会拆开交易机器人的完整开发周期——从数据采集、清洗、样本扩增、特征工程，到模型筛选与训练，再用 Python 落地系统并持续监控实盘。Python 在特征构造与迭代速度上优势明显，但把训练好的模型导成 ONNX 时，必须复用与 Python 端完全一致的特征生成逻辑，这一步极易出错，也是我坚持用 Python 直接做在线交易而非中途转格式的原因。
◍ 把盈利模型拆成可跑的流水线
想让 Python 机器学习模型在外汇和贵金属上持续产出正期望，先把目标拆成一条能落地的流水线：从 MT5 拉数据、做增强扩样本、给数据打标签，再做特征工程（生成、聚类、筛选），然后选模型训练，条件允许就做集成。这条链路里每一步都可在本地复现，外汇与贵金属杠杆高、滑点跳空频繁，任何一步失真都会放大实盘风险。 具体到本文这一截，先只做三件基础事：用 Python 调 MT5 接口收历史 tick 和分时，把价格、成交量等主特征先 load 进来；用重采样或加噪做数据增强，把样本量撑起来；按未来 N 根 K 线的收益符号给样本贴标签。特征工程里的聚类与选择留到下一节，这一节先把原料备齐。 工具面只认两样：Python 接 MT5 的官方库（提速且能直接下数据），以及 sklearn 这类机器学习库。别急着上复杂模型，先把数据管道跑通，你才能在 MT5 里验证后面每一版特征是否真的带来区分度。
「从 MT5 拉历史报价并造特征」
做价格行为建模，第一步是把 MT5 终端里的历史行情抓出来。脚本通过 mt5.initialize() 传入 terminal64.exe 的绝对路径建立连接，再用 mt5.copy_rates_range() 按品种、时间框、起止日期拉取 H1 数据；循环里塞了失败计数器与 1 秒延迟，避免终端没就绪就空跑。 依赖库先用 pip 一次性装齐：numpy、pandas 管数组与表，MetaTrader5 做桥接，datetime 处理日期，concurrent.futures 和 tqdm 留给后续多币种并行，sklearn 里的 train_test_split 管训练切分。装完 import 进环境，全局变量里点差成本 MARKUP 设 0.00001、最大同开 3 单、品种 EURUSD，回测窗口从 2000-01-01 拉到 2024-01-01。 retrieve_data() 是独立函数，连不上就重试最多 300 次，成功拿到 rates 后转成 DataFrame 并 drop 最后一根未闭合 K 线。接着直接算两组特征：10 与 20 周期 SMA、收盘价百分比变化；原文还列了标准偏差、量能变化、日月变动、中位数等，用来量化驱动价的因子。跑通后打印列名与最新行，就能进下一段逻辑。 外汇与贵金属杠杆高、点差滑点随时变，这套环境在真仓前务必用历史数据先验证连通性与特征合理性。
pip install numpy pandas MetaTrader5 concurrent-futures tqdm sklearn matplotlib imblearn class="kw">import numpy as np class="kw">import pandas as pd class="kw">import random from class="type">class="kw">datetime class="kw">import class="type">class="kw">datetime class="kw">import MetaTrader5 as mt5 class="kw">import time class="kw">import concurrent.futures from tqdm class="kw">import tqdm from sklearn.model_selection class="kw">import train_test_split class="kw">import matplotlib.pyplot as plt from sklearn.utils class="kw">import class_weight from imblearn.under_sampling class="kw">import RandomUnderSampler # GLOBALS MARKUP = class="num">0.00001 BACKWARD = class="type">class="kw">datetime(class="num">2000, class="num">1, class="num">1) FORWARD = class="type">class="kw">datetime(class="num">2010, class="num">1, class="num">1) EXAMWARD = class="type">class="kw">datetime(class="num">2024, class="num">1, class="num">1) MAX_OPEN_TRADES = class="num">3 symbol = "EURUSD" def retrieve_data(symbol, retries_limit=class="num">300): terminal_path = "C:/Program Files/MetaTrader class="num">5/Arima/terminal64.exe" attempt = class="num">0 raw_data = None while attempt < retries_limit: if not mt5.initialize(path=terminal_path): print("MetaTrader initialization failed") class="kw">return None instrument_count = mt5.symbols_total() if instrument_count > class="num">0: print(f"Number of instruments in the terminal: {instrument_count}") else: print("No instruments in the terminal") rates = mt5.copy_rates_range(symbol, mt5.TIMEFRAME_H1, BACKWARD, EXAMWARD) mt5.shutdown() if rates is None or len(rates) == class="num">0: print(f"Data for {symbol} not available(attempt {attempt+class="num">1})") attempt += class="num">1 time.sleep(class="num">1) else: raw_data = pd.DataFrame(rates[:-class="num">1], columns=[&class="macro">#x27;time&class="macro">#x27;, &class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;high&class="macro">#x27;, &class="macro">#x27;low&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27;, &class="macro">#x27;tick_volume&class="macro">#x27;]) raw_data[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(raw_data[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;) raw_data.set_index(&class="macro">#x27;time&class="macro">#x27;, inplace=True) break if raw_data is None: print(f"Data retrieval failed after {retries_limit} attempts") class="kw">return None # Add simple features raw_data[&class="macro">#x27;raw_SMA_10&class="macro">#x27;] = raw_data[&class="macro">#x27;close&class="macro">#x27;].rolling(window=class="num">10).mean() raw_data[&class="macro">#x27;raw_SMA_20&class="macro">#x27;] = raw_data[&class="macro">#x27;close&class="macro">#x27;].rolling(window=class="num">20).mean() raw_data[&class="macro">#x27;Price_Change&class="macro">#x27;] = raw_data[&class="macro">#x27;close&class="macro">#x27;].pct_change() * class="num">100 # Additional features
把原始K线展开成多周期特征矩阵
这段处理把单根K线扩成一张特征表，方便后续做价格行为归类。核心思路是用滚动窗口和位移，把「当下」和「过去N根」的关系量化出来。 先看波动与体量：20根收盘滚动标准差直接存进 raw_Std_Dev_Close；tick_volume 的环比变化百分比进 raw_Volume_Change。前1/7/30根收盘价差分别给出日、周、月级别的方向偏移量，对应 Prev_Day/Week/Month_Price_Change。 连涨连跌计数用分组累计实现：Price_Change 大于0时转int后按断点分组 cumsum，得到 Consecutive_Positive_Changes，负值对称处理。Price_Density 取10根收盘的去重个数，密度低说明价格在窄幅重复；Fractal_Analysis 在10根窗口里标1/-1/0，代表窗口内最高/最低/否。 还造了 Price_Volume_Ratio（收报÷成交量）、7与30根中位收盘，以及20根波动率的变异系数 Price_Volatility。最后 print 出尾部100行和列名，fillna 用均值回填后返回。外汇贵金属波动剧烈，这些特征仅描述历史分布，后续信号需结合实时风控验证。
raw_data[&class="macro">#x27;raw_Std_Dev_Close&class="macro">#x27;] = raw_data[&class="macro">#x27;close&class="macro">#x27;].rolling(window=class="num">20).std() raw_data[&class="macro">#x27;raw_Volume_Change&class="macro">#x27;] = raw_data[&class="macro">#x27;tick_volume&class="macro">#x27;].pct_change() * class="num">100 raw_data[&class="macro">#x27;raw_Prev_Day_Price_Change&class="macro">#x27;] = raw_data[&class="macro">#x27;close&class="macro">#x27;] - raw_data[&class="macro">#x27;close&class="macro">#x27;].shift(class="num">1) raw_data[&class="macro">#x27;raw_Prev_Week_Price_Change&class="macro">#x27;] = raw_data[&class="macro">#x27;close&class="macro">#x27;] - raw_data[&class="macro">#x27;close&class="macro">#x27;].shift(class="num">7) raw_data[&class="macro">#x27;raw_Prev_Month_Price_Change&class="macro">#x27;] = raw_data[&class="macro">#x27;close&class="macro">#x27;] - raw_data[&class="macro">#x27;close&class="macro">#x27;].shift(class="num">30) raw_data[&class="macro">#x27;Consecutive_Positive_Changes&class="macro">#x27;] = (raw_data[&class="macro">#x27;Price_Change&class="macro">#x27;] > class="num">0).astype(class="type">int).groupby((raw_data[&class="macro">#x27;Price_Change&class="macro">#x27;] > class="num">0).astype(class="type">int).diff().ne(class="num">0).cumsum()).cumsum() raw_data[&class="macro">#x27;Consecutive_Negative_Changes&class="macro">#x27;] = (raw_data[&class="macro">#x27;Price_Change&class="macro">#x27;] < class="num">0).astype(class="type">int).groupby((raw_data[&class="macro">#x27;Price_Change&class="macro">#x27;] < class="num">0).astype(class="type">int).diff().ne(class="num">0).cumsum()).cumsum() raw_data[&class="macro">#x27;Price_Density&class="macro">#x27;] = raw_data[&class="macro">#x27;close&class="macro">#x27;].rolling(window=class="num">10).apply(lambda x: len(set(x))) raw_data[&class="macro">#x27;Fractal_Analysis&class="macro">#x27;] = raw_data[&class="macro">#x27;close&class="macro">#x27;].rolling(window=class="num">10).apply(lambda x: class="num">1 if x.idxmax() else (-class="num">1 if x.idxmin() else class="num">0)) raw_data[&class="macro">#x27;Price_Volume_Ratio&class="macro">#x27;] = raw_data[&class="macro">#x27;close&class="macro">#x27;] / raw_data[&class="macro">#x27;tick_volume&class="macro">#x27;] raw_data[&class="macro">#x27;Median_Close_7&class="macro">#x27;] = raw_data[&class="macro">#x27;close&class="macro">#x27;].rolling(window=class="num">7).median() raw_data[&class="macro">#x27;Median_Close_30&class="macro">#x27;] = raw_data[&class="macro">#x27;close&class="macro">#x27;].rolling(window=class="num">30).median() raw_data[&class="macro">#x27;Price_Volatility&class="macro">#x27;] = raw_data[&class="macro">#x27;close&class="macro">#x27;].rolling(window=class="num">20).std() / raw_data[&class="macro">#x27;close&class="macro">#x27;].rolling(window=class="num">20).mean() print("\nOriginal columns:") print(raw_data[[&class="macro">#x27;close&class="macro">#x27;, &class="macro">#x27;high&class="macro">#x27;, &class="macro">#x27;low&class="macro">#x27;, &class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;tick_volume&class="macro">#x27;]].tail(class="num">100)) print("\nList of features:") print(raw_data.columns.tolist()) print("\nLast class="num">100 features:") print(raw_data.tail(class="num">100)) # Replace NaN values with the mean raw_data.fillna(raw_data.mean(), inplace=True) class="kw">return raw_data retrieve_data(symbol)