1. 程式人生 > >機器學習實戰(八)分類迴歸樹CART(Classification And Regression Tree)

機器學習實戰(八)分類迴歸樹CART(Classification And Regression Tree)

目錄

0. 前言

1. 迴歸樹

2. 模型樹

3. 剪枝(pruning)

3.1. 預剪枝

3.2. 後剪枝

4. 實戰案例

4.1. 迴歸樹

4.2. 模型樹


學習完機器學習實戰的分類迴歸樹,簡單的做個筆記。文中部分描述屬於個人消化後的理解,僅供參考。

所有程式碼和資料可以訪問 我的 github

如果這篇文章對你有一點小小的幫助,請給個關注喔~我會非常開心的~

0. 前言

分類迴歸樹(Classification And Regression Tree)主要用於複雜資料的迴歸。

  • 優點:可以對複雜和非線性的資料建模
  • 缺點:結果不易理解
  • 適用資料型別:數值型和標稱型在資料(標稱型資料需對映成二值型)

在 ID3 演算法中,每次選擇最佳的特徵分割資料,特徵有幾種取值,樹的結點就有幾棵子樹,而且連續型特徵需轉換為離散型。選擇過的特徵會被篩除,不會再次選擇。

在 CART 演算法中,每次選擇最佳的特徵分割資料,但是隻進行二元切分,產生兩棵子樹。選擇過的特徵,不會被篩除,仍有可能被選擇。

  • 迴歸樹:葉子結點為常數,即預測值
  • 模型樹:葉子結點為線性方程

1. 迴歸樹

在 ID3 演算法中,根據資訊增益定義資料的混亂度。

在 CART 演算法中,根據總方差(方差乘以樣本大小)定義資料的混亂度:

\sum_{i=1}^{m}(y^{(i)}-\mu_y)^2

在分類中,葉子節點表示的是對應的類別。

在迴歸中,葉子節點表示的是對應的預測值,在訓練模型的時候,使用資料結果的均值作為葉子節點

建立樹的虛擬碼如下表示:

每次選擇最佳特徵和特徵值時,採用誤差作為衡量標準,虛擬碼如下表示:

注:訓練資料結果相同、劃分後最小誤差和劃分前誤差相差不大、劃分後資料集很小,這幾種情況都直接返回葉子結點,不進行劃分。

2. 模型樹

模型樹在訓練的時候,當滿足返回葉子結點的條件的時候,對剩餘資料進行線性擬合,返回擬合引數,所以葉子結點是線性擬合的引數

模型樹的可解釋性是它優於迴歸樹的特點,當資料由分段函式組成的時候,模型樹可以更好的發揮它的作用。

在模型樹中,誤差的計算採用的是誤差平方和:

\sum_{i=1}^{m}(y^{(i)}-\hat{y}^{(i)})^2

3. 剪枝(pruning)

如果一棵樹的結點過多,可能會造成過擬合,需要對樹進行剪枝,去掉不必要的枝條,以降低複雜度。

  • 預剪枝(prepruning):在建立樹的時候,預先判斷,如果會造成過於複雜,則不擴充套件這個枝條
  • 後剪枝(postpruning):在樹建立了之後,對其進行測試,如果會造成過於複雜,則剪去這個枝條

一般地,為了達到更好的剪枝效果,會同時採用兩種剪枝方法

3.1. 預剪枝

在選擇最佳特徵的虛擬碼中,劃分後最小誤差和劃分前誤差相差不大、劃分後資料集很小,就直接返回葉子結點,不劃分資料擴充套件枝條,這就是預剪枝。

預剪枝對人為設定的引數比較敏感,例如最小誤差和劃分前誤差相差的閾值、資料集大小的閾值。

3.2. 後剪枝

在後剪枝中,將資料集分成訓練集和測試集,訓練集用於訓練樹,測試集用於剪枝。

後剪枝的思路是,從樹根進行遞迴,直到找到左結點和右結點都為葉子結點的時候,計算誤差平方和,再將兩個結點合併,計算誤差平方和,如果合併後誤差降低,則進行剪枝

後剪枝是從葉子結點從下往上合併,虛擬碼如下表示:

注:因測試資料和訓練資料的不同,可能會造成還未遞迴到葉子結點,測試資料就無法繼續劃分的情況。此時採用塌陷處理,即不斷遞迴返回結點的平均值,任一結點的平均值等於其左結點和右結點的平均。

4. 實戰案例

以下將展示書中案例的程式碼段,所有程式碼和資料可以在github中下載:

4.1. 迴歸樹

# coding:utf-8
from numpy import *

"""
迴歸樹
"""


