[人工智能]【学习笔记】集成学习（六）：Boosting

Datawhale组队学习第27期：集成学习
本次学习的指导老师萌弟的教学视频
本贴为学习记录帖，有任何问题欢迎随时交流~
部分内容可能还不完整，后期随着知识积累逐步完善。
开始时间：2021年7月23日
最新更新：2021年7月25日（Task6 Boosting）

一、Boosting概念

1. 思想

• 对同一组数据集反复学习
• 通过减少偏差的形式来降低测试误差（与Bagging的本质区别）

2. 常见模型

• AdaBoosting
• GradientBoosting
• Xgboost
• LightGBM
• Catboost

3. 理论来源

• 强可学习与弱可学习
  • 弱学习：识别错误率小于1/2，准确率略高于随机猜测
  • 强学习：识别准确率很高，并能在多项式时间内完成的学习算法
• PAC学习框架下，强可学习和弱可学习是等价的。
  • 意味着如果发现了弱可学习算法，可以通过提升方法使得学习增强
  • 大多数boosting都是改变训练数据集的概率分布（训练数据不同样本的权值），针对不同概率分布的数据调用弱分类算法学习一系列的弱分类器。
• 需要回答两个关键问题：
  • 每一轮学习如何改变数据的概率分布
  • 各个弱分类器如何组合

二、Adaboost

1. 基本思想

• 提高前一轮分类器错误分类样本的权重，降低正确分类样本的权重
• 各个弱分类器采用加权表决的方式组合
• Adaboost实际上是增加了模型的复杂度，减少了偏差

2. 基本步骤

• 训练器的权重初始化

• 训练多个分类器模型，计算训练集的分类误差率（分类错误加起来）

• 计算分类器在总模型的重要性

• 更新训练数据集的权重分布

  • 注意：$Z_m$是规范化因子，使得$D_{m+1}$称为概率分布

• 构建基本分类器的线性组合，得到最终分类器

3. 代码实现

• 选用数据集diabetes，二分类问题
• 定义弱分类器，单层决策树

# 定义单层决策树算法（弱分类器）
class DecisionStump:
    def __init__(self, x, y):
        self.x = np.array(x)
        self.y = np.array(y)
        self.n = self.x.shape[0]  # 样本量n
        self.w = None  # 权重初始化
        self.threshold_value = 0  # 阈值初始化
        self.threshold_pos = 0  # 参数下标初始化
        self.threshold_res = 0  # 取值初始化

    def train(self, w, steps=1000):
        """
        用于返回所有参数中阈值最小的那一个
        w 是长度为n的向量，表示n个样本的权值
        threshold_value 为阈值
        threshold_pos 为第几个参数（参数下标）
        threshold_tag 取值空间为{-1, +1}，大于阈值则分为某个threshold_tag，小于则相反（这里的如果相等？）
        """
        min_ = float("inf")  # 最小值初始化为无穷大inf
        # 注意区分与__init__函数self的属性区别
        threshold_value = 0
        threshold_pos = 0
        threshold_tag = 0
        self.w = np.array(w)  # 更新权重值，并将实际w转成ndarray类型

        # 分别以m个属性中的任意一个建立单层决策树
        for i in range(self.n):
            # value表示阈值，errcnt 表示错误的数量
            # 找到当前参数下最小的阈值（当前属性下决策树错误率最低）
            value, errcnt = self.findmin(i, 1, steps)  # 找最小值
            if errcnt < min_:
                min_ = errcnt
                threshold_value = value
                threshold_pos = i
                threshold_tag = 1

        # 对self属性值进行更新
        self.threshold_value = threshold_value
        self.threshold_pos = threshold_pos
        self.threshold_res = threshold_tag
        print(self.threshold_value, self.threshold_pos, self.threshold_res)

        return min_

    # 找出第i个参数的最小阈值，tag为1或-1
    def findmin(self, i, tag, steps):
        t = 0

        # 测试当前阈值情况下的预测能力（这里并没有作用，可以视为创建空间或初始化）
        tmp = self.predintrain(self.x, i, t, tag).transpose()
        errcnt = np.sum((tmp != self.y) * self.w)  # 计算当前阈值下错误的个数

        # 定义上下界（表示从第i个变量中最小值和最大值直接寻找阈值）
        bottom = np.min(self.x[i, :])  # 该属性下的最小值，即下界
        up = np.max(self.x[i, :])  # 该属性下的最大值，即上界

        # 初始化
        minerr = float("inf")  # 初始化错误的数量minerr为无穷大
        value = 0  # 初始化阈值为0
        st = (up - bottom) / steps  # 表示间隔

        # 寻找最小阈值value和分类误差率minerr
        for t in np.arange(bottom, up, st):
            tmp = self.predintrain(self.x, i, t, tag).transpose()  # 转置成(1,测试集样本量)
            errcnt = np.sum((tmp != self.y) * self.w)  # 返回分类误差率e_m
            if errcnt < minerr:
                minerr = errcnt
                value = t
        return value, minerr

    # 返回训练时按照阈值为t时预测的结果
    def predintrain(self, test_set, i, t, tag):
        # 测试集是(变量数, 样本量)的结构
        # 将测试集转成矩阵行数与样本量一致的数组
        test_set = np.array(test_set).reshape(self.n, -1)

        # 生成矩阵行数与测试集列数一致的且值为1的数组（这里测试集列数是测试集样本量）
        pre_y = np.ones((np.array(test_set).shape[1], 1))

        # 表示选择第i行（第i个变量）的数据，如果小于阈值t，则修改对应值为-1
        # 将对每个预测值进行判断，小于阈值的取-1，返回pre_y是(测试集样本量, 1)的结构
        pre_y[test_set[i, :] * tag < t * tag] = -1
        return pre_y

    # 弱分类器进行预测
    def pred(self, test_x):
        # 测试集转成行数与样本量一致的二维数组结构
        test_x = np.array(test_x).reshape(self.n, -1)
        pre_y = np.ones((np.array(test_x).shape[1], 1))  # 生成行数与样本量一致且值为1的二维数组
        pre_y[test_x[self.threshold_pos, :] * self.threshold_res <
              self.threshold_value * self.threshold_res] = -1
        return pre_y

