GRU代码预测出错，结果很差

Question

我写了一个GRU的时序模型预测的代码，当我通过30天的7个特征数据，预测第31天的7个维度数据。模型训练和预测的效果还可以。

但是当我开始利用滚动预测，也就是通过预测出来的31天的数据，填充进去再把第一天的数据剔除并预测，效果就开始差了。。

。具体的代码如下：不知道哪里有问题。

import torch
import torch.nn as nn
import numpy as np
import yfinance as yf
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
from torch.utils.data import DataLoader, TensorDataset
import matplotlib.pyplot as plt
from datetime import datetime
import warnings

warnings.filterwarnings('ignore')

# 下载并预处理数据
def download_preprocess_data(ticker, start_date, end_date):
    data = yf.download(ticker, start=start_date, end=end_date)
    data = data[['Open', 'Close', 'Volume']]
    data['MA5'] = data['Close'].rolling(window=5).mean()
    data['MA10'] = data['Close'].rolling(window=10).mean()
    data['MA20'] = data['Close'].rolling(window=20).mean()
    data['EMA'] = data['Close'].ewm(span=20, adjust=False).mean()
    data.dropna(inplace=True)
    return data

def normalize_data(data):
    scaler = MinMaxScaler(feature_range=(0, 1))
    return scaler.fit_transform(data), scaler

def create_sequences(data, look_back):
    X, y = [], []
    max_index = len(data) - look_back
    for i in range(max_index):
        X.append(data[i:(i + look_back)])  # 特征窗口：从第i天到第i+look_back-1天
        y_target = data[i + look_back]  # 预测下一个时间步的所有7个特征
        y.append(y_target)
    return np.array(X, dtype=np.float32), np.array(y, dtype=np.float32)



data = download_preprocess_data('AAPL', '2014-01-01', datetime.now().strftime('%Y-%m-%d'))

# x_data, y_data = create_sequences(data.values, look_back = 30, steps_ahead=30)
scaled_data, scaler = normalize_data(data.values)
look_back = 30
x_data, y_data = create_sequences(scaled_data, look_back)

# Split data into train and test sets
train_size = int(len(x_data) * 0.8)
x_train, x_test = x_data[:train_size], x_data[train_size:]
y_train, y_test = y_data[:train_size], y_data[train_size:]
print("x_train shape:", x_train.shape)
print("y_train shape:", y_train.shape)
print("Data types:", x_train.dtype, y_train.dtype)
# Convert to PyTorch tensors
train_dataset = TensorDataset(torch.from_numpy(x_train).float(), torch.from_numpy(y_train).float())
train_loader = DataLoader(train_dataset, batch_size=16, shuffle=False)

# Define the GRU model
class GRUModel(nn.Module):
    def __init__(self, input_size, hidden_size1=256, hidden_size2=128, output_size=7):
        super(GRUModel, self).__init__()
        self.gru1 = nn.GRU(input_size, hidden_size1, num_layers=1, batch_first=True)
        self.gru2 = nn.GRU(hidden_size1, hidden_size2, num_layers=1, batch_first=True)
        self.dropout = nn.Dropout(0.2)
        self.dense = nn.Linear(hidden_size2, output_size)

    def forward(self, x):
        gru_out1, _ = self.gru1(x)
        gru_out1 = self.dropout(gru_out1)
        gru_out2, _ = self.gru2(gru_out1)
        gru_out2 = gru_out2[:, -1, :]  # 只取最后一个时间步的输出
        predictions = self.dense(gru_out2)
        return predictions

model = GRUModel(input_size=x_train.shape[2], hidden_size1=256, hidden_size2=128, output_size=7)
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.0001)

# Train the model
epochs = 1
for epoch in range(epochs):
    for inputs, targets in train_loader:
        optimizer.zero_grad()
        outputs = model(inputs)
        loss = criterion(outputs, targets.unsqueeze(1))
        loss.backward()
        optimizer.step()
    print(f'Epoch {epoch+1}/{epochs}, Loss: {loss.item()}')

model.eval()

# 预测x_test的全部数据
with torch.no_grad():
    full_test_predictions = model(torch.from_numpy(x_test).float())  # 预测所有测试数据

# 准备滚动预测的初始输入：使用x_test的最后一个样本
last_inputs = x_test[-1]  # 这应该是一个[30, 7]的数组

# 使用模型进行初始预测
with torch.no_grad():
    last_inputs_tensor = torch.from_numpy(last_inputs).float().unsqueeze(0)  # 添加批次维度
    initial_prediction = model(last_inputs_tensor).numpy()

# 更新第30天的数据
current_input = np.vstack([last_inputs[1:], initial_prediction])  # 移除第一天，添加新预测的天

predicted_features = [initial_prediction.flatten()]  # 保存初始预测结果

# 滚动预测30次
for _ in range(30):
    with torch.no_grad():
        # 对当前输入进行预测
        current_input_tensor = torch.from_numpy(current_input[np.newaxis, :, :]).float()
        prediction = model(current_input_tensor).numpy()
        # 更新预测结果列表
        predicted_features.append(prediction.flatten())
        # 更新输入数据：移除第一个时间步，添加新的预测结果
        current_input = np.vstack([current_input[1:], prediction])

# 将预测结果转换为numpy数组
predicted_features = np.array(predicted_features)

print(predicted_features)

# 反归一化处理
full_predictions = np.zeros((predicted_features.shape[0], scaled_data.shape[1]))
full_predictions[:, 1] = predicted_features[:, 1]  # 取出Close价格特征
predictions_inv = scaler.inverse_transform(full_predictions)[:, 1]