# 載入資料集
def loadDataSet(fileName):
    dataMat = []
    fr = open(fileName)
    for line in fr.readlines():
        curLine = line.strip().split('\t')
        # 將資料對映為浮點型
        fltLine = list(map(float, curLine))
        dataMat.append(fltLine)
    return dataMat


# 根據特徵和特徵值,二元分割一個數據集
def binSplitDataSet(dataSet, feature, value):
    mat0 = dataSet[nonzero(dataSet[:, feature] > value)[0], :]
    mat1 = dataSet[nonzero(dataSet[:, feature] <= value)[0], :]
    return mat0, mat1


# 建立葉子結點時,採取所有剩餘資料的標籤的均值
def regLeaf(dataSet):
    return mean(dataSet[:, -1])


# 計算總方差
def regErr(dataSet):
    return var(dataSet[:, -1]) * shape(dataSet)[0]


# CART演算法選擇最佳劃分點
def chooseBestSplit(dataSet, leafType=regLeaf, errType=regErr, ops=(1, 4)):
    # 誤差改善的最小要求
    tolS = ops[0]
    # 資料集大小的最小要求
    tolN = ops[1]
    # 如果當前資料集結果標籤都是同一個值,則直接返回葉子節點
    if len(set(dataSet[:, -1].T.tolist()[0])) == 1:
        return None, leafType(dataSet)
    m, n = shape(dataSet)
    # 獲取當前資料集的誤差
    S = errType(dataSet)
    bestS = inf
    bestIndex = 0
    bestValue = 0
    # 一重迴圈遍歷所有特徵
    for featIndex in range(n - 1):
        # 二重迴圈遍歷所有特徵值
        for splitVal in set((dataSet[:, featIndex].T.A.tolist())[0]):
            # 劃分資料
            mat0, mat1 = binSplitDataSet(dataSet, featIndex, splitVal)
            # 如果劃分後資料集太小則返回
            if (shape(mat0)[0] < tolN) or (shape(mat1)[0] < tolN):
                continue
            # 計算新的誤差
            newS = errType(mat0) + errType(mat1)
            # 如果新的誤差小於當前最好的誤差,則替換
            if newS < bestS:
                bestIndex = featIndex
                bestValue = splitVal
                bestS = newS
    # 遍歷結束後,如果最佳的誤差與遍歷之前資料集的誤差改善不大,則直接返回
    if (S - bestS) < tolS:
        return None, leafType(dataSet)
    # 劃分兩個資料子集
    mat0, mat1 = binSplitDataSet(dataSet, bestIndex, bestValue)
    # 如果兩個子集太小,則直接返回
    if (shape(mat0)[0] < tolN) or (shape(mat1)[0] < tolN):
        return None, leafType(dataSet)
    return bestIndex, bestValue


# 遞迴建立樹
# dataSet: 資料集
# leafType: 返回葉子節點的時候引用的函式
# errType: 誤差計算引用的函式
# ops: 使用者定義的標準值
def createTree(dataSet, leafType=regLeaf, errType=regErr, ops=(1, 4)):
    # 選擇最佳的劃分點
    feat, val = chooseBestSplit(dataSet, leafType, errType, ops)
    # 當前為葉子節點
    if feat == None:
        return val
    # 記錄當前的劃分的特徵和特徵值
    retTree = {}
    retTree['spInd'] = feat
    retTree['spVal'] = val
    # 劃分兩個資料集
    lSet, rSet = binSplitDataSet(dataSet, feat, val)
    # 遞迴對兩個子集建立子樹
    retTree['left'] = createTree(lSet, leafType, errType, ops)
    retTree['right'] = createTree(rSet, leafType, errType, ops)
    return retTree


# 對樹進行後剪枝
# 判斷是否是子樹
def isTree(obj):
    return (type(obj).__name__ == 'dict')


# 對樹進行後剪枝
# 遞迴獲取當前節點的均值
# 在沒有測試資料的時候,對節點進行塌陷處理
def getMean(tree):
    if isTree(tree['right']): tree['right'] = getMean(tree['right'])
    if isTree(tree['left']): tree['left'] = getMean(tree['left'])
    return (tree['left'] + tree['right']) / 2.0


