带有预测性的三角套利(基础篇)
📘

带有预测性的三角套利(基础篇)

第 1/3 篇

「用深度学习给三角套利加预测层」

传统三角套利靠即时汇率闭环价差吃无风险利润,但 MT5 上裸跑这类策略常被滑点和延迟吃掉空间。2024-11-25 发布的这套思路不一样:它把深度学习模型嵌进三角套利 EA,用历史汇率序列预测下一窗口的价差方向,再决定是否触发闭环交易。 原文作者 Javier Santiago Gaston De Iriarte Cabrera 在文中给了两个基于深度学习的三角套利示例,模型和 EA 都放在文章附件里可直接取用。对外汇和贵金属来说,这类跨品种闭环仍属高风险,模型预测只提高触发概率,不消除流动性断裂的可能。 想在 MT5 验证,先下附件里的 EA,用欧美/镑美/美日三组符号跑历史回放,重点看模型输出信号与实际价差收敛的时差——这决定了你实盘要不要加缓冲点差。

◍ 三角套利在外汇里的乘法逻辑

体育博彩里两家公司赔率倒算胜率之和若小于100%,就能靠分配本金实现无风险搬砖,这也是账号被封的原因。外汇与贵金属市场没有这种显性套利空间,但三角循环用符号链条重写了同一思路。 在MT5里货币对写作 A/B、B/C、C/A,三者连续相乘构成闭环:(A/B)×(B/C)×(C/A)。乘积严格大于1时,沿该顺序做一次环交易倾向产生正价差;小于1则反向操作。外汇与贵金属带高杠杆,点差和滑点随时吞掉薄利,实际能跑通的概率随流动性剧烈波动。 这类循环在 EUR/USD、USD/JPY、EUR/JPY 上最容易被小布盯盘脚本扫到瞬时错位,但贵金属如 XAU/USD 穿插交叉盘时,报价源延迟会让乘积失真,验证前必须确认三家报价同时间戳。

点差账户做三角套利的实际约束

零点差账户里,三角套利能在 1 秒甚至更短时间内闭环成交;一旦账户带点差,这么短的窗口基本不可能覆盖价差成本。 我自己的账户不是零点差类型,所以下面所有示例都基于带点差的真实环境来跑。带点差不代表没法做,只是执行节奏和成交假设要重新校准,EA 在两种账户下都有盈利可能,但参数逻辑不能照搬。 开 MT5 后先查自己账户规格里的 spread 字段,若不是 0,就别用零点差那套 1 秒内闭环的回测设定去硬套,否则样本失真。

「把Python预测塞进MT5的实操链路」

EA的本质,是把Python里训练好的预测逻辑转成ONNX模型,再交给MT5调用。整套环境先从Microsoft商城装Python 3.10和Visual Studio Code,随后补一个Visual Studio 2019的C++运行库——这是装某些python库的前置条件。 装完要把python脚本目录加进系统PATH,并把“.py”补到PATHEXT变量里。之后在VSC新建终端,顺手装一批固定版本库:MetaTrader5==5.0.4200、pandas==2.2.1、tensorflow==2.15.0、tf2onnx==1.16.1等,版本锁死能少踩很多坑。 跑.py之前得先设品种、样本量、周期和结束日期。脚本会吐出三个ONNX模型加图表,txt里每行是RMSE、MSE、R2三个数。本例R2非常接近1,按回归定义这意味着模型覆盖了均值附近几乎全部波动,拟合倾向极强。 零点差账户可以试最小价格变动(Ticks)替代Periods,改两处设置即可拿到Bid/Ask的tick级数据;要更多tick就走免费的全量下载接口。 跑完把ONNX丢进MQL5/Files,EA代码里用一行指定路径读取。判断是否过拟合看第二张损失图:本例训练损失(蓝)快速下降后收敛,验证损失(绿)全程极低且稳,说明泛化好、记忆训练集的概率低。外汇与贵金属杠杆高,模型再漂亮也只是一种概率工具,实盘前务必在策略测试器用历史日期先跑一遍。

