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

《异常检测——从经典算法到深度学习》23 TimesNet: 用于常规时间序列分析的时间二维变化模型

news 文章来源：https://blog.csdn.net/smileyan9/article/details/132612181 2025/5/9 2:39:18

zz# 《异常检测——从经典算法到深度学习》

0 概论
1 基于隔离森林的异常检测算法
2 基于LOF的异常检测算法
3 基于One-Class SVM的异常检测算法
4 基于高斯概率密度异常检测算法
5 Opprentice——异常检测经典算法最终篇
6 基于重构概率的 VAE 异常检测
7 基于条件VAE异常检测
8 Donut: 基于 VAE 的 Web 应用周期性 KPI 无监督异常检测
9 异常检测资料汇总（持续更新&抛砖引玉）
10 Bagel: 基于条件 VAE 的鲁棒无监督KPI异常检测
11 ADS: 针对大量出现的KPI流快速部署异常检测模型
12 Buzz: 对复杂 KPI 基于VAE对抗训练的非监督异常检测
13 MAD: 基于GANs的时间序列数据多元异常检测
14 对于流数据基于 RRCF 的异常检测
15 通过无监督和主动学习进行实用的白盒异常检测
16 基于VAE和LOF的无监督KPI异常检测算法
17 基于 VAE-LSTM 混合模型的时间异常检测
18 USAD：多元时间序列的无监督异常检测
19 OmniAnomaly：基于随机循环网络的多元时间序列鲁棒异常检测
20 HotSpot：多维特征 Additive KPI 的异常定位
21 Anomaly Transformer: 基于关联差异的时间序列异常检测
22 Kontrast: 通过自监督对比学习识别软件变更中的错误
23TimesNet: 用于常规时间序列分析的时间二维变化模型

23. TimesNet: 用于常规时间序列分析的时间二维变化模型

论文名称：TIMESNET: TEMPORAL 2D-VARIATION MODELING FOR GENERAL TIME SERIES ANALYSIS
论文发表于 International Conference on Learning Representations 2023
论文下载：openreview
源码地址：https://github.com/thuml/TimesNet

23.1 论文概述

这篇论文的主要内容是介绍了一种新的方法来处理时间序列分析中的复杂时间模式。该方法将一维时间序列转换为二维张量，以便更好地发现多个周期内和周期间的变化。作者提出的TimesNet with TimesBlock 是一种通用的时间序列分析骨干（Backbone）网络，可以用于多种分析任务。该论文还介绍了一些实验结果和性能比较，以证明该方法的有效性和优越性。

该论文提出的方法主要包括以下技术：

TimesBlock：可以将一维时间序列自适应地转换为一组二维张量，并通过一个参数高效的Inception块来捕获二维空间中的周期内和周期间变化的模块。TimesBlock的设计使得它可以自适应地发现多个周期内和周期间的变化，并且可以通过Inception块来捕获这些变化。
TimesNet：这是一种通用的时间序列分析骨干网络，可以用于多种分析任务。TimesNet使用了TimesBlock模块来处理时间序列，并利用现有的视觉骨干网络来进一步提高性能。TimesNet的设计使得它可以自适应地发现多个周期内和周期间的变化，并且可以通过Inception块来捕获这些变化。
多周期性：该论文提出了一种新的思路，即将时间序列转换为二维张量，以便更好地发现多个周期内和周期间的变化。这种方法可以更好地捕获时间序列中的复杂模式，并且可以用于多种分析任务。
实验结果和性能比较：该论文通过实验结果和性能比较证明了TimesNet的有效性和优越性。作者在五个主流的时间序列分析任务中进行了实验，包括短期和长期预测、填充、分类和异常检测。实验结果表明，TimesNet在这些任务中均取得了一致的最先进性能。

23.2 相关技术

该论文提出的方法主要包括以下技术：

在该论文的RELATED WORK模块中，作者介绍了一些与他们的工作相关的先前研究。这些研究主要涉及时间序列分析和深度学习领域的一些关键技术和方法。以下是一些相关的研究：

时间序列分析：该部分介绍了时间序列分析的一些基本概念和方法，包括时间序列的基本特征、时间序列分解、周期性分析、自回归模型、移动平均模型等。
卷积神经网络（CNN）：该部分介绍了CNN在时间序列分析中的应用。CNN可以通过卷积操作来捕获时间序列中的局部模式，并且可以通过池化操作来减少参数数量和计算量。
循环神经网络（RNN）：该部分介绍了RNN在时间序列分析中的应用。RNN可以通过循环连接来处理时间序列中的时序信息，并且可以通过长短时记忆（LSTM）单元来处理长期依赖关系。
注意力机制：该部分介绍了注意力机制在时间序列分析中的应用。注意力机制可以通过对时间序列中不同部分的加权来提高模型的性能，并且可以通过自注意力机制来处理时间序列中的长期依赖关系。
时空卷积神经网络（ST-CNN）：该部分介绍了ST-CNN在时间序列分析中的应用。ST-CNN可以通过卷积操作来捕获时间序列中的时空信息，并且可以通过池化操作来减少参数数量和计算量。

