numpy实现adaboost算法复现

# -*- coding:utf-8 -*- # /usr/bin/python ''' ------------------------------------------------------------------------- @File Name : AdaBoost.py @Description : @Run Script : python AdaBoost.py @Envs : pip install @Change Activity: 1. 2021/12/22 下午2:19 : build ------------------------------------------------------------------------- @CodeStyle : standard, simple, readable, maintainable, and portable! @Author : Yan Errol 13075851954 @Email : 260187357@qq.com @Copyright : "Copyright 2021, Yan Errol" ------------------------------------------------------------------------- ''' import numpy as np import os class AdaBoost(): def __init__(self,nEstimators=50,learningRate=0.1): ''' 对象属性 :param nEstimators: 迭代代数 :param learningRate: 学习率 ''' self.nEstimators = nEstimators self.learningRate = learningRate def initArgs(self,datasets,labels): ''' 初始化参数 :param datasets: 数据集 :param labels: 标签 :return: ''' self.X = datasets self.Y = labels self.M,self.N = self.X.shape # 弱分类器 self.clfSets=[] # 初始化权值 self.weights = [1.0 / self.M]*self.M # 不同分类器的系数 self.alpha = [] def _G(self,features,labels,weights): m = len(features) print("m",m) error = 10000 bestV = 0 features_min = min(features) features_max = max(features) n_step = (features_max - features_min + self.learningRate) // self.learningRate print('n_step:{}'.format(n_step)) direct, compare_array = None, None for i in range(1, int(n_step)): v = features_min + self.learningRate * i if v not in features: # 误分类计算 compare_array_positive = np.array([1 if features[k] > v else -1 for k in range(m)]) print(compare_array_positive) weight_error_positive = sum([weights[k] for k in range(m) if compare_array_positive[k] != labels[k]]) print(weight_error_positive) compare_array_nagetive = np.array([-1 if features[k] > v else 1 for k in range(m)]) weight_error_nagetive = sum([weights[k] for k in range(m) if compare_array_nagetive[k] != labels[k]]) if weight_error_positive < weight_error_nagetive: weight_error = weight_error_positive _compare_array = compare_array_positive direct = 'positive' else: weight_error = weight_error_nagetive _compare_array = compare_array_nagetive direct = 'nagetive' print('v:{} error:{}'.format(v, weight_error)) if weight_error < error: error = weight_error compare_array = _compare_array bestV = v return bestV,direct,error,compare_array # 计算alpha def _alpha(self,error): ''' 更新alpha :param error: 误差 :return: 新的alpha ''' return 0.5 * np.log((1 - error) / error) def _Z(self,weights,a,clf): ''' 规范化因子 :param weights: 权值 :param a: 系数 :param clf: 类别 :return: ''' return sum([weights[i] * np.exp(-1 * a * self.Y[i] * clf[i]) for i in range(self.M)]) def _w(self,a,clf,Z): ''' 更新权值 :param a: 系数 :param clf: 类别 :param Z: :return: 规范因子 ''' self.weights = self.weights * np.exp(-1 * a * self.Y * clf) / Z # G(x)的线性组合 def _f(self, alpha, clf_sets): pass def G(self, x, v, direct): ''' 分类器 :param x: 输入特征 :param v: 阈值 :param direct: 正负倾向 :return: ''' if direct == 'positive': return 1 if x > v else -1 else: return -1 if x > v else 1 def fit(self,X,y): ''' 训练 :param X: x特征 :param y: y标签 :return: ''' self.initArgs(X,y) for epoch in range(self.nEstimators): best_clf_error, bestV, clf_result = 100000, None, None for j in range(self.N): features = self.X[:, j] # 分类阈值，分类误差，分类结果 v, direct, error, compare_array = self._G(features, self.Y, self.weights) if error < best_clf_error: best_clf_error = error bestV = v final_direct = direct clf_result = compare_array axis = j if best_clf_error == 0: break # 计算G(x)系数a a = self._alpha(best_clf_error) self.alpha.append(a) # 记录分类器 self.clfSets.append((axis, bestV, final_direct)) # 规范化因子 Z = self._Z(self.weights, a, clf_result) # 权值更新 self._w(a, clf_result, Z) def predict(self, feature): ''' 预测 :param feature: x特征 :return: ''' result = 0.0 for i in range(len(self.clfSets)): axis, clf_v, direct = self.clfSets[i] f_input = feature[axis] result += self.alpha[i] * self.G(f_input, clf_v, direct) # sign return 1 if result > 0 else -1 def score(self, X_test, y_test): ''' 分数计算 :param X_test: 测试数据集的x特征 :param y_test: 测试数据集的y特征 :return: ''' right_count = 0 for i in range(len(X_test)): feature = X_test[i] if self.predict(feature) == y_test[i]: right_count += 1 return right_count / len(X_test) X = np.arange(10).reshape(10,1) y = np.array([1, 1, 1, -1, -1, -1, 1, 1, 1, -1]) clf = AdaBoost(nEstimators=10,learningRate=0.9) clf.fit(X,y) testX = np.array([[0],[9],[2]]) testY = np.array([1,-1,1]) print(clf.score(testX,testY))