基于LSTM+Transformer对时间序列进行预测包含多步和单步预测

项目描述与实现：基于LSTM+Transformer对时间序列进行预测包含多步和单步预测
创新点：
1、利用LSTM+Transformer模型来获取时间序列特征。
整体流程：
1、多变量数据特征制作与数据预处理
2、LSTM-Transformer模型制作
3、模型训练预测及可视化展示

1

1

📈 基于LSTM+Transformer的时间序列预测：多步与单步预测实现

---

🚀 创新点概述

1. LSTM+Transformer模型融合

• LSTM：捕捉时间序列的短期依赖关系，适用于局部特征提取。
• Transformer：利用自注意力机制捕捉长期依赖关系，增强全局信息理解。

2. 多步与单步预测结合

• 单步预测：预测下一个时间点的值。
• 多步预测：预测未来多个时间点的值，更贴近实际应用场景。

---

📊 整体流程详解

1. 多变量数据特征制作与数据预处理

1.1 数据加载与可视化

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

# 加载数据（示例）
data = np.load('your_data.npy')  # 替换为你的数据路径

plt.figure(figsize=(14, 6))
plt.plot(data, color='green')
plt.title('Time Series Data')
plt.xlabel('Time Steps')
plt.ylabel('Values')
plt.grid(True)
plt.show()

1.2 数据标准化与滑动窗口划分

from sklearn.preprocessing import StandardScaler

def create_dataset(data, window_size):
    X, y = [], []
    for i in range(len(data) - window_size):
        X.append(data[i:i + window_size])
        y.append(data[i + window_size])
    return np.array(X), np.array(y)

# 数据标准化
scaler = StandardScaler()
scaled_data = scaler.fit_transform(data.reshape(-1, 1))

# 滑动窗口划分
window_size = 24  # 示例窗口大小
X, y = create_dataset(scaled_data, window_size)

2. LSTM-Transformer模型制作

2.1 导入必要的库

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset

2.2 定义LSTM-Transformer模型

class LSTMTransformer(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers, output_size, num_heads, dropout=0.1):
        super(LSTMTransformer, self).__init__()
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.transformer = nn.TransformerEncoder(
            nn.TransformerEncoderLayer(d_model=hidden_size, nhead=num_heads, dropout=dropout),
            num_layers=num_layers
        )
        self.fc = nn.Linear(hidden_size, output_size)

    def forward(self, x):
        lstm_out, _ = self.lstm(x)
        transformer_out = self.transformer(lstm_out.permute(1, 0, 2)).permute(1, 0, 2)
        output = self.fc(transformer_out[:, -1, :])
        return output

# 模型参数设置
input_size = 1
hidden_size = 64
num_layers = 2
output_size = 1
num_heads = 8

model = LSTMTransformer(input_size, hidden_size, num_layers, output_size, num_heads)

3. 模型训练预测及可视化展示

3.1 数据集准备与模型训练

# 划分训练集和测试集
train_size = int(len(X) * 0.8)
X_train, X_test = X[:train_size], X[train_size:]
y_train, y_test = y[:train_size], y[train_size:]

# 转换为PyTorch张量
X_train_tensor = torch.tensor(X_train, dtype=torch.float32)
y_train_tensor = torch.tensor(y_train, dtype=torch.float32)
X_test_tensor = torch.tensor(X_test, dtype=torch.float32)
y_test_tensor = torch.tensor(y_test, dtype=torch.float32)

train_dataset = TensorDataset(X_train_tensor, y_train_tensor)
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)

# 训练模型
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

num_epochs = 100
for epoch in range(num_epochs):
    model.train()
    for batch_X, batch_y in train_loader:
        optimizer.zero_grad()
        outputs = model(batch_X)
        loss = criterion(outputs, batch_y)
        loss.backward()
        optimizer.step()
    print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {loss.item():.4f}')

3.2 预测与结果可视化

# 单步预测
model.eval()
with torch.no_grad():
    test_outputs = model(X_test_tensor)

# 反标准化
predicted = scaler.inverse_transform(test_outputs.numpy())
actual = scaler.inverse_transform(y_test)

# 绘制预测结果
plt.figure(figsize=(14, 6))
plt.plot(actual, label='Actual', color='blue')
plt.plot(predicted, label='Predicted', color='red')
plt.title('Single-Step Prediction')
plt.xlabel('Time Steps')
plt.ylabel('Values')
plt.legend()
plt.grid(True)
plt.show()

# 多步预测（示例代码，需根据具体需求调整）
multi_step_predictions = []
current_input = X_test_tensor[0].unsqueeze(0)

for _ in range(len(X_test)):
    with torch.no_grad():
        output = model(current_input)
    multi_step_predictions.append(output.item())
    current_input = torch.cat([current_input[:, 1:, :], output.unsqueeze(0).unsqueeze(0)], dim=1)

multi_step_predictions = scaler.inverse_transform(np.array(multi_step_predictions).reshape(-1, 1))

