如何在Sklearn多估计器Pipeline中为各估计器指定不同特征集

一、为StackingClassifier各估计器分配不同特征集 #

Sklearn的StackingClassifier本身不直接支持多特征集输入，但可以给每个基学习器/元学习器套一层特征选择子Pipeline，让每个模型自动从全局特征矩阵中提取自己需要的特征集。

实现步骤 #

1. 明确各特征集的列名或索引（假设特征矩阵为DataFrame格式）
2. 为每个模型构建包含特征选择逻辑的子Pipeline
3. 将这些子Pipeline传入StackingClassifier的estimators参数

代码示例 #

from sklearn.ensemble import StackingClassifier, RandomForestClassifier
from sklearn.neighbors import KNeighborsClassifier
from sklearn.svm import SVC
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
import pandas as pd

# 自定义特征集（根据实际业务调整列名）
featureset1 = ["feat1", "feat2", "feat3"]
featureset2 = ["feat4", "feat5"]
featureset3 = ["feat6", "feat7", "feat8"]

# 为随机森林构建带特征选择的子Pipeline
rf_pipeline = Pipeline([
    ("select_feat1", ColumnTransformer([("select", "passthrough", featureset1)])),
    ("rf", RandomForestClassifier())
])

# 为KNN构建带特征选择的子Pipeline
knn_pipeline = Pipeline([
    ("select_feat2", ColumnTransformer([("select", "passthrough", featureset2)])),
    ("knn", KNeighborsClassifier())
])

# 为SVC构建带特征选择的子Pipeline（需开启probability=True以输出概率）
svc_pipeline = Pipeline([
    ("select_feat3", ColumnTransformer([("select", "passthrough", featureset3)])),
    ("svc", SVC(probability=True))
])

# 构建二级StackingClassifier
stackingclassifier2 = StackingClassifier(
    estimators=[("knn", knn_pipeline)],
    final_estimator=svc_pipeline,
    passthrough=False
)

# 构建一级StackingClassifier
stackingclassifier1 = StackingClassifier(
    estimators=[("rf", rf_pipeline)],
    final_estimator=stackingclassifier2,
    passthrough=False
)

# 训练时直接传入完整特征矩阵与标签
# stackingclassifier1.fit(X_train, y_train)

说明：ColumnTransformer的passthrough模式会精准保留指定特征列，过滤其他无关列，确保每个子Pipeline仅处理目标特征集。

二、概率传递式Pipeline（无堆叠的顺序方案） #

如果实际需求是前一个模型输出概率，传递给下一个模型（可结合指定特征集），最终输出概率，可以用FunctionTransformer自定义概率生成逻辑，结合Pipeline和ColumnTransformer实现串联式流程。

方案1：前模型概率作为后模型唯一输入 #

适用于后模型仅依赖前序模型的概率输出：

from sklearn.preprocessing import FunctionTransformer
from sklearn.pipeline import Pipeline

# 定义提取模型概率的工具函数
def extract_proba(model, X):
    return model.predict_proba(X)

# 第一步：用featureset1训练随机森林
rf_step = Pipeline([
    ("select_feat1", ColumnTransformer([("select", "passthrough", featureset1)])),
    ("rf", RandomForestClassifier())
])

# 自定义转换器：提取随机森林的概率输出
rf_proba_transformer = FunctionTransformer(
    extract_proba,
    kw_args={"model": rf_step.named_steps["rf"]}
)

# 第二步：用随机森林概率训练KNN
knn_step = Pipeline([
    ("rf_proba", rf_proba_transformer),
    ("knn", KNeighborsClassifier())
])

# 第三步：用KNN概率+featureset3训练SVC，输出最终概率
final_concat = ColumnTransformer([
    ("knn_proba", FunctionTransformer(lambda X: knn_step.predict_proba(X)), slice(None)),
    ("select_feat3", "passthrough", featureset3)
])

final_pipeline = Pipeline([
    ("rf", rf_step),
    ("knn", knn_step),
    ("concat_features", final_concat),
    ("svc", SVC(probability=True))
])

# 分步拟合（后步骤依赖前序模型的训练结果）
rf_step.fit(X_train, y_train)
rf_proba_train = rf_step.predict_proba(X_train)
knn_step.fit(rf_proba_train, y_train)
knn_proba_train = knn_step.predict_proba(rf_proba_train)
concat_train = pd.concat([pd.DataFrame(knn_proba_train), X_train[featureset3]], axis=1)
final_pipeline.named_steps["svc"].fit(concat_train, y_train)

# 自定义预测函数
def predict_final_proba(X):
    rf_proba = rf_step.predict_proba(X)
    knn_proba = knn_step.predict_proba(rf_proba)
    concat_X = pd.concat([pd.DataFrame(knn_proba), X[featureset3]], axis=1)
    return final_pipeline.named_steps["svc"].predict_proba(concat_X)

方案2：拼接原始特征与前模型概率（更灵活） #

如果后模型需要同时使用原始特征集和前序模型的概率，可改用FeatureUnion实现特征拼接：

from sklearn.pipeline import FeatureUnion

# 第一步：随机森林概率 + featureset2 作为KNN的输入
knn_input = FeatureUnion([
    ("rf_proba", FunctionTransformer(lambda X: rf_step.predict_proba(X))),
    ("select_feat2", ColumnTransformer([("select", "passthrough", featureset2)]))
])

knn_pipeline = Pipeline([
    ("input", knn_input),
    ("knn", KNeighborsClassifier())
])

# 第二步：KNN概率 + featureset3 作为SVC的输入
svc_input = FeatureUnion([
    ("knn_proba", FunctionTransformer(lambda X: knn_pipeline.predict_proba(X))),
    ("select_feat3", ColumnTransformer([("select", "passthrough", featureset3)]))
])

final_pipeline = Pipeline([
    ("rf", rf_step),
    ("knn", knn_pipeline),
    ("svc_input", svc_input),
    ("svc", SVC(probability=True))
])

# 分步拟合
rf_step.fit(X_train, y_train)
knn_pipeline.fit(X_train, y_train)
final_pipeline.fit(X_train, y_train)

# 直接预测概率
# final_pipeline.predict_proba(X_test)

说明：这种分步拟合的方式完全贴合“概率逐模型传递”的需求，且每个模型可独立指定专属特征集，灵活性高。

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