提升算法的理论参考《统计学习方法》，本文的部分代码参考《机器学习实战》实现的。《机器学习实战》这本书上的代码很多时候是跑不通的，而且某些算法关键部分来得莫名其妙，也没说明是怎么来的，为什么那么写。本人自己实现的代码关键部分是按照《统计学习方法》这本书的理论实现的，并且通俗易懂，完全对照算法的思路和公式就可看懂。好了，废话少说，直接开始吧！
简单训练数据
def load_simple_data():
data_mat = matrix([[1.0, 2.1],
[2.0, 1.1],
[1.3, 1.0],
[1.0, 1.0],
[2.0, 1.0]])
class_labels = [1.0, 1.0, -1.0, -1.0, 1.0]
return data_mat,class_labels
根据阈值判断每个特征的输出值，其中ret_array初始化时不能全-1或全1，其他任何值都可以。但是《机器学习实战》ret_array初始化为全1，我也跑过这本书上的算法，很遗憾没有得到正确的输出。
def stump_classify(data_matrix, dimen, threshval, thresh_ineq):
ret_array = zeros((shape(data_matrix)[0], 1))
if(thresh_ineq == "lt"):
ret_array[data_matrix[:, dimen] <= threshval] = -1.0
else:
ret_array[data_matrix[:, dimen] > threshval] = 1.0
return ret_array
单层决策树的实现：lt_predicted_arr存储小于等于阈值的输出值，gt_predicted_arr存储大于阈值的输出值，最后再综合起来得到该阈值下的输出值，最后计算权重和，用字典保存相关信息。在这里《机器学习实战》和本人实现的方式很不一样，仔细看了这本书上不同的那部分实现方式，第一：不懂为什么那样写，第二：感觉书上的实现方式错的，得不到正确的输出。哪位大神能否解答我的疑惑？
def build_stump(data_arr, class_labels, D):
data_matrix = mat(data_arr)
label_mat = mat(class_labels).T
m,n = shape(data_matrix)
num_steps = 10.0
best_stump = {}
best_class_est = mat(zeros((m, 1)))
min_error = inf
for i in range(n):
range_min = data_matrix[:, i].min()
range_max = data_matrix[:, i].max()
step_size = (range_max - range_min) * 1.0 / num_steps
for j in range(-1, int(num_steps) + 1):
thresh_val = (range_min + float(j) * step_size)
lt_predicted_arr = zeros((m, 1)) #获得小于不等号的值
gt_predicted_arr = zeros((m, 1)) #获得大于不等号的值
predicted_arr = zeros((m, 1)) #最终的预测值
for inequal in ["lt", "gt"]:
predicted_vals = stump_classify(data_matrix, i, thresh_val, inequal)
if(inequal == "lt"):
lt_predicted_arr = predicted_vals
else:
gt_predicted_arr = predicted_vals
for k in range(m):
predicted_arr[k] = lt_predicted_arr[k]
if(gt_predicted_arr[k] != 0):
predicted_arr[k] = gt_predicted_arr[k]
err_arr = mat(ones((m, 1)))
err_arr[predicted_arr == label_mat] = 0
weight_error = D.T * err_arr
print("min_error = %0.5f, split: dim %d, thresh %0.2f,\
the weighted error is %0.3f" %\
(min_error, i, thresh_val, weight_error))
if(weight_error < min_error):
min_error = weight_error
best_class_est = predicted_arr.copy()
best_stump["dim"] = i
best_stump["thresh"] = thresh_val
best_stump["class_est"] = best_class_est
return best_stump,min_error,best_class_est
根据单层决策树得到一个弱分类器保存在列表中，根据得到的弱分类器输出的预测值和原始值修改权重，减小误差小的点的权重，增大误差大的点的权重。最后误差率等于0则停止迭代。
def adaboost_train_DS(data_arr, class_labels, num_it = 40):
weak_class_arr = []
m = shape(data_arr)[0]
D = mat(ones((m, 1)) / m)
agg_class_est = mat(zeros((m, 1)))
counts = 0
for i in range(num_it):
best_stump,error,class_est = build_stump(data_arr, class_labels, D)
print("D: ", D.T)
alpha = float(0.5 * log((1.0 - error) / error))
best_stump["alpha"] = alpha
weak_class_arr.append(best_stump)
print("class_est: ", class_est.T)
expon = multiply(-1 * alpha * mat(class_labels).T, class_est)
D = multiply(D, exp(expon))
D = D / D.sum()
agg_class_est += alpha * class_est #预测值
print("agg_class_est: ", agg_class_est.T)
agg_errors = multiply(sign(agg_class_est) != mat(class_labels).T, ones((m, 1)))
error_rate = agg_errors.sum() / m
print("total error: ", error_rate)
counts += 1
if(error_rate == 0.0):
break
return weak_class_arr,counts
弱分类器：根据每一个弱分类器阈值，输出每个特征的输出值。
def Gx(data_matrix, reshval):
m = shape(data_matrix)[0]
result_data = mat(ones((m, 1)))
result_data[data_matrix[:, 0] <= reshval] = -1.0
result_data[data_matrix[:, 0] > reshval] = 1.0
return result_data
根据输出数据，输出实际预测值。
def ada_classfiy(input_data, weak_class_arr, counts):
data_matrix = mat(input_data)
m,n = shape(data_matrix)
agg_class_est = mat(zeros((m, 1)))
for j in range(n):
for i in range(len(weak_class_arr)):
agg_class_est += (weak_class_arr[i]["alpha"] * (Gx(input_data[:, j], weak_class_arr[i]["thresh"])).T).T
print(agg_class_est)
return sign(agg_class_est)
画出实验结果图函数
def experiment_plot(data_matrix, agg_class_est):
data_arr_in = data_matrix.getA()
label_arr_in = agg_class_est.getA()
m,n = shape(data_matrix)
for i in range(m):
if(label_arr_in[i, 0] == -1):
plt.plot(data_arr_in[i, 0], data_arr_in[i, 1], "ob")
elif(label_arr_in[i, 0] == 1):
plt.plot(data_arr_in[i, 0], data_arr_in[i, 1], "or")
plt.xlabel("X")
plt.ylabel("Y")
plt.show()
主函数
def main():
#D = mat(ones((5, 1)) / 5.0)
data_mat,class_labels = load_simple_data()
#build_stump(data_mat, class_labels, D)
weak_class_arr,counts = adaboost_train_DS(data_mat, class_labels)
print("**************************")
data_mat2 = matrix([[1.2, 2.0],
[1.0, 1.0],
[0.6, 1.2],
[0.8, 1.1],
[1.8, 1.0],
[2.0, 0.4],
[1.7, 0.8],
[3.5, 0.2],
[2.5, 0.8],
[2.8, 0.9]])
agg_class_est = ada_classfiy(data_mat2, weak_class_arr, counts)
print("--------------------------")
print(agg_class_est)
experiment_plot(data_mat2, agg_class_est)
main()
实验结果：
本文的实验结果
4.PNG
《机器学习实战》这本书的结果
5.PNG
两者对比可知，结果是一样的，但是本文的实现方法是完全不同的。
完整代码如下：
from numpy import *
import numpy as np
import matplotlib.pyplot as plt
def load_simple_data():
data_mat = matrix([[1.0, 2.1],
[2.0, 1.1],
[1.3, 1.0],
[1.0, 1.0],
[2.0, 1.0]])
class_labels = [1.0, 1.0, -1.0, -1.0, 1.0]
return data_mat,class_labels
def stump_classify(data_matrix, dimen, threshval, thresh_ineq):
ret_array = zeros((shape(data_matrix)[0], 1))
if(thresh_ineq == "lt"):
ret_array[data_matrix[:, dimen] <= threshval] = -1.0
else:
ret_array[data_matrix[:, dimen] > threshval] = 1.0
return ret_array
def build_stump(data_arr, class_labels, D):
data_matrix = mat(data_arr)
label_mat = mat(class_labels).T
m,n = shape(data_matrix)
num_steps = 10.0
best_stump = {}
best_class_est = mat(zeros((m, 1)))
min_error = inf
for i in range(n):
range_min = data_matrix[:, i].min()
range_max = data_matrix[:, i].max()
step_size = (range_max - range_min) * 1.0 / num_steps
for j in range(-1, int(num_steps) + 1):
thresh_val = (range_min + float(j) * step_size)
lt_predicted_arr = zeros((m, 1)) #获得小于不等号的值
gt_predicted_arr = zeros((m, 1)) #获得大于不等号的值
predicted_arr = zeros((m, 1)) #最终的预测值
for inequal in ["lt", "gt"]:
predicted_vals = stump_classify(data_matrix, i, thresh_val, inequal)
if(inequal == "lt"):
lt_predicted_arr = predicted_vals
else:
gt_predicted_arr = predicted_vals
for k in range(m):
predicted_arr[k] = lt_predicted_arr[k]
if(gt_predicted_arr[k] != 0):
predicted_arr[k] = gt_predicted_arr[k]
err_arr = mat(ones((m, 1)))
err_arr[predicted_arr == label_mat] = 0
weight_error = D.T * err_arr
print("min_error = %0.5f, split: dim %d, thresh %0.2f,\
the weighted error is %0.3f" %\
(min_error, i, thresh_val, weight_error))
if(weight_error < min_error):
min_error = weight_error
best_class_est = predicted_arr.copy()
best_stump["dim"] = i
best_stump["thresh"] = thresh_val
best_stump["class_est"] = best_class_est
return best_stump,min_error,best_class_est
def adaboost_train_DS(data_arr, class_labels, num_it = 40):
weak_class_arr = []
m = shape(data_arr)[0]
D = mat(ones((m, 1)) / m)
agg_class_est = mat(zeros((m, 1)))
counts = 0
for i in range(num_it):
best_stump,error,class_est = build_stump(data_arr, class_labels, D)
print("D: ", D.T)
alpha = float(0.5 * log((1.0 - error) / error))
best_stump["alpha"] = alpha
weak_class_arr.append(best_stump)
print("class_est: ", class_est.T)
expon = multiply(-1 * alpha * mat(class_labels).T, class_est)
D = multiply(D, exp(expon))
D = D / D.sum()
agg_class_est += alpha * class_est #预测值
print("agg_class_est: ", agg_class_est.T)
agg_errors = multiply(sign(agg_class_est) != mat(class_labels).T, ones((m, 1)))
error_rate = agg_errors.sum() / m
print("total error: ", error_rate)
counts += 1
if(error_rate == 0.0):
break
return weak_class_arr,counts
def Gx(data_matrix, reshval):
m = shape(data_matrix)[0]
result_data = mat(ones((m, 1)))
result_data[data_matrix[:, 0] <= reshval] = -1.0
result_data[data_matrix[:, 0] > reshval] = 1.0
return result_data
def ada_classfiy(input_data, weak_class_arr, counts):
data_matrix = mat(input_data)
m,n = shape(data_matrix)
agg_class_est = mat(zeros((m, 1)))
for j in range(n):
for i in range(len(weak_class_arr)):
agg_class_est += (weak_class_arr[i]["alpha"] * (Gx(input_data[:, j], weak_class_arr[i]["thresh"])).T).T
print(agg_class_est)
return sign(agg_class_est)
def experiment_plot(data_matrix, agg_class_est):
data_arr_in = data_matrix.getA()
label_arr_in = agg_class_est.getA()
m,n = shape(data_matrix)
for i in range(m):
if(label_arr_in[i, 0] == -1):
plt.plot(data_arr_in[i, 0], data_arr_in[i, 1], "ob")
elif(label_arr_in[i, 0] == 1):
plt.plot(data_arr_in[i, 0], data_arr_in[i, 1], "or")
plt.xlabel("X")
plt.ylabel("Y")
plt.show()
def main():
#D = mat(ones((5, 1)) / 5.0)
data_mat,class_labels = load_simple_data()
#build_stump(data_mat, class_labels, D)
weak_class_arr,counts = adaboost_train_DS(data_mat, class_labels)
print("**************************")
data_mat2 = matrix([[1.2, 2.0],
[1.0, 1.0],
[0.6, 1.2],
[0.8, 1.1],
[1.8, 1.0],
[2.0, 0.4],
[1.7, 0.8],
[3.5, 0.2],
[2.5, 0.8],
[2.8, 0.9]])
agg_class_est = ada_classfiy(data_mat2, weak_class_arr, counts)
print("--------------------------")
print(agg_class_est)
experiment_plot(data_mat2, agg_class_est)
main()
总结：在学习机器学习算法的过程中，先自己分析理论，看懂书上的代码为什么那么写，一步一步对照算法的思路理解代码。如果遇到无法运行的代码，自己根据算法的思路和理论修改代码，调试。有时间的话，可以自己把算法实现一遍。