23.3 核心方法

23.3.1 论文结构梳理

Section	内容概述
1. INTRODUCTION	相关背景介绍以及对本论文的概述
2. RELATED WORK	相关技术介绍
3. TIMESNET	论文主题部分
4. EXPERIMENTS	实验部分
5. CONCLUSION AND FUTURE WORK	总结以及未来工作

读者应当重点关注第3部分，如果需要对论文复现的话考虑读一些第4部分。

请一定不要太在意论文的结果复现时与你本地测试的结果不一致的问题，因为导致结果有所差异的原因很多，我们不是审稿员，也没有必要太较真实验结果。主要还是理清楚核心过程。

23.3.2 INTRODUCTION

本文从多周期性这一新的维度对时间序列进行分析，以解决复杂的时间变化问题。首先，我们观察到现实世界的时间序列通常呈现多周期性，例如天气观测的日变化和年变化，电力消耗的周变化和季度变化。这些多个时期相互重叠和相互作用，使得变化建模变得棘手。其次，对于每个时段，我们发现每个时间点的变化不仅受其相邻区域的时间模式的影响，而且与其相邻时段的变化高度相关。为了清楚起见，我们将这两种时间变化分别命名为期内变化和期间变化。前者表示一个时期内的短期时间模式。后者可以反映连续不同时期的长期趋势。注意，对于没有明确周期性的时间序列，变化将由周期内变化主导，并且等价于具有无限周期长度的时间序列。

由于不同的周期会导致不同的周期内和周期间的变化，多周期性可以自然地衍生出一个模块化的架构，用于时间变化建模，在那里我们可以捕获由某个特定的周期在一个模块中产生的变化。此外，这种设计使复杂的时间模式被解开，有利于时间变化建模。然而，值得注意的是，一维时间序列很难同时明确地呈现两种不同类型的变化。为了解决这个障碍，我们扩展到二维空间的时间变化的分析。具体来说，如图1所示，我们可以将一维时间序列重塑为二维张量，其中每列包含一个周期内的时间点，每行包含不同周期中同一阶段的时间点。因此，通过将一维时间序列变换为二维张量，可以突破一维空间的表征能力瓶颈，成功地将二维空间的周期内和周期间变化统一起来，得到时间二维变化。

在技术上，基于上述动机，我们超越了以前的骨干，提出了TimesNet作为一个新的任务通用模型的时间序列分析。通过TimesBlock的授权，TimesNet可以发现时间序列的多周期性，并在模块化架构中捕获相应的时间变化。具体来说，TimesBlock可以根据学习的周期自适应地将一维时间序列转换为一组二维张量，并通过参数高效的初始块进一步捕获二维空间中的周期内和周期间变化。在实验中，TimesNet在五个主流分析任务中实现了一致的最先进水平，包括短期和长期预测，插补，分类和异常检测。我们的贡献可归纳为三个方面：

基于多周期性和周期内、周期间复杂的相互作用，本文提出了一种时间变化建模的模块化方法。通过将一维时间序列转换到二维空间，我们可以同时呈现期内和期间的变化。
我们提出了TimesNet与TimesBlock发现多个周期和捕获时间的二维变化从变换的二维张量的参数有效的起始块。
作为一个通用的任务基础模型，TimesNet在五个主流的时间序列分析任务中达到了一致的先进水平。包括详细和有见地的可视化。

23.3.3 模型结构 1 —— 将一维变分变换为二维变分

如图1所示，每个时间点同时涉及两种类型的时间变化，即与其相邻区域的时间变化和不同时期之间的相同相位的时间变化，即期内变化和期间变化。然而，这种原始的时间序列的一维结构只能呈现相邻时间点之间的变化。为了解决这个问题，我们探索了时间变化的二维结构，它可以显式地呈现周期内和周期之间的变化，从而在表示能力方面具有更多优势，并有利于后续的表示学习。

具体地说，对于 $C$ 个长度为 $T$ 的时间序列记录变量（recorded variates），其原始一维组织形式为 $\mathbf{X}_\text{1D} \in \mathbb{R}^{T×C}$ 。为了表示周期间的变化，我们需要首先发现周期。从技术上讲，我们通过快速傅里叶变换（Fast Fourier Transform，FFT）在频域中分析时间序列如下：

