将ML模型与策略测试器集成（结论）：实现价格预测的回归模型·进阶篇
「用决策树回归给 MACD 配一个价格预期模型」
这段脚本把 MACD 线、信号线与 30 周期 EMA 作为特征，用决策树回归器去拟合收盘价。特征先经 MinMaxScaler 归一化，再按 8:2 不洗牌切分训练集与测试集，避免未来信息泄漏。 基础模型用 DecisionTreeRegressor(random_state=42) 跑完，会打印 MAE、MSE、RMSE 与 R² 四项指标；外汇与贵金属波动大，这类回归结果只反映样本内拟合倾向，实盘高风险，别直接当入场信号。 接着用 GridSearchCV 在 max_depth 3~8、min_samples_split 2~4、min_samples_leaf 1~3 的网格里做 5 折交叉验证，以负 MSE 为评分选最优超参。优化后同四项指标会再打印一次，通常 RMSE 与 R² 会比基础模型更稳。 最后 denormalize_price 函数借原 scaler 把归一化预测还原成价格量纲，方便和 y_test_denorm 对照画图验证偏差。开 MT5 导出的 CSV 跑一遍，重点看 R² 是否随深度增加而过拟合。
class="kw">return macd_line, signal_line # Function to calculate EMA def ema(df, period=class="num">30): class="kw">return df[&class="macro">#x27;close&class="macro">#x27;].ewm(span=period).mean() # Calculating MACD and the signal line smoothed_df[&class="macro">#x27;macd&class="macro">#x27;], smoothed_df[&class="macro">#x27;signal&class="macro">#x27;] = macd(smoothed_df) # Calculating EMA smoothed_df[&class="macro">#x27;ema&class="macro">#x27;] = ema(smoothed_df) # Selecting the variables selected_df = smoothed_df[[&class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;macd&class="macro">#x27;, &class="macro">#x27;ema&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27;]].dropna() scaler = MinMaxScaler() normalized_df = pd.DataFrame(scaler.fit_transform(selected_df), columns=selected_df.columns, index=selected_df.index) X = normalized_df[[&class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;macd&class="macro">#x27;, &class="macro">#x27;ema&class="macro">#x27;]] y = normalized_df[&class="macro">#x27;close&class="macro">#x27;] X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=class="num">0.2, shuffle=False) from sklearn.tree class="kw">import DecisionTreeRegressor from sklearn.metrics class="kw">import mean_squared_error, r2_score regressor = DecisionTreeRegressor(random_state=class="num">42) regressor.fit(X_train, y_train) y_pred = regressor.predict(X_test) from sklearn.metrics class="kw">import mean_absolute_error, mean_squared_error, r2_score mae = mean_absolute_error(y_test, y_pred) mse = mean_squared_error(y_test, y_pred) rmse = np.sqrt(mse) r2 = r2_score(y_test, y_pred) print(f"MAE: {mae:.4f}") print(f"MSE: {mse:.4f}") print(f"RMSE: {rmse:.4f}") print(f"R²: {r2:.4f}") from sklearn.model_selection class="kw">import GridSearchCV from sklearn.tree class="kw">import DecisionTreeRegressor regressor = DecisionTreeRegressor(random_state=class="num">42) param_grid = { &class="macro">#x27;max_depth&class="macro">#x27;: [class="num">3, class="num">4, class="num">5, class="num">6, class="num">7, class="num">8], &class="macro">#x27;min_samples_split&class="macro">#x27;: [class="num">2, class="num">3, class="num">4], &class="macro">#x27;min_samples_leaf&class="macro">#x27;: [class="num">1, class="num">2, class="num">3] } grid_search = GridSearchCV(estimator=regressor, param_grid=param_grid, scoring=&class="macro">#x27;neg_mean_squared_error&class="macro">#x27;, cv=class="num">5, n_jobs=-class="num">1) grid_search.fit(X_train, y_train) best_params = grid_search.best_params_ print(f"Melhores hiperparâmetros: {best_params}") best_regressor = DecisionTreeRegressor(**best_params, random_state=class="num">42) best_regressor.fit(X_train, y_train) y_pred_optimized = best_regressor.predict(X_test) mae_optimized = mean_absolute_error(y_test, y_pred_optimized) mse_optimized = mean_squared_error(y_test, y_pred_optimized) rmse_optimized = np.sqrt(mse_optimized) r2_optimized = r2_score(y_test, y_pred_optimized) print(f"MAE otimizado: {mae_optimized:.4f}") print(f"MSE otimizado: {mse_optimized:.4f}") print(f"RMSE otimizado: {rmse_optimized:.4f}") print(f"R² otimizado: {r2_optimized:.4f}") class="kw">import matplotlib.pyplot as plt # Function for denormalizing prices def denormalize_price(scaler, normalized_price, column_name): dummy_df = pd.DataFrame(np.zeros((len(normalized_price), len(selected_df.columns))), columns=selected_df.columns) dummy_df[column_name] = normalized_price denormalized_df = scaler.inverse_transform(dummy_df) class="kw">return denormalized_df[:, selected_df.columns.get_loc(column_name)] # Denormalize actual and forecast prices y_test_denorm = denormalize_price(scaler, y_test, &class="macro">#x27;close&class="macro">#x27;) y_pred_optimized_denorm
用单例文件类打通 Python 与 MT5 测试器
把 Python 训练的回归模型放进 MT5 策略测试器，核心不是模型本身，而是两边怎么交换数据。原文方案是建一个 Python 端的 File 类，通过 CSV 文件做桥梁：测试器一启动就把品种价格写进文件，Python 轮询读取后做标准化、去异常值、造附加特征，再喂给已加载的回归模型出预测。 这个类被设计成 Singleton（元类控制），保证整个进程只有一个实例，避免多个句柄同时读写同一份 CSV 把通信搞乱。__init_file 方法用 Path.is_file() 探文件，不在就 sleep(1) 返回 False，等于给了 1 秒宽限期等 MT5 侧先落盘。 异常处理只认三类：EPERM / EACCES 映射成 PermissionError，ENOENT 映射成 FileNotFoundError，其余 errno 走 Unknown error 分支打印。check_init_param 会死循环重试，直到读到文件并取出 DataFrame 里 typerun 列的第 0 行值——这个返回值通常用来告诉 Python 当前是训练态还是预测态。 数据预处理阶段，标准化把特征压到 0~1 区间，让价格和成交量同尺度；金融序列里的跳空或错价会被检测剔除，否则容易带偏回归。附加特征如移动平均也在这步拼进去。只有 DataFrame 干净了才送进基于决策树的回归模型。 等模型在 MT5 可控环境里反复跑稳了，再导成 ONNX 格式做原生集成，能少踩直接写 MQL5 模型的坑。外汇和贵金属波动大、杠杆高，这套验证流程只是降低出错概率，不保证实盘收益。
class Singleton(type): _instances = {} def __call__(cls, *args, **kwargs): if cls not in cls._instances: cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs) class="kw">return cls._instances[cls] from pathlib class="kw">import Path from time class="kw">import sleep from typing class="kw">import Tuple class="kw">import pandas as pd from pandas.core.frame class="kw">import DataFrame class="kw">import os from errno class="kw">import EACCES, EPERM, ENOENT class="kw">import sys class Singleton(type): _instances = {} def __call__(cls, *args, **kwargs): if cls not in cls._instances: cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs) class="kw">return cls._instances[cls] CSV_SEPARATOR = &class="macro">#x27;;&class="macro">#x27; class File(metaclass=Singleton): def __init__(self) -> None: pass def __init_file(self, name_arq: str) -> class="type">bool: class="kw">return Path(name_arq).is_file() or sleep(class="num">1) or False def __handle_error(self, e, name_arq: str): ERRORS = { EPERM: "PermissionError", EACCES: "PermissionError", ENOENT: "FileNotFoundError" } print(f"{ERRORS.get(e.errno, &class="macro">#x27;Unknown error&class="macro">#x27;)} error({e.errno}): {e.strerror} for:\n{name_arq}") def check_init_param(self, name_arq : str) -> Tuple[str]: class="kw">while True: try: if self.__init_file(name_arq): df = pd.read_csv(name_arq, sep=CSV_SEPARATOR) class="kw">return (df.typerun.values[class="num">0]) except(IOError, OSError) as e: self.__handle_error(e, name_arq) except: print(&class="macro">#x27;Unexpected error:&class="macro">#x27;, sys.exc_info()[class="num">0])
◍ 用 Python 封装 CSV 读写与容错
把 MT5 导出的复盘数据落盘，最怕文件被占用或路径不存在。下面这段 Python 类方法用无限重试绕开临时 IO 错误，直到 __init_file 确认文件可用才用 pd.read_csv 按自定义分隔符读入。
check_open_file 里 try/except (IOError, OSError) 只捕获系统级异常，其余未知错误走 sys.exc_info()[0] 打印，避免脚本静默死循环。
写盘分两层：save_file_csv 直接吐 DataFrame，save 则裸写字符串；删文件用 os.remove 包空 except，文件不存在也不报错。实盘外接小布分析脚本时，这种封装能省掉一半路径调试时间。
def check_open_file(self, name_arq: str) -> pd.DataFrame(): class="kw">while True: try: if self.__init_file(name_arq): class="kw">return pd.read_csv(name_arq, sep=CSV_SEPARATOR) except(IOError, OSError) as e: self.__handle_error(e, name_arq) except: print(&class="macro">#x27;Unexpected error:&class="macro">#x27;, sys.exc_info()[class="num">0]) @staticmethod def save_file_csv(name_arq: str, dataset:DataFrame = pd.DataFrame({&class="macro">#x27;col&class="macro">#x27;:[&class="macro">#x27;ok&class="macro">#x27;]})): dataset.to_csv(name_arq, sep=CSV_SEPARATOR) @staticmethod def save(name_arq:str, data:str): with open(name_arq, &class="macro">#x27;w&class="macro">#x27;) as f: f.write(data) @staticmethod def delete_file(name_arq: str): try: os.remove(name_arq) except: pass
「把模型塞进 MT5 只是起点」
前几节走完了回归模型从选型、清洗、训练评估到 ONNX 导出的全链路，这一节真正落地的动作，是把训练好的模型接进 MetaTrader 5 策略测试器。Python 侧训完、MQL5 侧调用，中间靠 ONNX 做格式桥接，避免在终端里重跑训练逻辑。 作者把配套代码放在了 GitHub 仓库（github.com/jowpereira/mql5-tutoriais），一共 4 篇系列文，本篇是第 4 篇。直接 clone 下来用 MT5 策略测试器跑一遍，比看文字快得多。 外汇和贵金属杠杆高、滑点跳价频繁，模型在历史数据上表现好不等于实盘能复制，接进测试器先跑 3 个月 tick 回测再谈下一步。