Python機(jī)器學(xué)習(xí)之AdaBoost算法

from numpy import *
import matplotlib.pyplot as plt


# 加載數(shù)據(jù)集
def loadDataSet(fileName):
    numFeat = len(open(fileName).readline().split('\t'))
    dataMat = []
    labelMat = []
    fr = open(fileName)
    for line in fr.readlines():
        lineArr = []
        curLine = line.strip().split('\t')
        for i in range(numFeat - 1):
            lineArr.append(float(curLine[i]))
        dataMat.append(lineArr)
        labelMat.append(float(curLine[-1]))
    return dataMat, labelMat


# 返回分類預(yù)測(cè)結(jié)果  根據(jù)閾值所以有兩種返回情況
def stumpClassify(dataMatrix, dimen, threshVal, threshIneq):
    retArray = ones((shape(dataMatrix)[0], 1))
    if threshIneq == 'lt':
        retArray[dataMatrix[:, dimen] = threshVal] = -1.0
    else:
        retArray[dataMatrix[:, dimen] > threshVal] = -1.0
    return retArray


# 返回 該弱分類器單層決策樹的信息  更新D向量的錯(cuò)誤率 更新D向量的預(yù)測(cè)目標(biāo)
def buildStump(dataArr, classLabels, D):
    dataMatrix = mat(dataArr)
    labelMat = mat(classLabels).T
    m, n = shape(dataMatrix)
    numSteps = 10.0
    bestStump = {}  # 字典用于保存每個(gè)分類器信息
    bestClasEst = mat(zeros((m, 1)))
    minError = inf  # 初始化最小誤差最大
    for i in range(n):  # 特征循環(huán)  （三層循環(huán)，遍歷所有的可能性）
        rangeMin = dataMatrix[:, i].min()
        rangeMax = dataMatrix[:, i].max()
        stepSize = (rangeMax - rangeMin) / numSteps  # (大-小)/分割數(shù)  得到最小值到最大值需要的每一段距離
        for j in range(-1, int(numSteps) + 1):  # 遍歷步長(zhǎng) 最小值到最大值的需要次數(shù)
            for inequal in ['lt', 'gt']:  # 在大于和小于之間切換
                threshVal = (rangeMin + float(j) * stepSize)  # 最小值+次數(shù)*步長(zhǎng)  每一次從最小值走的長(zhǎng)度
                predictedVals = stumpClassify(dataMatrix, i, threshVal,
                                              inequal)  # 最優(yōu)預(yù)測(cè)目標(biāo)值  用于與目標(biāo)值比較得到誤差
                errArr = mat(ones((m, 1)))
                errArr[predictedVals == labelMat] = 0
                weightedError = D.T * errArr
                if weightedError  minError:  # 選出最小錯(cuò)誤的那個(gè)特征
                    minError = weightedError  # 最小誤差 后面用來更新D權(quán)值的
                    bestClasEst = predictedVals.copy()  # 最優(yōu)預(yù)測(cè)值

                    bestStump['dim'] = i  # 特征
                    bestStump['thresh'] = threshVal  # 到最小值的距離 （得到最優(yōu)預(yù)測(cè)值的那個(gè)距離）
                    bestStump['ineq'] = inequal  # 大于還是小于 最優(yōu)距離為-1
    return bestStump, minError, bestClasEst


# 循環(huán)構(gòu)建numIt個(gè)弱分類器
def adaBoostTrainDS(dataArr, classLabels, numIt=40):
    weakClassArr = []  # 保存弱分類器數(shù)組
    m = shape(dataArr)[0]
    D = mat(ones((m, 1)) / m)  # D向量 每條樣本所對(duì)應(yīng)的一個(gè)權(quán)重
    aggClassEst = mat(zeros((m, 1)))  # 統(tǒng)計(jì)類別估計(jì)累積值
    for i in range(numIt):
        bestStump, error, classEst = buildStump(dataArr, classLabels, D)
        alpha = float(0.5 * log((1.0 - error) / max(error, 1e-16)))
        bestStump['alpha'] = alpha
        weakClassArr.append(bestStump)  # 加入單層決策樹

        # 得到運(yùn)算公式中的向量+/-α，預(yù)測(cè)正確為-α，錯(cuò)誤則+α。每條樣本一個(gè)α
        # multiply對(duì)應(yīng)位置相乘  這里很聰明，用-1*真實(shí)目標(biāo)值*預(yù)測(cè)值，實(shí)現(xiàn)了錯(cuò)誤分類則-，正確則+
        expon = multiply(-1 * alpha * mat(classLabels).T, classEst)
        D = multiply(D, exp(expon))  # 這三步為更新概率分布D向量 拆分開來了，每一步與公式相同
        D = D / D.sum()

        # 計(jì)算停止條件錯(cuò)誤率=0 以及計(jì)算每次的aggClassEst類別估計(jì)累計(jì)值
        aggClassEst += alpha * classEst
        # 很聰明的計(jì)算方法 計(jì)算得到錯(cuò)誤的個(gè)數(shù)，向量中為1則錯(cuò)誤值
        aggErrors = multiply(sign(aggClassEst) != mat(classLabels).T, ones((m, 1)))  # sign返回?cái)?shù)值的正負(fù)符號(hào)，以1、-1表示
        errorRate = aggErrors.sum() / m  # 錯(cuò)誤個(gè)數(shù)/總個(gè)數(shù)
        # print("錯(cuò)誤率：", errorRate)
        if errorRate == 0.0:
            break
    return weakClassArr, aggClassEst