# 准备绘图数据
start_date = data.index[train_size + look_back]
valid_dates = pd.date_range(start=start_date, periods=len(predicted_features), freq='D')

# 创建DataFrame
predictions_df = pd.DataFrame({
    'Predictions': predictions_inv,
    'Actual': data['Close'][train_size + look_back:train_size + look_back + len(predicted_features)].values
}, index=valid_dates)

# Visualization
plt.figure(figsize=(16, 8))
plt.title('Future Close Price Predictions Including Rolling Predictions')
plt.xlabel('Date')
plt.ylabel('Close Price USD ($)')

# 绘制预测结果和实际数据
plt.plot(predictions_df['Predictions'], label='Predicted Future Close', color='red', linestyle='--')
plt.plot(predictions_df['Actual'], label='Actual Close Price', color='blue', alpha=0.5)
plt.legend()
plt.show()

Answer 1

以下回复引自免费微信小程序：皆我百晓生。 如问题解决，还请采纳。如仍有疑问，请回复留言或私信博主，以便获得更精准的帮助。

要解决这个问题，首先需要了解如何使用滚动预测在PYTHON中实现GRU模型，并且理解滚动预测的概念。滚动预测是一种特殊的GRU模型，它可以根据输入数据的变化来更新参数值，从而达到更好的预测效果。以下是一个基本的步骤：

1. 理解滚动预测

滚动预测（ROLLING GRADIENTS）是一种在GRU网络中的一种特殊形式，它允许在每次迭代中更新参数。滚动预测在GRU中起着关键作用，因为它可以帮助模型更好地适应输入变化的影响。

2. 使用滚动预测进行预测

例如，在一个GRU神经网络中，可以有多个GRUS分别对输入进行处理。每个GRU都有自己的权重和偏置，用于计算输出。滚动预测允许在每次迭代中更新这些权重和偏置，以确保在不同时间点上得到相似的结果。

3. 滚动预测的具体实现

滚动预测通常涉及以下几个步骤：

• 初始化：在每次迭代之前，根据输入数据调整模型的参数。
• 更新：在每次迭代后，从当前状态中减去前一次的状态，然后乘以一个滚动因子，即ROLLING_FACTOR。
• 求导：应用滚动因子来计算新的梯度，这个新梯度会随着滚动因子的增加而增长。

4. 在PYTHON中的滚动预测示例

假设我们有一个简单的GRU神经网络，其中包含三个GRUS，每个GRU都负责对输入进行不同的操作：

CLASS GRU(NN.MODULE):
    DEF __INIT__(SELF, INPUT_SIZE, HIDDEN_SIZE, OUTPUT_SIZE):
        SUPER(GRU, SELF).__INIT__()
        SELF.INPUT_SIZE = INPUT_SIZE
        SELF.HIDDEN_SIZE = HIDDEN_SIZE
        SELF.OUTPUT_SIZE = OUTPUT_SIZE
        
        # 初始化权重和偏置
        SELF.GRU = NN.GRU(INPUT_SIZE, HIDDEN_SIZE)
        
    DEF FORWARD(SELF, X):
        # 进行前向传播
        H_N, C_N = SELF.GRU(X)
        # 将隐藏状态和偏置添加到输出
        RETURN H_N, (C_N,)

在这个例子中，INPUT_SIZE是输入的维度，HIDDEN_SIZE是每个GRU的输入维度，OUTPUT_SIZE是每个GRU的输出维度。

5. 测试滚动预测

现在我们可以测试滚动预测的实际效果了。假设我们的数据集已经准备好了，包括输入、输出以及一些标签信息。我们可以这样使用滚动预测：

DEF PREDICT(MODEL, DATA_LOADER, ROLLING_FACTOR=1):
    MODEL.EVAL()
    
    WITH TORCH.NO_GRAD():
        FOR INPUTS, TARGETS IN DATA_LOADER:
            OUTPUTS, (H_N, C_N) = MODEL(INPUTS)
            
            IF ROLLING_FACTOR > 0:
                OUTPUTS = OUTPUTS * ROLLING_FACTOR
                
            # 对于每一个目标，计算期望输出
            PREDICTIONS = []
            FOR I IN RANGE(LEN(TARGETS)):
                PRED = OUTPUTS[I].ARGMAX(DIM=-1).ITEM()
                
                # 添加目标对应的预测结果到列表
                PREDICTIONS.APPEND(PRED)

            # 更新滚动因子
            ROLLING_FACTOR += 1
            
            # 返回最近的预测
            RETURN PREDICTIONS[-1]

# 测试函数
IF __NAME__ == "__MAIN__":
    DEVICE = TORCH.DEVICE("CUDA" IF TORCH.CUDA.IS_AVAILABLE() ELSE "CPU")
    
    MODEL = GRU(64, 256, 1)
    MODEL.TO(DEVICE)
    
    DATA_LOADER = DATALOADER(TRAIN_DATASET, BATCH_SIZE=32, SHUFFLE=TRUE)
    ROLLING_FACTOR = 1
    
    PRINT(F"STARTING PREDICTION ON ROLLING FACTOR {ROLLING_FACTOR}")
    PREDICTIONS = PREDICT(MODEL, DATA_LOADER, ROLLING_FACTOR)

这段代码展示了如何使用滚动预测在PYTHON中预测特定的时间步数。请注意，这只是一个基本的例子，实际的预测过程可能需要更复杂的逻辑和更多的细节。