MQL5 / C++
https:<span class="comment">class=class="str">"cmt">//learn.microsoft.com/en-US/cpp/windows/latest-supported-vc-redist?view=msvc-class="num">170class="macro">#visual-studio-class="num">2015-class="num">2017-class="num">2019-and-class="num">2022</span>
pip install MetaTrader5==<span class="number">class="num">5.0</span>.<span class="number">class="num">4200</span>
pip install pandas==<span class="number">class="num">2.2</span>.<span class="number">class="num">1</span>
pip install scipy==<span class="number">class="num">1.12</span>.<span class="number">class="num">0</span>
pip install statsmodels==<span class="number">class="num">0.14</span>.<span class="number">class="num">1</span>
pip install numpy==<span class="number">class="num">1.26</span>.<span class="number">class="num">4</span>
pip install tensorflow==<span class="number">class="num">2.15</span>.<span class="number">class="num">0</span>
pip install tf2onnx==<span class="number">class="num">1.16</span>.<span class="number">class="num">1</span>
pip install scikit-learn==<span class="number">class="num">1.4</span>.<span class="number">class="num">1</span>.post1
pip install keras==<span class="number">class="num">2.15</span>.<span class="number">class="num">0</span>
pip install matplotlib==<span class="number">class="num">3.8</span>.<span class="number">class="num">3</span>
# python libraries
class="kw">import MetaTrader5 <span class="keyword">as</span> mt5
class="kw">import tensorflow <span class="keyword">as</span> tf
class="kw">import numpy <span class="keyword">as</span> np
class="kw">import pandas <span class="keyword">as</span> pd
class="kw">import tf2onnx
from <span class="keyword">class="type">class="kw">datetime</span> class="kw">import timedelta, <span class="keyword">class="type">class="kw">datetime</span>
# <span class="keyword">input</span> parameters
symbol1 = "EURGBP"
symbol2 = "GBPUSD"
symbol3 = "EURUSD"
sample_size1 = <span class="number">class="num">200000</span>
optional = "_M1_test"
timeframe = mt5.TIMEFRAME_M1
class="macro">#end_date = <span class="keyword">class="type">class="kw">datetime</span>.now()
end_date = <span class="keyword">class="type">class="kw">datetime</span>(<span class="number">class="num">2024</span>, <span class="number">class="num">3</span>, <span class="number">class="num">4</span>, <span class="number">class="num">0</span>)
inp_history_size = <span class="number">class="num">120</span>
sample_size = sample_size1
symbol = symbol1
optional = optional
inp_model_name = str(symbol)+"_"+str(optional)+".onnx"
<span class="keyword">if</span> not mt5.initialize():
&nbsp;&nbsp;&nbsp;&nbsp;print("initialize() failed, error code =",mt5.last_error())
&nbsp;&nbsp;&nbsp;&nbsp;quit()
# we will save generated onnx-file near our script to use <span class="keyword">as</span> resource
from sys class="kw">import argv
data_path=argv[<span class="number">class="num">0</span>]
last_index=data_path.rfind("\\")+<span class="number">class="num">1</span>
data_path=data_path[<span class="number">class="num">0</span>:last_index]
print("data path to save onnx model",data_path)
# and save to MQL5\Files folder to use <span class="keyword">as</span> 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 <span class="keyword">for</span> history data
class="macro">#end_date = <span class="keyword">class="type">class="kw">datetime</span>.now()
class="macro">#end_date = <span class="keyword">class="type">class="kw">datetime</span>(<span class="number">class="num">2024</span>, <span class="number">class="num">5</span>, <span class="number">class="num">1</span>, <span class="number">class="num">0</span>)
start_date = end_date - timedelta(days=inp_history_size*<span class="number">class="num">20</span>)
# 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, timeframe , end_date, sample_size )
# create dataframe
df=pd.DataFrame()
df = pd.DataFrame(eurusd_rates)
print(df)
# Extraer los precios de cierre directamente
datas = df[&class="macro">#x27;close&class="macro">#x27;].values
"""# Calcular la inversa de cada valor
inverted_data = <span class="number">class="num">1</span> / datas
# Convertir los datos invertidos a un array de numpy si es necesario
data = inverted_data.values"""
data = datas.reshape(-<span class="number">class="num">1</span>,<span class="number">class="num">1</span>)
# Imprimir los resultados
"""data = datas"""
# scale data
from sklearn.preprocessing class="kw">import MinMaxScaler
scaler=MinMaxScaler(feature_range=(<span class="number">class="num">0</span>,<span class="number">class="num">1</span>))
scaled_data = scaler.fit_transform(data)
# training size is <span class="number">class="num">80</span>% of the data
training_size = <span class="keyword">class="type">int</span>(len(scaled_data)*<span class="number">class="num">0.80</span>)
print("Training_size:",training_size)

◍ 把归一化序列喂给 CNN-LSTM 并导成 ONNX

外汇与贵金属行情用 LSTM 做一步预测前,得先把缩放后的序列切成样本。下面这段 Python 把训练集前 training_size 行作训练、其余作测试,再用滑动窗口拆成 (X, y),窗口长度由 inp_history_size 决定。 train_data_initial = scaled_data[0:training_size,:] test_data_initial = scaled_data[training_size:,: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)-1: 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) 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[0],x_train.shape[1],1) x_test = x_test.reshape(x_test.shape[0],x_test.shape[1],1) # define model from keras.models import Sequential from keras.layers import Dense, Activation, Conv1D, MaxPooling1D, Dropout, Flatten, LSTM from keras.metrics import RootMeanSquaredError as rmse from tensorflow.keras import callbacks model = Sequential() model.add(Conv1D(filters=256, kernel_size=2, activation='relu',padding = 'same',input_shape=(inp_history_size,1))) model.add(MaxPooling1D(pool_size=2)) model.add(LSTM(100, return_sequences = True)) model.add(Dropout(0.3)) model.add(LSTM(100, return_sequences = False)) model.add(Dropout(0.3)) model.add(Dense(units=1, activation = 'sigmoid')) model.compile(optimizer='adam', loss= 'mse' , metrics = [rmse()]) # Set up early stopping early_stopping = callbacks.EarlyStopping( monitor='val_loss', patience=5, restore_best_weights=True, ) # model training for 300 epochs history = model.fit(x_train, y_train, epochs = 300 , validation_data = (x_test,y_test), batch_size=32, callbacks=[early_stopping], verbose=2) # evaluate training data train_loss, train_rmse = model.evaluate(x_train,y_train, batch_size = 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 = 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() #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(-1, 1) # Selecciona solo el ultimo elemento de cada muestra de prueba plot_y_train = y_train.reshape(-1,1) train_predict = model.predict(x_train) #transform data to real values value1=scaler.inverse_transform(plot_y_test) value2 = scaler.inverse_transform(test_predict.reshape(-1, 1)) #calc score from sklearn import metrics from sklearn.metrics import r2_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)) model.summary() value11=pd.DataFrame(value1) value22=pd.DataFrame(value2) value111=value11.iloc[:,:] value222=value22.iloc[:,:] print("longitud salida (tandas de 1 minuto): ",len(value111) ) print("en horas son " + str(((len(value111)))/60)+ " horas") 逐行看关键处:split_sequence 用 for i 从头滑窗,end_ix 超出 len-1 就 break,保证 y 总取窗口后一根 K。Conv1D 用 256 个滤波器、kernel_size=2 提取局部形态,接 MaxPooling1D(2) 压缩,再叠两层 LSTM(100) 且中间 Dropout(0.3) 抑制过拟合。 训练设 300 epoch、batch_size=32,但 EarlyStopping 监控 val_loss、patience=5,实际可能在验证损失连续 5 轮不降时就停。外汇与贵金属波动剧烈,这套结构在 1 分钟样本上仅给出概率性方向,杠杆交易风险高,回测 RMSE 需结合 inverse_transform 还原价格后再判。 导 ONNX 时代码把模型存了 data_path 和 file_path 两处,小布盯盘端加载的应是后者。末尾用 scaler.inverse_transform 把预测还原成真实报价,输出长度除以 60 得到「小时数」——这直接对应你 MT5 上能往前看的分钟 K 根数。

