| 63 |
TensorFlow利用LSTM训练MQL5的数据 |
<p>1、Tensorflow在windows上2.10版本过后就不支持GPU训练了</p>
<p>2、Python关于MT5的库也只支持windows。</p>
<p>3、综上所述，本篇笔记推荐的环境是win10的系统、4090以下的显卡。但很明显这样终将被淘汰。目前已经转PyTorch。但MT5对PyTorch转的onnx模型的GPU运行支持度还不够完美，期待MT5更新。</p>
<pre class=""><code class="language-python hljs"># python libraries
import MetaTrader5 as mt5
import tensorflow as tf
import numpy as np
import pandas as pd
import tf2onnx
# input parameters
inp_model_name = "model.eurusd.H1.120_old.onnx"
inp_history_size = 120
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 import argv
data_path = argv[0]
last_index = data_path.rfind("\\") + 1
data_path = data_path[0:last_index]
print("data path to save onnx model", data_path)
# set start and end dates for history data
from datetime import timedelta, datetime
# end_date = datetime.now()
end_date = datetime(2023, 1, 1, 0)
start_date = end_date - timedelta(days=inp_history_size)
# print start and end dates
print("data start date =", start_date)
print("data end date =", end_date)
# get rates
eurusd_rates = mt5.copy_rates_range("EURUSD", mt5.TIMEFRAME_H1, start_date, end_date)
# create dataframe
df = pd.DataFrame(eurusd_rates)
# get close prices only
data = df.filter(['close']).values
# scale data
from sklearn.preprocessing import MinMaxScaler
# MinMaxScaler：一个用于数据归一化的工具，将数据的值缩放到指定的范围，这里是[0, 1]。
# fit_transform()：将scaler适配到数据并对其进行变换。
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(data)
# training_size：定义训练集的大小，占总数据的80%。
# 将数据分为训练集（train_data_initial）和测试集（test_data_initial）。
# training size is 80% of the data
training_size = int(len(scaled_data) * 0.80)
print("Training_size:", training_size)
train_data_initial = scaled_data[0:training_size, :]
test_data_initial = scaled_data[training_size:, :1]
# split_sequence()：将时间序列数据分割成输入（X）和输出（y）序列。n_steps是输入序列的长度，输出是下一个时间点的预测值。
# 例如，输入的时间步长为n_steps，输出为时间步长后的下一个值。
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_sequence 函数，将数据转换为时间序列输入输出对。
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)
# 调整数据形状以适应LSTM
# LSTM模型要求输入的数据形状为 [样本数, 时间步长, 特征数]，这里每个数据点只有一个特征（即收盘价），因此将输入数据的形状调整为 (样本数, 时间步长, 1)。
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
# Sequential：初始化一个线性堆叠的模型。
# Conv1D：一维卷积层，用于提取时间序列中的局部特征。
# MaxPooling1D：最大池化层，用于减少数据的维度。
# LSTM：长短期记忆层，擅长处理时间序列数据。
# Dropout：丢弃层，防止过拟合。
# Dense：全连接层，用于输出预测值。
# compile：编译模型，指定优化器、损失函数和评估指标（均方误差和RMSE）。
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()])
# model.fit()：训练模型，使用训练数据训练300个周期，并在每个周期后使用验证数据进行验证。
# epochs=300：训练300个周期。
# batch_size=32：每次梯度更新时使用32个样本。
# verbose=2：显示每个周期的训练进度。
history = model.fit(x_train, y_train, epochs=300, validation_data=(x_test, y_test), batch_size=32, verbose=2)
# 评估模型在训练集和测试集上的表现
# 使用训练集和测试集对模型进行评估，输出损失值和根均方误差（RMSE）
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}")
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}")
# finish
mt5.shutdown()
</code></pre> |
1 |
2 |
1 |
3/6/26, 7:44 PM |
3/6/26, 7:44 PM |
View Edit Delete |