基于神经网络的外汇预测

import torch import torch.nn as nn import torch.optim as optim import numpy as np import pandas as pd import matplotlib.pyplot as plt from sklearn.model_selection import train_test_split from sklearn import metrics # 数据查看与缺失处理 df = pd.read_csv("./debug_data/EURUSD_M1_202012150519_202103251444.csv", sep="\t") df.dropna(inplace=True) values = df['<CLOSE>'].values.tolist() # 数据集划分 X = [] y = [] # for i in range(60, len(values)): # X.append(values[i - 60:i]) # y.append(values[i]) for i in range(60, 100): X.append(values[i - 60:i]) y.append(values[i]) X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=0) # 将数据转化为张量 X_train = torch.tensor(X_train).float() y_train = torch.tensor(y_train).float() X_test = torch.tensor(X_test).float() y_test = torch.tensor(y_test).float() # 定义神经网络模型 class ForexNet(nn.Module): def __init__(self): super(ForexNet, self).__init__() self.conv1 = nn.Conv1d(in_channels=1, out_channels=32, kernel_size=3, padding=1) self.fc1 = nn.Linear(in_features=960, out_features=1) def forward(self, x): x = x.unsqueeze(1) x = self.conv1(x) x = nn.functional.relu(x) x = nn.functional.max_pool1d(x, kernel_size=2) x = x.view(x.size(0), -1) x = self.fc1(x) return x # 训练模型 net = ForexNet() criterion = nn.MSELoss() optimizer = optim.Adam(net.parameters(), lr=0.001) train_loss = [] val_loss = [] for epoch in range(100): running_loss = 0.0 net.train() optimizer.zero_grad() outputs = net(X_train) # loss = criterion(outputs, y_train) loss = criterion(outputs.squeeze(), y_train) # loss = F.mse_loss(outputs.squeeze(), y_train) loss.backward() optimizer.step() train_loss.append(loss.item()) net.eval() with torch.no_grad(): val_outputs = net(X_test) val_loss.append(criterion(val_outputs.squeeze(), y_test)) print('Epoch %d, Train Loss: %.3f, Valid Loss: %.3f' % (epoch + 1, train_loss[-1], val_loss[-1])) # 模型预测 net.eval() y_pred = net(X_test) # 模型评估 print('平均绝对误差:', metrics.mean_absolute_error(y_test.detach().numpy(), y_pred.detach().numpy())) print('均方误差:', metrics.mean_squared_error(y_test.detach().numpy(), y_pred.detach().numpy())) print('均方根误差:', np.sqrt(metrics.mean_squared_error(y_test.detach().numpy(), y_pred.detach().numpy()))) print('回归决定系数 R2:', metrics.r2_score(y_test.detach().numpy(), y_pred.detach().numpy())) # 可视化损失函数 plt.plot(train_loss, label='Training loss') plt.plot(val_loss, label='Validation loss') plt.legend() plt.title('Training and Validation Loss') plt.xlabel('Epoch') plt.ylabel('Loss') plt.show() # 可视化预测结果 plt.plot(y_test.detach().numpy(), label='True value') plt.plot(y_pred.detach().numpy(), label='Predicted value') plt.legend() plt.title('Forex Rate Prediction') plt.xlabel('Time') plt.ylabel('Rate') plt.show() # import pandas as pd # import numpy as np # from sklearn.model_selection import train_test_split # from sklearn.linear_model import LinearRegression # from sklearn import metrics # import tabulate # import matplotlib.pyplot as plt # import torch # # # 数据查看与缺失处理 # df = pd.read_csv("./debug_data/EURUSD_M1_202012150519_202103251444.csv", sep="\t") # df.dropna(inplace=True) # values = df['<CLOSE>'].values.tolist() # # # 数据可视化 # plt.figure(figsize=(8, 6)) # plt.plot(df['<CLOSE>'][2000:2400]) # plt.xlabel('<TIME>') # plt.ylabel('<CLOSE>') # plt.title('data') # plt.legend() # plt.show() # # # # 数据集划分 # X = [] # 特征集 # y = [] # 标签集 # for i in range(60, len(values)): # X.append(values[i - 60:i]) # y.append(values[i]) # # X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=0) # # # 使用sklearn库中的线性回归进行训练 # lr_model = LinearRegression() # lr_model.fit(X_train, y_train) # # print(lr_model.score(X_train, y_train)) # lr_model.fit(X_test, y_test) # # print(lr_model.score(X_test, y_test)) # # # 模型评估 # y_pred = lr_model.predict(X_test) # # table = [['<真值>', '<预测值>']] # for i in range(len(y_test)): # table.append([y_test[i], y_pred[i]]) # print(tabulate.tabulate(table[0:10], headers='firstrow', tablefmt='grid')) # # print('平均绝对误差:', metrics.mean_absolute_error(y_test, y_pred)) # print('均方误差:', metrics.mean_squared_error(y_test, y_pred)) # print('均方根误差:', np.sqrt(metrics.mean_squared_error(y_test, y_pred))) # print('回归决定系数 R2:', metrics.r2_score(y_test, y_pred)) # # history = [] # for i in range(len(y_test)): # a = torch.tensor(y_test[i]) # b = torch.tensor(y_pred[i]) # loss_fn = torch.nn.MSELoss() # loss = loss_fn(b, a) # z = loss.detach().numpy() # history.append(z) # # plt.figure(figsize=(8, 6)) # plt.plot(history[120:240]) # plt.xlabel('<x>') # plt.ylabel('<MSELoss>') # plt.legend() # plt.title("MSE") # plt.show()