MQL5 / C++
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:
            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;))
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">5,
    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 class="kw">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">#transform data to real values
value1=scaler.inverse_transform(plot_y_test)
value2 = scaler.inverse_transform(test_predict.reshape(-class="num">1, class="num">1))
class="macro">#calc score
from sklearn class="kw">import metrics
from sklearn.metrics class="kw">import r2_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))
model.summary()
value11=pd.DataFrame(value1)
value22=pd.DataFrame(value2)
value111=value11.iloc[:,:]
value222=value22.iloc[:,:]
print("longitud salida(tandas de class="num">1 minuto): ",len(value111) )
print("en horas son " + str(((len(value111)))/class="num">60)+ " horas")

把训练好的预测模型落盘成 ONNX 与回测图

模型跑完并不是终点。先把预测误差摊开看:用散点图把每个样本的 error = 真实值减预测值画出来,零轴拉一条红色虚线,肉眼就能判断误差是对称散布还是系统性偏高。EURJPY 这类交叉盘在消息面跳动时,误差散点常往正半轴堆,说明模型对跳空段倾向低估。 回测指标必须落文件。RMSE、MSE、R² 三个数写进 results.txt,一行一个,后面接 MT5 加载做横向对比。训练集和验证集的 RMSE 随迭代变化的曲线分开存图(图1),loss 同理(图2);再用 inverse_transform 把量纲还原,画原始价 vs 预测价,训练集图3、测试集图4。测试集图里绿线若长期贴着蓝线但滞后几小时,说明模型抓趋势但抓不准拐点。 换品种时直接改 symbol 和 optional 参数,脚本会重新连 MT5 拉数据。注意 history_size 设 120、回看天数按 120×20=2400 天倒推,样本量 sample_size 不够时 copy_rates_from 会截断,导致图3图4 横轴变短。外汇与贵金属杠杆高,模型回测吻合不代表实盘概率同分布,上线前先用小仓验证。 ONNX 导出路径有讲究:脚本取自身 argv[0] 截出目录存一份,再读 terminal_info.data_path 拼出 MQL5\Files\ 存一份。后者才是 EA 里用 FileOpen 能直接读的位置,漏了这一步 EA 加载模型会报文件不存在。

