当前位置：首页 > news >正文

时间序列预测实操

news 来源：原创 2025/5/22 19:47:18

#入相关库

import torch
import torch.nn as nn
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler

# 加载数据，并查看

data = pd.read_csv('stock.csv')
print(data)

# 仅保留 'Open'、'High'、'Low'、'Close'、'Volume' 列
data = data[['open', 'high', 'low', 'close', 'volume']]
print(data)

# 处理缺失值

# 缺失值统计
missing_values = data.isnull().sum()
print("缺失值统计：\n", missing_values)

# 查看包含缺失的行
missing_rows = data[data.isnull().any(axis=1)]
print("包含缺失值的行：\n", missing_rows)

# 填充缺失值

data = data.fillna(method='ffill')

# 检查每列的缺失值
missing_values = data.isnull().sum()
print("缺失值统计：\n", missing_values)

# 检查每行的缺失值
missing_rows = data[data.isnull().any(axis=1)]
print("包含缺失值的行：\n", missing_rows)

# 数据归一化

# 将数据缩放到 [0, 1] 范围
scaler = MinMaxScaler(feature_range=(0, 1))
data_scaled = scaler.fit_transform(data)
print(data_scaled)

# 转换为 PyTorch Tensor
data_scaled = torch.FloatTensor(data_scaled)
print(data_scaled.size())

# 数据拆分

def create_sequences(data, seq_length):
    sequences = []
    labels = []
    for i in range(len(data) - seq_length):
        seq = data[i:i+seq_length]
        label = data[i+seq_length:i+seq_length+1][:, 3]  # 'Close' 列作为目标变量
        sequences.append(seq)
        labels.append(label)
    return torch.stack(sequences), torch.stack(labels)


# 设置序列长度和划分训练集与测试集
sequence_length = 10
train_size = int(len(data_scaled) * 0.8)


# 创建训练集和测试集
train_data, test_data = data_scaled[:train_size], data_scaled[train_size:]
print(train_data.size())
print(test_data.size())


# 创建输入序列和标签
train_sequences, train_labels = create_sequences(train_data, sequence_length)
test_sequences, test_labels = create_sequences(test_data, sequence_length)
print(train_sequences.size())
print(train_labels.size())
print(test_sequences.size())
print(test_labels.size())

# 选择预测模型

# 定义 LSTM 模型
class LSTMModel(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers, output_size):
        super(LSTMModel, self).__init__()
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, output_size)

    def forward(self, x):
        out, _ = self.lstm(x) # LSTM层
        out = self.fc(out[:, -1, :]) # 全连接层
        return out

# 定义超参数
input_size = 5
hidden_size = 64
num_layers = 1
output_size = 1
learning_rate = 0.001
epochs = 500


# 实例化模型
model = LSTMModel(input_size, hidden_size, num_layers, output_size)

# 定义损失函数和优化器

loss_function = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)

# 训练模型

for epoch in range(epochs):
    outputs = model(train_sequences)
    optimizer.zero_grad()
    loss = loss_function(outputs, train_labels)
    loss.backward()
    optimizer.step()
    print(f'Epoch [{epoch+1}/{epochs}], Loss: {loss.item()}')

# 测试模型

model.eval()  # 设置模型为 evaluation 模式
test_loss = 0.0
predictions, true_values = [], []
with torch.no_grad():
        outputs = model(test_sequences)
        # 注意：在测试阶段不需要进行反向传播和参数更新
        # 保存预测结果和真实值
        predictions.append(outputs.numpy())
        true_values.append(test_labels.numpy())

# 逆归一化

predictions = np.array(predictions).reshape(-1, 1)
predictions = np.repeat(predictions, 5, axis=-1)
predictions =scaler.inverse_transform(predictions)[:,0]

true_values = np.array(true_values).reshape(-1, 1)
true_values = np.repeat(true_values, 5, axis=-1)
true_values =scaler.inverse_transform(true_values)[:,0]

# 绘制预测结果

plt.plot(predictions, label='Predictions')
plt.plot(true_values, label='True Values')
plt.title('Stock Price Prediction using LSTM')
plt.xlabel('Time')
plt.ylabel('Stock Price')
plt.legend()
plt.show()

# 导入实际数据

newdata = pd.read_csv('actual_data.csv')
newdata = newdata[['open', 'high', 'low', 'close', 'volume']]
scaler = MinMaxScaler(feature_range=(0, 1))
data_scaled = scaler.fit_transform(newdata)
data_scaled = torch.FloatTensor(data_scaled)

# 保存并加载训练后的模型

torch.save(model.state_dict(), 'trained_model.pth')
trained_model = LSTMModel(input_size, hidden_size, num_layers, output_size)
trained_model.load_state_dict(torch.load('trained_model.pth'))
trained_model.eval()

# 预测

x = data_scaled[-sequence_length:].unsqueeze(0)
with torch.no_grad():
    prediction = trained_model(x)
prediction = np.repeat(prediction, 5, axis=-1)
prediction =scaler.inverse_transform(np.reshape(prediction,(len(prediction),5)))[:,0]
print(f'Predicted value for the next day: {prediction}')

代码和数据文件链接：

https://download.csdn.net/download/qq_14845119/90460926