用Python处理MIT-BIH-AF房颤数据集：从文件读取到信号预处理的完整实战指南

用Python处理MIT-BIH-AF房颤数据集：从文件读取到信号预处理的完整实战指南

在生物医学信号处理领域，MIT-BIH房颤数据库（MIT-BIH-AF）是研究心律失常特别是房颤现象的黄金标准数据集。这个包含23名患者长期心电图记录的宝贵资源，为开发智能诊断算法提供了真实世界的数据基础。但对于刚接触该数据集的研究者来说，面对.dat、.atr、.qrs等专业格式文件时，往往会遇到数据解析和预处理的挑战。

本文将手把手带你用Python实现从原始文件读取到信号预处理的完整流程，特别适合生物医学工程、AI医疗和信号处理领域需要快速上手的开发者和研究者。我们将重点使用WFDB（WaveForm DataBase）这个专业生物医学信号处理库，配合Matplotlib可视化和小波分析技术，构建端到端的处理管道。

1. 环境准备与数据集获取

1.1 安装必要的Python库

处理生理信号需要专门的工具链，以下是核心依赖库及其作用：

pip install wfdb matplotlib numpy scipy PyWavelets

• wfdb：专门用于读取PhysioNet各类生理信号数据的库
• PyWavelets：提供小波变换实现，用于信号去噪和特征提取
• scipy：提供信号重采样等科学计算功能

1.2 数据集下载与结构解析

MIT-BIH-AF数据集可通过PhysioNet官网获取，包含以下关键文件类型：

数据集采用双导联（通常为II和V1导联）记录，采样率为250Hz，每个记录包含约920万个数据点。建议先通过PhysioNet的LightWave可视化工具在线浏览数据特征。

2. 数据读取与基础解析

2.1 使用WFDB读取原始信号

以下代码演示如何加载单个患者的完整记录：

import wfdb # 设置数据路径（示例使用04015号患者） record_path = 'path/to/04015' # 读取信号数据（physical=True获取物理量值） record = wfdb.rdrecord(record_path, physical=True) annotations = wfdb.rdann(record_path, 'atr') rpeaks = wfdb.rdann(record_path, 'qrs') # 获取第一导联信号和R峰位置 ecg_signal = record.p_signal[:, 0] r_peaks = rpeaks.sample

2.2 理解标注数据结构

.atr文件中的标注包含重要的临床信息，主要关注aux_note字段：

# 提取房颤相关标注 afib_annotations = [ (sample, note) for sample, note in zip(annotations.sample, annotations.aux_note) if '(AFIB' in note ]

常见标注类型包括：

• (N：正常窦性心律
• (AFIB：房颤发作
• (AFL：房扑
• (J：交界性心律

3. 信号可视化与分析

3.1 基础ECG波形绘制

使用Matplotlib实现专业级心电图展示：

import matplotlib.pyplot as plt plt.figure(figsize=(12, 4)) plt.plot(ecg_signal[10000:11000], linewidth=0.5) plt.title('ECG Segment with AFIB Episode') plt.xlabel('Samples (250Hz)') plt.ylabel('Amplitude (mV)') # 标记R峰位置 for peak in [p for p in r_peaks if 10000 <= p <= 11000]: plt.axvline(peak-10000, color='r', linestyle='--', alpha=0.3)

3.2 心率变异性(HRV)分析

房颤患者的关键特征是其不规则的心率：

rr_intervals = np.diff(r_peaks) / 250 # 转换为秒单位 plt.hist(rr_intervals, bins=50) plt.title('RR Interval Distribution') plt.xlabel('Interval (s)') plt.ylabel('Count')

提示：正常窦性心律的RR间期分布相对集中，而房颤发作时会出现典型的"RR间期绝对不齐"特征

4. 高级预处理技术

4.1 小波去噪实战

ECG信号常受到肌电噪声、基线漂移等干扰，小波变换能有效分离噪声成分：

import pywt def wavelet_denoise(signal, wavelet='db4', level=3): coeffs = pywt.wavedec(signal, wavelet, level=level) sigma = np.median(np.abs(coeffs[-level])) / 0.6745 uthresh = sigma * np.sqrt(2*np.log(len(signal))) coeffs[1:] = [pywt.threshold(c, uthresh, mode='soft') for c in coeffs[1:]] return pywt.waverec(coeffs, wavelet)

4.2 信号重采样标准化

不同长度的ECG片段需要统一尺寸才能输入机器学习模型：

from scipy.signal import resample def uniform_resample(signal, target_length): current_length = len(signal) return resample(signal, target_length) if current_length != target_length else signal

5. 特征工程与数据集构建

5.1 时域特征提取

构建包含临床意义的特征集：

def extract_features(rr_segment): features = { 'mean_rr': np.mean(rr_segment), 'std_rr': np.std(rr_segment), 'rmssd': np.sqrt(np.mean(np.diff(rr_segment)**2)), 'nn50': sum(np.abs(np.diff(rr_segment)) > 0.05) } return features

5.2 创建训练数据集

将原始信号转换为适合机器学习的形式：

def create_dataset(ecg_signal, r_peaks, annotations, window_size=10): X, y = [], [] for i in range(len(r_peaks) - window_size): rr_segment = np.diff(r_peaks[i:i+window_size+1]) / 250 features = extract_features(rr_segment) # 检查窗口内是否有房颤标注 window_start = r_peaks[i] window_end = r_peaks[i+window_size] afib_in_window = any( window_start <= sample <= window_end for sample, note in zip(annotations.sample, annotations.aux_note) if '(AFIB' in note ) X.append(list(features.values())) y.append(1 if afib_in_window else 0) return np.array(X), np.array(y)

6. 实战技巧与性能优化

6.1 内存高效处理策略

对于长达10小时的记录，内存管理至关重要：

def chunked_processing(record_path, chunk_size=300000): signals = [] for i in range(0, record.sig_len, chunk_size): chunk = wfdb.rdrecord(record_path, sampfrom=i, sampto=i+chunk_size).p_signal signals.append(process_chunk(chunk)) return np.concatenate(signals)

6.2 多进程加速

利用Python的multiprocessing加速特征提取：

from multiprocessing import Pool def parallel_feature_extraction(r_peaks_chunks): with Pool(processes=4) as pool: results = pool.map(extract_features, r_peaks_chunks) return results

7. 质量评估与可视化验证

7.1 预处理效果对比

fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 6)) ax1.plot(raw_signal[5000:6000]) ax1.set_title('Raw Signal') ax2.plot(processed_signal[5000:6000]) ax2.set_title('After Denoising & Detrending')

7.2 房颤检测结果可视化

plt.figure(figsize=(12, 3)) plt.plot(ecg_signal[30000:35000]) for ann_sample, ann_note in zip(annotations.sample, annotations.aux_note): if 30000 <= ann_sample <= 35000 and '(AFIB' in ann_note: plt.axvline(ann_sample-30000, color='r', alpha=0.3)

在实际项目中，处理MIT-BIH-AF数据集时最常见的坑是时区转换问题——记录中的时间戳需要与采样点正确对应。另一个经验是，对于房颤检测任务，建议至少使用10个连续RR间期作为分析窗口，太短的窗口难以捕捉心律不齐的特征。