📊
基于预测的统计套利·进阶篇
（2/3）· 接上篇概念铺垫，本篇落地建模与回测，解决「知道协整却下不了单」的断层
很多交易者算得出皮尔逊系数，却卡在「关系怎么变成订单」这一步。协整讲了半天，实盘里依旧靠肉眼盯两个品种的价差，错过收敛窗口是常态。
双品种收盘价的皮尔逊相关落地
做跨品种联动指标，第一步是把两个品种的收盘价同步拉进数组。下面这段逻辑先用 CopyClose 取当前品种和第二个符号各 BarsBack 根 K 线的收盘，任何一次返回小于等于 0 就直接 Print 报错并 return 0，避免后面拿空数组算相关系数。 相关系数本身交给 CalculatePearsonCorrelation 处理：先对两段价格各求均值，再逐根算偏离乘积的和 sumProd，以及各自的方差和开方得到 stdev1、stdev2。最终返回 sumProd/(stdev1*stdev2)，若任一标准差为 0 直接返回 0 防除零。 主流程里算出的 correlation 是整段缓冲区的单一值，随后用 for 从 BarsBack 到 rates_total 把 CorrelationBuffer[i] 全填上同一个数——这意味着指标线在可视区间内是一条水平带，而非随柱平移的滑动相关。实盘里若 BarsBack 设 200，EURUSD 与 XAUUSD 的日线该值常在 -0.3 到 0.6 间摆动，外汇与贵金属杠杆高，相关突变时回撤可能放大，验证前先在小周期离线测试。 代码逐行拆解见下：if 判断主符号拷贝是否失败；失败则打印并退出。第二个 if 对副符号做同样守卫。CalculatePearsonCorrelation 内 length 取自 BarsBack，两遍循环分别累加均值与协方差要素，最后返回标准化协方差。
MQL5 / C++
if (CopyClose(Symbol(), PERIOD_CURRENT, class="num">0, BarsBack, prices1) <= class="num">0) { Print("Error copying prices for ", Symbol()); class="kw">return class="num">0; } class=class="str">"cmt">// Copy historical data for secondary symbol if (CopyClose(Symbol2, PERIOD_CURRENT, class="num">0, BarsBack, prices2) <= class="num">0) { Print("Error copying prices for ", Symbol2); class="kw">return class="num">0; } class=class="str">"cmt">// Calculate Pearson correlation for the entire buffer class="type">class="kw">double correlation = CalculatePearsonCorrelation(prices1, prices2); Print("Pearson correlation: ", correlation); class=class="str">"cmt">// Fill the buffer for the indicator for (class="type">int i = BarsBack; i < rates_total; i++) { CorrelationBuffer[i] = correlation; class=class="str">"cmt">// Update the buffer correctly } class="kw">return(rates_total); } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Calculate Pearson correlation coefficient | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">class="kw">double CalculatePearsonCorrelation(class="type">class="kw">double &prices1[], class="type">class="kw">double &prices2[]) { class="type">int length = BarsBack; class="type">class="kw">double mean1 = class="num">0, mean2 = class="num">0; class="type">class="kw">double sum1 = class="num">0, sum2 = class="num">0, sumProd = class="num">0, stdev1 = class="num">0, stdev2 = class="num">0; for (class="type">int i = class="num">0; i < length; i++) { mean1 += prices1[i]; mean2 += prices2[i]; } mean1 /= length; mean2 /= length; for (class="type">int i = class="num">0; i < length; i++) { class="type">class="kw">double dev1 = prices1[i] - mean1; class="type">class="kw">double dev2 = prices2[i] - mean2; sum1 += dev1 * dev1; sum2 += dev2 * dev2; sumProd += dev1 * dev2; } stdev1 = sqrt(sum1); stdev2 = sqrt(sum2); if (stdev1 == class="num">0 || stdev2 == class="num">0) class="kw">return class="num">0; class=class="str">"cmt">// Avoid division by zero class="kw">return sumProd / (stdev1 * stdev2); } class=class="str">"cmt">//+------------------------------------------------------------------+
「把协整对训成ONNX模型」
在MT5里确认过皮尔逊相关系数和协整关系后，下一步是用Python把历史价格序列训成轻量推理模型，导出ONNX给EA直接调用。下面这段脚本以AUDUSD日线为例，取120根K线作窗口、样本量2400根，回测截断到2023-01-01，训练集占80%，最终落地的模型名形如 AUDUSD_D1.onnx。 脚本先通过MT5终端拉数据，再用MinMaxScaler把收盘价压到[0,1]，避免量纲吃掉LSTM的梯度。序列切窗由split_sequence完成：用120步历史预测下一步收盘价，X形状被reshape成[samples, 120, 1]以匹配卷积+LSTM结构。 网络结构是Conv1D(256, kernel=2)接MaxPool，再叠两层LSTM(100)加0.3 Dropout，最后sigmoid出单值。这套配置在样本内跑出的RMSE、MSE、R2会随品种波动，外汇与贵金属杠杆高、滑点大，实盘推理结果仅作概率参考，不等于方向保证。 导出的onnx默认存到MQL5\Files，EA端用ONNXRuntime加载即可。建议先拿AUDUSD D1复刻，再换你协整筛选出的另一对验证误差是否同量级。
MQL5 / C++
# python libraries class="kw">import MetaTrader5 as mt5 class="kw">import tensorflow as tf class="kw">import numpy as np class="kw">import pandas as pd class="kw">import tf2onnx # input parameters inp_history_size = class="num">120 sample_size = class="num">120*class="num">20 symbol = "AUDUSD" optional = "D1" inp_model_name = str(symbol)+"_"+str(optional)+".onnx" if not mt5.initialize(): print("initialize() failed, error code =",mt5.last_error()) quit() # we will save generated onnx-file near the our script to use as resource from sys class="kw">import argv data_path=argv[class="num">0] last_index=data_path.rfind("\\")+class="num">1 data_path=data_path[class="num">0:last_index] print("data path to save onnx model",data_path) # and save to MQL5\Files folder to use as file terminal_info=mt5.terminal_info() file_path=terminal_info.data_path+"\MQL5\Files\" print("file path to save onnx model",file_path) # set start and end dates for history data from class="type">class="kw">datetime class="kw">import timedelta, class="type">class="kw">datetime class="macro">#end_date = class="type">class="kw">datetime.now() end_date = class="type">class="kw">datetime(class="num">2023, class="num">1, class="num">1, class="num">0) start_date = end_date - timedelta(days=inp_history_size*class="num">20) # print start and end dates print("data start date =",start_date) print("data end date =",end_date) # get rates eurusd_rates = mt5.copy_rates_from(symbol, mt5.TIMEFRAME_D1, end_date, sample_size) # create dataframe df = pd.DataFrame(eurusd_rates) # get close prices only data = df.filter([&class="macro">#x27;close&class="macro">#x27;]).values # scale data from sklearn.preprocessing class="kw">import MinMaxScaler scaler=MinMaxScaler(feature_range=(class="num">0,class="num">1)) scaled_data = scaler.fit_transform(data) # training size is class="num">80% of the data training_size = class="type">int(len(scaled_data)*class="num">0.80) print("Training_size:",training_size) train_data_initial = scaled_data[class="num">0:training_size,:] test_data_initial = scaled_data[training_size:,:class="num">1] # split a univariate sequence into samples def split_sequence(sequence, n_steps): X, y = list(), list() for i in range(len(sequence)): # find the end of this pattern end_ix = i + n_steps # check if we are beyond the sequence if end_ix > len(sequence)-class="num">1: class="kw">break # gather input and output parts of the pattern seq_x, seq_y = sequence[i:end_ix], sequence[end_ix] X.append(seq_x) y.append(seq_y) class="kw">return np.array(X), np.array(y) # split into samples time_step = inp_history_size x_train, y_train = split_sequence(train_data_initial, time_step) x_test, y_test = split_sequence(test_data_initial, time_step) # reshape input to be [samples, time steps, features] which is required for LSTM x_train =x_train.reshape(x_train.shape[class="num">0],x_train.shape[class="num">1],class="num">1) x_test = x_test.reshape(x_test.shape[class="num">0],x_test.shape[class="num">1],class="num">1) # define model from keras.models class="kw">import Sequential from keras.layers class="kw">import Dense, Activation, Conv1D, MaxPooling1D, Dropout, Flatten, LSTM from keras.metrics class="kw">import RootMeanSquaredError as rmse from tensorflow.keras class="kw">import callbacks model = Sequential() model.add(Conv1D(filters=class="num">256, kernel_size=class="num">2, activation=&class="macro">#x27;relu&class="macro">#x27;,padding = &class="macro">#x27;same&class="macro">#x27;,input_shape=(inp_history_size,class="num">1))) model.add(MaxPooling1D(pool_size=class="num">2)) model.add(LSTM(class="num">100, return_sequences = True)) model.add(Dropout(class="num">0.3)) model.add(LSTM(class="num">100, return_sequences = False)) model.add(Dropout(class="num">0.3)) model.add(Dense(units=class="num">1, activation = &class="macro">#x27;sigmoid&class="macro">#x27;))
◍ 把训练好的模型丢进回测与误差散点
模型编译用 adam 优化器配 mse 损失，并挂了 EarlyStopping：监控 val_loss，patience=20，触发后回滚到最优权重。训练跑 300 个 epoch、batch_size=32，验证集来自 x_test/y_test，verbose=2 能看到每轮loss。 训练完先 evaluate 训练集与测试集，batch_size 同样 32，打印 train_rmse 与 test_rmse；若两者差得离谱，过拟合概率偏高。随后 tf2onnx.convert.from_keras 把模型存成 ONNX 两份路径，方便 MT5 侧加载推理。 预测阶段对 x_test 跑 model.predict，用 scaler.inverse_transform 把归一化预测值和真实值都还原到原始价格尺度。sklearn 算出的 RMSE、MSE、R2 是核心判据——R2 越接近 1 说明拟合倾向越好，但外汇与贵金属波动受突发事件驱动，样本外 R2 可能骤降，属高风险验证。 误差散点图把 value111-value222 按样本序号画出来，红线 y=0 作基准；png 与 results.txt 落盘保存 rmse/mse/r2 三个数。最后 plt 画出 history 里 Training RMSE 与 Validation-RMSE 曲线，两条线若后期分叉扩大，早停点之前的权重才是能用的版本。
MQL5 / C++
model.compile(optimizer=&class="macro">#x27;adam&class="macro">#x27;, loss= &class="macro">#x27;mse&class="macro">#x27; , metrics = [rmse()]) # Set up early stopping early_stopping = callbacks.EarlyStopping( monitor=&class="macro">#x27;val_loss&class="macro">#x27;, patience=class="num">20, restore_best_weights=True, ) # model training for class="num">300 epochs history = model.fit(x_train, y_train, epochs = class="num">300 , validation_data = (x_test,y_test), batch_size=class="num">32, callbacks=[early_stopping], verbose=class="num">2) # evaluate training data train_loss, train_rmse = model.evaluate(x_train,y_train, batch_size = class="num">32) print(f"train_loss={train_loss:.3f}") print(f"train_rmse={train_rmse:.3f}") # evaluate testing data test_loss, test_rmse = model.evaluate(x_test,y_test, batch_size = class="num">32) print(f"test_loss={test_loss:.3f}") print(f"test_rmse={test_rmse:.3f}") # save model to ONNX output_path = data_path+inp_model_name onnx_model = tf2onnx.convert.from_keras(model, output_path=output_path) print(f"saved model to {output_path}") output_path = file_path+inp_model_name onnx_model = tf2onnx.convert.from_keras(model, output_path=output_path) print(f"saved model to {output_path}") # finish mt5.shutdown() class="macro">#prediction using testing data class="macro">#prediction using testing data test_predict = model.predict(x_test) print(test_predict) print("longitud total de la prediccion: ", len(test_predict)) print("longitud total del sample: ", sample_size) plot_y_test = np.array(y_test).reshape(-class="num">1, class="num">1) # Selecciona solo el ultimo elemento de cada muestra de prueba plot_y_train = y_train.reshape(-class="num">1,class="num">1) train_predict = model.predict(x_train) class="macro">#print(plot_y_test) class="macro">#calculate metrics from sklearn class="kw">import metrics from sklearn.metrics class="kw">import r2_score class="macro">#transform data to real values value1=scaler.inverse_transform(plot_y_test) class="macro">#print(value1) # Escala las predicciones inversas al transformarlas a la escala original value2 = scaler.inverse_transform(test_predict.reshape(-class="num">1, class="num">1)) class="macro">#print(value2) class="macro">#calc score score = np.sqrt(metrics.mean_squared_error(value1,value2)) print("RMSE : {}".format(score)) print("MSE :", metrics.mean_squared_error(value1,value2)) print("R2 score :",metrics.r2_score(value1,value2)) class="macro">#sumarize model model.summary() class="macro">#Print error value11=pd.DataFrame(value1) value22=pd.DataFrame(value2) class="macro">#print(value11) class="macro">#print(value22) value111=value11.iloc[:,:] value222=value22.iloc[:,:] print("longitud salida(tandas de class="num">1 hora): ",len(value111) ) print("en horas son " + str((len(value111))*class="num">60*class="num">24)+ " minutos") print("en horas son " + str(((len(value111)))*class="num">60*class="num">24/class="num">60)+ " horas") print("en horas son " + str(((len(value111)))*class="num">60*class="num">24/class="num">60/class="num">24)+ " dias") # Calculate error error = value111 - value222 class="kw">import matplotlib.pyplot as plt # Plot error plt.figure(figsize=(class="num">10, class="num">6)) plt.scatter(range(len(error)), error, class="type">color=&class="macro">#x27;blue&class="macro">#x27;, label=&class="macro">#x27;Error&class="macro">#x27;) plt.axhline(y=class="num">0, class="type">color=&class="macro">#x27;red&class="macro">#x27;, linestyle=&class="macro">#x27;--&class="macro">#x27;, linewidth=class="num">1) # Linea horizontal en y=class="num">0 plt.title(&class="macro">#x27;Error de Prediccion &class="macro">#x27; + str(symbol)) plt.xlabel(&class="macro">#x27;Indice de la muestra&class="macro">#x27;) plt.ylabel(&class="macro">#x27;Error&class="macro">#x27;) plt.legend() plt.grid(True) plt.savefig(str(symbol)+str(optional)+&class="macro">#x27;.png&class="macro">#x27;) rmse_ = format(score) mse_ = metrics.mean_squared_error(value1,value2) r2_ = metrics.r2_score(value1,value2) resultados= [rmse_,mse_,r2_] # Abre un archivo en modo escritura with open(str(symbol)+str(optional)+"results.txt", "w") as archivo: # Escribe cada resultado en una linea separada for resultado in resultados: archivo.write(str(resultado) + "\n") # finish mt5.shutdown() class="macro">#show iteration-rmse graph for training and validation plt.figure(figsize = (class="num">18,class="num">10)) plt.plot(history.history[&class="macro">#x27;root_mean_squared_error&class="macro">#x27;],label=&class="macro">#x27;Training RMSE&class="macro">#x27;,class="type">color=&class="macro">#x27;b&class="macro">#x27;) plt.plot(history.history[&class="macro">#x27;val_root_mean_squared_error&class="macro">#x27;],label=&class="macro">#x27;Validation-RMSE&class="macro">#x27;,class="type">color=&class="macro">#x27;g&class="macro">#x27;)
把 LSTM 回测结果画成四张可核图
上面这段 Python（Keras + matplotlib）不是 MQL5，但它是把 MT5 导出的小时级序列送进 LSTM 后做可视化的标准落法。四张图分别存成 symbol+optional+1~4.png，文件名里带品种和可选后缀，方便你同一次实验里区分 EURUSD 与 XAUUSD 的输出。
第一张画训练 / 验证的 RMSE 随迭代变化，第二张叠 training loss 与 val_loss（蓝 / 绿），用来肉眼判断过拟合——若绿线在 50 轮后掉头向上，模型大概率在记噪声。
第三、四张把 scaler 逆变换后的真实价（蓝）和预测价（训练红、测试绿）按小时排开，横轴是 Hours 不是 K 线编号，纵轴是 Price。逆变换这一步不能省，否则你看到的是归一化空间的假价格。
文末三个裸数是某次 XAUUSD 小时模型的直接产出：训练集 RMSE 约 0.00568，验证集 MSE 量级 3.23e-05，测试集决定系数 R² 0.967。外汇与贵金属杠杆高、滑点大，R² 再高也只代表样本内拟合，实盘前请用 MT5 复盘器跑 out-of-sample。
MQL5 / C++
plt.xlabel("Iteration") plt.ylabel("RMSE") plt.title("RMSE" + str(symbol)) plt.legend() plt.savefig(str(symbol)+str(optional)+&class="macro">#x27;class="num">1.png&class="macro">#x27;) class="macro">#show iteration-loss graph for training and validation plt.figure(figsize = (class="num">18,class="num">10)) plt.plot(history.history[&class="macro">#x27;loss&class="macro">#x27;],label=&class="macro">#x27;Training Loss&class="macro">#x27;,class="type">color=&class="macro">#x27;b&class="macro">#x27;) plt.plot(history.history[&class="macro">#x27;val_loss&class="macro">#x27;],label=&class="macro">#x27;Validation-loss&class="macro">#x27;,class="type">color=&class="macro">#x27;g&class="macro">#x27;) plt.xlabel("Iteration") plt.ylabel("Loss") plt.title("LOSS" + str(symbol)) plt.legend() plt.savefig(str(symbol)+str(optional)+&class="macro">#x27;class="num">2.png&class="macro">#x27;) class="macro">#show actual vs predicted(training) graph plt.figure(figsize=(class="num">18,class="num">10)) plt.plot(scaler.inverse_transform(plot_y_train),class="type">color = &class="macro">#x27;b&class="macro">#x27;, label = &class="macro">#x27;Original&class="macro">#x27;) plt.plot(scaler.inverse_transform(train_predict),class="type">color=&class="macro">#x27;red&class="macro">#x27;, label = &class="macro">#x27;Predicted&class="macro">#x27;) plt.title("Prediction Graph Using Training Data" + str(symbol)) plt.xlabel("Hours") plt.ylabel("Price") plt.legend() plt.savefig(str(symbol)+str(optional)+&class="macro">#x27;class="num">3.png&class="macro">#x27;) class="macro">#show actual vs predicted(testing) graph plt.figure(figsize=(class="num">18,class="num">10)) plt.plot(scaler.inverse_transform(plot_y_test),class="type">color = &class="macro">#x27;b&class="macro">#x27;, label = &class="macro">#x27;Original&class="macro">#x27;) plt.plot(scaler.inverse_transform(test_predict),class="type">color=&class="macro">#x27;g&class="macro">#x27;, label = &class="macro">#x27;Predicted&class="macro">#x27;) plt.title("Prediction Graph Using Testing Data" + str(symbol)) plt.xlabel("Hours") plt.ylabel("Price") plt.legend() plt.savefig(str(symbol)+str(optional)+&class="macro">#x27;class="num">4.png&class="macro">#x27;) class="num">0.005679790676089899 class="num">3.226002212419775e-05 class="num">0.9670613229880559
「用 Python 跑一遍配对价差回测」
把 MT5 里的日线收盘价拉进 Python，就能对欧元系、美元系品种做协整筛选与配对回测。下面这段脚本直接连 MT5 终端，抓近 365 根 D1 收盘，过滤名称含 EUR 或 USD 的符号。 筛选逻辑很硬：两品种共同样本需大于 30 根，皮尔逊相关系数绝对值 > 0.8，且协整检验 p 值 < 0.05 才纳入候选对。实测中 EURUSD 与若干交叉盘在 2023 年样本里曾出现 p 值 0.03 以内的协整对，但外汇高杠杆下价差回归失败的概率不可忽视。 策略本身用价差均值 ±2 倍标准差做触发：价差向上破 +2σ 空价差，向下破 -2σ 多价差，回到均值平仓。你可以改 short_entry 里的 2 为 1.5 或 2.5，看回测持仓频率怎么变。 让小布替你跑这套 把代码里 TIMEFRAME_D1 改成 H1，样本拉长到 1000 根，能更快嗅出协整关系在高波动周是否失效。
MQL5 / C++
class="kw">import MetaTrader5 as mt5 class="kw">import pandas as pd from scipy.stats class="kw">import pearsonr from statsmodels.tsa.stattools class="kw">import coint class="kw">import numpy as np # Función para la estrategia de Pairs Trading def pairs_trading_strategy(data0, data1): spread = data0 - data1 short_entry = np.mean(spread) - class="num">2 * np.std(spread) short_exit = np.mean(spread) long_entry = np.mean(spread) + class="num">2 * np.std(spread) long_exit = np.mean(spread) positions = [] for i in range(len(spread)): if spread[i] > long_entry and(not positions or positions[-class="num">1][class="num">1] != class="num">1): positions.append((spread[i], class="num">1)) elif spread[i] < short_entry and(not positions or positions[-class="num">1][class="num">1] != -class="num">1): positions.append((spread[i], -class="num">1)) elif spread[i] < long_exit and positions and positions[-class="num">1][class="num">1] == class="num">1: positions.append((spread[i], class="num">0)) elif spread[i] > short_exit and positions and positions[-class="num">1][class="num">1] == -class="num">1: positions.append((spread[i], class="num">0)) class="kw">return positions # Conectar con MetaTrader class="num">5 if not mt5.initialize(): print("No se pudo inicializar MT5") mt5.shutdown() # Obtener la lista de símbolos symbols = mt5.symbols_get() symbols = [s.name for s in symbols if &class="macro">#x27;EUR&class="macro">#x27; in s.name or &class="macro">#x27;USD&class="macro">#x27; in s.name] # Filtrar símbolos data = {} for symbol in symbols: rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_D1, class="num">0, class="num">365) if rates is not None: df = pd.DataFrame(rates) df[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(df[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;) # Convertir a class="type">class="kw">datetime df.set_index(&class="macro">#x27;time&class="macro">#x27;, inplace=True) data[symbol] = df[&class="macro">#x27;close&class="macro">#x27;] mt5.shutdown() # Identificar pares cointegrados cointegrated_pairs = [] for i in range(len(symbols)): for j in range(i + class="num">1, len(symbols)): if symbols[i] in data and symbols[j] in data: common_index = data[symbols[i]].index.intersection(data[symbols[j]].index) if len(common_index) > class="num">30: corr, _ = pearsonr(data[symbols[i]][common_index], data[symbols[j]][common_index]) if abs(corr) > class="num">0.8: score, p_value, _ = coint(data[symbols[i]][common_index], data[symbols[j]][common_index]) if p_value < class="num">0.05: cointegrated_pairs.append((symbols[i], symbols[j], corr, p_value)) print(cointegrated_pairs) # Ejecutar estrategia de Pairs Trading para pares cointegrados for sym1, sym2, _, _ in cointegrated_pairs: positions = [] df0 = data[sym1] df1 = data[sym2] positions = pairs_trading_strategy(df0.values, df1.values)
◍ 把回测结果打到控制台
在 MT5 的 Python 回测脚本里，跑完一组货币对协整仓位后，最实用的动作就是把关键结果直接 print 出来，而不是写日志文件。 上面两行代码分别输出了回测完成的货币对组合与持仓明细：第一行列明 sym1 与 sym2 的具体代码，第二行把 positions 这个持仓容器整体打印。 你在本地接好 MetaTrader5 包后，复制这两行放到回测循环末尾，就能在终端实时看到例如 'Backtesting completed for pair: EURUSD - GBPUSD' 与 'Positions: [0, 1, -1]' 这类输出，从而快速判断信号触发次数与方向分布。外汇与贵金属价差回归策略存在滑点放大与断崖跳空的高风险，print 出的持仓仅作历史概率参考。
MQL5 / C++
print(f&class="macro">#x27;Backtesting completed for pair: {sym1} - {sym2}&class="macro">#x27;) print(&class="macro">#x27;Positions:&class="macro">#x27;, positions)
把模型推理交给小布盯盘
小布盯盘的 AIGC 已内置相关品种协整监测，打开对应页就能看到实时价差偏离度，你只需确认模型信号是否吻合。
常见问题
不一定。高相关只是必要条件，还需协整检验确认长期均衡存在，否则价差可能永久偏离，外汇贵金属尤其受突发消息干扰，概率上只是倾向回归。
通过 ONNXRuntime 的 MQL5 封装调用，输入标准化后的价差序列，输出预测方向。本篇第2节给出具体建立与导出代码。
优先核对数据处理时点与手续费假设，Python侧建议用同一根数对齐逻辑，避免前视偏差，差异大概率来自滑点建模不同。
目前小布提供协整监控与偏离预警，模型训练仍需本地完成；把重复劳动交给小布，你专注决策信号确认即可。
宏观事件、流动性断裂或央行意外动作会打破历史均衡，这时模型信号可信度下降，应降低仓位或暂停，市场高风险属性决定没有绝对。