$\mathbf{A}=\text{Avg}\left(\text{ Amp }\left(\text{ FFT}(\mathbf{X}_{\mathrm{lD}})\right)\right),\{f_1,\cdots,f_k\}=\text{ arg Topk }\left(\mathbf{A}\right),p_i=\left\lceil\frac{T}{f_i}\right\rceil,i\in\{1,\cdots,k\}.\tag{1}$

其中， $\text{FFT}(\cdot)$ 和 $\text{Amp}(\cdot)$ 表示FFT和幅度值的计算。 $\mathbf{A}\in \mathbb{R}^T$ 表示每个频率的计算振幅，这是从 $C$ 维平均 $\text{Avg}(\cdot)$ 。注意，第 $j$ 个值 $A_j$ 表示与周期长度对应的频率 $j$ 周期基函数的强度 $\lceil {\frac{T}{j}} \rceil$ 。考虑到频域的稀疏性和避免无意义的高频带来的噪声，我们只选择前 $k$ 个幅度值 $f_1,...,f_k$ 并获得最重要的频率 $\{\mathbf{A}_{f_1},....,\mathbf{A}_{f_k}\}$ ，其中 $k$ 是超参数。这些选择的频率也对应于 $k$ 个周期长度 ${p_i, ..., p_k\}$ 。由于频域的共轭性，我们只考虑 $\{1, ..., [\frac{T}{2}]\}$ 内的频率。我们将等式1总结如下：

$\mathbf{A},\{f_{1},\cdots,f_{k}\},\{p_{1},\cdots,p_{k}\}=\mathrm{Period}(\mathbf{X}_{1\text{D}} ) \tag{2}$

基于选定的频率 ${f_1, ..., f_k\}$ 和相应的周期长度 ${p_1, ..., p_k\}$ ，我们可以通过以下等式将一维时间序列 $\mathbf{X}_{1D} \in \mathbb{R}^{T\times C}$ 整形为多个二维张量：

$\mathbf{X}_{2 \mathrm{D}}^i=\operatorname{Reshape}_{p_i, f_i}\left(\operatorname{Padding}\left(\mathbf{X}_{1 \mathrm{D}}\right)\right), i \in\{1, \cdots, k\} \tag{3}$

其中 $\text{Padding}(\cdot)$ 是将时间序列沿着时间维度扩展零，以使其与 $\text{Reshape}_{p_i, f_i}$ $(\cdot)$ 兼容，其中 $p_i$ 和 $f_i$ 分别表示变换后的二维张量的行数和列数。注意， $\mathbf{X}^i_{\text{2D}} \in \mathbb{R}^{p_i \times f_i \times C}$ 表示基于频率 $f_i$ 的第 $i$ 个整型时间序列，其列和行分别表示相应周期长度 $p_i$ 下的周期内变化和周期间变化。最终，如图2 所示，基于所选择的频率和估计的周期，我们获得一组二维张量 $\mathbf{X}^1_{2D},...,\mathbf{X}^k_{2D}$ ，其指示由不同周期导出的 $k$ 个不同的时间二维张量。

同样值得注意的是，这种变换为变换后的2维张量带来了两种类型的局部，即相邻时间点（列，周期内变化）和相邻周期（行，周期间变化）之间的局部。因此，时间二维变换可以容易地由二维内核处理。

23.3.4 模型结构 2 —— TIMESBLOCK

如图3所示，我们以残差方式组织 TimesBlock。具体地说，对于长度为 $T$ 的一维输入时间序列 $\mathbf{X}_{\text{1D}} \in \mathbb{R}^{T \times C}$ ，我们首先通过嵌入层 $\mathbf{X}^0 _{\text{1D}} = \text{Embed}(\textbf{X}_{1\text{D}})$ 将原始输入投影到深度特征 $\mathbf{X}^0_{\text{1D}} \in \mathbb{R}^{T\times d}$ 模型中。对于 TimesNet 的第 $l$ 层，输入是 $\mathbf{X}^{l−1}_{\text{1D}} \in \mathbb{R}^{T\times d_{model}}$ ，该过程可以形式化为：

$\mathbf{X}_{1 \mathrm{D}}^l=\text { TimesBlock }\left(\mathbf{X}_{1 \mathrm{D}}^{l-1}\right)+\mathbf{X}_{1 \mathrm{D}}^{l-1} \tag{4}$

如图3所示，对于第 $L$ 个TimesBlock，整个过程包括两个连续的部分：捕获时间二维变化和自适应地聚合来自不同时期的表示。

