【深度学习】使用Pytorch实现的用于时间序列预测的各种深度学习模型类

偏执的太偏执、 2024-02-21 12:06 28阅读 0赞

#### 深度学习模型类 ####

*  简介
 *  按滑动时间窗口切割数据集
 *  模型类
 *   *  CNN
     *  GRU
     *  LSTM
     *  MLP
     *  RNN
     *  TCN
     *  Transformer
     *  Seq2Seq

## 简介 ##

本文所定义模型类的输入数据的形状shape统一为 `[batch_size, time_step，n_features]`，batch\_size为批次大小，time\_step为时间步长，n\_features为特征数量。另外该模型类同时适用于单特征与多特征

本项目代码统一了训练方式，只需在models文件夹中加入下面模型类，即可使用该模型，而不需要重新写训练模型等的代码，减少了代码的冗余。

代码有注释，不加解释  
声明：转载请标明出处  
![在这里插入图片描述][2655c97aa74d496f8ceacd86e8a0be6a.png]  
超参数只需通过字典定义传入即可，所有训练方式一样的模型通用  
![在这里插入图片描述][203a745d3288405490f793ec429a1ebe.png]

## 按滑动时间窗口切割数据集 ##

import os
    
    import pandas as pd
    import torch
    from sklearn.preprocessing import MinMaxScaler
    from torch.utils.data import Dataset, DataLoader
    
    
    class TimeSeriesDataset(Dataset):
        """
        自定义的时间序列数据集类，用于处理时间序列数据的加载和预处理。目的是将时间序列数据准备成适合机器学习模型训练的格式(按滑动窗口划分)。
    
        Args:
            data (torch.Tensor): 包含时间序列数据的张量，形状为 [1, n_features, data_len]
            args.time_step (int): 输入数据的时间步。
            args.skip (int): 输入数据的跳跃步。
    
         Returns:
                tuple: 包含输入数据 x 和目标数据 y 的元组。
                X: 输入数据的批次，形状为 [time_step, n_features]
                Y: 目标数据的批次，形状为 [1]
    
        """
    
        def __init__(self, data, args):
            self.data = data
            self.time_step = args.time_step
            self.skip = args.skip
    
        def __len__(self):
            n = self.data.shape[-1] - self.time_step + 1 - self.skip
            return n
    
        def __getitem__(self, idx):
            x = self.data[:, :, idx:idx + self.time_step].permute(2, 1, 0).squeeze(-1)
            y = self.data[:, :1, idx + self.time_step + self.skip - 1].view(-1)
            return x, y
    
    
    def Dataset_Custom(args, if_Batching=True):
        """
        创建自定义时间序列数据集
    
        Args:
            args (Namespace): 包含所有必要参数的命名空间。
            if_Batching: 是否批次化，类似XGBoost算法不需要
    
        Returns:
            Tuple[TimeSeriesDataset, TimeSeriesDataset, TimeSeriesDataset]: 训练、验证和测试数据集
        """
    
        # 读取数据
        data = pd.read_csv(os.path.join(args.root_path, args.data_path))
    
        # 检查是否存在 "date" 列，如果存在则删除
        if 'date' in data.columns:
            data = data.drop('date', axis=1)
    
        # 确保 "load" 列是第一列，如果不是，将其移到第一列
        if args.target in data.columns:
            data = data[[args.target] + [col for col in data.columns if col != args.target]]
    
        # 定义数据集划分比例（例如，70% 训练集，10% 验证集，20% 测试集）
        data_len = len(data)
        num_train = int(data_len * 0.7)
        num_test = int(data_len * 0.2)
        num_vali = data_len - num_train - num_test
    
        train_data = data[:num_train]
        vali_data = data[num_train:num_train + num_vali]
        test_data = data[num_train + num_vali:]
    
        # 归一化
        scaler = MinMaxScaler(feature_range=(0, 1))
    
        scaler.fit(train_data)
        train_data = scaler.transform(train_data)
        vali_data = scaler.transform(vali_data)
        test_data = scaler.transform(test_data)
    
        # 转换为张量并添加维度
        train_data = torch.from_numpy(train_data).float()
        vali_data = torch.from_numpy(vali_data).float()
        test_data = torch.from_numpy(test_data).float()
    
        # 将其变为[1, n_features, data_len]
        train_data = train_data.unsqueeze(0).permute(0, 2, 1)
        vali_data = vali_data.unsqueeze(0).permute(0, 2, 1)
        test_data = test_data.unsqueeze(0).permute(0, 2, 1)
    
        # 按滑动时间窗口转成机器学习的数据格式
        training_dataset = TimeSeriesDataset(train_data, args)
        valiing_dataset = TimeSeriesDataset(vali_data, args)
        testing_dataset = TimeSeriesDataset(test_data, args)
        print(f"train:{
          len(training_dataset)},vali:{
          len(valiing_dataset)},test:{
          len(testing_dataset)}")
        if if_Batching:
            # 创建数据加载器，用于批量加载数据
            train_loader = DataLoader(training_dataset, shuffle=True, drop_last=True, batch_size=args.batch_size)
            vali_loader = DataLoader(valiing_dataset, shuffle=True, drop_last=True, batch_size=args.batch_size)
            test_loader = DataLoader(testing_dataset, shuffle=False, drop_last=False, batch_size=len(testing_dataset))
            return train_loader, vali_loader, test_loader
        else:
            return training_dataset, valiing_dataset, testing_dataset

