3、数据预处理流水线：带通滤波、独立成分分析(ICA)、共平均参考(CAR)、基线校正、分段与标注

说实话，数据预处理这事儿，看着不起眼，但我在实际项目中栽过不少跟头。你想想看，原始脑电信号那叫一个乱——工频干扰、眼电伪迹、肌电噪声，全混在一起。模型再牛，喂进去的是垃圾，出来的还是垃圾。所以这一章，咱们就把预处理流水线拆开揉碎了讲。

3.1 带通滤波：把不要的频率筛掉

带通滤波，说白了就是只保留你关心的频段。比如做运动想象，我们通常关注8-30Hz（mu节律和beta节律）。低于0.5Hz的基线漂移，高于50Hz的肌电噪声，统统干掉。

我个人习惯用Butterworth滤波器，零相位延迟，不会把信号搞歪。阶数选4阶，再高容易震荡。

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

def bandpass_filter(data, lowcut=0.5, highcut=45.0, fs=250, order=4):
    nyquist = 0.5 * fs
    low = lowcut / nyquist
    high = highcut / nyquist
    b, a = butter(order, [low, high], btype='band')
    # 用filtfilt实现零相位滤波
    filtered = filtfilt(b, a, data, axis=-1)
    return filtered

# 示例：250Hz采样率，保留0.5-45Hz
eeg_data = np.random.randn(64, 1000)  # 64通道，4秒数据
clean_data = bandpass_filter(eeg_data)

3.2 独立成分分析(ICA)：把伪迹揪出来

ICA这东西，原理上挺玄乎——把混合信号拆成统计独立的成分。但在工程上，我们用它干一件事：干掉眼电和心电伪迹。

我曾经在某个BCI竞赛数据集上，跑完ICA发现前两个成分明显是眨眼和水平眼动。手动剔除后，分类准确率从72%直接跳到85%。你想想看，这差距有多大。

实际操作分三步：

1. 跑ICA：用FastICA或Infomax算法，成分数一般等于通道数
2. 识别伪迹成分：看时域波形（有没有陡峭的尖峰）、看地形图（是不是集中在额叶或眼周）
3. 剔除并重建：把伪迹成分置零，反变换回原始空间

from sklearn.decomposition import FastICA

def ica_denoise(data, n_components=None):
    if n_components is None:
        n_components = data.shape[0]  # 等于通道数
    ica = FastICA(n_components=n_components, random_state=42)
    # 分解
    sources = ica.fit_transform(data.T)  # 注意维度：样本数×成分数
    # 这里假设我们手动标记了第0、1成分为伪迹
    sources[:, :2] = 0
    # 重建
    clean = ica.inverse_transform(sources).T
    return clean

3.3 共平均参考(CAR)：把全局噪声抹平

CAR的原理很简单：每个通道减去所有通道的平均值。为什么要这么做？因为脑电记录时，参考电极的位置会影响信号。共平均参考相当于做了一个虚拟的、全脑平均的参考。

嗯，这里要注意：CAR对全局性噪声（比如工频干扰、头皮电位的整体漂移）特别有效。但对局部信号（比如某个脑区的ERD/ERS）可能会削弱。所以不是所有场景都适合。

def common_average_reference(data):
    """
    data: 通道数 × 时间点
    """
    avg = np.mean(data, axis=0, keepdims=True)
    car_data = data - avg
    return car_data

参考方式	优点	缺点
共平均参考(CAR)	抑制全局噪声，实现简单	可能削弱局部信号
双极参考	突出局部差异	通道数减半
拉普拉斯参考	空间分辨率高	计算复杂，边缘通道效果差

3.4 基线校正：把信号拉回零点

基线校正，说白了就是让每个trial的起始点对齐到零。为什么需要？因为脑电信号有直流漂移，不同trial的基线可能不一样。如果不校正，模型会学到「这个trial整体偏高」这种没用的特征。

我常用的做法是：取刺激前200ms作为基线，计算平均值，然后整个trial减去这个值。

def baseline_correction(trial, baseline_window=(-0.2, 0)):
    """
    trial: 时间点数组（假设采样率250Hz，基线窗口-200ms到0ms）
    """
    fs = 250
    start_idx = int(baseline_window[0] * fs)
    end_idx = int(baseline_window[1] * fs)
    baseline_mean = np.mean(trial[start_idx:end_idx])
    corrected = trial - baseline_mean
    return corrected

3.5 分段与标注：把连续信号切成样本

最后一步，把连续脑电信号切成一个个trial。每个trial对应一个实验事件（比如「左手运动想象」或「右手运动想象」）。

分段时要注意两个参数：

• t_start：刺激前多久开始（通常-0.5s到-0.2s，包含基线）
• t_end：刺激后多久结束（运动想象任务通常是2-4秒）

标注就是把事件编码转换成标签。比如：左手→0，右手→1，舌头→2。

def epoch_and_label(continuous_data, events, fs=250, t_start=-0.5, t_end=3.0):
    """
    continuous_data: 通道数 × 总时间点
    events: 列表，每个元素是 (时间点, 事件类型)
    """
    start_offset = int(t_start * fs)
    end_offset = int(t_end * fs)
    epochs = []
    labels = []
    
    for onset, event_type in events:
        start = onset + start_offset
        end = onset + end_offset
        if start < 0 or end > continuous_data.shape[1]:
            continue  # 超出边界就跳过
        epoch = continuous_data[:, start:end]
        epochs.append(epoch)
        labels.append(event_type)
    
    return np.array(epochs), np.array(labels)

3.6 完整流水线：把一切串起来

下面这张图，是我自己项目里用的预处理流程。你照着这个顺序走，基本不会出大问题。

好了，这一章的内容就这些。数据预处理是深度学习模型的基础，花时间把这一步做好，后面模型训练会省心很多。