机器学习中的集成方法（4）-- Stacking（堆叠法）

一、概念理解

Stacking 就是当用初始训练数据学习出若干个基学习器后，将这几个学习器的预测结果作为新的训练集，来学习一个新的学习器。Stacking 的基础层通常包括不同的学习算法，因此stacking ensemble往往是异构的。

二、执行步骤

假设有1000个样本，70%的样本作为训练集，30%的样本作为测试集。
STEP1：在训练集上采用算法A、B、C等训练出一系列基学习器。
STEP2：用这些基学习器的输出结果组成新的训练集，在其上训练一个元学习器（meta-classifier，通常为线性模型LR），用于组织利用基学习器的答案，也就是将基层模型的答案作为输入，让元学习器学习组织给基层模型的答案分配权重

三、使用mlxtend库实现Stacking方法

最基本的使用方法，即使用前面分类器产生的特征输出或者概率输出作为meta-classifier的输入数据

#-*- coding:utf-8 -*-
'''
Stacking方法
'''
from sklearn.datasets import load_iris
from sklearn.preprocessing import StandardScaler
from sklearn import model_selection
from sklearn.linear_model import LogisticRegression
from sklearn.neighbors import KNeighborsClassifier
from sklearn.naive_bayes import GaussianNB 
from sklearn.ensemble import RandomForestClassifier
from mlxtend.classifier import StackingClassifier
from sklearn.metrics import accuracy_score
from sklearn.metrics import confusion_matrix
from sklearn.metrics import classification_report
import warnings; warnings.filterwarnings(action='ignore')

#========================================================
#  载入iris数据集
#========================================================

iris = load_iris()
X = iris.data[:,:5]
y = iris.target

print('feature=',X)
print('target=',y)

#========================================================
#  实现Stacking集成
#========================================================

def StackingMethod(X, y):
    '''
    Stacking方法实现分类
    INPUT -> 特征, 分类标签
    '''
    scaler = StandardScaler() # 标准化转换
    scaler.fit(X)  # 训练标准化对象
    traffic_feature= scaler.transform(X)   # 转换数据集
    feature_train, feature_test, target_train, target_test = model_selection.train_test_split(X, y, test_size=0.3, random_state=0)

    clf1 = LogisticRegression(random_state=1)
    clf2 = RandomForestClassifier(random_state=1)
    clf3 = GaussianNB()

    sclf = StackingClassifier(classifiers=[clf1, clf2, clf3],
                              # use_probas=True, 类别概率值作为meta-classfier的输入
                              # average_probas=False,  是否对每一个类别产生的概率值做平均
                              meta_classifier=LogisticRegression())

    sclf.fit(feature_train, target_train)

    # 模型测试
    predict_results = sclf.predict(feature_test)
    print(accuracy_score(predict_results, target_test))
    conf_mat = confusion_matrix(target_test, predict_results)
    print(conf_mat)
    print(classification_report(target_test, predict_results))

    # 5折交叉验证
    for clf, label in zip([clf1, clf2, clf3, sclf], ['Logistic Regression', 'Random Forest', 'naive Bayes', 'StackingModel']):
        scores = model_selection.cross_val_score(clf, X, y, cv=5, scoring='accuracy')
        print("Accuracy: %0.2f (+/- %0.2f) [%s]" % (scores.mean(), scores.std(), label))

    return sclf

#========================================================
#  主程序
#========================================================

if __name__ == '__main__':

    model = StackingMethod(X, y)

另外一种方法是对训练集中的特征维度进行操作的，这次不是给每一个基分类器全部的特征，而是给不同的基分类器分配不同的特征，即比如基分类器1训练前半部分特征，基分类器2训练后半部分特征（可以通过sklearn 的pipelines 实现）。最终通过StackingClassifier组合起来。

#-*- coding:utf-8 -*-
'''
Stacking方法
'''
from sklearn.datasets import load_iris
from sklearn.preprocessing import StandardScaler
from sklearn import model_selection
from sklearn.linear_model import LogisticRegression
from mlxtend.feature_selection import ColumnSelector
from mlxtend.classifier import StackingClassifier
from sklearn.pipeline import make_pipeline
from sklearn.metrics import accuracy_score
from sklearn.metrics import confusion_matrix
from sklearn.metrics import classification_report
import warnings; warnings.filterwarnings(action='ignore')

#========================================================
#  载入iris数据集
#========================================================

iris = load_iris()
X = iris.data[:,:5]
y = iris.target

print('feature=',X)
print('target=',y)

#========================================================
#  实现Stacking集成
#========================================================

def StackingMethod(X, y):
    '''
    Stacking方法实现分类
    INPUT -> 特征, 分类标签
    '''
    scaler = StandardScaler() # 标准化转换
    scaler.fit(X)  # 训练标准化对象
    traffic_feature= scaler.transform(X)   # 转换数据集
    feature_train, feature_test, target_train, target_test = model_selection.train_test_split(X, y, test_size=0.3, random_state=0)

    pipe1 = make_pipeline(ColumnSelector(cols=(0, 1)),  
                          LogisticRegression())
    pipe2 = make_pipeline(ColumnSelector(cols=(2, 3, 4)),  
                          LogisticRegression())

    sclf = StackingClassifier(classifiers=[pipe1, pipe2],
                              meta_classifier=LogisticRegression())

    sclf.fit(feature_train, target_train)

    # 模型测试
    predict_results = sclf.predict(feature_test)
    print(accuracy_score(predict_results, target_test))
    conf_mat = confusion_matrix(target_test, predict_results)
    print(conf_mat)
    print(classification_report(target_test, predict_results))

    return sclf

#========================================================
#  主程序
#========================================================

if __name__ == '__main__':

    model = StackingMethod(X, y)

四、使用源码实现Stacking方法

# coding=utf-8
import imp
import numpy as np
from sklearn.datasets import load_iris

iris = load_iris()
x = iris.data
y = iris.target

import pandas as pd

x = pd.DataFrame(x, columns=['s_len','s_wid','p_len','p_wid'])

from sklearn.pipeline import Pipeline
from sklearn.base import BaseEstimator, ClassifierMixin, TransformerMixin
from sklearn.svm import SVC

class StackingAveragedModels(BaseEstimator, ClassifierMixin, TransformerMixin):
    def __init__(self, base_models, meta_model):
        self.base_models = base_models
        self.meta_model = meta_model

    def fit(self, X, y):

        # 训练第一级学习器
        layer1_predictions = np.zeros((X.shape[0], len(self.base_models)))
        for i, model in enumerate(self.base_models):
            model.fit(X, y)
            y_pred = model.predict(X)
            layer1_predictions[:, i] = y_pred

        # 训练第二级学习器
        self.meta_model.fit(layer1_predictions, y)
        return self

    def predict(self, X):
        temp = np.column_stack([model.predict(X) for model in self.base_models])
        return self.meta_model.predict(temp)

clf = StackingAveragedModels(base_models=[SVC(gamma='scale'),
                                          SVC(gamma='scale')],
                             meta_model=SVC(gamma='scale'))

pipe = Pipeline(steps=[
              ('Clf', clf)
              ])

from sklearn.model_selection import train_test_split

x_train, x_test, y_train, y_test = train_test_split(x, y, test_size= 0.5, random_state= 666)
pipe.fit(x_train, y_train)

print('Test accuracy is %.3f' % pipe.score(x_test, y_test))

拓展阅读1
拓展阅读2