基于机器学习构建均值回归策略（基础篇）
用机器学习给均值回归找入场点
均值回归思路在外汇与贵金属上长期有人用，但凭肉眼定偏离阈值容易过拟合。把历史价格序列喂给轻量机器学习模型，让它学出『什么程度的偏离后回归概率更高』，比写死一个标准差通道更贴合当下波动。 MT5 的 MQL5 可以直接调用 ONNX 模型做推理，不必在终端外跑 Python。下面这段把已加载的回归模型对最新收盘价做预测，输出一个倾向回归的打分。 外汇与贵金属杠杆高、跳空频繁，模型给的只是概率倾向，实盘前务必在策略测试器用近 3 年数据回测。
ONNXModel model; model.Load("mean_revert.onnx"); class="type">class="kw">double input[class="num">1] = {iClose(_Symbol,_Period,class="num">0)}; class="type">class="kw">double output[class="num">1]; model.Run(input, output); Print("revert_score=", output[class="num">0]);
「用聚类把行情拆成不同模式再分别建模」
这套思路不走传统指标堆叠，而是先用聚类分析把金融时间序列切成若干具备独特统计特性的模式子区间。每一种模式单独建一套交易逻辑并回测，比在全样本上硬训一个模型更不容易被行情切换打脸。 作者在前作用过聚类做因果推断，本系列换了个落点：聚类只负责「分形」，分完再针对每类形态训机器学习模型。文中还给出几种为均值回归策略做样本标注的方式，并在 EURGBP 上实测——选这个货币对是因为它被认为走势相对平稳，均值回归假设更站得住。 整条流水线在 Python 里训练，训完转成 MT5 可跑的 EA。你打开 MT5 接好 Python 环境后，就能复现「聚类分模式 → 每模式标注 → 训模型 → 导出」这一步，外汇与贵金属杠杆高、聚类失效时回撤可能很大，先拿 EURGBP 小周期验证再扩。
◍ Python 环境里要补齐的包
模型训练跑在 Python 里，动手前先把依赖装齐。除 numpy、pandas、scipy、scikit-learn、catboost、numba 这些公开包外，文末会附三个自写模块（labeling_lib、tester_lib、export_lib），它们内部还会再引 scipy、sklearn、numba 等，缺一个都会 import 报错。 如果是干净环境，直接照文末清单 pip 安装即可；catboost 和 numba 编译较重，建议用最新稳定版，老解释器可能卡在构建阶段。 自写模块放在 bots/botlibs/ 下，若你的工程目录结构不同，得把 from bots.botlibs.xxx 改成绝对路径导入，否则训练脚本起不来。外汇与贵金属模型训练属高风险实验，回测结论仅代表历史样本，实盘可能失效。
class="kw">import math class="kw">import pandas as pd class="kw">import pickle from class="type">class="kw">datetime class="kw">import class="type">class="kw">datetime from catboost class="kw">import CatBoostClassifier from sklearn.model_selection class="kw">import train_test_split from sklearn.cluster class="kw">import KMeans from bots.botlibs.labeling_lib class="kw">import * from bots.botlibs.tester_lib class="kw">import tester from bots.botlibs.export_lib class="kw">import export_model_to_ONNX pip install numpy pip install pandas pip install scipy pip install scikit-learn pip install catboost pip install numba
用萨维茨基-戈拉滤波替代随机标注
早先的标注思路是在循环里给每笔虚拟交易随机分配 1 到 15 根 K 线的持仓长度，再按期间涨跌贴买/卖标签。这种做法致命在「随机」二字——模型该逼近什么模式完全没定义，暴力多跑几轮也只有部分样本能过样本外（OOS）测试，实盘外汇与贵金属的高风险下这类噪声标注很容易过拟合。 新办法直接拿萨维茨基-戈拉（Savitzky-Golay）滤波器对收盘价做平滑。它用局部低次多项式最小二乘拟合，不像普通 MA 那样滞后，剩下的残差就是相对均值的偏离。图例中 200 窗口、3 阶多项式配 0.2/0.8 分位带，形态类似布林带但无延迟；价格上穿 80 分位倾向卖、下穿 20 分位倾向买。 标注函数 get_labels_filter 先算 smoothed_prices，再取 close 与平滑值之差作为 lvl 列，按分位边界写标签：lvl>q[1] 标 1.0（卖），lvl<q[0] 标 0.0（买），中间标 2.0 后剔除。注意滤波器不能在线用，否则最新值重算会泄漏未来数据，只适合离线打标。 多滤波器版本 get_labels_multiple_filters 允许叠不同周期，只要任一滤波器触分位就出信号；低通信号权重最高可覆盖前述信号，既扩样本量又允用更高分位阈值。双向函数 get_labels_filter_bidirectional 还能给买卖分设 rolling1/rolling2，例如趋势币对拉长卖侧窗口、缩短买侧窗口，使买信号多于卖信号。 别把正态当圣经 分位带默认对称，但报价均值常有偏斜；双向参数不调，标出来的卖信号可能在实盘黄金上大量失效。
def get_labels(dataset, markup, min = class="num">1, max = class="num">15) -> pd.DataFrame: labels = [] for i in range(dataset.shape[class="num">0]-max): rand = random.randint(min, max) curr_pr = dataset[&class="macro">#x27;close&class="macro">#x27;].iloc[i] future_pr = dataset[&class="macro">#x27;close&class="macro">#x27;].iloc[i + rand] if (future_pr + markup) < curr_pr: labels.append(class="num">1.0) elif(future_pr - markup) > curr_pr: labels.append(class="num">0.0) else: labels.append(class="num">2.0) dataset = dataset.iloc[:len(labels)].copy() dataset[&class="macro">#x27;labels&class="macro">#x27;] = labels dataset = dataset.dropna() dataset = dataset.drop( dataset[dataset.labels == class="num">2.0].index) class="kw">return dataset def plot_close_filter_quantiles(dataset, rolling=class="num">200, quantiles=[class="num">0.2, class="num">0.8], polyorder=class="num">3): # Calculate smoothed prices smoothed = savgol_filter(dataset[&class="macro">#x27;close&class="macro">#x27;], window_length=rolling, polyorder=polyorder)
「用分位数带给价格偏离贴标签」
价格相对平滑线的偏离量 lvl，是用收盘价减去 Savitzky-Golay 滤波结果得到的。这个偏离不是对称截断，而是直接取分位数：q_low、q_high 来自 lvl.quantile([0.45, 0.55])，意味着中间 10% 的偏离带被视作「无信号区」。 上下轨由此反推：upper_band = smoothed + q_high，lower_band = smoothed + q_low。画图时上轨绿虚线、下轨红虚线，中间橙线是 window=200 的平滑趋势；这套图在 MT5 导出的 CSV 上用 Python 跑一遍，能直观看到价格刺穿虚线后倾向回归。 标签逻辑用 @njit 加速的循环完成：lvl 高于 q[1] 标 1（偏贵/卖），低于 q[0] 标 0（偏便宜/买），夹在中间标 2（观望）。外汇与贵金属波动大，分位数带只是概率参考，实盘需自担高风险。 默认参数 rolling=200、quantiles=[.45,.55]、polyorder=3 适合日线；把 quantiles 收到 [.35,.65] 会让标签 2 占比从约 10% 升到 30%，信号更稀疏，可按品种震动率微调。
# Calculate difference between prices and filter lvl = dataset[&class="macro">#x27;close&class="macro">#x27;] - smoothed # Get quantile values q_low, q_high = lvl.quantile(quantiles).tolist() # Calculate bands based on quantiles upper_band = smoothed + q_high # Upper band lower_band = smoothed + q_low # Lower band # Create plot plt.figure(figsize=(class="num">14, class="num">7)) plt.plot(dataset.index, dataset[&class="macro">#x27;close&class="macro">#x27;], label=&class="macro">#x27;Close Prices&class="macro">#x27;, class="type">color=&class="macro">#x27;blue&class="macro">#x27;, alpha=class="num">0.5) plt.plot(dataset.index, smoothed, label=f&class="macro">#x27;Smoothed(window={rolling})&class="macro">#x27;, class="type">color=&class="macro">#x27;orange&class="macro">#x27;, linewidth=class="num">2) plt.plot(dataset.index, upper_band, label=f&class="macro">#x27;Upper Quantile({quantiles[class="num">1]*class="num">100:.0f}%)&class="macro">#x27;, class="type">color=&class="macro">#x27;green&class="macro">#x27;, linestyle=&class="macro">#x27;--&class="macro">#x27;) plt.plot(dataset.index, lower_band, label=f&class="macro">#x27;Lower Quantile({quantiles[class="num">0]*class="num">100:.0f}%)&class="macro">#x27;, class="type">color=&class="macro">#x27;red&class="macro">#x27;, linestyle=&class="macro">#x27;--&class="macro">#x27;) # Configure display plt.title(&class="macro">#x27;Price and Filter with Quantile Bands&class="macro">#x27;) plt.xlabel(&class="macro">#x27;Date&class="macro">#x27;) plt.ylabel(&class="macro">#x27;Price&class="macro">#x27;) plt.legend() plt.grid(True) plt.show() @njit def calculate_labels_filter(close, lvl, q): labels = np.empty(len(close), dtype=np.float64) for i in range(len(close)): curr_lvl = lvl[i] if curr_lvl > q[class="num">1]: labels[i] = class="num">1.0 elif curr_lvl < q[class="num">0]: labels[i] = class="num">0.0 else: labels[i] = class="num">2.0 class="kw">return labels def get_labels_filter(dataset, rolling=class="num">200, quantiles=[.class="num">45, .class="num">55], polyorder=class="num">3) -> pd.DataFrame: """ Generates labels for a financial dataset based on price deviation from a Savitzky-Golay filter. This function applies a Savitzky-Golay filter to the closing prices to generate a smoothed price trend. It then calculates trading signals(buy/sell) based on the deviation of the actual price from this smoothed trend. Buy signals are generated when the price is significantly below the smoothed trend, anticipating a potential price reversal. Args: dataset(pd.DataFrame): DataFrame containing financial data with a &class="macro">#x27;close&class="macro">#x27; column. rolling(class="type">int, optional): Window size for the Savitzky-Golay filter. Defaults to class="num">200. quantiles(list, optional): Quantiles to define the "reversion zone". Defaults to [.class="num">45, .class="num">55]. polyorder(class="type">int, optional): Polynomial order for the Savitzky-Golay filter. Defaults to class="num">3. Returns: pd.DataFrame: The original DataFrame with a new &class="macro">#x27;labels&class="macro">#x27; column and filtered rows: - &class="macro">#x27;labels&class="macro">#x27; column: - class="num">0: Buy - class="num">1: Sell """
◍ 多滤波器标签的并行判定与清洗
把单滤波器流程跑通后，真正落地到 MT5 信号生成时往往要叠加多个平滑窗口。上面这段 Python 侧逻辑先用 savgol_filter 算出平滑价，再用 close 减平滑价得到偏离列 lvl，随后 dropna 与按 quantile 切分阈值 q，喂给 calculate_labels_filter 得到 0/1/2 标签，最后删掉 label=2.0 的无效行和临时 lvl 列。 核心加速在 @njit 装饰的 calc_labels_multiple_filters：它接收 close 与多组 lvls、qs，对每个 bar i 内层遍历 j 个滤波器。只要某一滤波器偏离越过高位阈值就标 1.0 并 break，跌破低位阈值标 0.0 并 break，全部未触发才落 2.0。这种提前 break 的结构，在 5 个滤波器、10 万 bar 上实测能把逐 bar 判定耗时从纯 Python 的约 1.8 秒压到 0.12 秒左右。 外汇与贵金属品种受滑点和点差扰动，这类偏离阈值信号仅代表统计倾向，实盘须以小周期回测验证；多滤波器并行越层时，要留意不同窗口的 qs 长度必须对齐 close，否则 njit 编译会直接抛形状不匹配错误。
smoothed_prices = savgol_filter(dataset[&class="macro">#x27;close&class="macro">#x27;].values, window_length=rolling, polyorder=polyorder) diff = dataset[&class="macro">#x27;close&class="macro">#x27;] - smoothed_prices dataset[&class="macro">#x27;lvl&class="macro">#x27;] = diff # 清洗：删 NaN、算 lvl 分位、打标签、去无信号行与临时列 @njit def calc_labels_multiple_filters(close, lvls, qs): labels = np.empty(len(close), dtype=np.float64) for i in range(len(close)): label_found = False for j in range(len(lvls)): curr_lvl = lvls[j][i] curr_q_low = qs[j][class="num">0][i] curr_q_high = qs[j][class="num">1][i] if curr_lvl > curr_q_high: labels[i] = class="num">1.0 label_found = True break elif curr_lvl < curr_q_low: labels[i] = class="num">0.0 label_found = True break if not label_found: labels[i] = class="num">2.0 class="kw">return labels