利用Tushare获取交易信息，对收盘价建立简单的LSTM模型

本人双非院校，学校买的国泰安数据库居然在去年年底到期了！毕业论文急着要数据，发现了python中伟大的Tushare大数据社区，里面的金融经济数据很多，甚至比学校在国泰安订购的数据多，并且在python中可以及时调取，两三行命令就可以调取所需要的内容（tushare ID :454006）

pro = ts.pro_api('4b6e7c4f534f358a10b10a3edfdc4397c14f2fc633b3101ecfcca430')
data = pro.daily_basic(ts_code='600030.SH', start_date='20170601',end_date='20211231',fields='trade_date,close,turnover_rate,pe')
data.columns = ['Date','Close','Turnover_rate','PE']

接口均是可以更新的，三行命令获取到数据。虽然是取了两个特征，但通过实证发现，PE对预测收益率的影响不大，所以在实证中被删去了。下面对数据进行简单的归一化处理和划分数据集（划分数据集部分就省略了）

from sklearn.preprocessing import MinMaxScaler

scaler = MinMaxScaler(feature_range=(-1, 1))
price['Close'] = scaler.fit_transform(price['Close'].values.reshape(-1,1))

price['Turnover_rate'] = scaler.fit_transform(price['Turnover_rate'].values.reshape(-1,1))

接着进行模型构建

##构建模型
import torch
import torch.nn as nn

x_train = torch.from_numpy(x_train).type(torch.Tensor)
x_test = torch.from_numpy(x_test).type(torch.Tensor)
y_train_lstm = torch.from_numpy(y_train).type(torch.Tensor)
y_test_lstm = torch.from_numpy(y_test).type(torch.Tensor)
y_train_gru = torch.from_numpy(y_train).type(torch.Tensor)
y_test_gru = torch.from_numpy(y_test).type(torch.Tensor)

##设置一些参数，根据不同数据进行变化
input_dim = 2
hidden_dim = 32
num_layers = 2
output_dim = 1
num_epochs = 100

class LSTM(nn.Module):
    def __init__(self, input_dim, hidden_dim, num_layers, output_dim):
        super(LSTM, self).__init__()
        self.hidden_dim = hidden_dim
        self.num_layers = num_layers
        
        self.lstm = nn.LSTM(input_dim, hidden_dim, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_dim, output_dim)

    def forward(self, x):
        h0 = torch.zeros(self.num_layers, x.size(0), self.hidden_dim).requires_grad_()
        c0 = torch.zeros(self.num_layers, x.size(0), self.hidden_dim).requires_grad_()
        out, (hn, cn) = self.lstm(x, (h0.detach(), c0.detach()))
        out = self.fc(out[:, -1, :]) 
        return out

model = LSTM(input_dim=input_dim, hidden_dim=hidden_dim, output_dim=output_dim, num_layers=num_layers)
criterion = torch.nn.MSELoss()
optimiser = torch.optim.Adam(model.parameters(), lr=0.01)

对模型的训练、测试结果进行评价

import math, time
from sklearn.metrics import mean_squared_error
from sklearn.metrics import r2_score
from sklearn.metrics import mean_absolute_error
# make predictions
y_test_pred = model(x_test)

# invert predictions
y_train_pred = scaler.inverse_transform(y_train_pred.detach().numpy())
y_train = scaler.inverse_transform(y_train_lstm.detach().numpy())
y_test_pred = scaler.inverse_transform(y_test_pred.detach().numpy())
y_test = scaler.inverse_transform(y_test_lstm.detach().numpy())

# calculate root mean squared error
trainScore = math.sqrt(mean_squared_error(y_train[:,0], y_train_pred[:,0]))
print('Train Score: %.5f RMSE' % (trainScore))
testScore = math.sqrt(mean_squared_error(y_test[:,0], y_test_pred[:,0]))
print('Test Score: %.5f RMSE' % (testScore))

##结果可视化
import plotly.express as px
import plotly.graph_objects as go

fig = go.Figure()
fig.add_trace(go.Scatter(go.Scatter(x=result.index, y=result[0],
                    mode='lines',
                    name='Train prediction')))
fig.add_trace(go.Scatter(x=result.index, y=result[1],
                    mode='lines',
                    name='Test prediction'))
fig.add_trace(go.Scatter(go.Scatter(x=result.index, y=result[2],
                    mode='lines',
                    name='Actual Value')))
fig.update_layout(
    xaxis=dict(
        showline=True,
        showgrid=True,
        showticklabels=False,
        linecolor='black',
        linewidth=2
    ),
    yaxis=dict(
        title_text='Close (CITIC)',
        titlefont=dict(
            family='Rockwell',
            size=12,
            color='black',
        ),
        showline=True,
        showgrid=True,
        showticklabels=True,
        linecolor='black',
        linewidth=2,
        ticks='outside',
        tickfont=dict(
            family='Rockwell',
            size=12,
            color='black',
        ),
    ),
    showlegend=True,
    template = 'plotly_white'

)



annotations = []
annotations.append(dict(xref='paper', yref='paper', x=0.0, y=1.05,
                              xanchor='left', yanchor='bottom',
                              text='Results (LSTM)',
                              font=dict(family='Rockwell',
                                        size=26,
                                        color='black'),
                              showarrow=False))
fig.update_layout(annotations=annotations)

fig.show()

trainScore_R = r2_score(y_train[:,0], y_train_pred[:,0])
print('Train Score_R: %.5f R^2' % (trainScore_R))
testScore_R = r2_score(y_test[:,0], y_test_pred[:,0])
print('Test Score_R: %.5f R^2' % (testScore_R))

trainScore_mae = mean_absolute_error(y_train[:,0], y_train_pred[:,0])
print('Train Score_MAE: %.5f MAE' % (trainScore_mae))
testScore_mae = mean_absolute_error(y_test[:,0], y_test_pred[:,0])
print('Test Score_MAE: %.5f MAE' % (testScore_mae))

lstm.append(trainScore)
lstm.append(testScore)
lstm.append(training_time)

 结果图如下

拟合效果还不错，但存在一定的滞后性，之后有待各位大佬进行改进。