带有预测性的三角套利(基础篇)
「用深度学习给三角套利加预测层」
传统三角套利靠即时汇率闭环价差吃无风险利润,但 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代码里用一行指定路径读取。判断是否过拟合看第二张损失图:本例训练损失(蓝)快速下降后收敛,验证损失(绿)全程极低且稳,说明泛化好、记忆训练集的概率低。外汇与贵金属杠杆高,模型再漂亮也只是一种概率工具,实盘前务必在策略测试器用历史日期先跑一遍。
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(): print("initialize() failed, error code =",mt5.last_error()) 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 根数。
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 加载模型会报文件不存在。
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