深度学习_循环神经网络_预测平安中国股价(文末附带数据集下载链接, 长期有效, 如果有大佬愿意帮忙, 我先在这磕一个,感谢)

news2024/10/28 13:21:10

简介:
        使用循环神经网络RNN对股价进行预测,
        也就是搭建循环神经网络对中国平安的收盘股价进行预测
深度学习训练流程
        1.数据集导入

2.数据预处理
3.模型训练
模型结构要求:
单层简单RNN, 神经元=5, 每次使用前八个预测第九个数据
输出层: 因为是对股价进行拟合预测, 使用线性回归
4.预测和评估, 并可视化结果

===============================================================

代码实现(人工版)

效果预览图

# ====================================================================
# 1.导入数据集
import pandas as pd
zgpa_data=pd.read_csv(r"C:\Users\鹰\Desktop\zgpa.csv")        # 格式: r+文件地址
zgpa_data.shape
# ============================================================================
# 2.数据预处理
# 2.1.数据基本处理, 数据很完善, 不需要进行缺失值/异常值处理
# 2.2.特征工程
# 2.2.1确定特征值和目标值
price=zgpa_data.loc[:,'close']
# 2.2.2特征预处理-对数据进行归一化, 可以加快函数收敛速度
price_norm=price/max(price)
# 2.2.3对处理后的结果进行可视化, 主要是确定一下处理的怎么样, 可视化嘛, 就是方便给人看的嘛
from matplotlib import pyplot as plt
price_fig=plt.figure(figsize=(4,3))
plt.plot(price)
plt.xlabel('time')
plt.ylabel('price')
plt.title('zgpa_price')
plt.show()

# 这一步是将之前归一化之后的数据进行分段截取, 每段截取8个数
# extract_data()函数的作用就是根据输入的数据:price和截取长度time_step=8, 实现上述功能 
import numpy as np # 在这里导入numpy的作用是调用numpy来将数据转化为数组格式
def extract_data(data, time_step):
    x=[]
    y=[]
    for i in range(len(data)-time_step):
        x.append([a for a in data[i:i+time_step]])
        y.append(data[i+time_step])
    # 加上这两行, 有什么区别?
    x=np.array(x)
    y=np.array(y)
    # 下面这一步给数据增加维度, 是为了让数据能够进入神经网络层, 因为simpleRNN需要接受的是三维数据
    x=x.reshape(x.shape[0], x.shape[1], 1)
    return x, y

time_step=8   # 每次截取的数据长度==8
x_all, y_all=extract_data(price_norm, time_step)


# 将处理好的输入数据进行数据分割
from sklearn.model_selection import train_test_split
x_train, x_test, y_train, y_test = train_test_split(x_all, y_all, test_size=0.3,random_state=10)
print(x_train.shape, y_train.shape)

# =============================================================================
# 3.训练模型-SimpleRNN
# 选择模型框架
from keras.models import Sequential
price_model=Sequential()
# 填充框架
from keras.layers import SimpleRNN, Dense
price_model.add(SimpleRNN(units=50, input_shape=(time_step, 1), activation='relu'))
# price_model.add(SimpleRNN(units=50, input_shape=(time_step, 1), activation='relu', return_sequences=True))
# price_model.add(SimpleRNN(units=50, activation='relu'))
price_model.add(Dense(units=1, activation='linear'))
# 编译模型--主要是对模型进行的配置
price_model.compile(loss='mean_squared_error', optimizer='adam')
# 模型概述--就是可以简单查看一下模型怎么样
price_model.summary()                

# 训练模型--输入训练集特征值x, 目标值y, 每次喂给模型的样本数batch_size=30, 训练次数epochs=200
# batch_size越小, 模型泛化能力越好,但是训练起来需要的时间和算力也就更多, 
# epoch是指使用全部样本训练一次, 每个epoch中包含多个batch_size,
# 两者的关系: 设有一百样本, batch_size=20, epochs=10, 那么系统会将100样本分成五次投喂给模型, 而这, 就完成了一次epoch, 类似操作还需要重复九次
price_model.fit(x_train, y_train, batch_size=30, epochs=10)


