from random import seed,randrange,random
from sklearn.model_selection import train_test_split
import numpy as np
# 導入csv文件
def loadDataSet(filename):
dataset = []
with open(filename, 'r') as fr:
for line in fr.readlines():
if not line:
continue
lineArr = []
for featrue in line.split(','):
# strip()返回移除字符串頭尾指定的字符生成的新字符串
str_f = featrue.strip()
# isdigit 如果是浮點型數值，就是 false，所以換成 isalpha() 函數
# if str_f.isdigit(): # 判斷是否是數字
if str_f.isalpha(): # 如果是字母，說明是標簽
# 添加分類標簽
lineArr.append(str_f)
else:
# 將數據集的第column列轉換成float形式
lineArr.append(float(str_f))
dataset.append(lineArr)
return dataset
def cross_validation_split(dataset, n_folds):
"""cross_validation_split(將數據集進行抽重抽樣 n_folds 份，數據可以重復重復抽取，每一次list的元素是無重復的)
Args:
dataset 原始數據集
n_folds 數據集dataset分成n_flods份
Returns:
dataset_split list集合，存放的是：將數據集進行抽重抽樣 n_folds 份，數據可以重復重復抽取，每一次list的元素是無重復的
"""
dataset_split = list()
dataset_copy = list(dataset) # 復制一份 dataset,防止 dataset 的內容改變
fold_size = len(dataset) / n_folds
for i in range(n_folds):
fold = list() # 每次循環 fold 清零，防止重復導入 dataset_split
while len(fold) < fold_size: # 這里不能用 if，if 只是在第一次判斷時起作用，while 執行循環，直到條件不成立
# 有放回的隨機采樣，有一些樣本被重復采樣，從而在訓練集中多次出現，有的則從未在訓練集中出現，此則自助采樣法。從而保證每棵決策樹訓練集的差異性
index = randrange(len(dataset_copy))
# 將對應索引 index 的內容從 dataset_copy 中導出，並將該內容從 dataset_copy 中刪除。
# pop() 函數用於移除列表中的一個元素（默認最后一個元素），並且返回該元素的值。
# fold.append(dataset_copy.pop(index)) # 無放回的方式
fold.append(dataset_copy[index]) # 有放回的方式
dataset_split.append(fold)
# 由dataset分割出的n_folds個數據構成的列表，為了用於交叉驗證
return dataset_split
# Split a dataset based on an attribute and an attribute value # 根據特征和特征值分割數據集
def test_split(index, value, dataset):
left, right = list(), list()
for row in dataset:
if row[index] < value:
left.append(row)
else:
right.append(row)
return left, right
def gini_index(groups, class_values): # 個人理解：計算代價，分類越准確，則 gini 越小
gini = 0.0
D = len(groups[0]) + len(groups[1])
for class_value in class_values: # class_values = [0, 1]
for group in groups: # groups = (left, right)
size = len(group)
if size == 0:
continue
proportion = [row[-1] for row in group].count(class_value) / float(size)
gini += float(size)/D * (proportion * (1.0 - proportion)) # 個人理解：計算代價，分類越准確，則 gini 越小
return gini
# 找出分割數據集的最優特征，得到最優的特征 index，特征值 row[index]，以及分割完的數據 groups（left, right）
def get_split(dataset, n_features):
class_values = list(set(row[-1] for row in dataset)) # class_values =[0, 1]
b_index, b_value, b_score, b_groups = 999, 999, 999, None
features = list()
while len(features) < n_features:
index = randrange(len(dataset[0])-1) # 往 features 添加 n_features 個特征（ n_feature 等於特征數的根號），特征索引從 dataset 中隨機取
if index not in features:
features.append(index)
for index in features: # 在 n_features 個特征中選出最優的特征索引，並沒有遍歷所有特征，從而保證了每課決策樹的差異性
for row in dataset:
groups = test_split(index, row[index], dataset) # groups=(left, right), row[index] 遍歷每一行 index 索引下的特征值作為分類值 value, 找出最優的分類特征和特征值
gini = gini_index(groups, class_values)
# 左右兩邊的數量越一樣，說明數據區分度不高，gini系數越大
if gini < b_score:
b_index, b_value, b_score, b_groups = index, row[index], gini, groups # 最后得到最優的分類特征 b_index,分類特征值 b_value,分類結果 b_groups。b_value 為分錯的代價成本
# print b_score
return {'index': b_index, 'value': b_value, 'groups': b_groups}
# Create a terminal node value # 輸出group中出現次數較多的標簽
def to_terminal(group):
outcomes = [row[-1] for row in group] # max() 函數中，當 key 參數不為空時，就以 key 的函數對象為判斷的標准
return max(set(outcomes), key=outcomes.count) # 輸出 group 中出現次數較多的標簽
# Create child splits for a node or make terminal # 創建子分割器，遞歸分類，直到分類結束
def split(node, max_depth, min_size, n_features, depth): # max_depth = 10, min_size = 1, n_features=int(sqrt((len(dataset[0])-1)
left, right = node['groups']
del(node['groups'])
# check for a no split 類別完全相同，不用划分了，直接返回類別中最多的那個類
if not left or not right:
node['left'] = node['right'] = to_terminal(left + right)
return
# check for max depth
if depth >= max_depth: # max_depth=10 表示遞歸十次，若分類還未結束，則選取數據中分類標簽較多的作為結果，使分類提前結束，防止過擬合
node['left'], node['right'] = to_terminal(left), to_terminal(right)
return
# process left child
if len(left) <= min_size:
node['left'] = to_terminal(left)
else:
node['left'] = get_split(left, n_features) # node['left']是一個字典，形式為{'index':b_index, 'value':b_value, 'groups':b_groups}，所以node是一個多層字典
split(node['left'], max_depth, min_size, n_features, depth+1) # 遞歸，depth+1計算遞歸層數
# process right child
if len(right) <= min_size:
node['right'] = to_terminal(right)
else:
node['right'] = get_split(right, n_features)
split(node['right'], max_depth, min_size, n_features, depth+1)
# Build a decision tree
def build_tree(train, max_depth, min_size, n_features):
"""build_tree(創建一個決策樹)
Args:
train 訓練數據集
max_depth 決策樹深度不能太深，不然容易導致過擬合
min_size 葉子節點的大小
n_features 選取的特征的個數
Returns:
root 返回決策樹
"""
# 返回最優列和相關的信息
root = get_split(train, n_features)
# 對左右2邊的數據 進行遞歸的調用，由於最優特征使用過，所以在后面進行使用的時候，就沒有意義了
# 例如： 性別-男女，對男使用這一特征就沒任何意義了
split(root, max_depth, min_size, n_features, 1)
return root
# Make a prediction with a decision tree
def predict(node, row): # 預測模型分類結果
if row[node['index']] < node['value']:
if isinstance(node['left'], dict): # isinstance 是 Python 中的一個內建函數。是用來判斷一個對象是否是一個已知的類型。
return predict(node['left'], row)
else:
return node['left']
else:
if isinstance(node['right'], dict):
return predict(node['right'], row)
else:
return node['right']
# Make a prediction with a list of bagged trees
def bagging_predict(trees, row):
"""bagging_predict(bagging預測)
Args:
trees 決策樹的集合
row 測試數據集的每一行數據
Returns:
返回隨機森林中，決策樹結果出現次數做大的
"""
# 使用多個決策樹trees對測試集test的第row行進行預測，再使用簡單投票法判斷出該行所屬分類
predictions = [predict(tree, row) for tree in trees]
return max(set(predictions), key=predictions.count)
# Create a random subsample from the dataset with replacement
def subsample(dataset, ratio): # 創建數據集的隨機子樣本
"""random_forest(評估算法性能，返回模型得分)
Args:
dataset 訓練數據集
ratio 訓練數據集的樣本比例
Returns:
sample 隨機抽樣的訓練樣本
"""
sample = list()
# 訓練樣本的按比例抽樣。
# round() 方法返回浮點數x的四舍五入值。
n_sample = round(len(dataset) * ratio)
while len(sample) < n_sample:
# 有放回的隨機采樣，有一些樣本被重復采樣，從而在訓練集中多次出現，有的則從未在訓練集中出現，此則自助采樣法。從而保證每棵決策樹訓練集的差異性
index = randrange(len(dataset))
sample.append(dataset[index])
return sample
# Random Forest Algorithm
def random_forest(train, test, max_depth, min_size, sample_size, n_trees, n_features):
"""random_forest(評估算法性能，返回模型得分)
Args:
train 訓練數據集
test 測試數據集
max_depth 決策樹深度不能太深，不然容易導致過擬合
min_size 葉子節點的大小
sample_size 訓練數據集的樣本比例
n_trees 決策樹的個數
n_features 選取的特征的個數
Returns:
predictions 每一行的預測結果，bagging 預測最后的分類結果
"""
trees = list()
# n_trees 表示決策樹的數量
for i in range(n_trees):
# 隨機抽樣的訓練樣本， 隨機采樣保證了每棵決策樹訓練集的差異性
sample = subsample(train, sample_size)
# 創建一個決策樹
tree = build_tree(sample, max_depth, min_size, n_features)
trees.append(tree)
# 每一行的預測結果，bagging 預測最后的分類結果
predictions = [bagging_predict(trees, row) for row in test]
return predictions
# Calculate accuracy percentage
def accuracy_metric(actual, predicted): # 導入實際值和預測值，計算精確度
correct = 0
for i in range(len(actual)):
if actual[i] == predicted[i]:
correct += 1
return correct / float(len(actual)) * 100.0
# 評估算法性能，返回模型得分
def evaluate_algorithm(dataset, algorithm, n_folds, *args):
"""evaluate_algorithm(評估算法性能，返回模型得分)
Args:
dataset 原始數據集
algorithm 使用的算法
n_folds 數據的份數
*args 其他的參數
Returns:
scores 模型得分
"""
# 將數據集進行抽重抽樣 n_folds 份，數據可以重復重復抽取，每一次 list 的元素是無重復的
folds = cross_validation_split(dataset, n_folds)
scores = list()
# 每次循環從 folds 從取出一個 fold 作為測試集，其余作為訓練集，遍歷整個 folds ，實現交叉驗證
for fold in folds:
train_set = list(folds)
train_set.remove(fold)
train_set = sum(train_set, [])
test_set = list()
# fold 表示從原始數據集 dataset 提取出來的測試集
for row in fold:
row_copy = list(row)
row_copy[-1] = None
test_set.append(row_copy)
predicted = algorithm(train_set, test_set, *args)
actual = [row[-1] for row in fold]
# 計算隨機森林的預測結果的正確率
accuracy = accuracy_metric(actual, predicted)
scores.append(accuracy)
return scores
if __name__ == '__main__':
# 加載數據
dataset = loadDataSet('./data/sonar_data.txt')
# print dataset
n_folds = 5 # 分成5份數據，進行交叉驗證
max_depth = 20 # 調參（自己修改） #決策樹深度不能太深，不然容易導致過擬合
min_size = 1 # 決策樹的葉子節點最少的元素數量
sample_size = 1.0 # 做決策樹時候的樣本的比例
# n_features = int((len(dataset[0])-1))
n_features = 15 # 調參（自己修改） #准確性與多樣性之間的權衡
for n_trees in [1,10]: # 理論上樹是越多越好
scores = evaluate_algorithm(dataset, random_forest, n_folds, max_depth, min_size, sample_size, n_trees, n_features)
# 每一次執行本文件時都能產生同一個隨機數
seed(1)
print('random=', random())
print('Trees: %d' % n_trees)
print('Scores: %s' % scores)
print('Mean Accuracy: %.3f%%' % (sum(scores)/float(len(scores))))
運行結果：
random= 0.13436424411240122 Trees: 1 Scores: [90.47619047619048, 80.95238095238095, 69.04761904761905, 92.85714285714286, 61.904761904761905] Mean Accuracy: 79.048% random= 0.13436424411240122 Trees: 10 Scores: [90.47619047619048, 78.57142857142857, 83.33333333333334, 90.47619047619048, 85.71428571428571] Mean Accuracy: 85.714%