## 模型类 ##

### CNN ###

import torch
    from torch import nn
    
    
    
    
    
    class Model(nn.Module):
        def __init__(self, configs):
            super(Model, self).__init__()
            self.input_size = configs.input_size  # 输入特征的大小
            self.output_size = configs.output_size  # 预测结果的维度
            self.time_step = configs.time_step  # 时间步数
            self.kernel_size = configs.kernel_size  # 卷积核的大小
    
            self.relu = nn.ReLU(inplace=True)  # ReLU激活函数
    
            # 第一个卷积层
            self.conv1 = nn.Sequential(
                nn.Conv1d(in_channels=self.input_size, out_channels=64, kernel_size=self.kernel_size),
                # 输入特征维度为input_size，输出通道数为64，卷积核大小为kernel_size
                nn.ReLU(),  # ReLU激活函数
                nn.MaxPool1d(kernel_size=self.kernel_size, stride=1)  # 最大池化，池化窗口大小为kernel_size，步长为1
            )
    
            # 第二个卷积层
            self.conv2 = nn.Sequential(
                nn.Conv1d(in_channels=64, out_channels=128, kernel_size=2),
                # 输入通道数为64，输出通道数为128，卷积核大小为2
                nn.ReLU(),  # ReLU激活函数
                nn.MaxPool1d(kernel_size=self.kernel_size, stride=1)  # 最大池化，池化窗口大小为kernel_size，步长为1
            )
    
            # 根据卷积操作后的数据格式和输出大小计算线性层的输入维度
            conv_output_size = self._calculate_conv_output_size()
    
            # 线性层1，输入维度为卷积层输出大小，输出维度为50
            self.linear1 = nn.Linear(conv_output_size, 50)
    
            # 线性层2，输入维度为50，输出维度为预测结果的维度
            self.linear2 = nn.Linear(50, self.output_size)
    
        def forward(self, x):
            x = x.transpose(1, 2)
            x = self.conv1(x)
            x = self.conv2(x)
            x = x.view(x.size(0), -1)
            x = self.linear1(x)
            x = self.relu(x)
            x = self.linear2(x)
            x = x.view(x.shape[0], -1)
            return x
    
        def _calculate_conv_output_size(self):
            """自动计算卷积层输出形状，此操作避免手算该参数 参数:input_size( 输入特征维度 )返回值: conv_output_size (卷积层输出的特征维度)"""
            input_tensor = torch.zeros(1, self.input_size, self.time_step)  # 创建输入零张量，维度为(1, input_size, time_step)
            conv1_output = self.conv1(input_tensor)
            conv2_output = self.conv2(conv1_output)
            conv_output_size = conv2_output.view(conv2_output.size(0), -1).size(1)
            return conv_output_size