plt.figure(figsize=(14, 6))
plt.plot(actual, label='Actual', color='blue')
plt.plot(multi_step_predictions, label='Multi-Step Predicted', color='orange')
plt.title('Multi-Step Prediction')
plt.xlabel('Time Steps')
plt.ylabel('Values')
plt.legend()
plt.grid(True)
plt.show()

---

📚 详细代码实现：LSTM + Transformer 时间序列预测

---

📑 数据加载与预处理

1. dataloader 函数实现

import torch
from torch.utils.data import DataLoader, TensorDataset

def dataloader(batch_size, workers=0):
    """
    数据加载函数: dataloader
    功能: 加载预处理好的训练集和测试集，并转换为PyTorch的DataLoader格式
    参数:
        - batch_size: 每个batch包含多少样本
        - workers: 多进程加载数据的线程数，默认0表示不使用多线程
    返回值:
        - train_loader: 训练集的DataLoader
        - test_loader: 测试集的DataLoader
    """

    # 示例数据加载，实际应用中请替换为你的数据加载逻辑
    X_train = torch.randn(1000, 12, 7)  # 假设训练集有1000个样本，每个样本长度为12，特征维度为7
    y_train = torch.randn(1000, 1)      # 假设训练集标签为1维输出
    X_test = torch.randn(200, 12, 7)    # 假设测试集有200个样本
    y_test = torch.randn(200, 1)        # 假设测试集标签为1维输出

    # 创建TensorDataset
    train_dataset = TensorDataset(X_train, y_train)
    test_dataset = TensorDataset(X_test, y_test)

    # 创建DataLoader
    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=workers)
    test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False, num_workers=workers)

    return train_loader, test_loader

---

🧠 LSTM + Transformer 混合模型定义

2. LSTMTransformer 类实现

import torch
import torch.nn as nn

class LSTMTransformer(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim, lstm_layers, transformer_layers, num_heads, dropout=0.1):
        super(LSTMTransformer, self).__init__()
        
        # LSTM层
        self.lstm = nn.LSTM(input_size=input_dim, hidden_size=hidden_dim, num_layers=lstm_layers, batch_first=True)
        
        # Transformer层
        encoder_layer = nn.TransformerEncoderLayer(d_model=hidden_dim, nhead=num_heads, dropout=dropout)
        self.transformer_encoder = nn.TransformerEncoder(encoder_layer, num_layers=transformer_layers)
        
        # 全连接层
        self.fc = nn.Linear(hidden_dim, output_dim)
        
    def forward(self, x):
        # LSTM提取时序局部依赖关系
        lstm_out, _ = self.lstm(x)
        
        # Transformer建模全局依赖
        transformer_out = self.transformer_encoder(lstm_out.permute(1, 0, 2)).permute(1, 0, 2)
        
        # 取最后一个时间步的输出作为最终输出
        final_output = transformer_out[:, -1, :]
        
        # 全连接层输出
        output = self.fc(final_output)
        
        return output

---

🚀 模型训练与预测

3. 模型训练与预测流程

import torch.optim as optim

# 超参数设置
input_dim = 7
hidden_dim = 64
output_dim = 1
lstm_layers = 2
transformer_layers = 2
num_heads = 8
dropout = 0.1
batch_size = 32
epochs = 50
learning_rate = 0.001

# 初始化模型、损失函数和优化器
model = LSTMTransformer(input_dim, hidden_dim, output_dim, lstm_layers, transformer_layers, num_heads, dropout)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

# 获取数据加载器
train_loader, test_loader = dataloader(batch_size)

# 训练模型
for epoch in range(epochs):
    model.train()
    running_loss = 0.0
    
    for inputs, labels in train_loader:
        optimizer.zero_grad()
        
        outputs = model(inputs)
        loss = criterion(outputs, labels)
        
        loss.backward()
        optimizer.step()
        
        running_loss += loss.item()
    
    print(f'Epoch [{epoch+1}/{epochs}], Loss: {running_loss/len(train_loader):.4f}')

# 测试模型
model.eval()
test_loss = 0.0
predictions = []

with torch.no_grad():
    for inputs, labels in test_loader:
        outputs = model(inputs)
        loss = criterion(outputs, labels)
        test_loss += loss.item()
        
        predictions.extend(outputs.numpy().flatten())

print(f'Test Loss: {test_loss/len(test_loader):.4f}')

---

📊 结果可视化

4. 预测结果可视化

import matplotlib.pyplot as plt

# 假设y_test是真实标签
y_test = [item[0] for item in test_loader.dataset.tensors[1].numpy()]

plt.figure(figsize=(14, 6))
plt.plot(y_test, label='Actual', color='blue')
plt.plot(predictions, label='Predicted', color='red')
plt.title('Time Series Prediction')
plt.xlabel('Time Steps')
plt.ylabel('Values')
plt.legend()
plt.grid(True)
plt.show()

---