MQL5 / C++
print("en horas son " + str(((len(value111)))/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">class="kw">color=&class="macro">#x27;blue&class="macro">#x27;, label=&class="macro">#x27;Error&class="macro">#x27;)
plt.axhline(y=class="num">0, class="type">class="kw">color=&class="macro">#x27;red&class="macro">#x27;, linestyle=&class="macro">#x27;--&class="macro">#x27;, linewidth=class="num">1)  # Línea horizontal en y=class="num">0
plt.title(&class="macro">#x27;Error de Predicción &class="macro">#x27; + str(symbol))
plt.xlabel(&class="macro">#x27;Índice 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 línea 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">class="kw">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">class="kw">color=&class="macro">#x27;g&class="macro">#x27;)
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">class="kw">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">class="kw">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">class="kw">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">class="kw">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">class="kw">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">class="kw">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;)
################################################################################################ EURJPY class="num">1
# 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 = sample_size1
symbol = symbol2
optional = optional
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 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()
class="macro">#end_date = class="type">class="kw">datetime(class="num">2024, class="num">5, 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_rates2 = mt5.copy_rates_from(symbol, timeframe ,  end_date, sample_size)
# create dataframe
df=pd.DataFrame()
df2 = pd.DataFrame(eurusd_rates2)
print(df2)
# Extraer los precios de cierre directamente
datas2 = df2[&class="macro">#x27;close&class="macro">#x27;].values
"""inverted_data = class="num">1 / datas
# Convertir los datos invertidos a un array de numpy si es necesario
data = inverted_data.values"""
data2 = datas2.reshape(-class="num">1,class="num">1)
# Convertir los datos invertidos a un array de numpy si es necesario
class="macro">#data = datas.values
# Imprimir los resultados
# scale data
from sklearn.preprocessing class="kw">import MinMaxScaler

常见问题

三角套利把三个货币对报价连乘成一个闭环,若乘积明显偏离1就存在无风险价差;普通对冲是两条腿抵消风险,不靠闭环乘积。实操先看 EUR/USD×USD/JPY×JPY/EUR 是否≈1。
点差账户每腿都收点差,三腿累计摩擦常大于理论套利空间。回测时要把实时点差算进成本,否则纸面盈利落地变亏损。
小布可接入你的品种页,把三腿报价乘积和预测偏离度直接标在盘口,省去手动算闭环。异常值会推送提醒,你再决定是否下单。
先在Python里把序列归一化喂给CNN-LSTM,训练完导出ONNX,再用软件侧的ONNX运行库加载做推理。回测图确认信号方向后再接实盘。
要做滑动窗口,否则时序结构会丢;基础篇用20根K线作窗口较稳。导成ONNX前在验证集看预测偏移是否收敛。