4、数据预处理技术：信号滤波（低通、高通、带通）、去噪、异常值检测与处理

各位同学，欢迎来到数据预处理这一章。

说实话，呼吸机采集到的原始信号，就像菜市场里的录音——什么声音都有。病人的呼吸波形、心跳干扰、管路抖动、甚至隔壁床按铃的声音都可能混进去。如果你直接拿这些数据做分析，那结果基本没法看。

我个人习惯把数据预处理比作「洗菜」。菜不洗干净，炒出来的菜再好看也没人敢吃。信号滤波、去噪、异常值处理，就是呼吸机数据的「清洗三件套」。今天咱们一个一个来拆解。

4.1 信号滤波：把有用的留下，没用的扔掉

滤波的本质很简单：让特定频率的信号通过，把其他频率的拦在外面。呼吸机信号里，我们关心的主要是呼吸频率范围——成人一般是 8-20 次/分钟，换算成频率大约是 0.13-0.33 Hz。但实际采集到的信号里，有 50 Hz 的工频干扰（电网带来的），有高频的肌肉震颤，还有低频的基线漂移。

所以我们需要三种滤波器：

• 低通滤波器：让低频通过，滤掉高频噪声。比如去掉 50 Hz 的工频干扰。
• 高通滤波器：让高频通过，滤掉低频漂移。比如去掉呼吸波形的基线缓慢变化。
• 带通滤波器：只让某个频率范围的信号通过。比如只保留 0.1-0.5 Hz 的呼吸信号。

下面是一个 Python 实现的低通滤波器示例，用的是 Butterworth 滤波器——这玩意儿在呼吸机领域很常见，因为它通带平坦，不会把波形搞变形。

import numpy as np
from scipy.signal import butter, filtfilt

def lowpass_filter(data, fs, cutoff=5.0, order=4):
    """
    低通滤波器
    :param data: 输入信号
    :param fs: 采样率 (Hz)
    :param cutoff: 截止频率 (Hz)
    :param order: 滤波器阶数
    """
    nyquist = 0.5 * fs
    normal_cutoff = cutoff / nyquist
    b, a = butter(order, normal_cutoff, btype='low', analog=False)
    # 用 filtfilt 做零相位滤波，避免波形偏移
    filtered = filtfilt(b, a, data)
    return filtered

# 使用示例
fs = 100  # 假设采样率 100 Hz
raw_signal = np.random.randn(1000)  # 模拟原始信号
clean_signal = lowpass_filter(raw_signal, fs, cutoff=5.0)

4.2 去噪：不只是滤波那么简单

滤波能解决大部分噪声问题，但有些噪声是「非平稳」的——比如病人突然咳嗽一下，或者管路被压了一下。这种噪声在频域上和呼吸信号重叠，用普通滤波器根本分不开。

这时候就需要更高级的去噪方法。我个人比较常用的是：

• 小波去噪：把信号分解成不同尺度，把噪声对应的小波系数置零，再重构信号。对突发性噪声效果很好。
• 中值滤波：对每个点取邻域内的中位数。对付「椒盐噪声」（比如传感器偶尔跳变）特别管用。
• 移动平均：简单粗暴，但会平滑掉波形的尖峰。我一般只在做趋势分析时用。

举个例子，小波去噪的代码长这样：

import pywt

def wavelet_denoise(data, wavelet='db4', level=4):
    """
    小波去噪
    :param data: 输入信号
    :param wavelet: 小波基
    :param level: 分解层数
    """
    # 小波分解
    coeffs = pywt.wavedec(data, wavelet, level=level)
    # 估计噪声标准差
    sigma = np.median(np.abs(coeffs[-1])) / 0.6745
    # 软阈值去噪
    threshold = sigma * np.sqrt(2 * np.log(len(data)))
    coeffs_thresh = [coeffs[0]]  # 近似系数保留
    for i in range(1, len(coeffs)):
        coeffs_thresh.append(pywt.threshold(coeffs[i], threshold, mode='soft'))
    # 重构信号
    denoised = pywt.waverec(coeffs_thresh, wavelet)
    return denoised

