从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数据当采样率从257HzINCART到1000HzPTB不等时怎样实现智能重采样面对单导联PhysioNet-2017和12导联CPSC-2018的混合数据该选择通道映射还是降维处理1. 数据格式解码从文件到NumPy数组1.1 处理PhysioNet标准格式WFDBWaveForm DataBase库是处理PhysioNet数据的瑞士军刀。以MIT-BIH数据集为例读取一条记录只需三行代码import wfdb record wfdb.rdrecord(mitdb/100, sampto1000) ecg_signal record.p_signal # 获取导联数据关键参数解析pn_dir指定PhysioNet子数据集名称如mitdbsampfrom/sampto控制读取的样本范围channel_names选择特定导联如[MLII]注意WFDB读取的信号值通常是原始ADC单位需要根据头文件中的gain参数转换为mV1.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, fs500): 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] # III加权 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, freq50): # 陷波滤波器 b, a signal.iirnotch(freq, 30, fs) ecg_clean signal.filtfilt(b, a, ecg) # LMS自适应滤波 lms AdaptiveFilter(len(ecg), mu0.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, waveletdb4, level5): coeffs pywt.wavedec(ecg, wavelet, levellevel) 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_fs500, target_leads12): 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的五个不同数据集。