机器学习 - [集成学习]
Bagging算法的编程实现
【导读】本文主要介绍如何实现Bagging算法，作为一种集成算法我们假设其基分类器已经写好。本文代码中，直接使用sklearn中的决策树分类器等基分类器。
1. 回顾知识点
从训练集中均匀随机有放回地采样不同的子集且训练不同的基础模型
给定集合 D
通过从D中均匀随机的有放回采样m个样本构建 Di
（drawn with replacement, 取出放回， 有放回采样)
希望 Di 会 遗漏掉 D 中的某些样本
步骤:
- 每轮随机抽出一个训练集（漏掉一些样本）
- 通过这些样本训练一个子分类器（只用同一种弱学习器为基分类器）
- 一共T（参数）轮
- 等权重投票，票多的分类标签获胜
for i in [1, 2, …, T]: 从S中拔靴采样产生 Dt 在 Dt 上训练一个分类器H t 分类一个新的样本x∈X 时，通过对 H t 多数投票(等权重)
2. 代码实现
from sklearn.tree import DecisionTreeClassifier # 用作 基分类器的 决策树分类器 from sklearn.svm import LinearSVC # 用作 基分类器的 线性支持向量机分类器 from sklearn.calibration import CalibratedClassifierCV # 将base_estimator拟合到交叉验证生成器的训练集上，并使用测试集进行校准。 # 然后将每个折叠的概率取平均值进行预测。 # 概率校准能 # 将非概率分类模型的输出转化为概率、 # 对概率分类模型的结果进行进一步修正 import copy import numpy as np from collections import Counter import scipy import sklearn from sklearn.feature_extraction.text import TfidfVectorizer word_model = TfidfVectorizer(stop_words='english') train_X = word_model.fit_transform(train_df['reviewText']) test_X = word_model.transform(test_df['reviewText']) train_X = scipy.sparse.hstack([train_X, train_df['overall'].values.reshape((-1, 1)) / 5]) test_X = scipy.sparse.hstack([test_X, test_df['overall'].values.reshape((-1, 1)) / 5]) X_train = train_X.tocsr() Y_train = train_df['label'] X_test = test_X.tocsr() def majority(nparray): """ 选取一个数组中出现次数最多的元素及其索引并返回 """ return Counter(nparray).most_common()[0] class BaggingClassifier(object): """ a Bagging Classifier """ def __init__(self, base_estimator="DCT", n_estimators=10, dct_parameters=None): """ Parameters ---------- base_estimator : str, default="DCT" 用于在一个随机抽取的子训练集上分类的分类器类型。 如果为"DCT"，则程序将实例化一个`sklearn.tree.DecisionTreeClassifier`； 如果为"SVM"，则程序将实例化一个`sklearn.svm.LinearSVC` 默认为`DecisionTreeClassifier`（决策树分类器）. n_estimators : int, default=10 继承的基分类器的个数。 dct_parameters: 当基分类器为决策树时的参数. """ self.base_estimator = base_estimator self.n_estimators = n_estimators if dct_parameters == None: self.dct_parameters={"max_depth":10,"min_samples_split":2} else: self.dct_parameters=dct_parameters self.instance_base_estimator() self.sub_estimators = [] infos = "------\nBaggingClassifier Parameters:\n 基分类器："+str(base_estimator)+";\n 基分类器数量："+str(n_estimators)+".\n------" print(infos) def instance_base_estimator(self): """ 实例化基分类器 """ if self.base_estimator=="DCT": clf = DecisionTreeClassifier( max_depth=self.dct_parameters["max_depth"], min_samples_split=self.dct_parameters["min_samples_split"], class_weight='balanced') elif self.base_estimator=="SVM": clf = LinearSVC() self.classifier = CalibratedClassifierCV(clf, cv=2, method='sigmoid') def sub_dataset(self,X_train,Y_train): """ 从当前dataset中随机取一部分返回 用作某一个子分类器（单独的基分类器）的训练集数据 """ data_size = len(Y_train) index=np.random.choice(np.arange(data_size), data_size, replace=True) return X_train[index],Y_train[index] def fit(self,X_train,Y_train): """ 训练 T=self.n_estimators 个基分类器 """ for t in range(self.n_estimators): print("正在训练第 i =",t+1,"个基分类器...") sub_train_X,sub_train_Y = self.sub_dataset(X_train,Y_train) # 抽取一个随机子数据集 classifier = copy.copy(self.classifier) classifier.fit(sub_train_X, sub_train_Y) # 随机子数据集上训练基分类器 self.sub_estimators.append(classifier) # 保存基分类器 def predict(self, X_test): """ 为X_test中的每条评论数据预测分类类别 """ return [majority(one)[0] for one in np.array([clf.predict(X_test) for clf in self.sub_estimators]).T] def predict_proba(self, X_test): """ 为X_test中的每条数据预测为高质量的概率 基分类器的predict_proba()方法，第i行第j列上的数值是模型预测第i个预测样本的标签为j的概率 """ proba = np.zeros(X_test.shape[0]) # 以评论的数目生成一个全0一维向量，用于累加各基分类器对各评论预测为高质量概率和 for clf in self.sub_estimators: # 对于每个基分类器 proba = proba + clf.predict_proba(X_test)[:, 1] # 逐个累加所有基分类器在同一条评论预测为高质量的概率和 average_proba = proba / self.n_estimators # 按个数算平均概率，作为整体预测为高质量的概率 return average_proba