捕获时间2D变化 类似于等式1，我们可以通过 $\text{Period}(\cdot)$ 估计深度特征 $\mathbf{X}^{l-1}_{\text{1D}}$ 的周期长度。基于估计的周期长度，我们可以将一维时间序列变换到二维空间，得到一组二维张量，从这组张量中我们可以得到二维空间的参数有效信息表示，并得到一组二维张量，从这组张量中我们可以方便地得到参数有效起始块的信息表示。该过程形式化如下：

$\begin{aligned} \mathbf{A}^{l-1},\left\{f_1, \cdots, f_k\right\},\left\{p_1, \cdots, p_k\right\} & =\operatorname{Period}\left(\mathbf{X}_{1 \mathrm{D}}^{l-1}\right) \\ \mathbf{X}_{2 \mathrm{D}}^{l, i} & =\operatorname{Reshape}_{p_i, f_i}\left(\operatorname{Padding}\left(\mathbf{X}_{1 \mathrm{D}}^{l-1}\right)\right), i \in\{1, \cdots, k\} \\ \widehat{\mathbf{X}}_{2 \mathrm{D}}^{l, i} & =\operatorname{Inception}\left(\mathbf{X}_{2 \mathrm{D}}^{l, i}\right), i \in\{1, \cdots, k\} \\ \widehat{\mathbf{X}}_{1 \mathrm{D}}^{l, i} & =\operatorname{Trunc}\left(\operatorname{Reshape}_{1,\left(p_i \times f_i\right)}\left(\widehat{\mathbf{X}}_{2 \mathrm{D}}^{l, i}\right)\right), i \in\{1, \cdots, k\}, \tag{5} \end{aligned}$

其中 $\mathbf{X}_{\text{2D}}^{l,i}\in\mathbb{R}^{p_i\times f_i \times d_{model}}$ 表示第 $i$ 个变换而得的二维向量。转换完成以后，我们通过参数有效的初始块将2D张量处理为初始 $\text{Inspection}(\cdot)$ ，该初始块涉及多维2D核（multi-scale 2D kernels），是最知名的视觉骨干之一。然后，我们将学习到的2D表示 $\widehat{\mathbf{X}}_{2 \mathrm{D}}^{l, i}$ 转换回一维空间 $\widehat{\mathbf{X}}_{1 \mathrm{D}}^{l, i} \in \mathbb{R}^{T \times d_{\text {model }}}$ 从而用于聚合，其中我们使用 $\text{Trunc}(\cdot)$ 将长度为 $(p_i \times f_i)$ 的填充序列截断为原始长度 $T$ 。

注意，得益于一维时间序列的转换，起始块中的2D核可以同时聚合多尺度周期内变化（列）和周期间变化（行），覆盖相邻时间点和相邻周期。此外，我们对不同的重构 2D 张量 $\left\{\mathbf{X}_{2 \mathrm{D}}^{l, 1}, \cdots, \mathbf{X}_{2 \mathrm{D}}^{l, k}\right\}$ 采用了共享的起始块来提高参数效率，这可以使模型大小对超参数k的选择保持不变。

自适应聚合 最后，我们需要为下一层融合 $k$ 个不同的 1D 表示 $\left\{\widehat{\mathbf{X}}_{1 \mathrm{D}}^{l, 1}, \cdots, \widehat{\mathbf{X}}_{1 \mathrm{D}}^{l, k}\right\}$ 。受自相关的启发，幅度 $A$ 可以反映所选择的频率和周期的相对重要性，从而对应于每个变换的2D张量的重要性。因此，我们基于振幅聚合1D表示：

$\begin{aligned} \widehat{\mathbf{A}}_{f_1}^{l-1}, \cdots, \widehat{\mathbf{A}}_{f_k}^{l-1} & =\operatorname{Softmax}\left(\mathbf{A}_{f_1}^{l-1}, \cdots, \mathbf{A}_{f_k}^{l-1}\right) \\ \mathbf{X}_{1 \mathrm{D}}^l & =\sum_{i=1}^k \widehat{\mathbf{A}}_{f_i}^{l-1} \times \widehat{\mathbf{X}}_{1 \mathrm{D}}^{l, i} \end{aligned} \tag{6}$

由于周期内和周期之间的变化已经涉及到多个高度结构化的2D张量，TimesBlock可以同时完全捕获多尺度时间2D变化。因此，TimesNet可以实现比直接从1D时间序列更有效的表示学习。

2D视觉主干的通用性 受益于1D时间序列到时间2D变化的转换，我们可以选择各种计算机视觉骨干来代替表示学习的初始块，例如广泛使用的 ResNet 和ResNeXt，先进的 ConvNeXt 和基于注意力的模型。因此，我们的时间二维变化设计也将一维时间序列与蓬勃发展的二维视觉骨干连接起来，使时间序列分析能够利用计算机视觉社区的发展。一般来说，用于表示学习的更强大的2D主干将带来更好的性能。考虑到性能和效率（图4右），我们基于参数高效的初始块进行主要实验，如公式5所示。
在这里插入图片描述

