一、概念理解
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))
网友评论