CEEMDAN-Transformer 实战：如何解决非平稳时序信号预测中的模态混叠问题

1次阅读

没有评论

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

非平稳时序信号预测在电力负荷、金融时间序列等领域具有重要应用价值。传统方法如 STFT 和 Wavelet 变换虽然能提供时频分析，但存在以下问题：

模态混叠：不同频率成分在时频平面上重叠，导致特征提取困难
窗口效应：固定窗口大小难以适应非平稳信号的局部特性
信噪比敏感：低信噪比环境下高频分量容易被噪声淹没

相比于 STFT 的固定分辨率和小波变换的基函数选择依赖，CEEMDAN 通过自适应噪声添加策略，实现了更优的模态分离效果。其核心优势在于：

噪声辅助分析避免模式混淆
自动确定本征模态函数 (IMF) 数量
残差分解更彻底

算法通过迭代添加自适应白噪声来改进 EMD 的模态混叠问题。关键步骤数学描述如下：

$$IMF_1 = \frac{1}{I}\sum_{i=1}^I E_1(x + \epsilon w_i)$$
$$r_k = r_{k-1} – IMF_k$$

其中：
– $E_k(\cdot)$ 表示第 k 阶 IMF 提取算子
– $w_i$ 为高斯白噪声
– $\epsilon$ 控制噪声强度

停止条件采用相对标准差(RDS)：

$$RDS = \frac{\sum_{t=1}^T |h_{k-1}(t) – h_k(t)|^2}{\sum_{t=1}^T h_{k-1}^2(t)} < \delta$$

处理 IMF 分量的结构示意：

[Input] -> [IMF1] -> [IMF2] -> ... -> [IMFn] 
           |           |             |
       [Embedding] [Embedding]   [Embedding]
           |           |             |
       [Multi-Head Attention] <- [Positional Encoding]
                     |
                [Prediction]

多头注意力机制允许模型同时关注不同 IMF 分量的时频特征，其计算公式：

$$Attention(Q,K,V) = softmax(\frac{QK^T}{\sqrt{d_k}})V$$

def check_stopping_criterion(h_prev, h_current, threshold=0.05):
    """
    计算相对标准差 (RDS) 判断是否停止分解
    Args:
        h_prev (np.array): 前次迭代结果
        h_current (np.array): 当前迭代结果
        threshold (float): RDS 阈值
    Returns:
        bool: 是否停止分解
    """
    numerator = np.sum((h_prev - h_current)**2)
    denominator = np.sum(h_prev**2)
    rds = np.sqrt(numerator / denominator)
    return rds < threshold

class IMF2Tensor(nn.Module):
    def __init__(self, d_model):
        super().__init__()
        # 每个 IMF 分量独立嵌入
        self.embedding = nn.Linear(1, d_model)  

    def forward(self, imfs):
        """
        Args:
            imfs (list): 可变长度的 IMF 分量列表
        Returns:
            torch.Tensor: [seq_len, batch, d_model]
        """
        # 对每个 IMF 进行零填充和嵌入
        max_len = max(imf.shape[0] for imf in imfs)
        embedded = []
        for imf in imfs:
            padded = F.pad(imf, (0, max_len - imf.shape[0]))
            embedded.append(self.embedding(padded.unsqueeze(-1)))
        return torch.stack(embedded, dim=1)

class AdaptivePositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super().__init__()
        position = torch.arange(max_len).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2) * (-math.log(10000.0) / d_model))
        pe = torch.zeros(max_len, d_model)
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        self.register_buffer('pe', pe)

    def forward(self, x):
        """
        Args:
            x: [seq_len, batch, d_model]
        """
        # 动态截取实际序列长度
        return x + self.pe[:x.size(0)]

使用 UCI 电力负荷数据集，对比模型包括：