23.4 论文实验

特此声明：如果不打算基于这篇论文的源码开发，亦或者不需要了解代码细节，完全不需要运行所有源码。主要把论文思路梳理清楚就差不多了。因为复现这些源码比较麻烦，至少需要一张还过得去的显卡。不过有条件、有时间、感兴趣的同学可以试试。

23.4.1 数据准备

前去本论文的提到的github地址可以找到对应的谷歌云盘、清华云盘（当前已经失效）以及百度云盘的下载链接，如果觉得麻烦的话，可以考虑访问我的夸克云盘，地址如下：

夸克云盘
链接：https://pan.quark.cn/s/b167f0d17234
提取码：wQKX

数据文件解压的相对地址等，将在后面介绍。

23.4.2 源码准备

前去本论文提到的github地址，把源码clone或下载到本地，然后把前面下载的数据集压缩包接下，注意相对路径，而且需要把解压后的文件夹重命名为 dataset。如图所示：
在这里插入图片描述

23.4.3 安装相关依赖

为了尽可能地减少麻烦，先保证使用的python版本与论文实验一致，即 3.8，其次确保安装好了与显卡对应的驱动，确保cuda可用，这些方面的坑太多太多，所以为了避免继续踩坑，尽可能与原论文的版本保持一致。如图所示：
在这里插入图片描述

23.4.4 执行脚本

安装依赖以后，我们不妨写几行代码，确保 cuda 可用。

import torchprint(torch.cuda.is_available())

在这里插入图片描述
接着可以直接执行作者已经给我们写好的脚本，注意，可能出现GPU不能用的情况

在这里插入图片描述
编辑这个 sh 脚本文件，并如图所示，修改使用GPU的索引为0，因为我只有一张显卡，只能使用索引为0的显卡。

在这里插入图片描述

其他脚本皆是如此，如果只有一张显卡，就把脚本第一行最后一个值改为 0。

其他的脚本运行方法均是如此，唯一需要注意的就是 export CUDA_VISIBLE_DEVICES=0 确保是自己想用的显卡的序号。

23.5 速读源码

23.5.1 layers/Conv_Blocks.py

这段代码定义了两个不同版本的 Inception 模块（Inception_Block_V1 和 Inception_Block_V2），这些模块包含多个卷积核，用于从输入数据中提取不同尺度的特征。这些模块可以用于深度学习模型中，以提高特征提取的多样性。模块中的卷积核具有不同的大小和填充，以捕获不同尺度的信息。在前向传播中，模块对输入数据应用这些卷积核，并将它们的输出平均在一起，以生成最终的输出。

import torch
import torch.nn as nn# 定义一个名为 Inception_Block_V1 的 PyTorch 模型类
class Inception_Block_V1(nn.Module):def __init__(self, in_channels, out_channels, num_kernels=6, init_weight=True):super(Inception_Block_V1, self).__init__()# 初始化模块的参数self.in_channels = in_channelsself.out_channels = out_channelsself.num_kernels = num_kernels# 创建卷积核列表，用于构建多个不同尺寸的卷积核kernels = []for i in range(self.num_kernels):# 添加卷积层，kernel_size为2 * i + 1，padding为ikernels.append(nn.Conv2d(in_channels, out_channels, kernel_size=2 * i + 1, padding=i))# 使用 nn.ModuleList 将卷积核列表转化为模块列表self.kernels = nn.ModuleList(kernels)# 如果 init_weight 为 True，则初始化模块的权重if init_weight:self._initialize_weights()def _initialize_weights(self):# 初始化模型参数的函数for m in self.modules():if isinstance(m, nn.Conv2d):# 使用 kaiming_normal_ 初始化权重，适用于 ReLU 激活函数nn.init.kaiming_normal_(m.weight, mode='fan_out', nonlinearity='relu')if m.bias is not None:# 初始化偏置项为零nn.init.constant_(m.bias, 0)# 前向传播方法，接受输入 xdef forward(self, x):res_list = []# 对每个卷积核进行前向传播，并将结果添加到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# 定义另一个名为 Inception_Block_V2 的 PyTorch 模型类
class Inception_Block_V2(nn.Module):def __init__(self, in_channels, out_channels, num_kernels=6, init_weight=True):super(Inception_Block_V2, self).__init__()# 初始化模块的参数self.in_channels = in_channelsself.out_channels = out_channelsself.num_kernels = num_kernelskernels = []for i in range(self.num_kernels // 2):# 创建两种不同尺寸的卷积核kernels.append(nn.Conv2d(in_channels, out_channels, kernel_size=[1, 2 * i + 3], padding=[0, i + 1]))kernels.append(nn.Conv2d(in_channels, out_channels, kernel_size=[2 * i + 3, 1], padding=[i + 1, 0]))# 添加一个额外的卷积核，kernel_size为1kernels.append(nn.Conv2d(in_channels, out_channels, kernel_size=1))# 使用 nn.ModuleList 将卷积核列表转化为模块列表self.kernels = nn.ModuleList(kernels)# 如果 init_weight 为 True，则初始化模块的权重if init_weight:self._initialize_weights()def _initialize_weights(self):# 初始化模型参数的函数for m in self.modules():if isinstance(m, nn.Conv2d):# 如果当前模块是卷积层# 使用 Kaiming 初始化（适用于 ReLU 激活函数的初始化）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 = []# 对每个卷积核进行前向传播，并将结果添加到res_list中for i in range(self.num_kernels + 1):res_list.append(self.kernels[i](x))# 在最后一个维度上堆叠结果，然后计算平均值res = torch.stack(res_list, dim=-1).mean(-1)return res

