作者：chen_h
微信號 & QQ：862251340
微信公眾號：coderpai
（一）機器學習中的集成學習入門
（二）bagging 方法
（三）使用Python進行交易的隨機森林算法
（四）Python中隨機森林的實現與解釋
（五）如何用 Python 從頭開始實現 Bagging 算法
決策樹是一種簡單而強大的預測建模技術，但它們存在高方差。這意味著在給定不同的訓練數據的情況下，樹可以得到非常不同的結果。為了使決策樹更加健壯并實現更好性能，我們會采用集成學習方法，其中一種是 Bagging 方法。
在本教程中，您將了解如何使用 Python從頭開始使用決策樹的 bagging 過程。完成本教程后，您將了解：
- 如何創建數據集的自舉過程；
- 如何使用自舉模型進行預測；
- 如何將 bagging 算法應用到你的預測模型中；
Bootstrap Aggregation 算法
Bootstrap 是一種有放回的數據采集方式。這還意味著一個新的數據集是從原來數據中進行隨機采用得到的，并且會把數據進行放回，然后進行下一次采樣。
當我們在估算一個非常龐大的數據集的時候，這種估算方式是非常好的。我們可以通過計算一個有限集合的均值從而來得到整個數據集的均值。這種方法我們一般都是和一些具有高方差的算法一起使用，比如決策樹。我們通過對每個自舉樣本進行單獨模型計算，然后輸出多個模型結果的平均值。這種技術稱為 bootstrap 或者 bagging。
方差意味著算法的性能對訓練數據敏感，高方差表明訓練數據的變化越多，算法的性能就越差。我們可以通過訓練許多樹并且取其預測的平均值，可以改善諸如未修剪的決策樹之類的高方差機器學習算法的性能。模型取得的結果通常會優于單個決策樹的表現。
除了提高性能之外，bagging 的另一個好處是它不會過度擬合問題，我們可以通過繼續添加樹木，知道達到最佳性能。
Sonar 數據集
在本教程中我們使用的是 Sonar 數據集。這是一個描述聲吶信號從不同表面反彈的數據集。輸入數據是由 60 個特征數據組成的，輸出數據是一個二分類，來判斷物體表面是巖石還是金屬圓柱。數據一共有 208 條。這是一個非常簡單的數據集。所有的輸入變量都是連續的，取值在 0 到 1 之間。輸出變量是 M（金屬圓柱） 和 R（巖石），我們需要將這個分類結果轉變成 1 和 0。數據我們通過 UCI Machine Learing 進行下載。下載鏈接：https://archive.ics.uci.edu/ml/datasets/Connectionist+Bench+(Sonar,+Mines+vs.+Rocks)
實戰例子
本教程分為兩部分：
- Bootstrap 采樣；
- 聲吶數據分析；
這些步驟提供了數據采樣和算法編寫的基本功能，我們可以學習bagging算法是如何進行基礎工作的。
1. Bootstrap 采樣
讓我們首先深入了解 bootstrap 方法的工作原理。
我們可以通過從數據集中隨機選擇行數據，并將它們添加到新列表來創建數據集成為新樣本。我們可以針對固定數量的行重復進行此操作，或者知道新數據集的大小與原始數據集的大小的比率達到我們的要求。我們每采集一次數據，都會進行放回，然后再次采集。
下面是一個名為 subsample() 的函數，它實現了這個過程。隨機模塊中的 randrange() 函數用于選擇隨機行索引，以便在循環的每次迭代中添加到樣本中。樣本的默認數量大小是原始數據集的大小。
def
subsample
(
dataset
,
ratio
=
1.0
)
:
sample
=
list
(
)
n_sample
=
round
(
len
(
dataset
)
*
ratio
)
while
len
(
sample
)
<
n_sample
:
index
=
randrange
(
len
(
dataset
)
)
sample
.
append
(
dataset
[
index
]
)
return
sample
我們可以使用這個函數來評估一個人造的數據集的平均值。
首先，我們創建一個包含 20 行，里面的數字時 0 到 9 之間的隨機值，并且我們計算他們的平均值。
然后，我們可以制作原始數據集的自舉樣本集，我們不斷重復這個過程，直到我們有一個均值列表，然后計算平均值。這個平均值跟我們整個樣本的平均值是非常接近的。
下面列出了一個完整的示例。
每個自舉樣本是原始樣本的 10 %，也就是 2 個樣本。然后，我們通過創建原始數據集的 1個，10個，100個自舉樣本，計算他們的平均值，然后平均所有這些估計的平均值來進行實驗。
from
random
import
seed
from
random
import
random
from
random
import
randrange
# Create a random subsample from the dataset with replacement
def
subsample
(
dataset
,
ratio
=
1.0
)
:
sample
=
list
(
)
n_sample
=
round
(
len
(
dataset
)
*
ratio
)
while
len
(
sample
)
<
n_sample
:
index
=
randrange
(
len
(
dataset
)
)
sample
.
append
(
dataset
[
index
]
)
return
sample
# Calculate the mean of a list of numbers
def
mean
(
numbers
)
:
return
sum
(
numbers
)
/
float
(
len
(
numbers
)
)
seed
(
1
)
# True mean
dataset
=
[
[
randrange
(
10
)
]
for
i
in
range
(
20
)
]
print
(
'True Mean: %.3f'
%
mean
(
[
row
[
0
]
for
row
in
dataset
]
)
)
# Estimated means
ratio
=
0.10
for
size
in
[
1
,
10
,
100
]
:
sample_means
=
list
(
)
for
i
in
range
(
size
)
:
sample
=
subsample
(
dataset
,
ratio
)
sample_mean
=
mean
(
[
row
[
0
]
for
row
in
sample
]
)
sample_means
.
append
(
sample_mean
)
print
(
'Samples=%d, Estimated Mean: %.3f'
%
(
size
,
mean
(
sample_means
)
)
)
運行該示例將打印我們要估計的原始數據平均值。
然后我們可以從各種不同數量的自舉樣本中看到估計的平均值。我們可以看到，通過 100 個樣本，我們可以很好的估計平均值。
True
Mean
:
4.450
Samples
=
1
,
Estimated Mean
:
4.500
Samples
=
10
,
Estimated Mean
:
3.300
Samples
=
100
,
Estimated Mean
:
4.480
我們可以為每個子樣本創建一個模型，而不是簡單的計算平均值。
接下來，讓我們看看如何組合多個 bootstrap 模型的預測。
2. 聲吶數據集案例研究
在這一節中，我們將隨機森林算法應用于聲吶數據集。
首先，我們需要導入數據集，將字符串值轉換為數值型，并將輸出列從字符串轉換為 0 和 1 的整數值。這是通過輔助函數 load_csv() ，str_column_to_float() 和 str_column_to_int() 來實現的，以便預處理數據集。
我們將使用 k-fold 交叉驗證來估計學習模型在未知數據上的性能。這意味著我們將構建和評估 k 個模型，并將性能估計為平均模型誤差。分類精度將評估每個模型，這些算法都在 cross_validation_split() ，accuracy_metric() 和 evaluate_algoritm() 函數中得到解決。
我們使用 CART 算法來實現我們的 bagging 過程，在實現的過程中我們還設計了一些輔助函數：test_split() 函數將數據集拆分成組，gini_index() 用于評估拆分點，get_split() 用于查找最佳拆分點，to_terminal()，split() 和 build_tree() 用于創建單個決策樹，predict() 用于使用決策樹進行預測，并使用前一步驟中描述的 subsample() 函數來創建訓練的子樣本訓練集。
我們還開發了一個 bagging_predict() 函數，該函數負責使用每個決策樹進行預測并將預測組合成單個返回值。這是 bagging 方法常用的一種模式。
最后，我們設計一個新的 bagging() 函數，負責創建訓練數據集的樣本，在每個樣本上訓練決策樹，然后使用bagging() 列表對測試數據集進行預測。
下面給出了完整代碼：
# Bagging Algorithm on the Sonar dataset
from
random
import
seed
from
random
import
randrange
from
csv
import
reader
# Load a CSV file
def
load_csv
(
filename
)
:
dataset
=
list
(
)
with
open
(
filename
,
'r'
)
as
file
:
csv_reader
=
reader
(
file
)
for
row
in
csv_reader
:
if
not
row
:
continue
dataset
.
append
(
row
)
return
dataset
# Convert string column to float
def
str_column_to_float
(
dataset
,
column
)
:
for
row
in
dataset
:
row
[
column
]
=
float
(
row
[
column
]
.
strip
(
)
)
# Convert string column to integer
def
str_column_to_int
(
dataset
,
column
)
:
class_values
=
[
row
[
column
]
for
row
in
dataset
]
unique
=
set
(
class_values
)
lookup
=
dict
(
)
for
i
,
value
in
enumerate
(
unique
)
:
lookup
[
value
]
=
i
for
row
in
dataset
:
row
[
column
]
=
lookup
[
row
[
column
]
]
return
lookup
# Split a dataset into k folds
def
cross_validation_split
(
dataset
,
n_folds
)
:
dataset_split
=
list
(
)
dataset_copy
=
list
(
dataset
)
fold_size
=
int
(
len
(
dataset
)
/
n_folds
)
for
i
in
range
(
n_folds
)
:
fold
=
list
(
)
while
len
(
fold
)
<
fold_size
:
index
=
randrange
(
len
(
dataset_copy
)
)
fold
.
append
(
dataset_copy
.
pop
(
index
)
)
dataset_split
.
append
(
fold
)
return
dataset_split
# 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
# Evaluate an algorithm using a cross validation split
def
evaluate_algorithm
(
dataset
,
algorithm
,
n_folds
,
*
args
)
:
folds
=
cross_validation_split
(
dataset
,
n_folds
)
scores
=
list
(
)
for
fold
in
folds
:
train_set
=
list
(
folds
)
train_set
.
remove
(
fold
)
train_set
=
sum
(
train_set
,
[
]
)
test_set
=
list
(
)
for
row
in
fold
:
row_copy
=
list
(
row
)
test_set
.
append
(
row_copy
)
row_copy
[
-
1
]
=
None
predicted
=
algorithm
(
train_set
,
test_set
,
*
args
)
actual
=
[
row
[
-
1
]
for
row
in
fold
]
accuracy
=
accuracy_metric
(
actual
,
predicted
)
scores
.
append
(
accuracy
)
return
scores
# 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
# Calculate the Gini index for a split dataset
def
gini_index
(
groups
,
classes
)
:
# count all samples at split point
n_instances
=
float
(
sum
(
[
len
(
group
)
for
group
in
groups
]
)
)
# sum weighted Gini index for each group
gini
=
0.0
for
group
in
groups
:
size
=
float
(
len
(
group
)
)
# avoid divide by zero
if
size
==
0
:
continue
score
=
0.0
# score the group based on the score for each class
for
class_val
in
classes
:
p
=
[
row
[
-
1
]
for
row
in
group
]
.
count
(
class_val
)
/
size
score
+=
p
*
p
# weight the group score by its relative size
gini
+=
(
1.0
-
score
)
*
(
size
/
n_instances
)
return
gini
# Select the best split point for a dataset
def
get_split
(
dataset
)
:
class_values
=
list
(
set
(
row
[
-
1
]
for
row
in
dataset
)
)
b_index
,
b_value
,
b_score
,
b_groups
=
999
,
999
,
999
,
None
for
index
in
range
(
len
(
dataset
[
0
]
)
-
1
)
:
for
row
in
dataset
:
# for i in range(len(dataset)):
# row = dataset[randrange(len(dataset))]
groups
=
test_split
(
index
,
row
[
index
]
,
dataset
)
gini
=
gini_index
(
groups
,
class_values
)
if
gini
<
b_score
:
b_index
,
b_value
,
b_score
,
b_groups
=
index
,
row
[
index
]
,
gini
,
groups
return
{
'index'
:
b_index
,
'value'
:
b_value
,
'groups'
:
b_groups
}
# Create a terminal node value
def
to_terminal
(
group
)
:
outcomes
=
[
row
[
-
1
]
for
row
in
group
]
return
max
(
set
(
outcomes
)
,
key
=
outcomes
.
count
)
# Create child splits for a node or make terminal
def
split
(
node
,
max_depth
,
min_size
,
depth
)
:
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
:
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
)
split
(
node
[
'left'
]
,
max_depth
,
min_size
,
depth
+
1
)
# process right child
if
len
(
right
)
<=
min_size
:
node
[
'right'
]
=
to_terminal
(
right
)
else
:
node
[
'right'
]
=
get_split
(
right
)
split
(
node
[
'right'
]
,
max_depth
,
min_size
,
depth
+
1
)
# Build a decision tree
def
build_tree
(
train
,
max_depth
,
min_size
)
:
root
=
get_split
(
train
)
split
(
root
,
max_depth
,
min_size
,
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
)
:
return
predict
(
node
[
'left'
]
,
row
)
else
:
return
node
[
'left'
]
else
:
if
isinstance
(
node
[
'right'
]
,
dict
)
:
return
predict
(
node
[
'right'
]
,
row
)
else
:
return
node
[
'right'
]
# Create a random subsample from the dataset with replacement
def
subsample
(
dataset
,
ratio
)
:
sample
=
list
(
)
n_sample
=
round
(
len
(
dataset
)
*
ratio
)
while
len
(
sample
)
<
n_sample
:
index
=
randrange
(
len
(
dataset
)
)
sample
.
append
(
dataset
[
index
]
)
return
sample
# Make a prediction with a list of bagged trees
def
bagging_predict
(
trees
,
row
)
:
predictions
=
[
predict
(
tree
,
row
)
for
tree
in
trees
]
return
max
(
set
(
predictions
)
,
key
=
predictions
.
count
)
# Bootstrap Aggregation Algorithm
def
bagging
(
train
,
test
,
max_depth
,
min_size
,
sample_size
,
n_trees
)
:
trees
=
list
(
)
for
i
in
range
(
n_trees
)
:
sample
=
subsample
(
train
,
sample_size
)
tree
=
build_tree
(
sample
,
max_depth
,
min_size
)
trees
.
append
(
tree
)
predictions
=
[
bagging_predict
(
trees
,
row
)
for
row
in
test
]
return
(
predictions
)
# Test bagging on the sonar dataset
seed
(
1
)
# load and prepare data
filename
=
'sonar.all-data.csv'
dataset
=
load_csv
(
filename
)
# convert string attributes to integers
for
i
in
range
(
len
(
dataset
[
0
]
)
-
1
)
:
str_column_to_float
(
dataset
,
i
)
# convert class column to integers
str_column_to_int
(
dataset
,
len
(
dataset
[
0
]
)
-
1
)
# evaluate algorithm
n_folds
=
5
max_depth
=
6
min_size
=
2
sample_size
=
0.50
for
n_trees
in
[
1
,
5
,
10
,
50
]
:
scores
=
evaluate_algorithm
(
dataset
,
bagging
,
n_folds
,
max_depth
,
min_size
,
sample_size
,
n_trees
)
print
(
'Trees: %d'
%
n_trees
)
print
(
'Scores: %s'
%
scores
)
print
(
'Mean Accuracy: %.3f%%'
%
(
sum
(
scores
)
/
float
(
len
(
scores
)
)
)
)
k 值為 5 時用于交叉驗證，每次迭代評估的數據量為 208/5 = 41.6 或者直接使用 40 條進行循環迭代。
構建樹的最大深度，我們設為 6，每個節點為 2 的最小訓練行數。訓練數據集的樣本創建為原始數據集大小的 50% 。這是為了強制用于訓練每棵樹的訓練集子樣本中的某些變體。bagging 的默認設置是使樣本數據集的大小與原始訓練數據集的大小相匹配。
接下來我們打印每個類別的結果：
Trees
:
1
Scores
:
[
87.8048780487805
,
65.85365853658537
,
65.85365853658537
,
65.85365853658537
,
73.17073170731707
]
Mean Accuracy
:
71.707
%
Trees
:
5
Scores
:
[
60.97560975609756
,
80.48780487804879
,
78.04878048780488
,
82.92682926829268
,
63.41463414634146
]
Mean Accuracy
:
73.171
%
Trees
:
10
Scores
:
[
60.97560975609756
,
73.17073170731707
,
82.92682926829268
,
80.48780487804879
,
68.29268292682927
]
Mean Accuracy
:
73.171
%
Trees
:
50
Scores
:
[
63.41463414634146
,
75.60975609756098
,
80.48780487804879
,
75.60975609756098
,
85.36585365853658
]
Mean Accuracy
:
76.098
%
這種方法的一個難點是，即使我們構建了一定深度的樹，但是 bagging 樹得到的結果也是非常相似的。但是我們希望在訓練的過程中可以降低高方差。這是因為我們在構造的過程中選擇了相同或者相似的分裂節點，這是一種貪婪算法。
本教程試圖通過約束用于訓練每棵樹的樣本大小來重新計算方差。更強大的技術是約束在創建每個分割點時對特征的評估。這是隨機森林中使用的方法。
擴展
-
調整樹：調整樹的大小，深度，以及單個樹的配置；
-
bagging 中構建不同的樹結構：我們可以通過使用不同的算法進行平均預測，比如貝葉斯，決策樹，神經網絡等等；
更多文章、技術交流、商務合作、聯系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號聯系： 360901061
您的支持是博主寫作最大的動力，如果您喜歡我的文章，感覺我的文章對您有幫助，請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點擊下面給點支持吧，站長非常感激您！手機微信長按不能支付解決辦法：請將微信支付二維碼保存到相冊，切換到微信，然后點擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對您有幫助就好】元
