如何在Sklearn管道中计算变换后特征的排列重要性

问题背景 #

我通过Sklearn管道实现了完整的预处理+建模流程，现在需要计算预处理后特征的排列重要性，但默认sklearn.inspection.permutation_importance是针对原始特征计算的。同时我担心单独对测试集做预处理会导致数据泄露，想知道可行的实现技巧。

我的管道代码如下：

import numpy as np
import pandas as pd
import os
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.model_selection import train_test_split
from sklearn.decomposition import PCA
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LogisticRegression
from sklearn.inspection import permutation_importance

###
# 数据加载部分
###

X_train, X_test, y_train, y_test = train_test_split(all_features, diagnosis, test_size=0.25, random_state=42)

pca_conn = Pipeline(
    steps = [("group_whiten", StandardScaler()),
             ('pca_conn', PCA(n_components = 100)),
            ("pca_whiten", StandardScaler())]
)

pca_grad = Pipeline(
    steps = [("group_whiten", StandardScaler()),
             ('pca_grad', PCA(n_components = 100)),
            ("pca_whiten", StandardScaler())]
)

pca_centroid_disp_pca = Pipeline(
    steps = [("group_whiten", StandardScaler()),
             ('pca_grad', PCA(n_components = 10)),
            ("pca_whiten", StandardScaler())]
)

pca_cortex_disp_pca = Pipeline(
    steps = [("group_whiten", StandardScaler()),
             ('pca_grad', PCA(n_components = 100)),
            ("pca_whiten", StandardScaler())]
)

cat_encoder = Pipeline(
    steps = [("cat_encoder", OneHotEncoder(handle_unknown="ignore"))]
)
whiten = Pipeline(
    steps = [("whiten", StandardScaler())]
)

preprocessor = ColumnTransformer(
    transformers=[
        ("pca_conn", pca_conn, conn_cols),
        ("pca_grad", pca_grad, grad_cols),
        ("pca_centroid_disp_pca", pca_centroid_disp_pca, centroid_disp_cols),
        ("pca_cortex_disp_pca", pca_cortex_disp_pca, cortex_disp_cols),
        ("encode_dataset", cat_encoder, ["dataset"]),
        ("encode_sex", cat_encoder, ["sex"]),
        ("whiten_fd", whiten, ["mean_fd"]),
        ("whiten_age", whiten, ["age"])
    ]
)

lr = LogisticRegression(random_state=42, max_iter = 10000)
clf = Pipeline([('preprocessor', preprocessor),
                ('lr',lr)])

trained_logreg = clf.fit(X_train, y_train)
trained_logreg.score(X_test, y_test)

# 默认计算的是原始特征的重要性，不是我要的
perm_acc = permutation_importance(trained_logreg, X_test, y_test,n_repeats=100, random_state=42, n_jobs = -1)

可行解决方案 #

直接拆分训练好的管道，用训练阶段拟合完成的预处理步骤转换测试集，再对转换后的特征计算排列重要性，完全不会造成数据泄露：

步骤1：提取预处理后的测试集特征 #

用训练好的管道里的preprocessor直接转换测试集，这个步骤用的是训练集拟合的参数，没有用到测试集的任何统计信息：

# 获取训练好的预处理模块
preprocessor = trained_logreg.named_steps['preprocessor']
# 转换测试集得到预处理后的特征矩阵
X_test_transformed = preprocessor.transform(X_test)

步骤2：提取训练好的模型 #

从管道中取出已经训练完成的逻辑回归模型：

model = trained_logreg.named_steps['lr']

步骤3：计算变换后特征的排列重要性 #

直接对预处理后的特征和测试集标签计算排列重要性：

perm_acc_transformed = permutation_importance(
    model, 
    X_test_transformed, 
    y_test, 
    n_repeats=100, 
    random_state=42, 
    n_jobs=-1
)

补充：给变换后特征命名（方便解读） #

变换后的特征没有默认名称，需要手动对应每个预处理模块的输出：

# 手动生成变换后特征的名称列表
transformed_feature_names = []

# 处理每个transformer的输出特征名
for name, transformer, cols in preprocessor.transformers_:
    if name == 'pca_conn':
        # PCA输出100个特征
        transformed_feature_names += [f'pca_conn_{i}' for i in range(100)]
    elif name == 'pca_grad':
        transformed_feature_names += [f'pca_grad_{i}' for i in range(100)]
    elif name == 'pca_centroid_disp_pca':
        transformed_feature_names += [f'pca_centroid_disp_{i}' for i in range(10)]
    elif name == 'pca_cortex_disp_pca':
        transformed_feature_names += [f'pca_cortex_disp_{i}' for i in range(100)]
    elif name == 'encode_dataset':
        # 获取OneHotEncoder的类别名
        cat_names = transformer.named_steps['cat_encoder'].get_feature_names_out(cols)
        transformed_feature_names += list(cat_names)
    elif name == 'encode_sex':
        cat_names = transformer.named_steps['cat_encoder'].get_feature_names_out(cols)
        transformed_feature_names += list(cat_names)
    elif name == 'whiten_fd':
        transformed_feature_names += ['mean_fd_scaled']
    elif name == 'whiten_age':
        transformed_feature_names += ['age_scaled']

# 现在可以把重要性结果和特征名对应起来
importance_df = pd.DataFrame({
    'feature': transformed_feature_names,
    'mean_importance': perm_acc_transformed.importances_mean,
    'std_importance': perm_acc_transformed.importances_std
}).sort_values(by='mean_importance', ascending=False)

为什么不会数据泄露？ #

• 预处理模块是用训练集拟合的，转换测试集时没有重新计算任何统计量（比如均值、方差、PCA成分），完全遵循训练/测试数据隔离原则。
• 排列重要性本身是在测试集上评估特征打乱后的模型性能变化，这一步也不会用到训练集之外的信息。

内容的提问来源于stack exchange，提问作者victoris_93