### GRU ###

import torch
    from torch import nn
    from torch.autograd import Variable
    
    
    # GRU模型结构
    class Model(nn.Module):
        def __init__(self, configs):
            super(Model, self).__init__()
    
            # 初始化模型参数
            self.output_size = configs.output_size  # 输出类别的数量
            self.num_layers = configs.num_layers  # GRU层数
            self.input_size = configs.input_size  # 输入特征的维度
            self.hidden_size = configs.hidden_size  # 隐藏状态的维度
            self.dropout = configs.dropout  # 在非循环层之间应用的丢弃比例，默认为0.0（没有丢弃）。
            self.bidirectional = configs.bidirectional  # 是否使用双向GRU
    
            # 创建GRU层  batch_first=True：输入数据的维度顺序是 (batch_size, seq_len, input_size)
            self.gru = nn.GRU(input_size=self.input_size, hidden_size=self.hidden_size, num_layers=self.num_layers,
                              dropout=self.dropout, bidirectional=self.bidirectional, batch_first=True)
            # 创建全连接层用于输出预测结果
            self.fc = nn.Linear(self.hidden_size, self.output_size)
    
        def forward(self, x):
            # 初始化初始隐藏状态
            h_0 = Variable(torch.zeros(self.num_layers, x.size(0), self.hidden_size))
            # 通过GRU层进行前向传播
            out, h_0 = self.gru(x, h_0)
            # 取GRU的最后一个时间步的输出
            out = out[:, -1]
            # 通过全连接层进行分类预测
            out = self.fc(out)
            return out

### LSTM ###

import torch
    import torch.nn as nn
    from torch.autograd import Variable
    
    
    class Model(nn.Module):
        def __init__(self, configs):
            super(Model, self).__init__()
            self.input_size = configs.input_size  # 输入特征的大小。
            self.hidden_size = configs.hidden_size  # LSTM 隐藏状态的维度
            self.num_layers = configs.num_layers  # LSTM 层的堆叠层数
            self.output_size = configs.output_size  # 输出的大小（预测结果的维度）
            self.dropout = configs.dropout  # 在非循环层之间应用的丢弃比例，默认为0.0（没有丢弃）。
            self.bidirectional = configs.bidirectional  # 如果为True，LSTM将是双向的（包括前向和后向），默认为False。
    
            self.lstm = nn.LSTM(input_size=configs.input_size, hidden_size=configs.hidden_size,
                                num_layers=configs.num_layers, dropout=self.dropout, bidirectional=self.bidirectional,
                                batch_first=True)  # 定义 LSTM 层
            self.fc = nn.Linear(configs.hidden_size, configs.output_size)  # 定义线性层，将 LSTM 输出映射到预测结果的维度
    
        def forward(self, x):
            h_0 = Variable(torch.zeros(self.num_layers, x.size(0), self.hidden_size))  # 初始化 LSTM 的隐藏状态
            c_0 = Variable(torch.zeros(self.num_layers, x.size(0), self.hidden_size))  # 初始化 LSTM 的记忆状态
            ula, (h_out, _) = self.lstm(x, (h_0, c_0))  # 前向传播过程，返回 LSTM 层的输出序列和最后一个时间步的隐藏状态
            h_out = h_out[-1, :, :].view(-1, self.hidden_size)  # 提取最后一个时间步的隐藏状态并进行形状变换
            ''''使用LSTM的最后一个时间步的隐藏状态h_out作为线性层的输入，是因为模型将隐藏状态视为包含了序列信息的高层表示。
                在许多情况下，使用最后一个时间步的隐藏状态进行预测已经足够。'''
            out = self.fc(h_out)  # 将最后一个时间步的隐藏状态通过线性层进行预测
            return out

### MLP ###

