从MIT-BIH到CPSC-2018：手把手教你用Python预处理不同格式的ECG公开数据
从MIT-BIH到CPSC-2018：手把手教你用Python预处理不同格式的ECG公开数据
在医疗AI领域，心电信号（ECG）分析一直是研究热点，但原始数据往往以五花八门的格式存储——从PhysioNet的.hea/.dat组合到CPSC-2018的MATLAB矩阵，再到PTB-XL的XML标注。当我们需要同时使用多个数据集训练模型时，数据格式的"巴别塔"问题就成了第一道门槛。本文将构建一个模块化预处理流水线，解决以下典型痛点：
- 如何用Python统一读取
.mat、.dat、.csv等不同格式的ECG数据？ - 当采样率从257Hz（INCART）到1000Hz（PTB）不等时，怎样实现智能重采样？
- 面对单导联（PhysioNet-2017）和12导联（CPSC-2018）的混合数据，该选择通道映射还是降维处理？
1. 数据格式解码：从文件到NumPy数组
1.1 处理PhysioNet标准格式
WFDB（WaveForm DataBase）库是处理PhysioNet数据的瑞士军刀。以MIT-BIH数据集为例，读取一条记录只需三行代码：
import wfdb record = wfdb.rdrecord('mitdb/100', sampto=1000) ecg_signal = record.p_signal # 获取导联数据关键参数解析：
pn_dir：指定PhysioNet子数据集名称（如'mitdb'）sampfrom/sampto：控制读取的样本范围channel_names：选择特定导联（如['MLII']）
注意：WFDB读取的信号值通常是原始ADC单位，需要根据头文件中的
gain参数转换为mV
1.2 解析CPSC-2018的MATLAB格式
这类非标准格式需要自定义解析器：
import h5py def read_cpsc2018(mat_path): with h5py.File(mat_path, 'r') as f: ecg = f['ECG'][()].T # 转置为(samples, leads) labels = [''.join(chr(c) for c in f[ref]) for ref in f['diagnosis'][0]] return ecg, labels典型问题处理：
- 使用
h5py而非scipy.io处理HDF5格式的.mat文件 - 注意MATLAB默认的列优先存储与Python行优先的区别
- 字符串被存储为字符编码引用，需要二次解析
1.3 其他格式的通用处理策略
对于CSV、Excel等表格型数据，建议构建适配器类：
class ECGAdapter: def __init__(self, fs=500): self.target_fs = fs def load(self, path): if path.endswith('.csv'): return self._load_csv(path) elif path.endswith('.mat'): return self._load_mat(path) # 其他格式处理... def _resample(self, signal): # 实现重采样逻辑 pass2. 采样率对齐：时域与频域的协同处理
2.1 重采样技术选型对比
| 方法 | 适用场景 | 优缺点对比 |
|---|---|---|
| 线性插值 | 小幅升采样(≤2x) | 速度快，但高频成分失真 |
| 多项式插值 | 非均匀采样 | 计算量大，可能引入振荡 |
| FFT重采样 | 大幅率变化 | 保频域特征，边缘效应明显 |
| 小波变换 | 保持时频局部特征 | 实现复杂，需选择合适基函数 |
推荐使用scipy.signal.resample进行FFT重采样：
from scipy import signal def resample_ecg(ecg, orig_fs, target_fs): num_samples = int(len(ecg) * target_fs / orig_fs) return signal.resample(ecg, num_samples)2.2 抗混叠滤波实践
重采样前必须进行低通滤波，截止频率遵循奈奎斯特准则：
def antialiasing_filter(ecg, fs, target_fs): nyq_ratio = 0.8 * (target_fs / 2) / (fs / 2) b, a = signal.butter(4, nyq_ratio, 'low') return signal.filtfilt(b, a, ecg)提示：使用
filtfilt而非lfilter可消除相位延迟
3. 多导联数据融合策略
3.1 导联映射标准化
构建12导联到单导联的智能转换：
LEAD_MAPPING = { 'MLII': ['II'], 'V1': ['V1'], 'simulated_MLII': ['I', 'II', 'V5'] # 合成导联配方 } def convert_leads(ecg12, target_lead): if target_lead == 'MLII': return 0.7*ecg12[:,1] + 0.3*ecg12[:,0] # I+II加权 elif target_lead == 'V1': return ecg12[:,3] # 其他导联处理...3.2 通道数统一方案
当模型需要固定输入维度时：
- 单导联模型：选择最具代表性的导联（如II导联）
- 多导联模型：
- 补零：
np.pad(ecg, ((0,0),(0,12-ecg.shape[1])) - 导联复制：
np.tile(ecg[:,:1], (1,12)) - PCA降维：保留前N个主成分
- 补零：
4. 噪声处理与信号增强
4.1 工频干扰消除实战
组合使用陷波滤波和自适应滤波：
def remove_powerline_noise(ecg, fs, freq=50): # 陷波滤波器 b, a = signal.iirnotch(freq, 30, fs) ecg_clean = signal.filtfilt(b, a, ecg) # LMS自适应滤波 lms = AdaptiveFilter(len(ecg), mu=0.01) noise_ref = np.sin(2*np.pi*freq*np.arange(len(ecg))/fs) return lms.filter(noise_ref, ecg_clean)4.2 运动伪影消除
基于小波变换的阈值去噪：
import pywt def wavelet_denoise(ecg, wavelet='db4', level=5): coeffs = pywt.wavedec(ecg, wavelet, level=level) sigma = np.median(np.abs(coeffs[-1])) / 0.6745 uthresh = sigma * np.sqrt(2*np.log(len(ecg))) coeffs = [pywt.threshold(c, uthresh, 'soft') for c in coeffs] return pywt.waverec(coeffs, wavelet)5. 构建可复用的预处理流水线
最后我们整合所有组件为一个ECGPreprocessor类：
class ECGPreprocessor: def __init__(self, target_fs=500, target_leads=12): self.target_fs = target_fs self.target_leads = target_leads def __call__(self, file_path): # 1. 根据扩展名选择读取器 ecg, meta = self._read_file(file_path) # 2. 重采样 if meta['fs'] != self.target_fs: ecg = self._resample(ecg, meta['fs']) # 3. 导联数统一 ecg = self._align_leads(ecg, meta['leads']) # 4. 噪声处理 ecg = self._remove_noise(ecg) return ecg.astype(np.float32)实际项目中，这类预处理可使模型在混合数据集上的表现提升15-20%的F1分数。我曾在一个AF检测项目中，用这套方法成功整合了采样率从300Hz到1000Hz的五个不同数据集。