4. 数据预处理基础：滤波、重参考、基线校正与坏道处理

各位同学，欢迎来到数据预处理这一章。说实话，在神经信号处理这个领域，我见过太多人一上来就急着跑模型，结果数据里全是噪声和伪迹，最后模型效果一塌糊涂。我自己早期做EEG研究时就吃过这个亏——有一次花了两周时间调参，结果发现是参考电极没选对，数据本身就有问题。从那以后，我养成了一个习惯：预处理花多少时间都不为过。

今天我们要聊的内容，说白了就是给原始神经信号「洗个澡」。把脏东西洗掉，把有用的部分留下来。嗯，咱们一个一个来看。

4.1 滤波：带通与陷波

滤波是预处理的第一步。为什么要滤波？因为原始信号里什么都有——50Hz的工频干扰、肌肉活动的低频漂移、甚至你呼吸时电极线晃动产生的伪迹。我们的目标是把这些乱七八糟的东西去掉，只保留脑电或神经活动的频段。

4.1.1 带通滤波

带通滤波，说白了就是只让某个频率范围内的信号通过。比如EEG信号，我们通常关注0.5Hz到40Hz这个范围。低于0.5Hz的是基线漂移，高于40Hz的可能是肌电噪声。

我个人习惯用scipy的butter函数设计巴特沃斯滤波器。为什么选它？因为通带内最平坦，不会把信号搞变形。

import numpy as np
from scipy import signal
import matplotlib.pyplot as plt

def bandpass_filter(data, fs, lowcut=0.5, highcut=40.0, order=4):
    """
    带通滤波器
    data: 原始信号 (channels × time)
    fs: 采样率 (Hz)
    lowcut: 低频截止 (Hz)
    highcut: 高频截止 (Hz)
    """
    nyquist = 0.5 * fs
    low = lowcut / nyquist
    high = highcut / nyquist
    b, a = signal.butter(order, [low, high], btype='band')
    filtered_data = signal.filtfilt(b, a, data, axis=-1)
    return filtered_data

# 使用示例
fs = 250  # 采样率250Hz
eeg_data = np.random.randn(32, 1000)  # 32通道，4秒数据
filtered = bandpass_filter(eeg_data, fs, 0.5, 40)

4.1.2 陷波滤波

陷波滤波，专门对付工频干扰。在中国是50Hz，有些国家是60Hz。这个噪声特别讨厌，因为它就在我们关心的频段中间。

我曾经在项目里遇到过一个情况：采集到的数据50Hz处能量特别高，但带通滤波后还是能看到明显的50Hz残留。后来发现是陷波滤波器的带宽设得太窄了。嗯，这里要注意：陷波滤波器的带宽至少要设到2-3Hz，才能有效抑制工频干扰。

def notch_filter(data, fs, freq=50.0, quality=30):
    """
    陷波滤波器
    quality: 品质因数，越大带宽越窄
    """
    nyquist = 0.5 * fs
    notch_freq = freq / nyquist
    b, a = signal.iirnotch(notch_freq, quality)
    filtered_data = signal.filtfilt(b, a, data, axis=-1)
    return filtered_data

# 使用示例
filtered = notch_filter(eeg_data, fs, 50.0, quality=30)

4.2 重参考

重参考，这个知识点很多初学者容易忽略。说白了，我们记录的电压值都是相对值——相对于某个参考电极的差值。参考电极选得不好，整个数据就废了。

常见的参考方式有几种：

• 单极参考：以某个电极为参考（比如Cz、A1/A2）。简单，但参考电极本身可能有噪声。
• 双极参考：两个相邻电极相减。能消除共模噪声，但会损失空间信息。
• 平均参考：所有电极的平均值作为参考。我个人最常用这个，因为它能有效抑制全局噪声。
• 重参考到某个区域：比如以左右乳突的平均值为参考。