import torch.nn as nn
    
    
    # MLP模型结构
    class Model(nn.Module):
        def __init__(self, configs):
            super(Model, self).__init__()
    
            self.input_size = configs.input_size  # 输入特征的大小，即输入层的维度。
            self.output_size = configs.output_size  # 输出的大小，即预测结果的维度。
            self.channel_sizes = [int(size) for size in configs.channel_sizes.split(',')]  # 因为输入的是字符串，转成列表
    
            # 一个整数列表，指定每个隐藏层的单元数量。列表的长度表示隐藏层的层数，每个元素表示对应隐藏层的单元数量。
            self.time_step = configs.time_step
            layers = []
            input_adjust_size = self.input_size * self.time_step  # 将数据展开后的维度为 特征数量*时间步长
            # 遍历channel_sizes列表
            for i in range(len(self.channel_sizes)):
                if i == 0:
                    # 对于第一层，创建一个从input_size到channel_sizes[i]的线性层
                    self.linear = nn.Linear(input_adjust_size, self.channel_sizes[i])
                    self.init_weights()  # 初始化线性层的权重
                    layers += [self.linear, nn.ReLU()]  # 将线性层和ReLU激活函数添加到layers列表中
                else:
                    # 对于后续层，创建一个从channel_sizes[i-1]到channel_sizes[i]的线性层
                    self.linear = nn.Linear(self.channel_sizes[i - 1], self.channel_sizes[i])
                    self.init_weights()  # 初始化线性层的权重
                    layers += [self.linear, nn.ReLU()]  # 将线性层和ReLU激活函数添加到layers列表中
            # 创建最后一个线性层，从channel_sizes[-1]到output_size
            self.linear = nn.Linear(self.channel_sizes[-1], self.output_size)
            self.init_weights()  # 初始化线性层的权重
            layers += [self.linear]  # 将最后一个线性层添加到layers列表中
            # 使用layers列表创建一个Sequential网络
            self.network = nn.Sequential(*layers)
    
        def init_weights(self):
            # 使用均值为0，标准差为0.01的正态分布初始化线性层的权重
            self.linear.weight.data.normal_(0, 0.01)
    
        def forward(self, x):
            # X输入的shape为 [batch_size, time_step，n_features]
    
            # 将输入数据的维度展平，然后传递给线性层。
            # 无论输入数据的特征数和时间步数如何，都能适应到模型中。这个修改应该可以适用于不同维度的输入数据。
            x = x.view(x.size(0), -1)
            return self.network(x)

### RNN ###

import torch
    import torch.nn as nn
    from torch.autograd import Variable
    
    
    # RNN模型结构
    class Model(nn.Module):
        def __init__(self, configs):
            super(Model, self).__init__()
            self.input_size = configs.input_size  # 输入特征的大小
            self.hidden_size = configs.hidden_size  # 隐藏状态的维度
            self.num_layers = configs.num_layers  # RNN的层数
            self.output_size = configs.output_size  # 输出的大小，即预测结果的维度
            self.dropout = configs.dropout  # 在非循环层之间应用的丢弃比例，默认为0.0（没有丢弃）。
            self.bidirectional = configs.bidirectional  # 如果为True，LSTM将是双向的（包括前向和后向），默认为False。
            # 定义RNN结构，输入特征大小、隐藏状态维度、层数等参数
            self.rnn = nn.RNN(input_size=self.input_size, hidden_size=self.hidden_size, dropout=self.dropout,
                              bidirectional=self.bidirectional, num_layers=self.num_layers, batch_first=True)
            # 将RNN的输出压缩到与输出大小相同的维度
            self.fc = nn.Linear(self.hidden_size, self.output_size)
    
        def forward(self, x):
            # 创建初始隐藏状态h_0，维度为(num_layers, batch_size, hidden_size)
            h_0 = Variable(torch.zeros(self.num_layers, x.size(0), self.hidden_size))
            # 通过RNN传播输入数据，获取输出out和最终隐藏状态h_0
            out, h_0 = self.rnn(x, h_0)
            # 取RNN最后一个时间步的输出，将其输入到全连接层进行预测
            out = self.fc(out[:, -1, :])
            return out

### TCN ###

