1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155
| """ SHAP (SHapley Additive exPlanations) 在碰撞严重度预测中的实现 基于博弈论的模型解释框架 """
import numpy as np from sklearn.ensemble import RandomForestClassifier from sklearn.model_selection import train_test_split from dataclasses import dataclass from typing import List, Dict import warnings warnings.filterwarnings('ignore')
try: import shap HAS_SHAP = True except ImportError: HAS_SHAP = False print("SHAP 未安装,使用自定义实现")
@dataclass class CrashFeature: """碰撞特征定义""" name: str description: str value_range: tuple typical_impact: str
CRASH_FEATURES = [ CrashFeature("speed_limit", "道路限速 (km/h)", (20, 120), "正影响"), CrashFeature("vehicle_type", "车辆类型", (0, 5), "大型车更严重"), CrashFeature("collision_type", "碰撞类型", (0, 5), "正面>追尾>侧面"), CrashFeature("weather", "天气条件", (0, 3), "恶劣天气加剧"), CrashFeature("lighting", "光照条件", (0, 2), "夜间更严重"), CrashFeature("road_geometry", "道路几何", (0, 3), "弯道更危险"), CrashFeature("driver_age", "驾驶员年龄", (18, 80), "极端年龄更严重"), CrashFeature("alcohol_involved", "酒精涉及", (0, 1), "正影响"), CrashFeature("seatbelt_used", "安全带使用", (0, 1), "负影响"), CrashFeature("airbag_deployed", "气囊展开", (0, 1), "负影响"), ]
def generate_crash_data(n_samples=2000): """生成模拟碰撞数据""" np.random.seed(42) data = np.zeros((n_samples, len(CRASH_FEATURES))) for i, feat in enumerate(CRASH_FEATURES): low, high = feat.value_range if low == 0 and high == 1: data[:, i] = np.random.choice([0, 1], n_samples, p=[0.7, 0.3]) else: data[:, i] = np.random.uniform(low, high, n_samples) severity = np.zeros(n_samples) for i in range(n_samples): score = 0 score += (data[i, 0] - 20) / 100 * 0.3 score += data[i, 7] * 0.25 score += (1 - data[i, 8]) * 0.2 score += (data[i, 1] / 5) * 0.15 score += (data[i, 2] / 5) * 0.1 score += np.random.normal(0, 0.05) if score > 0.5: severity[i] = 2 elif score > 0.25: severity[i] = 1 else: severity[i] = 0 return data, severity
def simplified_shap_values(model, X_background, X_explain): """ 简化版 SHAP 值估算(实际应使用 shap 库) 基于 Kernel SHAP 的近似: 1. 从背景数据中采样 2. 逐个特征扰动,观察预测变化 """ n_features = X_explain.shape[1] n_background = len(X_background) shap_values = np.zeros_like(X_explain) base_value = np.mean([ np.mean(model.predict_proba(X_background)[0]) for _ in range(min(n_background, 100)) ]) for i in range(len(X_explain)): x = X_explain[i] for j in range(n_features): x_perturbed = x.copy() x_perturbed[j] = np.random.choice(X_background[:, j]) pred_orig = model.predict_proba([x])[0] pred_pert = model.predict_proba([x_perturbed])[0] shap_values[i, j] = np.mean(pred_orig - pred_pert) return shap_values
if __name__ == "__main__": X, y = generate_crash_data(2000) X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42 ) feature_names = [f.name for f in CRASH_FEATURES] rf = RandomForestClassifier(n_estimators=100, max_depth=8, random_state=42) rf.fit(X_train, y_train) accuracy = rf.score(X_test, y_test) print(f"模型准确率: {accuracy:.3f}") importances = rf.feature_importances_ print("\n=== 随机森林特征重要性 ===") for name, imp in sorted(zip(feature_names, importances), key=lambda x: -x[1]): print(f" {name:25s}: {imp:.4f}") if HAS_SHAP: explainer = shap.TreeExplainer(rf) shap_values = explainer.shap_values(X_test[:50]) print(f"\nSHAP 值 shape: {np.array(shap_values).shape}") if isinstance(shap_values, list): for cls in range(min(3, len(shap_values))): mean_shap = np.mean(np.abs(shap_values[cls]), axis=0) top_idx = np.argsort(-mean_shap)[:3] print(f"\n类别 {cls} 前三大特征:") for idx in top_idx: print(f" {feature_names[idx]:25s}: {mean_shap[idx]:.4f}") else: print("\n使用简化版 SHAP 估算...") shap_vals = simplified_shap_values(rf, X_train[:100], X_test[:10]) mean_abs = np.mean(np.abs(shap_vals), axis=0) print("=== 简化 SHAP 平均绝对值 ===") for name, val in sorted(zip(feature_names, mean_abs), key=lambda x: -x[1]): print(f" {name:25s}: {val:.4f}")
|