# 預(yù)測(cè) 累加 多個(gè)弱分類器獲得預(yù)測(cè)值*該alpha 得到結(jié)果
def adaClassify(datToClass, classifierArr):  # classifierArr是元組，所以在取值時(shí)需要注意
    dataMatrix = mat(datToClass)
    m = shape(dataMatrix)[0]
    aggClassEst = mat(zeros((m, 1)))
    # 循環(huán)所有弱分類器
    for i in range(len(classifierArr[0])):
        # 獲得預(yù)測(cè)結(jié)果
        classEst = stumpClassify(dataMatrix, classifierArr[0][i]['dim'], classifierArr[0][i]['thresh'],
                                 classifierArr[0][i]['ineq'])
        # 該分類器α*預(yù)測(cè)結(jié)果 用于累加得到最終的正負(fù)判斷條件
        aggClassEst += classifierArr[0][i]['alpha'] * classEst  # 這里就是集合所有弱分類器的意見，得到最終的意見
    return sign(aggClassEst)  # 提取數(shù)據(jù)符號(hào)

# ROC曲線，類別累計(jì)值、目標(biāo)標(biāo)簽
def plotROC(predStrengths, classLabels):
    cur = (1.0, 1.0)  # 每次畫線的起點(diǎn)游標(biāo)點(diǎn)
    ySum = 0.0  # 用于計(jì)算AUC的值 矩形面積的高度累計(jì)值
    numPosClas = sum(array(classLabels) == 1.0)  # 所有真實(shí)正例    確定了在y坐標(biāo)軸上的步進(jìn)數(shù)目
    yStep = 1 / float(numPosClas)  # 1/所有真實(shí)正例 y軸上的步長(zhǎng)
    xStep = 1 / float(len(classLabels) - numPosClas)  # 1/所有真實(shí)反例 x軸上的步長(zhǎng)
    sortedIndicies = predStrengths.argsort()  # 獲得累計(jì)值向量從小到大排序的下表index [50,88,2,71...]
    fig = plt.figure()
    fig.clf()
    ax = plt.subplot(111)
    # 循環(huán)所有的累計(jì)值 從小到大
    for index in sortedIndicies.tolist()[0]:
        if classLabels[index] == 1.0:
            delX = 0  # 若為一個(gè)真正例，則沿y降一個(gè)步長(zhǎng)，即不斷降低真陽率；
            delY = yStep  # 若為一個(gè)非真正例，則沿x退一個(gè)步長(zhǎng)，尖笑陽率
        else:
            delX = xStep
            delY = 0
            ySum += cur[1]  # 向下移動(dòng)一次，則累計(jì)一個(gè)高度。寬度不變，我們只計(jì)算高度
        ax.plot([cur[0], cur[0] - delX], [cur[1], cur[1] - delY], c='b')  # 始終會(huì)有一個(gè)點(diǎn)是沒有改變的
        cur = (cur[0] - delX, cur[1] - delY)
    ax.plot([0, 1], [0, 1], 'b--')
    plt.xlabel('False positive rate')
    plt.ylabel('True positive rate')
    plt.title('ROC curve for AdaBoost horse colic detection system')
    ax.axis([0, 1, 0, 1])
    plt.show()
    print("the Area Under the Curve is: ", ySum * xStep)  # AUC面積我們以 高*低 的矩形來計(jì)算


# 測(cè)試正確率
datArr, labelArr = loadDataSet('horseColicTraining2.txt')
classifierArr = adaBoostTrainDS(datArr, labelArr, 15)

testArr, testLabelArr = loadDataSet('horseColicTest2.txt')
prediction10 = adaClassify(testArr, classifierArr)

errArr = mat(ones((67, 1)))  # 一共有67個(gè)樣本
cnt = errArr[prediction10 != mat(testLabelArr).T].sum()
print(cnt / 67)

# 畫出ROC曲線
datArr, labelArr = loadDataSet('horseColicTraining2.txt')
classifierArray, aggClassEst = adaBoostTrainDS(datArr, labelArr, 10)
plotROC(aggClassEst.T, labelArr)