import torch.nn as nn
    from torch.nn.utils import weight_norm
    
    # TCN模型结构
    '''
    Chomp1d模块：用于从卷积层的输出中移除无效的时间步。
    '''
    
    
    class Chomp1d(nn.Module):
        def __init__(self, chomp_size):
            super(Chomp1d, self).__init__()
            self.chomp_size = chomp_size
    
        def forward(self, x):
            # Chomp1d模块的作用是从卷积层的输出中移除无效的时间步，即通过切片操作去掉最后的self.chomp_size个时间步。
            # 由于切片操作可能导致存储不连续，因此在返回结果之前，需要使用contiguous()方法确保存储连续性。
            return x[:, :, :-self.chomp_size].contiguous()
    
    
    '''
    TemporalBlock模块：包含两个卷积层和相应的正则化、激活函数和dropout操作。
    第一个卷积层使用权重归一化，通过Chomp1d组件移除无效的时间步，然后经过ReLU激活函数和dropout操作。
    第二个卷积层也经过相同的处理流程。通过残差连接将第二个卷积层的输出和输入进行相加，并通过ReLU激活函数得到最终的输出。
    '''
    
    
    class TemporalBlock(nn.Module):
        def __init__(self, n_inputs, n_outputs, kernel_size, stride, dilation, padding, dropout=0.2):
            super(TemporalBlock, self).__init__()
            # 第一次卷积
            self.conv1 = weight_norm(nn.Conv1d(n_inputs, 3096, kernel_size,
                                               stride=stride, padding=padding, dilation=dilation))
            self.chomp1 = Chomp1d(padding)
            self.relu1 = nn.ReLU()
            # 随机失活（dropout）。随机失活是一种常用的正则化技术，用于减少过拟合
            self.dropout1 = nn.Dropout(dropout)
    
            # 第二次卷积
            self.conv2 = weight_norm(nn.Conv1d(3096, n_outputs, kernel_size,
                                               stride=stride, padding=padding, dilation=dilation))
            self.chomp2 = Chomp1d(padding)
            self.relu2 = nn.ReLU()
            self.dropout2 = nn.Dropout(dropout)
    
            # 将两次卷积层按顺序组合成一个网络
            self.net = nn.Sequential(
                self.conv1, self.chomp1, self.relu1, self.dropout1,
                self.conv2, self.chomp2, self.relu2, self.dropout2,
            )
            # 如果输入通道数和输出通道数不相同，则需要进行下采样
            self.downsample = nn.Conv1d(n_inputs, n_outputs, 1) if n_inputs != n_outputs else None
            self.relu = nn.ReLU()
            # 初始化权重
            self.init_weights()
    
        def init_weights(self):
            # 使用均值为0，标准差为0.01的正态分布初始化权重
            self.conv1.weight.data.normal_(0, 0.01)
            self.conv2.weight.data.normal_(0, 0.01)
            if self.downsample is not None:
                self.downsample.weight.data.normal_(0, 0.01)
    
        def forward(self, x):
            # 前向传播
            out = self.net(x)  # 通过两个卷积层
            res = x if self.downsample is None else self.downsample(x)  # 下采样
            return self.relu(out + res)  # 残差连接
        # 残差连接（Residual connection）是一种在神经网络中引入跨层连接的技术。它的目的是解决深层神经网络训练中的梯度消失或梯度爆炸问题，并促进信息在网络中的流动。
        # 在TCN模型中，残差连接被用于将每个TemporalBlock的输出与输入进行相加。这种设计使得信息可以直接通过跨层连接流动，有助于梯度的传播和模型的训练。
        # 具体地，在TemporalBlock的forward方法中，首先通过两个卷积层进行特征提取和建模，然后将第二个卷积层的输出和输入进行相加。这个相加的操作实现了残差连接。最终，通过ReLU激活函数对相加的结果进行非线性变换。
        # 残差连接的好处是，即使在网络较深的情况下，梯度可以通过跨层连接直接传播到前面的层次，减少了梯度消失的问题。同时，它也提供了一种捕捉输入与输出之间的细微差异和变化的机制，有助于提高模型的性能。
    
    
    '''
    TemporalConvNet模块是 TCN 模型的核心，由多个TemporalBlock组成的网络层次结构。
    根据num_channels列表的长度，确定网络层次的深度。
    每个层次上的TemporalBlock的参数根据当前层次的位置和前一层的输出通道数进行确定。
    通过层次化的结构，模型可以捕捉序列中的长期依赖关系。
    '''
    
    
    class TemporalConvNet(nn.Module):
        def __init__(self, num_inputs, num_channels, kernel_size=2, dropout=0.2):
            super(TemporalConvNet, self).__init__()
            layers = []
            self.relu = nn.ReLU()
            num_levels = len(num_channels)
            for i in range(num_levels):
                dilation_size = 2 ** i
                in_channels = num_inputs if i == 0 else num_channels[i - 1]
                out_channels = num_channels[i]
                # 每个层次添加一个TemporalBlock
                layers += [TemporalBlock(in_channels, out_channels, kernel_size, stride=1, dilation=dilation_size,
                                         padding=(kernel_size - 1) * dilation_size, dropout=dropout)]
            # 将所有的TemporalBlock按顺序组合成一个网络
            self.network = nn.Sequential(*layers)
    
        def forward(self, x):
            return self.relu(self.network(x) + x[:, 0, :].unsqueeze(1))
    
    
    '''
    TCN模块：TCN模型的主体部分。
    它包括一个TemporalConvNet，一个线性层和一个下采样层。
    输入数据首先经过TemporalConvNet进行序列建模和特征提取，然后通过ReLU激活函数和残差连接进行处理。
    最后，通过线性层进行预测，并通过下采样层将输入数据的通道数降低到与TemporalBlock的输出通道数相同，以便在残差连接中使用。
    '''
    
    
    class Model(nn.Module):
        def __init__(self,configs):  # input_size 输入的不同的时间序列数目
            super(Model, self).__init__()
            self.input_size = configs.input_size  # 输入特征的大小
            self.output_size = configs.output_size  # 预测结果的维度
            self.num_channels = [configs.nhid] * configs.levels  # 卷积层通道数的列表，用于定义TemporalConvNet的深度
            self.kernel_size = configs.kernel_size  # 卷积核的大小
            self.dropout = configs.dropout  # 随机丢弃率
    
            # TemporalConvNet层
            self.tcn = TemporalConvNet(self.input_size, self.num_channels, kernel_size=self.kernel_size, dropout=self.dropout)
            # 线性层，用于预测
            self.linear = nn.Linear(self.num_channels[-1], self.output_size)
            # 下采样层，用于通道数降低
            self.downsample = nn.Conv1d(self.input_size, self.num_channels[0], 1)
            self.relu = nn.ReLU()
            # 初始化权重
            self.init_weights()
    
        def init_weights(self):
            self.linear.weight.data.normal_(0, 0.01)
            self.downsample.weight.data.normal_(0, 0.01)
    
        def forward(self, x):
            # 前向传播
            x = x.transpose(1, 2)
            # 通过TemporalConvNet进行序列建模和特征提取
            y1 = self.relu(self.tcn(x) + x[:, 0, :].unsqueeze(1))
            # 线性层进行预测
            return self.linear(y1[:, :, -1])

