04 信号预处理流水线：滤波、去噪与归一化工程实现

⭐信号预处理流水线：从原始波形到标准化输入

在信号识别任务中，原始传感器数据往往夹杂着工频干扰、基线漂移和随机噪声。如果直接将未经处理的信号送入模型，模型会花费大量容量去拟合噪声，导致泛化能力下降。本节的核心目标是构建一条可复用的信号预处理流水线（Signal Preprocessing Pipeline），并将其集成到PyTorch信号识别训练模板（PyTorch Signal Recognition Training Template）中。这条流水线将完成三个关键步骤：带通滤波、小波去噪与Z-score归一化，确保输入到PyTorch模型的信号是干净且尺度统一的

🧭预处理模块的工程实现：滤波、去噪与归一化

我们将预处理逻辑封装为一个独立的Python模块 preprocessor.py，它接收原始信号张量（Tensor），依次执行以下操作：

带通滤波：保留有效频带

对于大多数生物电信号（如ECG、EEG），有效信息集中在0.5Hz-45Hz之间。我们使用scipy.signal.butter设计一个4阶巴特沃斯带通滤波器，并通过filtfilt实现零相位滤波，避免相位偏移破坏波形时序。关键参数为低截止频率lowcut=0.5和高截止频率highcut=45.0，采样频率fs需根据实际采集设备设定（常见为250Hz或1000Hz）

小波去噪：抑制非平稳噪声

带通滤波无法去除频带内的随机噪声。我们采用小波阈值去噪（Wavelet Denoising），选择db4小波基，分解层数为5层。使用软阈值函数（soft thresholding）处理细节系数，阈值由np.sqrt(2 * np.log(n)) * median估算，其中n为信号长度。这一步能有效保留信号尖峰特征，同时平滑背景噪声

Z-score归一化：统一尺度

最后，对去噪后的信号进行Z-score归一化（Z-score Normalization）(x - mean) / std。注意，均值和标准差必须在训练集上计算并保存，验证/测试集使用相同的统计量，避免数据泄露。在Z-score信号识别训练模板（Z-score Signal Recognition Training Template）中，我们建议将归一化参数保存为一个JSON文件，随模型一起部署

import torch
import numpy as np
from scipy import signal
import pywt
class SignalPreprocessor:
    def __init__(self, fs=250.0, lowcut=0.5, highcut=45.0):
        self.fs = fs
        self.lowcut = lowcut
        self.highcut = highcut
        self.mean = None
        self.std = None
        # 设计带通滤波器系数
        nyquist = 0.5 * self.fs
        low = self.lowcut / nyquist
        high = self.highcut / nyquist
        self.b, self.a = signal.butter(4, [low, high], btype='band')
    def bandpass_filter(self, data: torch.Tensor) -> torch.Tensor:
        # 零相位滤波，保持波形对齐
        filtered = signal.filtfilt(self.b, self.a, data.numpy(), axis=-1)
        return torch.from_numpy(filtered.copy()).float()
    def wavelet_denoise(self, data: torch.Tensor) -> torch.Tensor:
        # 小波软阈值去噪
        coeffs = pywt.wavedec(data.numpy(), 'db4', level=5)
        sigma = np.median(np.abs(coeffs[-1])) / 0.6745
        threshold = sigma * np.sqrt(2 * np.log(data.shape[-1]))
        coeffs_th = [coeffs[0]] + [pywt.threshold(c, threshold, mode='soft') for c in coeffs[1:]]
        denoised = pywt.waverec(coeffs_th, 'db4')
        return torch.from_numpy(denoised.copy()).float()
    def fit_normalize(self, data: torch.Tensor):
        # 仅在训练集上调用
        self.mean = data.mean().item()
        self.std = data.std().item() + 1e-8
    def zscore_normalize(self, data: torch.Tensor) -> torch.Tensor:
        return (data - self.mean) / self.std
    def __call__(self, data: torch.Tensor, fit=False) -> torch.Tensor:
        x = self.bandpass_filter(data)
        x = self.wavelet_denoise(x)
        if fit:
            self.fit_normalize(x)
        return self.zscore_normalize(x)

阅读这段代码时，重点看信号预处理流水线的伪代码，以及它如何用稳定性与可观测性证明结果可复测。

上述代码展示了信号预处理流水线信号识别训练模板（Signal Preprocessing Pipeline Signal Recognition Training Template）的核心实现。注意fit_normalize方法只在训练阶段调用一次，而__call__中的fit=False确保推理时复用已保存的统计量。代码中axis=-1保证了多通道信号（如12导联ECG）能正确沿时间轴滤波

💻参数配置与故障排查：让流水线稳定运行

预处理模块的稳定性直接影响训练效果。以下是关键参数表与常见故障排查指南

常见故障与判断标准

当预处理输出异常时，可按以下检查清单快速定位：

💻集成到训练模板：流水线的调用契约

将SignalPreprocessor集成到PyTorch信号识别训练模板中时，需在Dataset的__getitem__中调用预处理逻辑。注意不要在DataLoader的多进程worker中重复计算滤波器系数，建议将预处理器作为Dataset的初始化参数传入

class SignalDataset(torch.utils.data.Dataset):
    def __init__(self, raw_data, preprocessor, fit=False):
        self.data = raw_data
        self.preprocessor = preprocessor
        self.fit = fit
        if self.fit:
            # 训练集：计算归一化参数
            for sample in self.data:
                self.preprocessor(sample, fit=True)
    def __getitem__(self, idx):
        x = self.data[idx]
        x = self.preprocessor(x, fit=False)  # 推理时不再更新统计量
        return x

阅读这段代码时，重点看信号预处理流水线的伪代码，以及它如何用稳定性与可观测性证明结果可复测。

上述代码中，fit=True仅在Dataset初始化时触发一次，确保所有训练样本都参与统计量计算。这种设计避免了在DataLoader的worker中重复计算，也防止了数据泄露

通过本节的学习，你应能独立构建一条可复用的信号预处理流水线，并将其无缝嵌入PyTorch训练流程。记住：参数配置是预处理模块的“灵魂”，而统计量的正确固化是归一化成功的保障