23.5.2 layers/Embed.py 其中的 DataEmbedding

这段代码定义了一个 DataEmbedding 模块，用于将输入数据进行嵌入处理，包括值嵌入、时间特征嵌入和位置编码。注释解释了每个函数的功能，包括初始化模块、创建不同类型的嵌入层、前向传播的操作以及如何应用 Dropout 以防止过拟合。根据 embed_type 参数的不同，可以选择不同的嵌入类型。根据是否提供时间特征 x_mark，可以选择不同的嵌入方式。

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.nn.utils import weight_norm
import math### 前面的代码 TimesNet用不到，这里没有复制过来，也不做解释### 以下代码是 TimesNet 用到的class DataEmbedding(nn.Module):def __init__(self, c_in, d_model, embed_type='fixed', freq='h', dropout=0.1):super(DataEmbedding, self).__init__()# 初始化 DataEmbedding 模块# c_in：输入通道数，通常是词汇表的大小# d_model：输出维度（模型的维度）# embed_type：嵌入类型，可以是 'fixed' 或 'timeF'# freq：时间频率，可以是 'h'（小时）、't'（分钟）、's'（秒）、'm'（月份）、'a'（年份）、'w'（星期）、'd'（日期）、'b'（工作日）# dropout：Dropout 概率，用于防止过拟合# 创建值嵌入层，用于将输入值（例如，词汇表中的词）映射为模型维度self.value_embedding = TokenEmbedding(c_in=c_in, d_model=d_model)# 创建位置编码层，用于表示输入序列中的位置信息self.position_embedding = PositionalEmbedding(d_model=d_model)# 创建时间特征嵌入层，用于表示时间特征（如小时、星期等）# 嵌入类型根据 embed_type 参数选择，可以是 'fixed' 或 'timeF'self.temporal_embedding = TemporalEmbedding(d_model=d_model, embed_type=embed_type, freq=freq) if embed_type != 'timeF' else TimeFeatureEmbedding(d_model=d_model, embed_type=embed_type, freq=freq)# 创建 Dropout 层，用于防止过拟合self.dropout = nn.Dropout(p=dropout)def forward(self, x, x_mark):# 前向传播函数# x：输入值张量，形状为 [batch_size, sequence_length, c_in]# x_mark：时间特征张量，形状取决于时间特征的嵌入类型if x_mark is None:# 如果没有时间特征，只使用值嵌入和位置编码x = self.value_embedding(x) + self.position_embedding(x)else:# 如果有时间特征，使用值嵌入、时间特征嵌入和位置编码x = self.value_embedding(x) + self.temporal_embedding(x_mark) + self.position_embedding(x)# 应用 Dropout 以减少过拟合return self.dropout(x)

23.5.3 models/TimesNet.py

以下这段代码定义了一个名为 TimesBlock 的 PyTorch 模型类，它实现了一种时间序列分析方法。该模块接受一个时间序列输入 x，通过快速傅里叶变换（FFT）分析其周期性，并使用卷积操作对不同周期的信号进行处理。模块的作用是对输入的时间序列进行周期性分析和特征提取，以捕捉时间序列中的周期性变化。最终输出经过自适应聚合和残差连接后的时间序列。

