干货 | 集成学习综述与代码实现

import os
import numpy as np
import pandas as pd
from sklearn.cross_validation import train_test_split
from sklearn.datasets import load_irisdef load_data():
   iris = load_iris()#导入数据
   #转换为dataframe
   df = pd.DataFrame(iris.data,columns=iris.feature_names)
   df['label'] = iris.target#导入标签
   #把0和1类拿出来
   df.columns = ['sepal length','sepal width','petal length','petal width','label']#重命名列名
   data = np.array(df.iloc[:100,[0,1,-1]])#前100行为分类为0和1两类
   #把为0的变成-1
   for i in range(len(data)):
       if data[i,-1]==0:#把0类的转换为-1类
           data[i,-1] = -1
   return data[:,0:2],data[:,-1]X,y = load_data()#导入数据
train_x,test_x,train_x,train_y = train_test_split(X,y,test_size=0.2,random_state=2333)#划分训练集和测试集
#定义Adaboost类
class Adaboost(object):
   def __init__(self,n_estimators=1,learning_rate = 0):
   '''
   n_estimators 分类器个数
   learning_rate 步长
   '''
       self.n_estimate = n_estimators
       self.learning_rate = learning_rate   def init_data(self,X,y):
   '''
   X：属性
   y：标签
   '''
       self.M,self.N = X.shape#属性一共有M行N列
       self.X = X#训练属性
       self.y = y#标签数据
       self.w = [1.0/self.M]*self.M#初始化权重为1/M
       self.alpha = []
       self.clf_set = []#模型参数集合   def _alpha(self,error):#误差越大 alpha值越小
       return 0.5*np.log((1-error)/error)   def _Z(self, weights, a, clf):
   '''
   a:各个分类器的权重
   weights:权重
   clf:compare_array
   '''
       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):#相当于给权重w做了一层softmax 归一化
   '''
   a:各个分类器的权重
   weights:权重
   clf:compare_array
   误分类self.y[i]*clf[i]=1权重增大 正确分类self.y[i]*clf[i]=-1权重变小
   '''
       for i in range(self.M):
           self.w[i] = self.w[i]*np.exp(-1*a*self.y[i]*clf[i])/ Z   #具体算法
   def G(self,feature,label,weights):#每次更新的是w
       #求出属性列中的最大值和最小值
       f_min = min(feature)
       f_max = max(feature)
       #在属性列上迭代的次数
       n_step = (f_max-f_min+self.learning_rate)//self.learning_rate
       error = 10000.0#定义为无穷大
       best_v = 0.0#表示属性中最好分类点的值
       direct,compare_array = None,None
       #direct表示优化方向
       #compare_array表示该方向对应的c_o_positive or c_o_negative数组
       for i in range(1,int(n_step)):
           v = f_min+self.learning_rate*i#每次增加的步长为learning_rate
           if v not in feature:#如果v不存在属性列中
               c_o_positive = np.array([1 if feature[k] > v else -1 for k in range(len(feature))])#大于v的label置为1 小于v的置为-1
               #print([weights[k] for k in range(len(feature)) if c_o_positive[k]!=label[k]])
               error_positive = sum([weights[k] for k in range(len(feature)) if c_o_positive[k]!=label[k]])#记录不等于label的weight并求和
               c_o_negative = np.array([-1 if feature[k] > v else 1 for k in range(len(feature))])#同上 但反向
               error_negative = sum([weights[k] for k in range(len(feature)) if c_o_negative[k]!=label[k]])
               if error_positive < error_negative:#如果正向误差小于反向误差
                   weight_error = error_positive#记录误差
                   _compare_array = c_o_positive
                   direct = 'positive'
               else:
                   weight_error = error_negative
                   _compare_array = c_o_negative
                   direct = 'negative'
               if weight_error < error:
                   error = weight_error#记录误差
                   best_v = v#记录最佳分类值
                   compare_array = _compare_array#记录label
       return best_v,error,compare_array,direct   def _G(self,x,v,direct):
       if direct=='positive':
           return 1 if x > v else -1
       else:
           return -1 if x>x else -1   def fit(self,X,y):
       self.init_data(X,y)#初始化类内参数
       for i in range(self.n_estimate):#迭代次数
           best_clf_error,best_v,clf_result = 100000,None,None
           #单个属性
           for j in range(self.N):#在属性列上迭代
               feature = self.X[:,j]#
               v,error,compare,direct = self.G(feature,self.y,self.w)
               if error<best_clf_error:
                   best_clf_error = error
                   best_v = v
                   final_direct = direct
                   clf_result = compare
                   axis = j
               if best_clf_error == 0:
                   break
           a = self._alpha(best_clf_error)#计算alpha 0.5*np.log((1-error)/error)
           self.alpha.append(a)#记录alpha
           z = self._Z(self.w,a,clf_result)
           self.clf_set.append((axis,best_v,final_direct))#clf_set集合
           self._w(a,clf_result,z)#修改self.w   def predict(self,feature):#预测函数（比较简单）
       result = 0.0
       for i in range(len(self.clf_set)):
           axis,best_v,final_direct = self.clf_set[i]
           f = feature[axis]
           result += self.alpha[i] * self._G(f,best_v,final_direct)
       return 1 if result >0 else -1   def score(self,X_test,y_test):#打分函数（比较简单）
       count = 0
       for i in range(len(X_test)):
           feature = X_test[i]
           if self.predict(feature) == y_test[i]:
               count += 1
       return count/len(X_test)