时间卷积网络（TCN）原理+代码详解

  在理解 TCN 的原理之前，我们可以先对传统的循环神经网络（RNN）进行简要回顾。RNN 是处理序列数据的常用方法，其核心思想是通过将前一个时间步的隐藏状态传递到下一个时间步，实现对序列依赖关系的建模。然而，RNN 在处理长序列时存在以下几个缺点：

• 无法并行计算：RNN 的计算依赖于时间步的顺序，导致无法高效利用 GPU 并行计算。

• 梯度消失/爆炸：在长时间依赖中，梯度在反向传播时会逐渐消失或变得不稳定。

• 短期记忆限制：由于计算依赖于序列的逐步传递，RNN 难以捕获远距离的时间依赖。

  TCN 正是在这样的背景下提出的。它通过因果卷积和扩张卷积，突破了 RNN 的这些瓶颈，特别适用于长时间序列数据。接下来，我们将详细解析 TCN 的原理。

一、TCN原理

1.1 因果卷积（Causal Convolution）

  在卷积操作中，卷积核在输入上滑动时会同时处理前后时间步的数据，导致当前时间步的输出可能依赖于未来的输入。然而，对于时间序列任务，我们通常希望模型只依赖于过去的输入，不“窥探”未来，这样的结构称为“因果性”。

  TCN 使用因果卷积来确保这一点。因果卷积是指每个时间步的输出仅依赖于它之前的时间步，而不依赖于未来。简单来说，当前时间步的输出只会考虑卷积核覆盖的前几个时间步的输入。

  TCN 通过适当的填充（padding）来实现这一点，使得每一层的卷积不会跨越未来时间步。因果卷积的示意图如下：

1.2 扩张卷积（Dilated Convolution）

  为了捕捉长时间依赖关系，TCN 通过 扩张卷积（Dilated Convolution 来扩展卷积核的感受野。扩张卷积通过在卷积核的元素之间插入“间隔”，从而在保持卷积核大小不变的情况下，扩大卷积的感受野。

  例如，假设卷积核大小为 3，当扩张率 dilation=2 时，卷积核的元素之间插入 1 个间隔，感受野可以从 3 扩展到 5。通过这种扩张卷积，TCN 在每一层可以通过指数扩展的方式增大感受野，使得模型能够捕捉到远距离的依赖关系。例如，TCN 中第 $i$ 层的感受野大小为 $2^i$，这样层数越深，感受野就越大。如下图所示：

二、代码实现

2.1 Chomp1d 模块

  TCN 使用填充操作来保证卷积后的时间步不丢失，但填充会导致额外的时间步，因此需要 Chomp1d 来修剪掉多余部分，保证输入输出的时间维度一致。

class Chomp1d(nn.Module):
    def __init__(self, chomp_size):
        super(Chomp1d, self).__init__()
        self.chomp_size = chomp_size

    def forward(self, x):
        return x[:, :, :-self.chomp_size].contiguous()

  Chomp1d 的作用是对卷积结果的最后几个时间步进行修剪，这确保了卷积核在时间序列两端不会额外输出冗余的步长。

2.2 TemporalBlock 模块

  TemporalBlock 是 TCN 的基本构建单元，包含两层扩张卷积，每层后接激活函数和 Chomp1d 操作。

class TemporalBlock(nn.Module):
    def __init__(self, n_inputs, n_outputs, kernel_size, stride, dilation, padding, dropout):
        super(TemporalBlock, self).__init__()
        # 第一层卷积
        self.ll_conv1 = nn.Conv1d(n_inputs, n_outputs, kernel_size, stride=stride, padding=padding, dilation=dilation)
        self.chomp1 = Chomp1d(padding)
        self.relu1 = nn.LeakyReLU()

        # 第二层卷积
        self.ll_conv2 = nn.Conv1d(n_outputs, n_outputs, kernel_size, stride=stride, padding=padding, dilation=dilation)
        self.chomp2 = Chomp1d(padding)
        self.relu2 = nn.LeakyReLU()

        # Dropout 作为正则化，防止过拟合
        self.dropout = nn.Dropout(dropout)

    def forward(self, x):
        # 第一个卷积、修剪、激活和 Dropout
        out = self.ll_conv1(x)
        out = self.chomp1(out)
        out = self.relu1(out)
        out = self.dropout(out)

        # 第二个卷积、修剪、激活和 Dropout
        out = self.ll_conv2(out)
        out = self.chomp2(out)
        out = self.relu2(out)
        out = self.dropout(out)

        return out

• ll_conv1 和 ll_conv2 是两层扩张卷积层，dilation 参数决定了每层的感受野大小。

• Chomp1d 保证卷积结果不会产生额外的时间步。

• LeakyReLU 是非线性激活函数，为模型引入非线性。

• Dropout 用于防止过拟合，通过随机丢弃一部分神经元。

2.3 TemporalConvNet 模块

  TemporalConvNet 是由多个 TemporalBlock 级联组成的模型，每一层的卷积感受野逐层递增。

