基于预定义类别特征子集的分类器集成改造及优化问询

以下最小可行示例（MWE）使用SelectKBest算法和RandomForest分类器构建集成方法：通过get_ensemble函数生成10个基于不同K值特征选择的基模型，再通过VotingClassifier实现硬投票集成，代码及使用示例如下：

# required import
import numpy as np
import pandas as pd
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.feature_selection import SelectKBest, f_classif
from sklearn.ensemble import RandomForestClassifier, VotingClassifier
from sklearn.pipeline import Pipeline

# ensemble created from features selected
def get_ensemble(n_features):
  # define base models
  models = []
  # enumerate the features in the training dataset
  for i in range(1, n_features + 1):
    # feature selection transform
    fs = SelectKBest(score_func=f_classif, k=i)
    # create the model
    model = RandomForestClassifier(n_estimators=50)
    # create the pipeline
    pipe = Pipeline([('fs', fs), ('m', model)])
    # list of tuple of models for voting
    models.append((str(i), pipe))

  # define the voting ensemble
  ensemble_clf = VotingClassifier(estimators=models, voting='hard')

  return ensemble_clf

使用示例：

# generate data for a 3-class classification
X, y = make_classification(n_samples=1000, n_features=10, n_classes=3,
                             n_informative=3)

X = pd.DataFrame(X, columns=list('ABCDEFGHIJ'))

X_train, X_test, y_train, y_test = train_test_split(X, y,
         test_size=0.3, random_state=42)

# get the ensemble model
ensemble_clssifier = get_ensemble(X_train.shape[1])

ensemble_clssifier.fit(X_train, y_train)

需替换get_ensemble中的SelectKBest特征选择逻辑，改为针对每个类别使用预定义最优特征子集构建3个基模型：

要求训练和预测时，各基模型自动选用对应特征子集，最终通过多数投票形成集成分类器。

自行编写了CustomEnsemble类，基于基模型的置信度训练最终集成模型，代码如下：

from sklearn.base import clone

class CustomEnsemble:
    def __init__(self, base_model, best_feature_subsets):
        self.base_models = {class_label: clone(base_model) for class_label in best_feature_subsets}
        self.best_feature_subsets = best_feature_subsets
        self.final_model = base_model

    def train_base_models(self, X_train, y_train):
        for class_label, features in self.best_feature_subsets.items():
            model = self.base_models[class_label]
            model.fit(X_train[features], (y_train == class_label))
        
        return self
    
    def train_final_model(self, X_train, y_train):
        """
        Probably better to implement the train methods (base models & ensemble)
        in one method suc as the train_base_models  altogether.
        """
        predictions = pd.DataFrame()

        for class_label, model in self.base_models.items():
            predictions[class_label] = model.predict_proba(X_train[self.best_feature_subsets[class_label]])[:, 1]

        self.final_model.fit(predictions, y_train)


    def predict_base_models(self, X_test):
        predictions = pd.DataFrame()

        for class_label, model in self.base_models.items():
            predictions[class_label] = model.predict_proba(X_test[self.best_feature_subsets[class_label]])[:, 1]

        return predictions

    def predict(self, X_test):
        base_model_predictions = self.predict_base_models(X_test)
        return self.final_model.predict(base_model_predictions)

    def predict_proba_base_models(self, X_test):
        predictions = pd.DataFrame()

        for class_label, model in self.base_models.items():
            predictions[class_label] = model.predict_proba(X_test[self.best_feature_subsets[class_label]])[:, 1]

        return predictions

    def predict_proba(self, X_test):
        base_model_predictions = self.predict_proba_base_models(X_test)
        return self.final_model.predict_proba(base_model_predictions)

使用示例：

optimal_features = {
    0: ['A', 'B', 'C'],
    1: ['D', 'E', 'F', 'G'],
    2: ['G', 'H', 'I', 'J']
}
classifier = RandomForestClassifier()
ensemble   = CustomEnsemble(classifier, optimal_features)
# first, train base models
ensemble.train_base_models(X_train, y_train)
# then, train the ensemble
ensemble.train_final_model(X_train, y_train)
yhat = ensemble.predict(X_test)
yhat_proba = ensemble.predict_proba(X_test)

疑问解答 #

1. 当前最终模型的训练逻辑是否正确？是否确为基于基模型输出而非原始输入特征？ #

你的最终模型训练逻辑是正确的，且完全基于基模型输出而非原始特征。

具体细节：

• 每个基模型是针对对应类别训练的二分类器，任务是判断样本是否属于该类别
• train_final_model中，先让每个基模型对训练集的对应特征子集输出属于该类别的概率（即predict_proba返回结果的第二列），再将这些概率值作为新的特征矩阵喂给最终模型训练
• 整个流程未使用原始输入特征，完全依赖基模型的输出结果，属于堆叠（Stacking）集成的典型实现

注意：你的实现是堆叠集成，和需求中提到的**多数投票（硬投票）**逻辑不同，如果要严格匹配需求，可参考后面的补充实现。

2. 是否应将train_base_models()和train_final_model()合并为单一训练方法？ #

非常建议合并成单一方法，原因如下：

• 两个步骤是强关联的连续流程，分开调用容易出现遗漏（比如忘记调用train_final_model导致最终模型未训练）
• 合并后可简化操作，用户只需调用一次方法就能完成整个集成模型的训练
• 符合scikit-learn的API设计习惯（统一用fit方法完成训练），降低使用门槛

合并后的实现示例：

def fit(self, X_train, y_train):
    # 训练基模型
    for class_label, features in self.best_feature_subsets.items():
        model = self.base_models[class_label]
        model.fit(X_train[features], (y_train == class_label))
    
    # 生成基模型的预测结果作为元特征
    predictions = pd.DataFrame()
    for class_label, model in self.base_models.items():
        predictions[class_label] = model.predict_proba(X_train[self.best_feature_subsets[class_label]])[:, 1]
    
    # 训练最终模型
    self.final_model.fit(predictions, y_train)
    return self

使用时只需调用ensemble.fit(X_train, y_train)即可，简洁且不易出错。

补充：匹配需求的硬投票实现 #

如果要严格实现需求中的多数投票逻辑，可以用VotingClassifier配合自定义特征选择的Pipeline，每个基模型是针对对应特征子集训练的多分类器，示例如下：

def get_custom_voting_ensemble(best_feature_subsets, base_model):
    models = []
    for class_label, features in best_feature_subsets.items():
        # 自定义特征选择：只保留指定特征
        def select_features(X):
            return X[features]
        # 用FunctionTransformer包装自定义特征选择逻辑
        from sklearn.preprocessing import FunctionTransformer
        fs = FunctionTransformer(select_features)
        # 构建Pipeline
        pipe = Pipeline([('fs', fs), ('m', clone(base_model))])
        models.append((f"class_{class_label}", pipe))
    
    # 硬投票集成
    ensemble_clf = VotingClassifier(estimators=models, voting='hard')
    return ensemble_clf

# 使用示例
optimal_features = {
    0: ['A', 'B', 'C'],
    1: ['D', 'E', 'F', 'G'],
    2: ['G', 'H', 'I', 'J']
}
base_clf = RandomForestClassifier(n_estimators=50)
voting_ensemble = get_custom_voting_ensemble(optimal_features, base_clf)
voting_ensemble.fit(X_train, y_train)
yhat = voting_ensemble.predict(X_test)

该实现完全匹配需求：每个基模型使用对应特征子集，最终通过多数投票得到分类结果。

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