# 對樹進行後剪枝,演算法
def prune(tree, testData):
    # 如果沒有測試資料了,則對樹進行塌陷處理
    if shape(testData)[0] == 0:
        return getMean(tree)
    # 如果左節點或者右節點是樹,則劃分測試資料集
    if (isTree(tree['right']) or isTree(tree['left'])):
        lSet, rSet = binSplitDataSet(testData, tree['spInd'], tree['spVal'])
    # 如果左節點或者右節點是樹,則遞迴後剪枝,直到葉子節點
    if isTree(tree['left']): tree['left'] = prune(tree['left'], lSet)
    if isTree(tree['right']): tree['right'] = prune(tree['right'], rSet)
    # 當前左節點和右節點都為葉子節點
    if not isTree(tree['left']) and not isTree(tree['right']):
        # 劃分測試資料集
        lSet, rSet = binSplitDataSet(testData, tree['spInd'], tree['spVal'])
        # 計算不合的誤差
        errorNoMerge = sum(power(lSet[:, -1] - tree['left'], 2)) + \
                       sum(power(rSet[:, -1] - tree['right'], 2))
        # 計算合併的誤差
        treeMean = (tree['left'] + tree['right']) / 2.0
        errorMerge = sum(power(testData[:, -1] - treeMean, 2))
        # 如果合併後誤差小,則合併
        if errorMerge < errorNoMerge:
            print("merging")
            return treeMean
        else:
            return tree
    else:
        return tree


# 測試函式
# 返回葉子節點浮點型別值
def regTreeEval(model, inDat):
    return float(model)


# 預測函式
# inData是一條資料向量矩陣
def treeForeCast(tree, inData, modelEval=regTreeEval):
    # 葉子節點
    if not isTree(tree):
        return modelEval(tree, inData)
    # 選擇左子樹還是右子樹
    if inData[tree['spInd']] > tree['spVal']:
        # 判斷是否是樹
        if isTree(tree['left']):
            return treeForeCast(tree['left'], inData, modelEval)
        else:
            return modelEval(tree['left'], inData)
    else:
        if isTree(tree['right']):
            return treeForeCast(tree['right'], inData, modelEval)
        else:
            return modelEval(tree['right'], inData)


# 預測函式測試
def createForeCast(tree, testData, modelEval=regTreeEval):
    m = len(testData)
    yHat = mat(zeros((m, 1)))
    for i in range(m):
        yHat[i, 0] = treeForeCast(tree, mat(testData[i]), modelEval)
    return yHat


if __name__ == '__main__':
    # myDat1 = loadDataSet('ex0.txt')
    # myMat1 = mat(myDat1)
    # tree1 = createTree(myMat1)
    # print(tree1)

    # myDat2 = loadDataSet('ex2.txt')
    # myMat2 = mat(myDat2)
    # tree2 = createTree(myMat2, ops=(0, 1))
    # myDat2Test = loadDataSet('ex2test.txt')
    # myMat2Test = mat(myDat2Test)
    # tree2 = prune(tree2, myMat2Test)
    # print(tree2)

    trainMat = mat(loadDataSet('bikeSpeedVsIq_train.txt'))
    testMat = mat(loadDataSet('bikeSpeedVsIq_test.txt'))
    myTree = createTree(trainMat, ops=(1, 20))
    yHat = createForeCast(myTree, testMat[:, 0])
    print(corrcoef(yHat, testMat[:, 1], rowvar=0)[0, 1])

4.2. 模型樹

# coding:utf-8
from numpy import *

"""
模型樹
"""


# 載入資料集
def loadDataSet(fileName):
    dataMat = []
    fr = open(fileName)
    for line in fr.readlines():
        curLine = line.strip().split('\t')
        # 將資料對映為浮點型
        fltLine = list(map(float, curLine))
        dataMat.append(fltLine)
    return dataMat


# 根據特徵和特徵值,二元分割一個數據集
def binSplitDataSet(dataSet, feature, value):
    mat0 = dataSet[nonzero(dataSet[:, feature] > value)[0], :]
    mat1 = dataSet[nonzero(dataSet[:, feature] <= value)[0], :]
    return mat0, mat1


# 對資料進行線性迴歸
def linearSolve(dataSet):
    m, n = shape(dataSet)
    X = mat(ones((m, n)))
    Y = mat(ones((m, 1)))
    # 需要x_0=1
    X[:, 1:n] = dataSet[:, 0:n - 1]
    Y = dataSet[:, -1]
    # 正規方程
    xTx = X.T * X
    if linalg.det(xTx) == 0.0:
        raise NameError('This matrix is singular, cannot do inverse,\n\
        try increasing the second value of ops')
    ws = xTx.I * (X.T * Y)
    return ws, X, Y


# 建立葉子結點時,採用線性函式,即權重ws
def modelLeaf(dataSet):
    ws, X, Y = linearSolve(dataSet)
    return ws


# 採用誤差平方和計算誤差
def modelErr(dataSet):
    ws, X, Y = linearSolve(dataSet)
    yHat = X * ws
    return sum(power(Y - yHat, 2))


