基于 Transformer 的时序数据建模与 Python 实现详解

这种模块化设计兼顾了可维护性与扩展性。

2. 核心组件详细分析

2.1 多头自注意力机制

这是 Transformer 的心脏，负责捕捉序列内部的依赖关系。下面是一个标准的 PyTorch 实现示例：

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

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, n_heads, dropout=0.1):
        super(MultiHeadAttention, self).__init__()
        assert d_model % n_heads == 0
        self.d_model = d_model
        self.n_heads = n_heads
        self.d_k = d_model // n_heads
        
        # 线性变换层
        self.W_q = nn.Linear(d_model, d_model, bias=False)
        self.W_k = nn.Linear(d_model, d_model, bias=False)
        self.W_v = nn.Linear(d_model, d_model, bias=False)
        self.W_o = nn.Linear(d_model, d_model)
        self.dropout = nn.Dropout(dropout)
        self.scale = math.sqrt(self.d_k)

    def forward(self, query, key, value, mask=None):
        batch_size = query.size(0)
        seq_len = query.size(1)
        
        # 线性变换并重塑为多头形式
        Q = self.W_q(query).view(batch_size, seq_len, self.n_heads, self.d_k).transpose(1, 2)
        K = self.W_k(key).view(batch_size, seq_len, self.n_heads, self.d_k).transpose(1, 2)
        V = self.W_v(value).view(batch_size, seq_len, self.n_heads, self.d_k).transpose(1, 2)
        
        # 计算注意力分数
        scores = torch.matmul(Q, K.transpose(-2, -1)) / self.scale
        
        # 应用掩码
        if mask is not None:
            scores = scores.masked_fill(mask == 0, -1e9)
            
        # 计算注意力权重
        attention_weights = F.softmax(scores, dim=-1)
        attention_weights = self.dropout(attention_weights)
        
        # 应用注意力权重
        context = torch.matmul(attention_weights, V)
        
        # 合并多头结果
        context = context.transpose(1, 2).contiguous().view(
            batch_size, seq_len, self.d_model
        )
        
        # 最终线性变换
        output = self.W_o(context)
        return output, attention_weights

这里的关键在于缩放因子 sqrt(d_k)，它能防止 softmax 进入饱和区，保证梯度的有效传播。

2.2 位置编码

由于自注意力机制本身是无序的，必须显式加入位置信息：

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_seq_length=5000, dropout=0.1):
        super(PositionalEncoding, self).__init__()
        self.dropout = nn.Dropout(dropout)
        
        pe = torch.zeros(max_seq_length, d_model)
        position = torch.arange(0, max_seq_length).unsqueeze(1).float()
        div_term = torch.exp(torch.arange(0, d_model, 2).float() * -(math.log(10000.0) / d_model))
        
        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):
        seq_length = x.size(1)
        x = x + self.pe[:, :seq_length]
        return self.dropout(x)

这种正弦余弦编码允许模型学习相对位置关系，且能处理未见过的序列长度。

2.3 Transformer 编码器块

编码器块集成了注意力、前馈网络和归一化：

class TransformerEncoderBlock(nn.Module):
    def __init__(self, d_model, n_heads, d_ff, dropout=0.1):
        super(TransformerEncoderBlock, self).__init__()
        self.self_attention = MultiHeadAttention(d_model, n_heads, dropout)
        
        self.feed_forward = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(d_ff, d_model),
            nn.Dropout(dropout)
        )
        
        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x, mask=None):
        # 多头自注意力 + 残差连接 + 层归一化
        attn_output, _ = self.self_attention(x, x, x, mask)
        x = self.norm1(x + self.dropout(attn_output))
        
        # 前馈网络 + 残差连接 + 层归一化
        ff_output = self.feed_forward(x)
        x = self.norm2(x + ff_output)
        return x

3. 完整的时序 Transformer 网络架构

整合上述组件，构建完整的时序模型：