### Transformer ###

import torch
    import torch.nn as nn
    
    from Models.layers.Transformer.decoder import Decoder
    from Models.layers.Transformer.encoder import Encoder
    from Models.layers.Transformer.utils import generate_original_PE, generate_regular_PE
    
    
    class Model(nn.Module):
        """基于Attention is All You Need的Transformer模型。
    
        适用于顺序数据的经典Transformer模型。
        嵌入（Embedding）已被替换为全连接层，
        最后一层softmax函数替换为sigmoid函数。
    
        属性
        ----------
        layers_encoding: :py:class:`list` of :class:`Encoder.Encoder`
            编码器层的堆叠。
        layers_decoding: :py:class:`list` of :class:`Decoder.Decoder`
            解码器层的堆叠。
    
        参数
        ----------
        d_input:
            模型输入的维度。
        d_model:
            输入向量的维度。
        d_output:
            模型输出的维度。
        q:
            查询和键的维度。
        v:
            值的维度。
        h:
            头数。
        N:
            要堆叠的编码器和解码器层数量。
        attention_size:
            应用注意力机制的反向元素数量。
            如果为 ``None``，则不激活。默认为 ``None``。
        dropout:
            每个多头自注意力（MHA）或前馈全连接（PFF）块之后的dropout概率。
            默认为 ``0.3``。
        chunk_mode:
            切块模式，可以是 ``'chunk'``、``'window'`` 或 ``None`` 之一。默认为 ``'chunk'``。
        pe:
            要添加的位置编码类型，可以是 ``'original'``、``'regular'`` 或 ``None`` 之一。默认为 ``None``。
        pe_period:
            如果使用 ``'regular'`` 位置编码，则可以定义周期。默认为 ``None``。
        """
    
        def __init__(self, configs):
            """根据Encoder和Decoder块创建Transformer结构。"""
            super(Model, self).__init__()
            d_input = configs.d_input
            d_model = configs.d_model
            d_output = configs.d_output
            q = configs.q
            v = configs.v
            h = configs.h
            N = configs.N
            attention_size = configs.attention_size
            dropout = configs.dropout
            chunk_mode = configs.chunk_mode
            pe = configs.pe
            pe_period = configs.pe_period
    
            self._d_model = d_model
    
            self.layers_encoding = nn.ModuleList([Encoder(d_model,
                                                          q,
                                                          v,
                                                          h,
                                                          attention_size=attention_size,
                                                          dropout=dropout,
                                                          chunk_mode=chunk_mode) for _ in range(N)])
            self.layers_decoding = nn.ModuleList([Decoder(d_model,
                                                          q,
                                                          v,
                                                          h,
                                                          attention_size=attention_size,
                                                          dropout=dropout,
                                                          chunk_mode=chunk_mode) for _ in range(N)])
    
            self._embedding = nn.Linear(d_input, d_model)
            self._linear = nn.Linear(d_model, d_output)
    
            pe_functions = {
        
                'original': generate_original_PE,
                'regular': generate_regular_PE,
            }
    
            if pe in pe_functions.keys():
                self._generate_PE = pe_functions[pe]
                self._pe_period = pe_period
            elif pe is None:
                self._generate_PE = None
            else:
                raise NameError(
                    f'未知的位置编码（PE）"{
          pe}"。必须为 {
          ", ".join(pe_functions.keys())} 或 None。')
    
            self.name = 'transformer'
    
        def forward(self, x: torch.Tensor) -> torch.Tensor:
            """通过Transformer进行输入传播。
    
            通过嵌入模块、编码器和解码器堆叠以及输出模块进行输入传播。
    
            参数
            ----------
            x:
                形状为 (batch_size, K, d_input) 的 torch.Tensor。
    
            返回
            -------
                形状为 (batch_size, K, d_output) 的输出张量。
            """
            K = x.shape[1]
    
            # 嵌入模块
            encoding = self._embedding(x)
    
            # 添加位置编码
            if self._generate_PE is not None:
                pe_params = {
        'period': self._pe_period} if self._pe_period else {
        }
                positional_encoding = self._generate_PE(K, self._d_model, **pe_params)
                positional_encoding = positional_encoding.to(encoding.device)
                encoding.add_(positional_encoding)
    
            # 编码器堆叠
            for layer in self.layers_encoding:
                encoding = layer(encoding)
    
            # 解码器堆叠
            decoding = encoding
    
            # 添加位置编码
            if self._generate_PE is not None:
                positional_encoding = self._generate_PE(K, self._d_model)
                positional_encoding = positional_encoding.to(decoding.device)
                decoding.add_(positional_encoding)
    
            for layer in self.layers_decoding:
                decoding = layer(decoding, encoding)
    
            # 输出模块
            output = self._linear(decoding)
            output = torch.sigmoid(output)
            return output[:, -1, :]