• 定义Adaboost

class AdaBoost:
    def __init__(self, x, y, weaker=DecisionStump):
        # 本函数用于初始化变量及参数
        # 初始化数据集(需要将训练集转成ndarray类型，这里本身已经是)
        self.x = np.array(x)
        self.y = np.array(y).flatten("F")  # flatten降至一维，这里没有用，F表示按列展开，原代码的有问题

        # 初始化分类器
        self.weaker = weaker

        # 初始化得分，计算组合分类器的预测得分总和
        self.sums = np.zeros(self.y.shape)  # 初始化与y.shape一样维度的零矩阵

        # 初始化权值w
        # 这里np.ones是根据样本量生成一个结构为(n,1)、值为1的二维数组，并降维到(n,)的一维数组
        # 生成的值皆为1的一维数组后，对每一个值除以样本量n（利用numpy的广播机制）
        # 最后得到的w的每个值都为1/n，即权重初始化是均匀分布的
        # self.w = np.ones((self.x.shape[1], 1)).flatten("F") / self.x.shape[1]
        self.w = np.ones(self.x.shape[1]) / self.x.shape[1]  # 效果同上，这里是简化代码

        # 初始化弱分类器
        self.q = 0  # 初始化弱分类器的实际个数
        self.g = {}  # 初始化所弱分类器的字典（原代码train函数中的放到此处）
        self.alpha = {}  # 初始化每个弱分类器的参数

    def train(self, m=5):
        # 给每个弱分类器的字典赋予默认key和默认value
        # g = {i: None, ...}  , alpha = {i: None, ...}
        for i in range(m):
            self.g.setdefault(i)
            self.alpha.setdefault(i)

        for i in range(m):
            self.g[i] = self.weaker(self.x, self.y)
            # 根据当前权值进行第i个弱分类器训练，g[i]是weaker，因此train是weaker的方法
            # 默认权重是均匀分布的，即权值都是1/n
            e = self.g[i].train(self.w)

            # 计算第i个分类器的系数（权重）
            self.alpha[i] = 1.0 / 2 * 2 * np.log((1 - e) / e)
            res = self.g[i].pred(self.x)  # res返回的是第i个分类器的输出（重点判断误分类）

            # 计算当前次数训练的精确度
            print("weak classfier acc", accuracy_score(self.y, res))
            print("======================================================")

            # 规范化因子
            z = self.w * np.exp(-self.alpha[i] * self.y * res.transpose())  # 不一样
            self.w = (z / z.sum()).flatten("F")  # 更新权值
            self.q = i

            # errorcnt返回分错的点数，当为0时表示没有分错
            if self.errorcnt(i) == 0:
                print("%d个弱分类器可以将错误率降到0" % (i + 1))
                break

    # 返回误分类的点
    def errorcnt(self, t):
        # 计算组合分类器的得分总和
        self.sums = self.sums + self.g[t].pred(self.x).flatten('F') * self.alpha[t]
        pre_y = np.zeros(np.array(self.sums).shape)
        pre_y[self.sums >= 0] = 1
        pre_y[self.sums < 0] = -1
        t = (pre_y != self.y).sum()
        return t

    # 模型预测（最终的分类器）
    def pred(self, test_x):
        test_x = np.array(test_x)
        sums = np.zeros(test_x.shape[1])
        for i in range(self.q + 1):
            sums = sums + self.g[i].pred(test_x).flatten("F") * self.alpha[i]
        pre_y = np.zeros(np.array(sums).shape)
        pre_y[sums >= 0] = 1
        pre_y[sums < 0] = -1
        return pre_y

• 主程序调用

 # 加载训练集和测试集
 data_set = pd.read_csv('diabetes.csv')
 x = data_set.values[:, :8]
 y = data_set.values[:, 8]

 # 划分数据集
 # test_size测试集样本比例; random_state随机种子
 x_train, x_test, y_train, y_test = train_test_split(x, y, test_size=0.2,
                                                     random_state=0)

 # 数组变换，方便矩阵运算；简化算法，修改y的取值空间为{-1,+1}
 x_train = x_train.transpose()
 y_train[y_train == 1] = 1
 y_train[y_train == 0] = -1
 x_test = x_test.transpose()
 y_test[y_test == 1] = 1
 y_test[y_test == 0] = -1

 # 模型训练
 print(x_train.shape, x_test.shape)
 ada = AdaBoost(x_train, y_train)  # 放进训练集的x和y，默认为弱分类器weak
 ada.train(10)  # 设置弱分类器的数量为10

 # 模型预测
 y_pre = ada.pred(x_test)
 print("total test", len(y_pre))
 print('true pred', len(y_pre[y_pre == y_test]))
 print('acc', accuracy_score(y_test, y_pre))

三、参考资料

1. https://github.com/datawhalechina/ensemble-learning
2. https://www.bilibili.com/video/BV1Mb4y1o7ck?t=470