CEEMDAN与Transformer融合实战：时间序列分解与预测的完整指南

1次阅读

共计 2348 个字符，预计需要花费 6 分钟才能阅读完成。

时间序列预测在实际应用中常常面临非平稳信号的难题。与平稳序列相比，非平稳信号具有时变的统计特性（time-varying statistics），比如趋势变化、季节性波动和突发性事件。传统方法如 STL（Seasonal-Trend decomposition using Loess）和小波变换（Wavelet Transform）虽然能进行信号分解，但存在明显局限：

STL 对非周期性的复杂模式处理能力有限
小波变换依赖基函数选择，且分解结果不具备自适应性
两者均难以与深度学习模型无缝衔接

CEEMDAN（Complete Ensemble Empirical Mode Decomposition with Adaptive Noise）是对 EEMD 的改进，其核心创新在于完全自适应噪声机制。算法流程可概括为：

对原始信号添加特定幅度的白噪声
执行 EMD 分解得到第一阶 IMF（本征模态函数 /Intrinsic Mode Function）
计算残差并重复过程直到满足停止条件

数学表达为：
$$IMF_1 = \frac{1}{N}\sum_{i=1}^N E_1(x+\epsilon w_i)$$
其中 $E_1(·)$ 表示提取第一阶 IMF 的操作，$\epsilon$ 控制噪声强度，$w_i$ 为高斯白噪声。

相较于 EEMD，CEEMDAN 的优势在于：

通过残差自适应调整噪声幅度
避免模态混叠（mode mixing）问题
分量数量更稳定

分解得到的 IMF 分量需要合理重组才能输入 Transformer：

高频 IMF（如 IMF1-3）反映细节特征
低频 IMF（如 IMF4+）包含趋势信息
残差项（Residue）代表长期变化

建议重组方式：

将 3 - 5 个相邻 IMF 合并为一个输入通道
对每个通道单独做标准化（Z-score）
按能量占比赋予不同权重

由于各 IMF 分量具有不同时间尺度，需调整 Transformer 的位置编码：

对高频分量使用较短的位置编码周期
低频分量采用更长周期的位置编码
残差项可省略位置编码

import numpy as np
from PyEMD import CEEMDAN
import torch
import torch.nn as nn

# CEEMDAN 分解模块
class CEEMDAN_Decomposer:
    def __init__(self, noise_scale=0.05, trials=100):
        self.ceemdan = CEEMDAN(noise_scale=noise_scale, trials=trials)

    def __call__(self, x):
        # x: [batch_size, seq_len]
        imfs = []
        for sample in x:
            imf = self.ceemdan(sample.cpu().numpy())
            imfs.append(torch.tensor(imf, dtype=torch.float32))
        return torch.stack(imfs)  # [batch_size, n_imfs, seq_len]

# 能量加权模块
class EnergyWeight(nn.Module):
    def forward(self, x):
        # x: [batch_size, n_imfs, seq_len]
        energy = torch.norm(x, dim=2, keepdim=True)  # [b, n_imfs, 1]
        weights = energy / energy.sum(dim=1, keepdim=True)
        return x * weights

# Transformer 编码器
class IMFTransformer(nn.Module):
    def __init__(self, n_imfs, d_model=64, nhead=4):
        super().__init__()
        self.proj = nn.Linear(n_imfs, d_model)
        encoder_layer = nn.TransformerEncoderLayer(d_model=d_model, nhead=nhead)
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=3)

    def forward(self, x):
        # x: [seq_len, batch_size, n_imfs]
        x = self.proj(x)  # [seq_len, batch_size, d_model]
        return self.transformer(x)

通过交叉验证选择最优噪声强度：