重构经典策略（第九部分）：多时间框架分析（第二部分）（基础篇）
◍ 用模型残差挑时间框架
选时间框架这件事，新手容易懵，老手也常吵不出结论。这里换了个客观口径：把“最优时间框架”定义成能让 AI 模型误差水平最小的那一档，而不是凭手感。 我们跑了 11 个时间框架的模型残差分布，看到月线和小时线各有一个低误差区；但整体分布没明显规律，小时框架上既出现了最大误差也出现了最小误差。外汇与贵金属波动受宏观事件驱动，这种高风险的品种上残差随市场状态漂移的概率不低。 要真回答“哪个周期最好用”，得先确认换一种市场残差分布不会乱变；后面还应对全部可用周期做完整搜索，再下结论。
多周期视角下的价格疲软信号
不管你看的是1分钟还是日线，某一瞬间市场只有一个成交价。但有经验的交易者会叠看多个周期来判断当下状态：当主周期趋势开始乏力，往往在更短的周期里先出现反向或纠结的走势，这种疲软从来都是小周期先露头，大周期后确认。 多数多时间框架策略的本质，是在大周期判情绪。比如有人在4小时图上等看涨吞噬这类经典形态，靠它有无来筛高概率入场。我们这次想用算法回答一个问题：预测欧元兑美元时，到底哪个周期能给到可靠的误差水平。 直觉上大家都认一个理——预测看得越远越难。但这次欧元兑美元的分析结果，从根本上摇了这根直觉。在你能判断我这话有没有道理前，得先搞清我用的方法是什么。
「同等样本量下的跨周期相关性实测」
要让回测不偏，先得保证每个周期拿到的数据条数一致。整条链路里最受限的是月线——满打满算 400 根月 K 也才覆盖约 33 年，能撑这么久历史的品种没几个，样本天然有偏。欧元兑美元是少有的长历史品种，这次就锚定它做基准集。 具体做法是：从 MT5 拉 400 行 EURUSD 月线收盘价，再拉对应的 400 行未来值；同样动作在周线、日线、H12、H8、H4、H1、M30、M15、M5、M1 上各重复一遍。输入表共 22 列，跑完得到的相关性矩阵很大，这里不铺开，只说结论：周期越接近相关性越强的直觉不成立，样本里整体只是中等相关。 值得单独拎出来的三组组合是：当前 H4 与未来 H8、当前 M1 与未来 H4、当前 M1 与未来 M5。建模后误差最低的反而是月线和小时线这两头，和「短线更好预测」的常识相反。目标设为预测未来 20 步：20 个月≈1 年 8 个月，模型跑长周期比跑 20 分钟准。 因为盯住月线和小时线低误差，把价格转成周期收益率后又做了格兰杰因果检验，小时收益率对月度收益率的 p 值显著，说明能用 VAR 拿小时收益去模拟月收益。再用时间序列变形库对齐相似点，相似段不少，信心够了就把月线模型调参、导出成 ONNX。 最后落地的 EA 逻辑是：月线出预期价位，小时线执行；平仓方式可在「AI 预测反转」和「均线」之间切，入场时机靠技术分析定。外汇与贵金属杠杆高、滑点跳空频繁，这类跨周期模型仅作概率参考，实盘前务必在 MT5 策略测试器用真实点差复核。
◍ 先把 EURUSD 多周期数据拉进 Pandas
做跨周期回归预测，第一步不是建模，而是把终端里的报价真正抓出来。MT5 的 Python 接口需要先 initialize() 连上本地终端，连不上后面全是空表。 代码里一次性列了 11 个时间框架，从月线到 1 分钟全塞进 time_frames 数组。fetch 设为 400，意味着每个周期抓 400 根 K 线的收盘价；look_ahead 设为 20，表示用向后偏移 20 根的位置做预测目标，这是典型的时序错位监督构造。 数据结构上，inputs 存各周期当前收盘价，target 存同名加「 Target」后缀的偏移价，columns 拼成 22 列，prices 表 index 为 0~399。error_levels 和 test_error_levels 则单独留一行存各周期误差，方便后续横向比哪个周期噪声小。 循环里 mt5.copy_rates_from_pos("EURUSD", tf, look_ahead, fetch) 取偏移后的收盘价填前半列，参数 0 取当前价填后半列（i+10）。跑完这段，prices 表里就同时有「现在」和「20 根后」的 EURUSD 多周期收盘价，外汇品种高杠杆高风险，样本仅代表历史窗口，预测倾向失效。
class="macro">#Import the libraries we need class="kw">import pandas as pd class="kw">import numpy as np class="kw">import seaborn as sns class="kw">import MetaTrader5 as mt5 from sklearn.model_selection class="kw">import cross_val_score,train_test_split,TimeSeriesSplit from sklearn.metrics class="kw">import mean_squared_error class="kw">import matplotlib.pyplot as plt from sklearn.linear_model class="kw">import LinearRegression class="macro">#Initialize the terminal mt5.initialize() class="macro">#Declare the time-frames we are interested in time_frames = [mt5.TIMEFRAME_MN1, mt5.TIMEFRAME_W1, mt5.TIMEFRAME_D1, mt5.TIMEFRAME_H12, mt5.TIMEFRAME_H8, mt5.TIMEFRAME_H4, mt5.TIMEFRAME_H1, mt5.TIMEFRAME_M30, mt5.TIMEFRAME_M15, mt5.TIMEFRAME_M5, mt5.TIMEFRAME_M1 ] class="macro">#How many bars should we fetch fetch = class="num">400 class="macro">#How far into the future should we forecast? look_ahead = class="num">20 class="macro">#Create our dataframe inputs = ["MN","W","D","H12","H8","H4","H1","M30","M15","M5","M1"] target = [] for i in np.arange(class="num">0,len(inputs)): target.append(inputs[i] + " Target") columns = inputs + target prices = pd.DataFrame(columns=columns,index=np.arange(class="num">0,fetch)) class="macro">#The columns for our error levels data frame. error_columns = [] for i in np.arange(class="num">0,len(inputs)): error_columns.append(inputs[i]) class="macro">#Create a dataframe to store our error levels error_levels = pd.DataFrame(columns=error_columns,index=[class="num">0]) test_error_levels = pd.DataFrame(columns=error_columns,index=[class="num">0]) for i in np.arange(class="num">0,len(time_frames)): print(i) prices.iloc[:,i] = pd.DataFrame(mt5.copy_rates_from_pos("EURUSD",time_frames[i],look_ahead,fetch)).loc[:,"close"] prices.iloc[:,i+class="num">10] = pd.DataFrame(mt5.copy_rates_from_pos("EURUSD",time_frames[i],class="num">0,fetch)).loc[:,"close"]
不同周期收益率的离散度藏着周期选择线索
先把各时间框架的收盘价拉出来算相关性矩阵，H12 与 H8 之间的相关系数明显偏高，说明这两个中周期走势高度同步，单独同时持仓意义不大。其余组合里，月度与周度收盘价的散点看不出干净方向，但整体重心偏向上方，价格序列存在慢牛式漂移。 把价格转成周期性收益率再画散点，分布就收敛到 0 轴附近，原先的趋势漂移被差分抹掉，剩下的是围绕均值波动的噪声。这一步是后面做均值回归或突破过滤的前置判断。 箱线图给的信息更硬：从月线往低周期切，收益率方差逐层缩小，所有帧的中位数都贴着 0。想抬组合收益期望，高时间框架的样本偏移更肥，低周期只是把波动压扁。外汇与贵金属杠杆品种里，这种方差收敛不代表风险消失，只是被时间压缩，实盘仍属高风险。
fig, ax = plt.subplots(figsize=(class="num">15,class="num">15)) sns.heatmap(prices.corr(),annot=True,ax=ax) sns.scatterplot(data=prices,x="MN Close",y="W Close") sns.scatterplot(data=prices.pct_change(),x="MN Close",y="W Close")
「切分多周期样本并跑通交叉验证」
把每个时间周期的特征列和目标列按 50% 不洗牌切分，是后续建模的底线操作。下面这段脚本对 MN 到 M1 共 11 个周期逐一调用 train_test_split，test_size=0.5 且 shuffle=False，意味着用后半段做测试，避免未来信息泄漏。 切完之后把训练集和测试集分别塞进 train_X / test_X 与 train_y / test_y 列表，方便循环里统一调度。注意这里目标值也按同样索引拆开，否则维度对不上会直接报 ValueError。 用 LinearRegression 做 5 折交叉验证，cv=5 得到的负分取均值就是各周期训练误差；再 fit 后拿测试集算均方误差写进 test_error_levels。实测误差表里 H1 周期测试 MSE 达到 40.406227，而 M30 仅 0.196244，说明低频周期在样本外泛化上差异极大，外汇与贵金属建模踩这坑概率不低。 最后用 seaborn 画条形图看误差分布，肉眼能确认哪些周期模型根本没学到东西。开 MT5 导出对应周期收盘价，复跑这套切分逻辑，就能知道自己数据的残差长什么样。
class="macro">#Create train test splits X_train_mn,X_test_mn,y_train_mn,y_test_mn = train_test_split(prices.loc[:,[“MN”]],prices.loc[:,“MN Target”],test_size=class="num">0.5,shuffle=False) X_train_w,X_test_w,y_train_w,y_test_w = train_test_split(prices.loc[:,[“W”]],prices.loc[:,“W Target”],test_size=class="num">0.5,shuffle=False) X_train_d,X_test_d,y_train_d,y_test_d = train_test_split(prices.loc[:,[“D”]],prices.loc[:,“D Target”],test_size=class="num">0.5,shuffle=False) X_train_h12,X_test_h12,y_train_h12,y_test_h12 = train_test_split(prices.loc[:,[“H12”]],prices.loc[:,“H12 Target”],test_size=class="num">0.5,shuffle=False) X_train_h8,X_test_h8,y_train_h8,y_test_h8 = train_test_split(prices.loc[:,[“H8”]],prices.loc[:,“H8 Target”],test_size=class="num">0.5,shuffle=False) X_train_h4,X_test_h4,y_train_h4,y_test_h4 = train_test_split(prices.loc[:,[“H4”]],prices.loc[:,“H4 Target”],test_size=class="num">0.5,shuffle=False) X_train_h1,X_test_h1,y_train_h1,y_test_h1 = train_test_split(prices.loc[:,[“H1”]],prices.loc[:,“H1 Target”],test_size=class="num">0.5,shuffle=False) X_train_m30,X_test_m30,y_train_m30,y_test_m30 = train_test_split(prices.loc[:,[“M30”]],prices.loc[:,“M30 Target”],test_size=class="num">0.5,shuffle=False) X_train_m15,X_test_m15,y_train_m15,y_test_m15 = train_test_split(prices.loc[:,[“M15”]],prices.loc[:,“M15 Target”],test_size=class="num">0.5,shuffle=False) X_train_m5,X_test_m5,y_train_m5,y_test_m5 = train_test_split(prices.loc[:,[“M5”]],prices.loc[:,“M5 Target”],test_size=class="num">0.5,shuffle=False) X_train_m1,X_test_m1,y_train_m1,y_test_m1 = train_test_split(prices.loc[:,[“M1”]],prices.loc[:,“M1 Target”],test_size=class="num">0.5,shuffle=False) train_X = [ X_train_mn, X_train_w, X_train_d, X_train_h12, X_train_h8, X_train_h4, X_train_h1, X_train_m30, X_train_m15, X_train_m5, X_train_m1 ] test_X = [ X_test_mn, X_test_w, X_test_d, X_test_h12, X_test_h8, X_test_h4, X_test_h1, X_test_m30, X_test_m15, X_test_m5, X_test_m1 ] train_y = [ y_train_mn, y_train_w, y_train_d, y_train_h12, y_train_h8, y_train_h4, y_train_h1, y_train_m30, y_train_m15, y_train_m5, y_train_m1, ] test_y = [ y_test_mn, y_test_w, y_test_d, y_test_h12, y_test_h8, y_test_h4, y_test_h1, y_test_m30, y_test_m15, y_test_m5, y_test_m1, ] class="macro">#Record our error for i in np.arange(class="num">0,len(train_X)): class="macro">#Fit the model model = LinearRegression() cv_score = cross_val_score(model,train_X[i],train_y[i],cv=class="num">5) error_levels.iloc[class="num">0,i] = np.mean(cv_score * -class="num">1) class="macro">#Record validation error model.fit(train_X[i],train_y[i]) test_error_levels.iloc[class="num">0,i] = mean_squared_error(test_y[i],model.predict(test_X[i])) error_levels fig, ax = plt.subplots(figsize=(class="num">7,class="num">4)) sns.barplot(error_levels,ax=ax)