4.3 异常值检测与处理：别让一个坏点毁了整个分析

呼吸机数据里，异常值太常见了。传感器接触不良、病人突然动了一下、管路积水……都会产生离谱的数据点。如果不处理，这些异常值会严重影响后续的呼吸频率计算、潮气量估算，甚至触发假报警。

异常值检测的方法有很多，我挑几个最实用的：

方法	原理	适用场景	缺点
3σ 法则	超出均值±3倍标准差的值视为异常	数据近似正态分布	对非正态分布效果差
IQR 方法	超出 Q1-1.5*IQR 或 Q3+1.5*IQR 的值视为异常	数据分布未知时	对尾部数据不敏感
移动窗口法	用滑动窗口内的局部统计量判断	信号有趋势变化时	窗口大小需要调参
Z-score 改进版	用中位数和 MAD 代替均值和标准差	数据有较多异常值时	计算量稍大

我个人最常用的是移动窗口法。为什么？因为呼吸机信号是非平稳的——病人呼吸深度会变，基线也会漂移。用全局的 3σ 法则，会把正常的变化也当成异常。而移动窗口法只跟局部比较，适应性好得多。

下面是一个移动窗口异常值检测的实现：

def moving_window_outlier_detection(data, window_size=50, threshold=3.0):
    """
    移动窗口异常值检测
    :param data: 输入信号
    :param window_size: 窗口大小
    :param threshold: 阈值倍数
    """
    outliers = np.zeros_like(data, dtype=bool)
    for i in range(len(data)):
        # 取当前点前后的窗口数据
        start = max(0, i - window_size // 2)
        end = min(len(data), i + window_size // 2)
        window = data[start:end]
        # 计算窗口内的均值和标准差
        mean = np.mean(window)
        std = np.std(window)
        # 判断当前点是否异常
        if abs(data[i] - mean) > threshold * std:
            outliers[i] = True
    return outliers

# 处理异常值：用窗口内的中位数替换
def replace_outliers(data, outliers, window_size=50):
    cleaned = data.copy()
    for i in np.where(outliers)[0]:
        start = max(0, i - window_size // 2)
        end = min(len(data), i + window_size // 2)
        cleaned[i] = np.median(data[start:end])
    return cleaned

4.4 预处理流程：把三件套串起来

好了，现在我们有滤波、去噪、异常值检测三件套了。但实际项目中，这三步不是随便做的。顺序很重要。

我的标准流程是：

1. 先做异常值检测与替换——因为异常值会干扰滤波器的计算（尤其是 IIR 滤波器，一个异常点能震半天）。
2. 再做滤波——去掉已知频率范围的噪声。
3. 最后做去噪——处理剩下的非平稳噪声。

为什么去噪放最后？因为滤波可能会改变噪声的统计特性，先滤波再去噪，小波阈值会更准确。

下面是一个完整的预处理函数：

def preprocess_respiratory_signal(data, fs):
    """
    呼吸机信号预处理完整流程
    """
    # 1. 异常值检测与替换
    outliers = moving_window_outlier_detection(data, window_size=50, threshold=3.5)
    data_clean = replace_outliers(data, outliers, window_size=50)
    
    # 2. 带通滤波：保留 0.1-0.5 Hz 的呼吸信号
    from scipy.signal import butter, filtfilt
    nyquist = 0.5 * fs
    low = 0.1 / nyquist
    high = 0.5 / nyquist
    b, a = butter(4, [low, high], btype='band')
    data_filtered = filtfilt(b, a, data_clean)
    
    # 3. 小波去噪
    data_denoised = wavelet_denoise(data_filtered, wavelet='db4', level=4)
    
    return data_denoised

下一章我们会讲特征提取，到时候你会发现——预处理做得好，特征提取事半功倍。预处理做得烂，后面再怎么折腾也是白搭。