# CART演算法選擇最佳劃分點
def chooseBestSplit(dataSet, leafType=modelLeaf, errType=modelErr, ops=(1, 10)):
    # 誤差改善的最小要求
    tolS = ops[0]
    # 資料集大小的最小要求
    tolN = ops[1]
    # 如果當前資料集結果標籤都是同一個值,則直接返回葉子節點
    if len(set(dataSet[:, -1].T.tolist()[0])) == 1:
        return None, leafType(dataSet)
    m, n = shape(dataSet)
    # 獲取當前資料集的誤差
    S = errType(dataSet)
    bestS = inf
    bestIndex = 0
    bestValue = 0
    # 一重迴圈遍歷所有特徵
    for featIndex in range(n - 1):
        # 二重迴圈遍歷所有特徵值
        for splitVal in set((dataSet[:, featIndex].T.A.tolist())[0]):
            # 劃分資料
            mat0, mat1 = binSplitDataSet(dataSet, featIndex, splitVal)
            # 如果劃分後資料集太小則返回
            if (shape(mat0)[0] < tolN) or (shape(mat1)[0] < tolN):
                continue
            # 計算新的誤差
            newS = errType(mat0) + errType(mat1)
            # 如果新的誤差小於當前最好的誤差,則替換
            if newS < bestS:
                bestIndex = featIndex
                bestValue = splitVal
                bestS = newS
    # 遍歷結束後,如果最佳的誤差與遍歷之前資料集的誤差改善不大,則直接返回
    if (S - bestS) < tolS:
        return None, leafType(dataSet)
    # 劃分兩個資料子集
    mat0, mat1 = binSplitDataSet(dataSet, bestIndex, bestValue)
    # 如果兩個子集太小,則直接返回
    if (shape(mat0)[0] < tolN) or (shape(mat1)[0] < tolN):
        return None, leafType(dataSet)
    return bestIndex, bestValue


# 遞迴建立樹
# dataSet: 資料集
# leafType: 返回葉子節點的時候引用的函式
# errType: 誤差計算引用的函式
# ops: 使用者定義的標準值
def createTree(dataSet, leafType=modelLeaf, errType=modelErr, ops=(1, 4)):
    # 選擇最佳的劃分點
    feat, val = chooseBestSplit(dataSet, leafType, errType, ops)
    # 當前為葉子節點
    if feat == None:
        return val
    # 記錄當前的劃分的特徵和特徵值
    retTree = {}
    retTree['spInd'] = feat
    retTree['spVal'] = val
    # 劃分兩個資料集
    lSet, rSet = binSplitDataSet(dataSet, feat, val)
    # 遞迴對兩個子集建立子樹
    retTree['left'] = createTree(lSet, leafType, errType, ops)
    retTree['right'] = createTree(rSet, leafType, errType, ops)
    return retTree


# 判斷是否是子樹
def isTree(obj):
    return (type(obj).__name__ == 'dict')


# 測試函式
# 返回葉子節點引數和資料相乘後的擬合結果
def modelTreeEval(model, inDat):
    n = shape(inDat)[1]
    # 因為存在x_0=1
    X = mat(ones((1, n + 1)))
    X[:, 1:n + 1] = inDat
    return float(X * model)


# 預測函式
# inData是一條資料向量矩陣
def treeForeCast(tree, inData, modelEval=modelTreeEval):
    # 葉子節點
    if not isTree(tree):
        return modelEval(tree, inData)
    # 選擇左子樹還是右子樹
    if inData[tree['spInd']] > tree['spVal']:
        # 判斷是否是樹
        if isTree(tree['left']):
            return treeForeCast(tree['left'], inData, modelEval)
        else:
            return modelEval(tree['left'], inData)
    else:
        if isTree(tree['right']):
            return treeForeCast(tree['right'], inData, modelEval)
        else:
            return modelEval(tree['right'], inData)


# 預測函式測試
def createForeCast(tree, testData, modelEval=modelTreeEval):
    m = len(testData)
    yHat = mat(zeros((m, 1)))
    for i in range(m):
        yHat[i, 0] = treeForeCast(tree, mat(testData[i]), modelEval)
    return yHat


if __name__ == '__main__':
    # myMat = mat(loadDataSet('exp2.txt'))
    # tree = createTree(myMat)
    # print(tree)

    trainMat = mat(loadDataSet('bikeSpeedVsIq_train.txt'))
    testMat = mat(loadDataSet('bikeSpeedVsIq_test.txt'))
    myTree = createTree(trainMat, ops=(1, 20))
    yHat = createForeCast(myTree, testMat[:, 0])
    print(corrcoef(yHat, testMat[:, 1], rowvar=0)[0, 1])

如果這篇文章對你有一點小小的幫助,請給個關注喔~我會非常開心的~