### Seq2Seq ###

import torch
    import torch.nn as nn
    
    # Seq2Seq 主类
    class Model(nn.Module):
        def __init__(self,configs ):
            super().__init__()
            self.input_size = configs.input_size  # 输入特征的大小
            self.output_size = configs.output_size  # 输出的大小（预测结果的维度）
            self.Encoder = Encoder(configs.input_size, configs.hidden_size, configs.num_layers, configs.batch_size)
            self.Decoder = Decoder(configs.input_size, configs.hidden_size, configs.num_layers, configs.output_size, configs.batch_size)
    
        def forward(self, input_seq):
            target_len = self.output_size  # 预测步长
            batch_size, seq_len, _ = input_seq.shape[0], input_seq.shape[1], input_seq.shape[2]
            h, c = self.Encoder(input_seq)
            outputs = torch.zeros(batch_size, self.input_size, self.output_size)
            decoder_input = input_seq[:, -1, :]  # 获取解码器的初始输入
            for t in range(target_len):
                decoder_output, h, c = self.Decoder(decoder_input, h, c)  # 解码器的前向传播
                outputs[:, :, t] = decoder_output
                decoder_input = decoder_output  # 将解码器的输出作为下一个时间步的输入
    
            return outputs[:, 0, :]  # 返回预测输出
    
    class Encoder(nn.Module):
        def __init__(self, input_size, hidden_size, num_layers, batch_size):
            super().__init__()
            self.input_size = input_size  # 输入特征的大小
            self.hidden_size = hidden_size  # LSTM 隐藏状态的维度
            self.num_layers = num_layers  # LSTM 层的堆叠层数
            self.num_directions = 1
            self.batch_size = batch_size
            # 定义编码器的LSTM层，接受输入序列
            self.lstm = nn.LSTM(self.input_size, self.hidden_size, self.num_layers, batch_first=True, bidirectional=False)
    
        def forward(self, input_seq):
            batch_size, seq_len = input_seq.shape[0], input_seq.shape[1]
            h_0 = torch.randn(self.num_directions * self.num_layers, batch_size, self.hidden_size)  # 初始化 LSTM 的隐藏状态
            c_0 = torch.randn(self.num_directions * self.num_layers, batch_size, self.hidden_size)  # 初始化 LSTM 的记忆状态
            output, (h, c) = self.lstm(input_seq, (h_0, c_0))  # 前向传播过程，返回 LSTM 层的输出序列和最后一个时间步的隐藏状态
    
            return h, c
    
    
    class Decoder(nn.Module):
        def __init__(self, input_size, hidden_size, num_layers, output_size, batch_size, *args, **kwargs):
            super().__init__(*args, **kwargs)
            self.input_size = input_size  # 输入特征的大小
            self.hidden_size = hidden_size  # LSTM 隐藏状态的维度
            self.num_layers = num_layers  # LSTM 层的堆叠层数
            self.output_size = output_size  # 输出的大小（预测结果的维度）
            self.num_directions = 1
            self.batch_size = batch_size
            # 定义解码器的LSTM层和线性层
            self.lstm = nn.LSTM(input_size, self.hidden_size, self.num_layers, batch_first=True, bidirectional=False)
            self.linear = nn.Linear(self.hidden_size, self.input_size)  # 线性层，将 LSTM 输出映射到预测结果的维度
    
        def forward(self, input_seq, h, c):
            # input_seq(batch_size, input_size)
            input_seq = input_seq.unsqueeze(1)  # 在输入序列中添加一个时间步的维度
            output, (h, c) = self.lstm(input_seq, (h, c))  # 前向传播过程，返回 LSTM 层的输出序列和最后一个时间步的隐藏状态
            pred = self.linear(output.squeeze(1))  # 使用线性层进行预测，pred(batch_size, 1, output_size)
    
            return pred, h, c

[2655c97aa74d496f8ceacd86e8a0be6a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/efca1961ab3946049232e9fc41510560.png
[203a745d3288405490f793ec429a1ebe.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/7677d41700a443f8808ea8ac040f8a44.png