class TemporalConvNet(nn.Module):
    def __init__(self, num_inputs, num_channels, kernel_size=2, dropout=0.0):
        super(TemporalConvNet, self).__init__()
        layers = []
        self.num_levels = len(num_channels)

        for i in range(self.num_levels):
            dilation_size = 2 ** i  # 每层的扩张率递增
            in_channels = num_inputs if i == 0 else num_channels[i - 1]
            out_channels = num_channels[i]
            layers.append(
                TemporalBlock(
                    in_channels, out_channels, kernel_size, stride=1, dilation=dilation_size,
                    padding=(kernel_size - 1) * dilation_size, dropout=dropout
                )
            )

        self.network = nn.Sequential(*layers)

    def forward(self, x):
        return self.network(x)

• TemporalConvNet 通过循环构建多层 TemporalBlock，每层的扩张率 dilation 是前一层的两倍，使得感受野指数级增长。

• 使用 nn.Sequential 将所有层级联在一起，模型最终输出序列数据经过所有层的处理结果。

2.4 完整代码示例

  在这个例子中，输入数据有 8 个样本，每个样本有 3 个特征，序列长度为 10。经过 TCN 网络的三层处理，输出的特征维度从 3 增加到 64，但时间维度（10）保持不变。

import torch.nn as nn
import torch.nn.functional as F
import torch

class Chomp1d(nn.Module):
    def __init__(self, chomp_size):
        super(Chomp1d, self).__init__()
        self.chomp_size = chomp_size

    def forward(self, x):
        return x[:, :, : -self.chomp_size].contiguous()


class TemporalBlock(nn.Module):
    def __init__(
        self, n_inputs, n_outputs, kernel_size, stride, dilation, padding, dropout
    ):
        super(TemporalBlock, self).__init__()
        self.n_inputs = n_inputs
        self.n_outputs = n_outputs
        self.kernel_size = kernel_size
        self.stride = stride
        self.dilation = dilation
        self.padding = padding
        self.dropout = dropout
        self.ll_conv1 = nn.Conv1d(
            n_inputs,
            n_outputs,
            kernel_size,
            stride=stride,
            padding=padding,
            dilation=dilation,
        )
        self.chomp1 = Chomp1d(padding)

        self.ll_conv2 = nn.Conv1d(
            n_outputs,
            n_outputs,
            kernel_size,
            stride=stride,
            padding=padding,
            dilation=dilation,
        )
        self.chomp2 = Chomp1d(padding)
        self.sigmoid = nn.Sigmoid()

    def net(self, x, block_num, params=None):
        layer_name = "ll_tc.ll_temporal_block" + str(block_num)
        if params is None:
            x = self.ll_conv1(x)
        else:
            x = F.conv1d(
                x,
                weight=params[layer_name + ".ll_conv1.weight"],
                bias=params[layer_name + ".ll_conv1.bias"],
                stride=self.stride,
                padding=self.padding,
                dilation=self.dilation,
            )

        x = self.chomp1(x)
        x = F.leaky_relu(x)

        return x

    def init_weights(self):
        self.ll_conv1.weight.data.normal_(0, 0.01)
        self.ll_conv2.weight.data.normal_(0, 0.01)

    def forward(self, x, block_num, params=None):
        out = self.net(x, block_num, params)
        return out


class TemporalConvNet(nn.Module):
    def __init__(self, num_inputs, num_channels, kernel_size=2, dropout=0.0):
        super(TemporalConvNet, self).__init__()
        layers = []
        self.num_levels = len(num_channels)

        for i in range(self.num_levels):
            dilation_size = 2 ** i
            in_channels = num_inputs if i == 0 else num_channels[i - 1]
            out_channels = num_channels[i]
            setattr(
                self,
                "ll_temporal_block{}".format(i),
                TemporalBlock(
                    in_channels,
                    out_channels,
                    kernel_size,
                    stride=1,
                    dilation=dilation_size,
                    padding=(kernel_size - 1) * dilation_size,
                    dropout=dropout,
                ),
            )

    def forward(self, x, params=None):

        for i in range(self.num_levels):
            temporal_block = getattr(self, "ll_temporal_block{}".format(i))
            x = temporal_block(x, i, params=params)
        return x


# 定义一个 TCN 模型，输入通道数为 3，输出通道分别为 16, 32, 64，核大小为 2
tcn = TemporalConvNet(num_inputs=3, num_channels=[16, 32, 64], kernel_size=2, dropout=0.2)

# 假设输入的张量形状为 (batch_size, num_inputs, sequence_length)
x = torch.randn(8, 3, 10)  # 8 个样本，3 个输入特征，序列长度为 10

# 通过 TCN 进行前向传播
output = tcn(x)

print(output.shape)  # 输出的形状为 (batch_size, 64, sequence_length)，即 (8, 64, 10)

参考文献

[1] https://github.com/locuslab/TCN

[2] 如何理解扩张卷积(dilated convolution)

[3] 【机器学习】详解 扩张/膨胀/空洞卷积 (Dilated / Atrous Convolution)