# =======================================================================
# 4.对模型进行预测
y_train_predict=price_model.predict(x_train)*max(price)
y=[i*max(price) for i in y_train]
# print(len(y_predict),len(y), y_predict, y)

# 对预测情况进行数据可视化: 
import matplotlib.pyplot as plt

fig_contrast = plt.figure(figsize=(4,3))  # 增大图形大小以便更好比较
plt.plot(y_train, label='y_real')                # 绘制实际值
plt.plot(y_train_predict, label='y_predict')     # 绘制预测值
plt.title('Y_Real VS Y_Predict')
plt.xlabel('Time')
plt.ylabel('Price')
plt.legend()                                # 添加图例
plt.show()

代码实现(AI版)

效果预览:

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense
from tensorflow.keras.callbacks import EarlyStopping
from sklearn.metrics import mean_squared_error, mean_absolute_error

# 读取数据
zgpa_data = pd.read_csv(r"C:\Users\鹰\Desktop\zgpa.csv")

# 选择特征
price = zgpa_data.loc[:, 'close']

# 特征预处理-对数据进行归一化
scaler = MinMaxScaler(feature_range=(0, 1))
price_norm = scaler.fit_transform(price.values.reshape(-1, 1)).flatten()

# 对处理后的结果进行可视化
plt.figure(figsize=(12, 6))
plt.plot(price)
plt.xlabel('time')
plt.ylabel('price')
plt.title('zgpa_price')
plt.show()

# 数据分段截取
def extract_data(data, time_step):
    x = []
    y = []
    for i in range(len(data) - time_step):
        x.append([a for a in data[i:i + time_step]])
        y.append(data[i + time_step])
    
    x = np.array(x)
    y = np.array(y)
    
    # 增加维度, 是为了让数据能够进入神经网络层, 因为LSTM需要接受的是三维数据
    x = x.reshape(x.shape[0], x.shape[1], 1)
    return x, y

time_step = 8  # 每次截取的数据长度==8
x, y = extract_data(price_norm, time_step)

# 划分训练集和测试集
split = int(0.8 * len(x))
x_train, x_test = x[:split], x[split:]
y_train, y_test = y[:split], y[split:]

# 构建LSTM模型
model = Sequential()
model.add(LSTM(units=50, input_shape=(time_step, 1), activation='relu', return_sequences=True))
model.add(LSTM(units=50, activation='relu'))
model.add(Dense(units=1, activation='linear'))

model.compile(loss='mean_squared_error', optimizer='adam')
model.summary()

# 使用早停法
early_stopping = EarlyStopping(monitor='val_loss', patience=10, restore_best_weights=True)

# 训练模型
history = model.fit(x_train, y_train, batch_size=32, epochs=200, validation_split=0.2, callbacks=[early_stopping])

# 预测
y_predict = model.predict(x_test)

# 反归一化
y_test_inv = scaler.inverse_transform(y_test.reshape(-1, 1)).flatten()
y_predict_inv = scaler.inverse_transform(y_predict).flatten()

# 评估
mse = mean_squared_error(y_test_inv, y_predict_inv)
rmse = np.sqrt(mse)
mae = mean_absolute_error(y_test_inv, y_predict_inv)

print(f'Mean Squared Error (MSE): {mse}')
print(f'Root Mean Squared Error (RMSE): {rmse}')
print(f'Mean Absolute Error (MAE): {mae}')

# 绘制结果
plt.figure(figsize=(4,3))
plt.plot(y_test_inv, label='Actual Price')
plt.plot(y_predict_inv, label='Predicted Price')
plt.title('Y_Real VS Y_Predict')
plt.xlabel('Time')
plt.ylabel('Price')
plt.legend()
plt.show()