python import neurokit2 as nk import pandas as pd import numpy as np from scipy.signal import butter, filtfilt from sklearn.ensemble import RandomForestRegressor from sklearn.model_selection import GroupKFold from sklearn.metrics import mean_absolute_error

预处理函数

def bandpass_filter(data, low, high, fs, order=4): nyq = 0.5 * fs low = low / nyq high = high / nyq b, a = butter(order, [low, high], btype='band') return filtfilt(b, a, data)

fs = 500

对 ECG 和 PPG 滤波

ecg_filtered = bandpass_filter(df['ecg'], 1, 40, fs) ppg_filtered = bandpass_filter(df['ppg'], 0.5, 10, fs)

使用 Neurokit 检测 R 峰

_, rpeaks = nk.ecg_peaks(ecg_filtered, sampling_rate=fs) r_times = rpeaks['ECG_R_Peaks'] / fs

使用 Neurokit 处理 PPG 提取特征点

ppg_signals, info = nk.ppg_process(ppg_filtered, sampling_rate=fs)

获取 PPG 峰值

ppg_peaks = info['PPG_Peaks']

计算 PAT

pat_peak = [] for r_time in r_times[:-1]: # 找到紧随 R 峰之后的 PPG 峰值 later_peaks = ppg_peaks[ppg_peaks / fs > r_time] if len(later_peaks) > 0: t_ppg = later_peaks[0] / fs pat_peak.append(t_ppg - r_time)

提取更多特征 ...

### 模型训练与验证

```python
features = ['pat_mean', 'pat_std', 'hr', 'ppg_amp', ...]
X = df_features[features]
y_sbp = df_features['sbp']
y_dbp = df_features['dbp']
groups = df_features['subject_id']

cv = GroupKFold(n_splits=5)
model = RandomForestRegressor(n_estimators=100, random_state=42)

mae_list = []
for train_idx, test_idx in cv.split(X, y_sbp, groups):
    X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
    y_train, y_test = y_sbp.iloc[train_idx], y_sbp.iloc[test_idx]
    model.fit(X_train, y_train)
    preds = model.predict(X_test)
    mae_list.append(mean_absolute_error(y_test, preds))

print(f"SBP MAE: {np.mean(mae_list):.2f} mmHg")

结果可视化

import matplotlib.pyplot as plt
# 地面真值 vs 预测散点图
plt.scatter(y_test, preds, alpha=0.4)
plt.plot([80, 180], [80, 180], 'r--')
plt.xlabel('参考 SBP (mmHg)')
plt.ylabel('预测 SBP (mmHg)')
plt.title('血压预测结果')
plt.grid()
plt.show()