import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.fft
from layers.Embed import DataEmbedding
from layers.Conv_Blocks import Inception_Block_V1def FFT_for_Period(x, k=2):# 使用快速傅里叶变换（FFT）分析时间序列 x 的周期性# x: 输入的时间序列 [B, T, C]xf = torch.fft.rfft(x, dim=1)# 通过振幅找到频率分量frequency_list = abs(xf).mean(0).mean(-1)frequency_list[0] = 0_, top_list = torch.topk(frequency_list, k)top_list = top_list.detach().cpu().numpy()period = x.shape[1] // top_list# 返回估计的周期和相应的频率成分return period, abs(xf).mean(-1)[:, top_list]# 定义一个名为 TimesBlock 的 PyTorch 模型类
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  # 周期估计的前 k 个频率成分# 创建一个卷积层序列，使用 Inception_Block_V1 模块self.conv = nn.Sequential(Inception_Block_V1(configs.d_model, configs.d_ff,num_kernels=configs.num_kernels),  # 第一个卷积层nn.GELU(),  # GELU 激活函数Inception_Block_V1(configs.d_ff, configs.d_model,num_kernels=configs.num_kernels)  # 第二个卷积层)# 前向传播方法，接受输入 xdef forward(self, x):B, T, N = x.size()  # 获取输入 x 的维度信息# 使用 FFT_for_Period 函数分析输入 x 的周期性，并获取周期列表和周期权重period_list, period_weight = FFT_for_Period(x, self.k)res = []  # 存储结果的列表for i in range(self.k):period = period_list[i]  # 获取第 i 个周期# 填充输入，使其长度能够整除周期if (self.seq_len + self.pred_len) % period != 0:length = (((self.seq_len + self.pred_len) // period) + 1) * periodpadding = 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# 重塑输入，将时间序列变成 2D 形式out = out.reshape(B, length // period, period, N).permute(0, 3, 1, 2).contiguous()# 应用卷积层，将1D变化转化为2D变化out = self.conv(out)# 重塑回原始形状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)  # 将结果堆叠在一起，形成一个多通道的输出period_weight = F.softmax(period_weight, dim=1)  # 对周期权重进行 softmax 归一化period_weight = period_weight.unsqueeze(1).unsqueeze(1).repeat(1, T, N, 1)  # 调整周期权重的形状res = torch.sum(res * period_weight, -1)  # 对多通道输出进行加权求和，进行自适应聚合res = res + x  # 添加残差连接return res  # 返回最终输出

这段代码定义了一个 PyTorch 模型类，用于处理不同类型的时间序列数据分析任务。根据任务类型，模型采用不同的网络架构和输出层。该模型类包括任务类型如长期预测、缺失值填充、异常检测和分类。根据不同的任务类型，模型进行数据标准化、数据嵌入、网络模型的运算、投影和反标准化等步骤，以满足任务要求。前向传播方法根据任务类型返回相应的结果。

