这是清华大学今年（2023年）刚出的一篇关于时间序列的论文TimesNet：https://openreview.net/pdf?id=ju_Uqw384Oq

 GitHub地址：GitHub - thuml/Time-Series-Library: A Library for Advanced Deep Time Series Models.

在论文的实验部分中，TimesNet在短期、长期预测、分类、异常检测以及缺失值处理这5个任务上都展现出了超越其他模型的效果，能够作为一个时间序列任务的通用基础模型（Foundation Model）。【Youth PhD Talk】ICLR 预讲会（三）_哔哩哔哩_bilibili 在B站上有论文一作的讲解，在差不多2:02:48那里。底下那个指路的评论就是我了。

接下来就从论文理解以及代码的角度来解释一下TimesNet。

一、时间序列的二维变化

和其他深度学习任务（图像以及自然语言处理）不同，尽管时间序列是连续记录的，然而每个时间点只记录了一些标量，语义信息不足，所以研究都集中在数据的时间变化上（temporal variation）。然而现实的时间序列数据，通常都是由各个有着不同周期的不同因素耦合在一起的，增加了建模难度。并且，时间点本身不仅受到本身缩在周期影响，相邻周期也会对这一周期的时间点产生影响。文中将这2种影响时间序列的变化分别称之为“期内变化”(intraperiod-variation)和“期间变化”(interperiod-variation)。为了将这2种变化区分开来，文中将一维的时间序列数据转换为了二维空间数据：

 其中，蓝色轴方向的店代表期间变化，可以看作是“不同周期，同一相位上时间点的数值”，而红色轴代表期内变化，即“统一周期上的不同时间点”。

这样做不仅将期间与期内变化分离了开来，同时二维数据还具有局部性（locality，如下图所示），使得一些用以作图像处理的方法也可以在这上面使用。

 

具体在实操时，使用傅里叶变换将时序转换到频域，观察前k个振幅的点，使用它们（频率）的倒数获得周期，以此确定不同周期用以作二维分解。

def FFT_for_Period(x, k=2):
    # [B, T, C]
    xf = torch.fft.rfft(x, dim=1)
    # find period by amplitudes
    frequency_list = abs(xf).mean(0).mean(-1) #展平
    frequency_list[0] = 0 #第0项代表周期正无穷，舍去
    _, top_list = torch.topk(frequency_list, k)#前k个振幅（能量）的点
    top_list = top_list.detach().cpu().numpy()
    period = x.shape[1] // top_list #周期
    return period, abs(xf).mean(-1)[:, top_list]#返回周期与振幅（权重）

具体将一维转化为二维的代码在下文的TimesBlock中。

二、TimesBlock

TimesNet的结构参考了ResNet，期望成为时间序列任务的Backbone之一。而TimesBlock也对标ResBlock，具有相似性：

 如图所示，每个TimesBlock的输入使用FFT_for_Period选取其topk个周期，然后循环对每个周期，将其展开为二维图像（即将时间序列每个周期的数据拆分为并排的n列），在这样的二维数据上，就可以使用2D-kernel的卷积核进行特征提取，不仅因为上文提到的locality能够让图像处理的结构也能很好地运作，而且卷积核本身速度也很快，增加了性能。论文中使用了GoogleNet中的Inception Block，将其称为“高效初始块”（parameter-efficient inception block）。

class Inception_Block_V1(nn.Module):
    #GoogleNet的Inception Block，降低参数量，网络结构稀疏但是能够产生稠密数据
    #将多个Conv2d并联起来，增加宽度，实现多尺度
    def __init__(self, in_channels, out_channels, num_kernels=6, init_weight=True):
        super(Inception_Block_V1, self).__init__()
        self.in_channels = in_channels
        self.out_channels = out_channels
        self.num_kernels = num_kernels
        kernels = []
        for i in range(self.num_kernels):
            kernels.append(nn.Conv2d(in_channels, out_channels, kernel_size=2 * i + 1, padding=i))
        self.kernels = nn.ModuleList(kernels)
        if init_weight:
            self._initialize_weights()

    def _initialize_weights(self):
        for m in self.modules():
            if isinstance(m, nn.Conv2d):
                nn.init.kaiming_normal_(m.weight, mode='fan_out', nonlinearity='relu')
                if m.bias is not None:
                    nn.init.constant_(m.bias, 0)
    def forward(self, x):
        res_list = []
        for i in range(self.num_kernels):
            res_list.append(self.kernels[i](x))
        res = torch.stack(res_list, dim=-1).mean(-1)
        return res