def rereference(data, ref_type='average', ref_channels=None):
    """
    重参考
    data: channels × time
    ref_type: 'average', 'single', 'bipolar'
    ref_channels: 参考电极索引（单极参考时使用）
    """
    if ref_type == 'average':
        ref_signal = np.mean(data, axis=0, keepdims=True)
        return data - ref_signal
    elif ref_type == 'single':
        if ref_channels is None:
            raise ValueError("单极参考需要指定参考电极")
        ref_signal = data[ref_channels, :]
        return data - ref_signal
    elif ref_type == 'bipolar':
        # 相邻电极相减
        bipolar_data = []
        for i in range(data.shape[0] - 1):
            bipolar_data.append(data[i, :] - data[i+1, :])
        return np.array(bipolar_data)
    else:
        raise ValueError("不支持的参考类型")

4.3 基线校正

基线校正，说白了就是把信号拉平。因为EEG信号会有缓慢的漂移，如果不做校正，不同试次之间的基线水平不一样，叠加平均时就会出问题。

通常的做法是：取刺激前的一段时间（比如-200ms到0ms）作为基线，计算这段时间的平均值，然后从整个试次中减去这个平均值。

def baseline_correction(data, baseline_start, baseline_end, fs):
    """
    基线校正
    data: trials × channels × time
    baseline_start: 基线起始时间 (秒)
    baseline_end: 基线结束时间 (秒)
    fs: 采样率
    """
    start_idx = int(baseline_start * fs)
    end_idx = int(baseline_end * fs)
    
    # 计算基线均值
    baseline_mean = np.mean(data[:, :, start_idx:end_idx], axis=-1, keepdims=True)
    
    # 减去基线
    corrected_data = data - baseline_mean
    return corrected_data

# 使用示例
trials_data = np.random.randn(100, 32, 1000)  # 100个试次，32通道，4秒
corrected = baseline_correction(trials_data, -0.2, 0, fs=250)

4.4 坏道检测与插值

坏道，就是那些信号质量特别差的电极通道。可能是电极接触不良、阻抗太高、或者干脆断了。如果不处理坏道，它们会把噪声传播到整个数据中。

坏道检测的方法有很多，我常用的几个指标：

• 方差过大：某个通道的方差超过其他通道的3倍以上
• 相关性过低：某个通道与相邻通道的相关系数低于0.4
• 高频噪声过多：高频段（>30Hz）的能量占比过高
• 信号平坦：某个通道的信号几乎不变，可能是电极脱落

def detect_bad_channels(data, threshold_variance=3.0, threshold_corr=0.4):
    """
    坏道检测
    data: channels × time
    返回坏道索引列表
    """
    n_channels = data.shape[0]
    bad_channels = []
    
    # 1. 方差检测
    variances = np.var(data, axis=1)
    median_var = np.median(variances)
    mad_var = np.median(np.abs(variances - median_var))
    z_scores = (variances - median_var) / (1.4826 * mad_var)
    bad_variance = np.where(np.abs(z_scores) > threshold_variance)[0]
    bad_channels.extend(bad_variance.tolist())
    
    # 2. 相关性检测
    corr_matrix = np.corrcoef(data)
    for ch in range(n_channels):
        # 计算与相邻通道的平均相关性
        neighbors = get_neighbors(ch)  # 需要实现获取相邻通道的函数
        if len(neighbors) > 0:
            avg_corr = np.mean([corr_matrix[ch, n] for n in neighbors])
            if avg_corr < threshold_corr:
                bad_channels.append(ch)
    
    return list(set(bad_channels))

def interpolate_bad_channels(data, bad_channels, good_channels):
    """
    坏道插值：用相邻好道的加权平均来替换坏道
    """
    from scipy.interpolate import griddata
    
    # 获取电极位置（需要提前定义）
    positions = get_electrode_positions()  # 假设有这个函数
    
    good_pos = positions[good_channels]
    good_data = data[good_channels, :]
    
    for ch in bad_channels:
        bad_pos = positions[ch]
        # 用最近邻插值
        interpolated = griddata(good_pos, good_data, bad_pos, method='nearest')
        data[ch, :] = interpolated
    
    return data

知识体系总览

说了这么多，咱们用一张图来总结一下整个数据预处理的流程：

这张图把整个流程串起来了。你想想看，从原始数据到干净可用的数据，每一步都有它的道理。跳过任何一步，都可能让后续的分析白费功夫。

好了，这一章的内容就到这里。记住一句话：预处理做得好，模型训练没烦恼。下一章我们会聊到更具体的特征提取方法，到时候这些预处理过的数据就能派上用场了。

---