本文實例為大家分享了樹回歸的具體代碼,供大家參考,具體內容如下
#-*- coding:utf-8 -*- #!/usr/bin/python ''''' 回歸樹 連續值回歸預測 的 回歸樹 ''' # 測試代碼 # import regTrees as RT RT.RtTreeTest() RT.RtTreeTest('ex0.txt') RT.RtTreeTest('ex2.txt') # import regTrees as RT RT.RtTreeTest('ex2.txt',ops=(10000,4)) # import regTrees as RT RT.pruneTest() # 模型樹 測試 # import regTrees as RT RT.modeTreeTest(ops=(1,10) # 模型回歸樹和普通回歸樹 效果比較 計算相關系數 # import regTrees as RT RT.MRTvsSRT() from numpy import * # Tab 鍵值分隔的數據 提取成 列表數據集 成浮點型數據 def loadDataSet(fileName): # dataMat = [] # 目標數據集 列表 fr = open(fileName) for line in fr.readlines(): curLine = line.strip().split('/t') fltLine = map(float,curLine) #轉換成浮點型數據 dataMat.append(fltLine) return dataMat # 按特征值 的數據集二元切分 特征(列) 對應的值 # 某一列的值大于value值的一行樣本全部放在一個矩陣里,其余放在另一個矩陣里 def binSplitDataSet(dataSet, feature, value): mat0 = dataSet[nonzero(dataSet[:,feature] > value)[0],:][0] # 數組過濾 mat1 = dataSet[nonzero(dataSet[:,feature] <= value)[0],:][0] # return mat0,mat1 # 常量葉子節點 def regLeaf(dataSet):# 最后一列為標簽 為數的葉子節點 return mean(dataSet[:,-1])# 目標變量的均值 # 方差 def regErr(dataSet): return var(dataSet[:,-1]) * shape(dataSet)[0]# 目標變量的平方誤差 * 樣本個數(行數)的得到總方差 # 選擇最優的 分裂屬性和對應的大小 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: # 特征剩余數量為1 則返回 return None, leafType(dataSet) #### 返回 1 #### 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]):# 遍歷對應 特性的 屬性值 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) #### 返回 2 #### mat0, mat1 = binSplitDataSet(dataSet, bestIndex, bestValue) if (shape(mat0)[0] < tolN) or (shape(mat1)[0] < tolN): #樣本數量 小于設定值,則不切分 return None, leafType(dataSet) #### 返回 3 #### return bestIndex,bestValue # 返回最佳的 分裂屬性 和 對應的值 # 創建回歸樹 numpy數組數據集 葉子函數 誤差函數 用戶設置參數(最小樣本數量 以及最小誤差下降間隔) 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 RtTreeTest(filename='ex00.txt',ops=(1,4)): MyDat = loadDataSet(filename) # ex00.txt y = w*x 兩維 ex0.txt y = w*x+b 三維 MyMat = mat(MyDat) print createTree(MyMat,ops=ops) # 判斷是不是樹 (按字典形式存儲) 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 pruneTest(): MyDat = loadDataSet('ex2.txt') MyMat = mat(MyDat) MyTree = createTree(MyMat,ops=(0,1)) # 為了得到 最大的樹 誤差設置為0 個數設置為1 即不進行預剪枝 MyDatTest = loadDataSet('ex2test.txt') MyMatTest = mat(MyDatTest) print prune(MyTree,MyMatTest) ######葉子節點為線性模型的模型樹######### # 線性模型 def linearSolve(dataSet): m,n = shape(dataSet) # 數據集大小 X = mat(ones((m,n))) # 自變量 Y = mat(ones((m,1))) # 目標變量 X[:,1:n] = dataSet[:,0:n-1]# 樣本數據集合 Y = dataSet[:,-1] # 標簽 # 線性模型 求解 xTx = X.T*X if linalg.det(xTx) == 0.0: raise NameError('行列式值為零,不能計算逆矩陣,可適當增加ops的第二個值') ws = xTx.I * (X.T * Y) return ws,X,Y # 模型葉子節點 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)) # 模型樹測試 def modeTreeTest(filename='ex2.txt',ops=(1,4)): MyDat = loadDataSet(filename) # MyMat = mat(MyDat) print createTree(MyMat,leafType=modelLeaf, errType=modelErr,ops=ops)#帶入線性模型 和相應 的誤差計算函數 # 模型效果計較 # 線性葉子節點 預測計算函數 直接返回 樹葉子節點 值 def regTreeEval(model, inDat): return float(model) def modelTreeEval(model, inDat): n = shape(inDat)[1] X = mat(ones((1,n+1)))# 增加一列 X[:,1:n+1]=inDat return float(X*model) # 返回 值乘以 線性回歸系數 # 樹預測函數 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 # 常量回歸樹和線性模型回歸樹的預測結果比較 def MRTvsSRT(): TestMat = mat(loadDataSet('bikeSpeedVsIq_test.txt')) TrainMat = mat(loadDataSet('bikeSpeedVsIq_train.txt')) # 普通回歸樹 預測結果 # 得到普通回歸樹樹 StaTree = createTree(TrainMat, ops=(1,20)) # 得到預測結果 StaYHat = createForeCast(StaTree, TestMat[:,0], regTreeEval)# 第一列為 自變量 # 預測結果和真實標簽的相關系數 StaCorr = corrcoef(StaYHat, TestMat[:,1], rowvar=0)[0,1] # NumPy 庫函數 # 模型回歸樹 預測結果 # 得到模型回歸樹 ModeTree = createTree(TrainMat,leafType=modelLeaf, errType=modelErr, ops=(1,20)) # 得到預測結果 ModeYHat = createForeCast(ModeTree, TestMat[:,0], modelTreeEval) # 預測結果和真實標簽的相關系數 ModeCorr = corrcoef(ModeYHat, TestMat[:,1], rowvar=0)[0,1] # NumPy 庫函數 print "普通回歸樹 預測結果的相關系數R2: %f" %(StaCorr) print "模型回歸樹 預測結果的相關系數R2: %f" %(ModeCorr) if ModeCorr>StaCorr: print "模型回歸樹效果優于普通回歸樹" else: print "回歸回歸樹效果優于模型普通樹"
|
新聞熱點
疑難解答