# 定义一个名为 Model 的 PyTorch 模型类
class Model(nn.Module):"""Paper link: https://openreview.net/pdf?id=ju_Uqw384Oq"""def __init__(self, configs):super(Model, self).__init__()# 初始化模型的参数self.configs = configsself.task_name = configs.task_name  # 任务名称self.seq_len = configs.seq_len  # 输入序列的长度self.label_len = configs.label_len  # 标签序列的长度self.pred_len = configs.pred_len  # 预测序列的长度# 创建一系列的 TimesBlock 模块，数量为配置中的 e_layersself.model = nn.ModuleList([TimesBlock(configs) for _ in range(configs.e_layers)])# 创建数据嵌入模块self.enc_embedding = DataEmbedding(configs.enc_in, configs.d_model, configs.embed, configs.freq,configs.dropout)self.layer = configs.e_layersself.layer_norm = nn.LayerNorm(configs.d_model)# 根据任务类型不同，初始化不同的输出层if self.task_name == 'long_term_forecast' or self.task_name == 'short_term_forecast':self.predict_linear = nn.Linear(self.seq_len, self.pred_len + self.seq_len)  # 用于预测的线性层self.projection = nn.Linear(configs.d_model, configs.c_out, bias=True)  # 投影线性层if self.task_name == 'imputation' or self.task_name == 'anomaly_detection':self.projection = nn.Linear(configs.d_model, configs.c_out, bias=True)  # 投影线性层if self.task_name == 'classification':self.act = F.gelu  # 激活函数self.dropout = nn.Dropout(configs.dropout)  # 丢弃层self.projection = nn.Linear(configs.d_model * configs.seq_len, configs.num_class)  # 分类线性层# 用于长期预测任务的方法def forecast(self, x_enc, x_mark_enc, x_dec, x_mark_dec):# 标准化输入数据means = x_enc.mean(1, keepdim=True).detach()x_enc = x_enc - meansstdev = torch.sqrt(torch.var(x_enc, dim=1, keepdim=True, unbiased=False) + 1e-5)x_enc /= stdev# 数据嵌入enc_out = self.enc_embedding(x_enc, x_mark_enc)# 预测线性层enc_out = self.predict_linear(enc_out.permute(0, 2, 1)).permute(0, 2, 1)# TimesNet 模型for i in range(self.layer):enc_out = self.layer_norm(self.model[i](enc_out))# 投影线性层dec_out = self.projection(enc_out)# 反标准化dec_out = dec_out * \(stdev[:, 0, :].unsqueeze(1).repeat(1, self.pred_len + self.seq_len, 1))dec_out = dec_out + \(means[:, 0, :].unsqueeze(1).repeat(1, self.pred_len + self.seq_len, 1))return dec_out# 用于缺失值填充任务的方法def imputation(self, x_enc, x_mark_enc, x_dec, x_mark_dec, mask):# 标准化输入数据means = torch.sum(x_enc, dim=1) / torch.sum(mask == 1, dim=1)means = means.unsqueeze(1).detach()x_enc = x_enc - meansx_enc = x_enc.masked_fill(mask == 0, 0)stdev = torch.sqrt(torch.sum(x_enc * x_enc, dim=1) /torch.sum(mask == 1, dim=1) + 1e-5)stdev = stdev.unsqueeze(1).detach()x_enc /= stdev# 数据嵌入enc_out = self.enc_embedding(x_enc, x_mark_enc)# TimesNet 模型for i in range(self.layer):enc_out = self.layer_norm(self.model[i](enc_out))# 投影线性层dec_out = self.projection(enc_out)# 反标准化dec_out = dec_out * \(stdev[:, 0, :].unsqueeze(1).repeat(1, self.pred_len + self.seq_len, 1))dec_out = dec_out + \(means[:, 0, :].unsqueeze(1).repeat(1, self.pred_len + self.seq_len, 1))return dec_out# 用于异常检测任务的方法def anomaly_detection(self, x_enc):# 标准化输入数据means = x_enc.mean(1, keepdim=True).detach()x_enc = x_enc - meansstdev = torch.sqrt(torch.var(x_enc, dim=1, keepdim=True, unbiased=False) + 1e-5)x_enc /= stdev# 数据嵌入enc_out = self.enc_embedding(x_enc, None)# TimesNet 模型for i in range(self.layer):enc_out = self.layer_norm(self.model[i](enc_out))# 投影线性层dec_out = self.projection(enc_out)# 反标准化dec_out = dec_out * \(stdev[:, 0, :].unsqueeze(1).repeat(1, self.pred_len + self.seq_len, 1))dec_out = dec_out + \(means[:, 0, :].unsqueeze(1).repeat(1, self.pred_len + self.seq_len, 1))return dec_out# 用于分类任务的方法def classification(self, x_enc, x_mark_enc):# 数据嵌入enc_out = self.enc_embedding(x_enc, None)# TimesNet 模型for i in range(self.layer):enc_out = self.layer_norm(self.model[i](enc_out))# 输出output = self.act(enc_out)output = self.dropout(output)output = output * x_mark_enc.unsqueeze(-1)output = output.reshape(output.shape[0], -1)output = self.projection(output)return output# 前向传播方法def forward(self, x_enc, x_mark_enc, x_dec, x_mark_dec, mask=None):if self.task_name == 'long_term_forecast' or self.task_name == 'short_term_forecast':dec_out = self.forecast(x_enc, x_mark_enc, x_dec, x_mark_dec)return dec_out[:, -self.pred_len:, :]  # 返回预测结果if self.task_name == 'imputation':dec_out = self.imputation(x_enc, x_mark_enc, x_dec, x_mark_dec, mask)return dec_out  # 返回填充结果if self.task_name == 'anomaly_detection':dec_out = self.anomaly_detection(x_enc)return dec_out  # 返回异常检测结果if self.task_name == 'classification':dec_out = self.classification(x_enc, x_mark_enc)return dec_out  # 返回分类结果return None  # 若任务名称不匹配，则返回空值

23.6 总结

本篇论文介绍了一种新的方法，通过将一维时间序列转换为二维张量来处理时间序列分析中的复杂时间模式。作者提出的TimesNet with TimesBlock是一种通用的时间序列分析模型，可以发现时间序列中的多个周期内和周期间的变化。该模型在五个主流时间序列分析任务中表现出了很好的通用性和性能。作者还提出了未来的研究方向，包括在大规模预训练中进一步探索利用TimesNet作为骨干的方法，以及在实际应用中的潜在应用。

论文提供的源码集成了一些其他的算法，对于需要做比对实验的人来说，简直是太方便了。并且，如果你设计了某个算法，可以基于TimesNet 源码架构进行开发，然后合并到代码仓库中作为其中一个算子，简单方便直观，给作者团队点赞 ~

Smileyan
2023.10.31 22:58