最后，针对于不同周期二维化后Conv2D的结果，我们还需要对它们进行自适应聚合（文中的adaptive aggregation）。具体而言就是使用了各个周期在FFT后对应的不同的能量（或是理解为振幅）将其经过Softmax函数后作为权重将各个结果加权求和。这样就得到了一个TimesBlock的结果。和ResNet一样，我们将TimesBlock的结果与输入相加，作为下一个TimesBlock的输入。

具体的代码如下，我在各个阶段加入了注释以便更容易读懂。

class TimesBlock(nn.Module):
    def __init__(self, configs):
        super(TimesBlock, self).__init__()
        self.seq_len = configs.seq_len
        self.pred_len = configs.pred_len
        self.k = configs.top_k
        # parameter-efficient design
        self.conv = nn.Sequential(
            Inception_Block_V1(configs.d_model, configs.d_ff,
                               num_kernels=configs.num_kernels),
            nn.GELU(),
            Inception_Block_V1(configs.d_ff, configs.d_model,
                               num_kernels=configs.num_kernels)
        )

    def forward(self, x):
        B, T, N = x.size()
        period_list, period_weight = FFT_for_Period(x, self.k)

        res = []
        for i in range(self.k):
            period = period_list[i]
            # padding 将最后不满足一个周期的部分填0
            if (self.seq_len + self.pred_len) % period != 0:
                length = (
                                 ((self.seq_len + self.pred_len) // period) + 1) * period
                padding = torch.zeros([x.shape[0], (length - (self.seq_len + self.pred_len)), x.shape[2]]).to(x.device)
                out = torch.cat([x, padding], dim=1)
            else:
                length = (self.seq_len + self.pred_len)
                out = x
            # reshape 改为2D,每个周期分开
            out = out.reshape(B, length // period, period,
                              N).permute(0, 3, 1, 2).contiguous()
            # 2D conv: from 1d Variation to 2d Variation 使用二维卷积核处理
            out = self.conv(out)
            # reshape back
            out = out.permute(0, 2, 3, 1).reshape(B, -1, N)
            res.append(out[:, :(self.seq_len + self.pred_len), :])
        res = torch.stack(res, dim=-1)
        # adaptive aggregation 根据振幅权重做自适应聚合
        period_weight = F.softmax(period_weight, dim=1)
        period_weight = period_weight.unsqueeze(
            1).unsqueeze(1).repeat(1, T, N, 1)
        res = torch.sum(res * period_weight, -1)
        # residual connection
        res = res + x
        return res

三、Embedding部分

这一部分实际上并非TimesNet的论文内容，但如果你看GitHub上的项目，你会发现作者为了方便使用者，将很多时序模型打包放在一起了，在使用时只要修改参数就可以了。这也是为什么TimesNet明明没有“encoding decoding”结构，你依然能够在Model函数中找到x_dec, x_mark_dec这2个参数（实际上这2个参数并没有用到）。TimesNet的DataEmbedding部分似乎直接采用了Informer的DataEmbedding策略，下面也顺带介绍一下此处DataEmbedding的三个组成部分：

1、TokenEmbedding

将序列上的每一个时间点数据都做编码，使用Conv1d进行，nn.Conv1d是对最后一个维度来做embed的，比直接用Linear多了个"kernal_size"可以做滑动窗口。

class TokenEmbedding(nn.Module):
    def __init__(self, c_in, d_model):
        super(TokenEmbedding, self).__init__()
        padding = 1 if torch.__version__ >= '1.5.0' else 2
        self.tokenConv = nn.Conv1d(in_channels=c_in, out_channels=d_model,
                                   kernel_size=3, padding=padding, padding_mode='circular', bias=False)
        for m in self.modules():
            if isinstance(m, nn.Conv1d):
                nn.init.kaiming_normal_(
                    m.weight, mode='fan_in', nonlinearity='leaky_relu')

    def forward(self, x):
        x = self.tokenConv(x.permute(0, 2, 1)).transpose(1, 2)
        return x

2、 PositionalEmbedding

实际上，PositionalEmbedding在Transformer中已经有了，这一部分的主要任务，是将说数据的位置信息使用正弦余弦函数加入到建模中。它不仅解决了Transformer对称性的问题，增加了位置信息，并且位置信息也会随距离增大逐渐衰减。

具体的数学原理可以参照这里：Transformer升级之路：1、Sinusoidal位置编码追根溯源 - 科学空间|Scientific Spaces

 

class PositionalEmbedding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super(PositionalEmbedding, self).__init__()
        # Compute the positional encodings once in log space.
        pe = torch.zeros(max_len, d_model).float()
        pe.require_grad = False

        position = torch.arange(0, max_len).float().unsqueeze(1)
        div_term = (torch.arange(0, d_model, 2).float()
                    * -(math.log(10000.0) / d_model)).exp()

        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)

        pe = pe.unsqueeze(0)
        self.register_buffer('pe', pe)

    def forward(self, x):
        return self.pe[:, :x.size(1)]

3、 TemporalEmbedding

针对年、月、日、小时等多个时间段使用多个不同的embedding层处理输入的时间戳，将结果相加。其中FixedEmbedding是使用sin cos作为位置编码以代替原本pytorch中的Embedding，注意FixedEmbedding里面所有的参数都不会在训练过程中更新。

class TemporalEmbedding(nn.Module):
    def __init__(self, d_model, embed_type='fixed', freq='h'):
        super(TemporalEmbedding, self).__init__()

        minute_size = 4
        hour_size = 24
        weekday_size = 7
        day_size = 32
        month_size = 13

        Embed = FixedEmbedding if embed_type == 'fixed' else nn.Embedding
        if freq == 't':
            self.minute_embed = Embed(minute_size, d_model)
        self.hour_embed = Embed(hour_size, d_model)
        self.weekday_embed = Embed(weekday_size, d_model)
        self.day_embed = Embed(day_size, d_model)
        self.month_embed = Embed(month_size, d_model)

    def forward(self, x):
        x = x.long()
        minute_x = self.minute_embed(x[:, :, 4]) if hasattr(
            self, 'minute_embed') else 0.
        hour_x = self.hour_embed(x[:, :, 3])
        weekday_x = self.weekday_embed(x[:, :, 2])
        day_x = self.day_embed(x[:, :, 1])
        month_x = self.month_embed(x[:, :, 0])

        return hour_x + weekday_x + day_x + month_x + minute_x

class FixedEmbedding(nn.Module):
    def __init__(self, c_in, d_model):
        super(FixedEmbedding, self).__init__()

        w = torch.zeros(c_in, d_model).float()
        w.require_grad = False

        position = torch.arange(0, c_in).float().unsqueeze(1)
        div_term = (torch.arange(0, d_model, 2).float()
                    * -(math.log(10000.0) / d_model)).exp()

        w[:, 0::2] = torch.sin(position * div_term)
        w[:, 1::2] = torch.cos(position * div_term)

        self.emb = nn.Embedding(c_in, d_model)
        self.emb.weight = nn.Parameter(w, requires_grad=False)

    def forward(self, x):
        return self.emb(x).detach()

四、总结

TimesNet通过傅里叶变换提取周期，将一维的时间序列转化为n个周期并排的二维序列，以此能够使用二维卷积以及类ResBlock的结构提取特征，在短期、长期预测、分类、异常检测以及缺失值处理这5个任务上都展现出了超越其他模型的效果。这个模型结构相较于其他Transformer-base的模型而言并不复杂，论文中主要篇幅都是在讲使用模型作的实验。这一架构模型依然显示出了比其他模型较优的性能，论文作者的理论以及代码水平可见一斑。