class TimeSeriesTransformer(nn.Module):
    def __init__(self, input_dim, d_model, n_heads, n_layers, d_ff, max_seq_length, num_classes, dropout=0.1):
        super(TimeSeriesTransformer, self).__init__()
        self.d_model = d_model
        
        self.input_embedding = nn.Linear(input_dim, d_model)
        self.positional_encoding = PositionalEncoding(d_model, max_seq_length, dropout)
        
        self.transformer_blocks = nn.ModuleList([
            TransformerEncoderBlock(d_model, n_heads, d_ff, dropout)
            for _ in range(n_layers)
        ])
        
        self.global_avg_pool = nn.AdaptiveAvgPool1d(1)
        
        self.classifier = nn.Sequential(
            nn.Linear(d_model, d_model // 2),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(d_model // 2, num_classes)
        )
        
        self._init_parameters()

    def _init_parameters(self):
        for module in self.modules():
            if isinstance(module, nn.Linear):
                nn.init.xavier_uniform_(module.weight)
                if module.bias is not None:
                    nn.init.zeros_(module.bias)

    def create_padding_mask(self, x, pad_token=0):
        mask = (x != pad_token).unsqueeze(1).unsqueeze(2)
        return mask

    def forward(self, x, mask=None):
        batch_size, seq_length, input_dim = x.shape
        
        # 输入嵌入
        x = self.input_embedding(x)
        x = x * math.sqrt(self.d_model)
        
        # 位置编码
        x = self.positional_encoding(x)
        
        # 通过编码器层
        attention_weights = []
        for transformer_block in self.transformer_blocks:
            x, attn_weights = transformer_block(x, mask)
            attention_weights.append(attn_weights)
        
        # 全局平均池化
        x = x.transpose(1, 2)
        x = self.global_avg_pool(x).squeeze(-1)
        
        # 分类预测
        output = self.classifier(x)
        return output, attention_weights

4. 模型配置与超参数设置

config = {
    'input_dim': 6,      # 输入特征维度
    'd_model': 256,      # 隐藏维度
    'n_heads': 8,        # 注意力头数
    'n_layers': 6,       # 编码器层数
    'd_ff': 1024,        # 前馈网络维度
    'max_seq_length': 512,
    'num_classes': 6,
    'dropout': 0.1
}

model = TimeSeriesTransformer(**config)
print(f"模型参数量：{sum(p.numel() for p in model.parameters() if p.requires_grad):,}")

三、技术细节与实现要点

1. 自注意力机制的计算复杂度优化

标准自注意力是 O(n²d)，长序列下开销巨大。可以通过降维 Key 来优化：

class EfficientAttention(nn.Module):
    def __init__(self, d_model, n_heads, dropout=0.1, max_seq_length=5000):
        super().__init__()
        self.d_model = d_model
        self.n_heads = n_heads
        self.d_k = d_model // n_heads
        self.reduced_dim = min(64, self.d_k)
        
        self.W_q = nn.Linear(d_model, n_heads * self.reduced_dim, bias=False)
        self.W_k = nn.Linear(d_model, n_heads * self.reduced_dim, bias=False)
        self.W_v = nn.Linear(d_model, d_model, bias=False)
        self.W_o = nn.Linear(d_model, d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, query, key, value, mask=None):
        B, L, D = query.shape
        
        Q = self.W_q(query).view(B, L, self.n_heads, self.reduced_dim).transpose(1, 2)
        K = self.W_k(key).view(B, L, self.n_heads, self.reduced_dim).transpose(1, 2)
        V = self.W_v(value).view(B, L, self.n_heads, self.d_k).transpose(1, 2)
        
        scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.reduced_dim)
        if mask is not None:
            scores.masked_fill_(mask == 0, -1e9)
        
        attn = F.softmax(scores, dim=-1)
        attn = self.dropout(attn)
        
        context = torch.matmul(attn, V).transpose(1, 2).contiguous().view(B, L, D)
        output = self.W_o(context)
        return output, attn

2. 位置编码的改进策略

针对时序特性，可尝试可学习或相对位置编码：

class LearnablePositionalEncoding(nn.Module):
    def __init__(self, d_model, max_seq_length=5000, dropout=0.1):
        super().__init__()
        self.dropout = nn.Dropout(dropout)
        self.pe = nn.Parameter(torch.randn(1, max_seq_length, d_model) * 0.1)

    def forward(self, x):
        seq_len = x.size(1)
        x = x + self.pe[:, :seq_len]
        return self.dropout(x)

class RelativePositionalEncoding(nn.Module):
    def __init__(self, d_model, max_relative_position=128):
        super().__init__()
        self.d_model = d_model
        self.max_relative_position = max_relative_position
        vocab_size = max_relative_position * 2 + 1
        self.relative_position_embeddings = nn.Embedding(vocab_size, d_model)

    def forward(self, length):
        range_vec = torch.arange(length)
        distance_mat = range_vec[None, :] - range_vec[:, None]
        distance_mat_clipped = torch.clamp(distance_mat, -self.max_relative_position, self.max_relative_position)
        final_mat = distance_mat_clipped + self.max_relative_position
        return self.relative_position_embeddings(final_mat)

3. Transformer 与传统方法的性能对比

Transformer 凭借自注意力机制实现了长距离依赖的直接捕捉，同时通过并行化大幅提升了训练效率。配合注意力权重的可视化，其决策逻辑也更加透明，适合处理复杂的时序预测与分析任务。