import pandas as pd import numpy as np import re import matplotlib.pyplot as plt from sklearn.model_selection import train_test_split, StratifiedKFold, cross_val_score from sklearn.preprocessing import StandardScaler, OneHotEncoder from sklearn.compose import ColumnTransformer from sklearn.linear_model import LogisticRegression from sklearn.ensemble import RandomForestClassifier from sklearn.svm import SVC from xgboost import XGBClassifier from sklearn.metrics import roc_auc_score, roc_curve, auc from sklearn.base import clone from imblearn.over_sampling import SMOTE, SMOTENC from imblearn.pipeline import make_pipeline from collections import defaultdict from tqdm import tqdm import warnings from sklearn.neighbors import KNeighborsClassifier from sklearn.tree import DecisionTreeClassifier from sklearn.ensemble import ExtraTreesClassifier, GradientBoostingClassifier, AdaBoostClassifier from sklearn.naive_bayes import GaussianNB from sklearn.neural_network import MLPClassifier warnings.filterwarnings('ignore') # 自定义路径处理 def safe_path(path): return path.replace('\\', '/') # 文件路径配置 FEATURE_PATH = safe_path(r'H:/taskPONV/stable/stable_features_dataBS.csv') LABEL_PATH = safe_path(r'H:/taskPONV/stable/label320_96.csv') # ID处理函数 def process_id(raw_id): clean_id = str(raw_id).lower().strip() match = re.match(r'(sub\d{3,4})(\.nii\.gz)?', clean_id) return match.group(1) if match else clean_id.split('.')[0] # 数据加载与预处理 def load_data(): features = pd.read_csv(FEATURE_PATH, dtype={'ID': 'string'}) labels = pd.read_csv(LABEL_PATH, dtype={'ID': 'string'}) features['ID'] = features['ID'].apply(process_id) labels['ID'] = labels['ID'].apply(process_id) data = pd.merge(features, labels, on='ID', how='inner') data['center'] = data['ID'].apply( lambda x: 'A' if re.fullmatch(r'sub\d{3}', x) else 'B' ) return data # 动态构建模型处理管道 def build_model_pipeline(model, categorical_indices, feature_indices, X_columns): """优化后的管道构建函数""" # 获取当前特征中的分类特征位置 current_cat_in_subset = [i for i, idx in enumerate(feature_indices) if idx in categorical_indices] # 判断是否需要SMOTE-NC if len(current_cat_in_subset) > 0: # 确保存在连续特征 num_features = len(feature_indices) - len(current_cat_in_subset) if num_features == 0: raise ValueError("必须至少包含一个连续特征") return make_pipeline( SMOTENC(categorical_features=current_cat_in_subset, random_state=42), StandardScaler(), model ) else: return make_pipeline( SMOTE(random_state=42), StandardScaler(), model ) # # 修改后的前向特征选择函数 def forward_feature_selection(X_train, y_train, X_val, y_val, categorical_indices, feature_names, max_features=31): """按特征顺序进行前向选择的优化版本""" ordered_features = list(range(len(feature_names))) # 按CSV列顺序生成特征索引 # 修改后的模型列表 selector_models = { 'SVM': SVC(probability=True, random_state=42), 'KNN': KNeighborsClassifier(algorithm='kd_tree'), 'DecisionTree': DecisionTreeClassifier(random_state=42), 'RandomForest': RandomForestClassifier(random_state=42), 'ExtraTrees': ExtraTreesClassifier(random_state=42), 'GradientBoosting': GradientBoostingClassifier(random_state=42), 'AdaBoost': AdaBoostClassifier(random_state=42), 'GaussianNB': GaussianNB(), 'LogisticRegression': LogisticRegression(max_iter=1000, random_state=42), 'MLP': MLPClassifier(max_iter=300, random_state=42), 'XGBoost': XGBClassifier(use_label_encoder=False, eval_metric='logloss', random_state=42) } selected = [] best_subset = [] best_val_auc = 0 history = [] # 强制包含至少一个连续特征 continuous_features = [i for i in ordered_features if i not in categorical_indices] if continuous_features: selected.append(continuous_features[0]) ordered_features.remove(continuous_features[0]) pbar = tqdm(total=min(max_features, len(ordered_features)), desc="特征选择进度", postfix={'best_auc': 0}) for feature in ordered_features: if len(selected) >= max_features: break temp_subset = selected + [feature] # 跳过全分类特征的情况 if all(f in categorical_indices for f in temp_subset): continue # 多模型加权评估 model_scores = [] for model_name, base_model in selector_models.items(): try: model = clone(base_model) pipeline = build_model_pipeline(model, categorical_indices, temp_subset, feature_names) # 交叉验证得分（权重0.4） cv_score = cross_val_score(pipeline, X_train[:, temp_subset], y_train, cv=StratifiedKFold(5), scoring='roc_auc', n_jobs=-1).mean() # 验证集得分（权重0.6） pipeline.fit(X_train[:, temp_subset], y_train) val_pred = pipeline.predict_proba(X_val[:, temp_subset])[:, 1] val_score = roc_auc_score(y_val, val_pred) model_scores.append(cv_score) except Exception as e: continue if not model_scores: continue avg_score = np.mean(model_scores) # 更新最佳子集 if avg_score > best_val_auc: best_val_auc = avg_score best_subset = temp_subset.copy() # 记录选择过程 history.append({ 'feature': feature_names[feature], 'score': avg_score }) selected.append(feature) pbar.update(1) pbar.set_postfix(best_auc=best_val_auc) pbar.close() return best_subset, history def main(): # 加载数据 data = load_data() # 划分中心数据 a_center = data[data['center'] == 'A'].drop(columns=['center']) b_center = data[data['center'] == 'B'].drop(columns=['center']) # 获取特征信息 feature_names = a_center.drop(columns=['label', 'ID']).columns.tolist() # 定义分类特征（确保这些列已经是数值编码） categorical_features = ['gender', 'smoke'] # 示例，根据实际数据调整 categorical_indices = [feature_names.index(col) for col in categorical_features] # 数据拆分 X_a = a_center.drop(columns=['label', 'ID']).values y_a = a_center['label'].values X_train, X_test, y_train, y_test = train_test_split( X_a, y_a, test_size=0.3, stratify=y_a, random_state=42 ) X_ext = b_center.drop(columns=['label', 'ID']).values y_ext = b_center['label'].values # 执行特征选择 best_subset, history = forward_feature_selection( X_train, y_train, X_ext, y_ext, categorical_indices, feature_names, max_features=31 ) # 打印结果 print("\n最佳特征子集（{}个）：".format(len(best_subset))) print([feature_names[i] for i in best_subset]) print("\n特征选择过程：") print("步骤 | 特征名称 | 综合得分") for i, record in enumerate(history, 1): print(f"{i:2d} | {record['feature']:15} | {record['score']:.3f}") # 模型评估（修正后的调用） evaluation_results = evaluate_models( X_train, y_train, X_test, y_test, # 添加测试集标签 X_ext, y_ext, # 添加外部验证集标签 best_subset, categorical_indices, feature_names ) # 输出评估结果 print("\n模型性能评估：") print(f"{'模型名称':<20} {'测试集AUC':<10} {'外部验证AUC':<10}") for name, metrics in evaluation_results.items(): print(f"{name:<20} {metrics['Test AUC']:.3f} {metrics['External AUC']:.3f}") if __name__ == "__main__": main()