---

前言

---

本文是对西瓜书中决策树章节的一个总结，将核心内容整理出来，帮助大家在短时间内快速建立一颗决策树。本文不侧重于帮大家理解，只是做一个知识点的总结，做一个“把书读薄”的工作。

---

1. 决策树结构

---

• 一棵决策树，有一个根节点，若干个内部节点，若干个子节点；
• 叶子结点对应于决策结果，其他每个节点都对应一个属性测试；
• 每个节点的样本集合，根据属性测试的结果被划分到了子节点中；
• 根节点包含样本全集。

---

2. 决策树的生成（注意区分属性和类别）

---

1. 生成过程：

• 一个递归过程，对于一个节点，每次选择一个最优的划分属性，划分出n个子节点。

• 有三种情况需要递归返回，并将当前的节点设为叶子结点，不再划分：

  • 当前节点包含的样本全属于同一类别，无需再分，将其判别结果设置为当前类别；
  • 当前属性集为空（属性全选一遍了，没有可选属性了），或 所有样本在属性上取值相同，将其判别结果设为该节点所含样本最多的类别；
  • 当前节点包含的样本集为空，将其判别结果设置为其父节点所含样本最多的类别。

2. 核心任务：

• 不难发现，生成过程中的一个核心任务，就是如何找到最优划分属性。

---

3. 划分选择

---

3.1 信息熵和信息增益

---

1. 信息熵（information entropy）

• 度量样本集合纯度的一种指标，假设当前样本集合 $D$ 中第 $K$ 类所含样本比例为 $P_k(k=1,2,...,|y|)$，则 $D$ 的信息熵为：
  $$Ent(D)=-\sum _{k=1}^{|y|}{P}_{k}lo{g}_2{P}_k$$
  $Ent(D)$ 的值越小，$D$ 的纯度越高（熵表示混乱程度，当然是越小越纯）。
• 注意：信息熵是对样本，对节点的概念。

2. 信息增益：

• 选择划分属性的依据，信息增益越大，则使用该属性进行划分得到的“纯度提升”越大。假设属性 $a$ 有 $V$ 个可能值 { a 1 , a 2 , . . . , a V } \{a^1,a^2,...,a^V\} {a1,a2,...,aV} （西瓜的颜色，就是一个属性，青绿和黄，就分别是两个取值）。用 $a$ 对样本集 $D$ 划分会产生 $V$ 个分支节点，其中第 $v$ 个分支包含了 $D$ 中所有在属性 $a$ 上取值为 $a^v$ 的样本，记为 $D^v$（$D^v$是第$v$个分支节点）。考虑到不同分支节点样本数不同，现对不同分支节点赋予权重 $|D^v|/|D|$（$D^v$节点上的样本数比上$D$节点中的样本数）。
• 可计算出用属性 $a$ 对样本集 $D$ 划分所得的“信息增益”（information gain）：
  $$Gain(D,a)=Ent(D)-\sum _{v=1}^V\frac{|D^v|}{|D|}Ent(D^v)$$
  父节点的信息熵 - 子节点信息熵带权加和。
• 信息增益是对属性的概念，选择信息增益最大的属性作为划分属性。
• ID3决策树就是以信息增益为准则来选择划分属性的，其中A是属性集合：
  $$a_{\ast }=\underset{a\in A}{argmaxGain(D,a)}$$

3. 计算信息熵的代码演示：

• 其中data_set是样本集合，包含目标值和特征值，不含标签。数据类型为list。

# 计算熵 -- 节点层面的概念，且只和目标值有关，和属性无关
def calcEntropy(data_set):# 1. 获取所有的样本数example_num = len(data_set)# 2. 计算每个类别出现的次数label_count = {}   # 一个字典for feat_vec in data_set:cur_label = feat_vec[-1] # 当前目标值if cur_label in label_count.keys():label_count[cur_label] += 1else:label_count[cur_label] = 1# 3. 计算熵值(对每个类别求熵值求和)entropy = 0   # 熵for key, value in label_count.items():# 每一个类别的概率值，即所占比例p = label_count[key] / example_num# 计算信息熵entropy += (-p * math.log(p, 2))return entropy

4. 根据信息增益选择划分属性的代码实现：

• feature_index是选中的特征索引，value是选中的特征值。

# 根据当前选中的属性和属性值去划分数据集
def splitDataSet(data_set, feature_index, value):ret_dataset = [] # 二维列表for feat_vec in data_set:if feat_vec[feature_index] == value:# 将 feature_index 那一列删除delete_feat_vec = feat_vec[:feature_index]delete_feat_vec.extend(feat_vec[feature_index + 1:])# 将删除后的样本追加到新的 data_set 中ret_dataset.append(delete_feat_vec)return ret_dataset# 选择最优划分属性，返回最优划分属性的索引
def chooseBestFeatureByEntropy(data_set):# 属性可取值数量(也可以把属性称为特征)num_features = len(data_set[0]) - 1# 计算该节点的熵(未划分时的熵值)cur_entropy = calcEntropy(data_set)# 信息增益best_info_gain = 0.0# 最优属性索引best_feature_index = -1# 遍历所有属性for i in range(num_features):# 拿到当前列的属性feat_list = [example[i] for example in data_set]# 获取唯一值unique_val = set(feat_list)# 分支节点信息熵带权加和sum_child_entropy = 0# 计算各分支(不同属性划分)的熵值for val in unique_val:# 根据当前属性划分 data_set, 得到分支节点 sub_data_setsub_data_set = splitDataSet(data_set, i, val)# 计算该节点权重weight = len(sub_data_set) / len(data_set)# 计算分支节点信息熵带权加和sum_child_entropy += (calcEntropy(sub_data_set) * weight)# 计算信息增益info_gain = cur_entropy - sum_child_entropy# 更新最大信息增益if best_info_gain < info_gain:best_info_gain = info_gainbest_feature_index = ireturn best_feature_index

---

3.2 增益率

---

1. 信息增益的弊端：

• 信息增益准则对值比较多的属性有所偏好；
• 为减少这种偏好可能带来的不利影响，C4.5决策树使用“增益率”来选择最优划分属性。

2. 定义：

$$Gain\_ratio(D,a)=\frac{Gain(D,a)}{IV(a)}$$

• 其中
  $$IV(a)=-\sum _{v=1}^V\frac{|D^v|}{|D|}{\log }_2\frac{|D^v|}{|D|}$$
  $IV(a)$ 是 $a$ 的“固有值”（intrinsic value），$a$ 的值越多，$IV(a)$ 越大。

3. 注意：

• 增益率准则对值较少的属性有所偏好，因此C4.5会先找出信息增益高于平均水平的属性，再从中选择增益率最高的属性。

---

3.3 基尼指数（鸡你指数）

---

1. 基尼值：
$$Gini(D)=\sum _{k=1}^{|y|}\sum _{k^{\prime }\ne k}{P}_k{P}_{k^{\prime }}=1-\sum _{k=1}^{|y|}{P}_k^2$$

• $Gini(D)$ 反映了从 $D$ 中随机取两个样本，其类别不一样的概率；
• $Gini(D)$ 越小，$D$ 的纯度越高；
• 基尼值也是对样本，对节点的概念。

2. 基尼指数：

$$Gini\_index(D,a)=\sum _{v=1}^V\frac{|D^v|}{|D|}Gini(D^v)$$

• CART决策树使用“基尼指数”来选择划分属性，选择 $Gini\_index(D,a)$ 最小的属性作为划分属性：
  $$a_{\ast }=\underset{a\in A}{argminGini\_index(D,a)}$$

3. 计算基尼值的代码实现：

def Gini(data_set):# 1. 获取所有的样本数example_num = len(data_set)# 2. 计算每个类别出现的次数label_count = {}  # 一个字典for feat_vec in data_set:cur_label = feat_vec[-1]  # 当前标签if cur_label in label_count.keys():label_count[cur_label] += 1else:label_count[cur_label] = 1# 3. 计算基尼值sum_p = 0for key, value in label_count.items():p = label_count[key] / example_num# 计算公式后面和的那一部分sum_p += p**2return 1 - sum_p

4. 根据基尼指数选择划分属性的代码实现：

# 根据当前选中的属性和属性值去划分数据集
def splitDataSet(data_set, feature_index, value):ret_dataset = [] # 二维列表for feat_vec in data_set:if feat_vec[feature_index] == value:# 将 feature_index 那一列删除delete_feat_vec = feat_vec[:feature_index]delete_feat_vec.extend(feat_vec[feature_index + 1:])# 将删除后的样本追加到新的 data_set 中ret_dataset.append(delete_feat_vec)return ret_dataset# 选择最优划分属性，返回最优划分属性的索引
def chooseBestFeatureByGini(data_set):# 属性可取值数量(也可以把属性称为特征)，列数num_features = len(data_set[0]) - 1# 当前基尼指数cur_gini_index = 0# 最优基尼指数best_gini_index = math.inf # 给最大值# 最优属性索引best_feature_index = -1for i in range(num_features):# 拿到当前列的属性feat_list = [example[i] for example in data_set]# 获取唯一值unique_val = set(feat_list)# 计算各分支的基尼值for val in unique_val:# 根据当前属性划分sub_data_set = splitDataSet(data_set, i, val)# 计算权重weight = len(sub_data_set) / len(data_set)# 计算分支基尼值，并加到基尼指数上cur_gini_index += weight * Gini(sub_data_set)# 更新最小的基尼指数if best_gini_index > cur_gini_index:best_gini_index = cur_gini_indexbest_feature_index = ireturn best_feature_index

---

4. 剪枝处理

• 剪枝是对付“过拟合”的主要手段，可以通过主动去掉一些分支来降低过拟合风险。

---

4.1 预剪枝

---

1. 原理：

• 在决策树生成过程中，若当前结点的划分不能带来决策树泛化性能的提升，则停止划分，并将当前节点标记为叶子结点。

2. 理解：

• 假定判断泛化性能是否提升采用留出法，将数据随机划分成训练集和验证集。若划分后的 验证集精度 $\le$ 划分前的验证集精度，就禁止划分。

3. 预剪枝的缺点：

• 有些分支的当前划分虽然不能提升泛化性能，甚至可能导致泛化性能暂时下降，但是在其基础上的后续划分却可能带来性能的显著提高；
• 预剪枝基于“贪心”禁止这些节点展开，给决策树带来了欠拟合的风险。

---

4.2 后剪枝

---

1. 原理：

• 后剪枝是先从训练集生成一颗完整的决策树，然后自底向上对非叶子结点进行考察，若将该节点对应的子树替换为叶子节点，能带来泛化性能的提升，则将该子树替换为叶子结点。

2. 理解：

• 假定判断泛化性能是否提升采用留出法，将数据随机划分成训练集和验证集。若 剪枝后的验证集精度 $\ge$ 剪枝前的验证集精度，就进行剪枝。

3. 优势和不足：

• 后剪枝决策树的欠拟合风险很小（因为其要自底向上对所有非叶子节点进行考察），泛化性能往往优于预剪枝决策树；
• 其训练时间开销比未剪枝决策树和预剪枝决策树大得多。

---

5. 连续值与缺失值处理

---

5.1 连续值处理

---

1. 连续值处理的必要性：

• 目前为止，我们只讨论了离散属性生成决策树，现实的学习任务中，常常会遇到连续属性。我们有必要对连续属性做离散化的处理。

2. 二分法处理连续属性：

• 给定样本集 $D$ 和连续属性 $a$ ，假设 $a$ 在 $D$ 上出现了 $n$ 个不同的取值，先将这些值从大到小排序，记为 { a 1 , a 2 , . . . , a n } \{a^1,a^2,...,a^n\} {a1,a2,...,an}，基于划分点 $t$ 可将 $D$ 分为子集 $D_t^{-}$ 和 $D_t^{+}$。其中 $D_t^{-}$包含那些在属性 $a$ 上取值不大于 $t$ 的样本，而 $D_t^{+}$ 则包含那些在属性 $a$ 上取值大于 $t$ 的样本。

• 对相邻的属性取值 $a^i$ 与 $a^{i+1}$ 来说，$t$ 在区间 $[a^i,a^i+1)$ 中取任意值所产生的划分结果相同。因此，对连续属性 $a$ ，我们可以得到一个划分点集合 $T_a$：
  T a = { a i + a i + 1 2 ∣ 1 ≤ i ≤ n − 1 } T_a=\{\frac{a^i+a^{i+1}}{2}|1\leq i \leq n-1\} Ta​={2ai+ai+1​∣1≤i≤n−1}

• 随后，我们就可以像考察离散属性一样来考察这些分类点，选择最优的划分点进行集合的划分：
  G a i n ( D , a ) = maxGain(D,a,t) ⁡ t ∈ T a = maxEnt(D) ⁡ t ∈ T a − ∑ λ ∈ { − , + } ∣ D t λ ∣ ∣ D ∣ E n t ( D t λ ) Gain(D,a)=\underset {t \in T_a} {\operatorname {maxGain(D,a,t)}}=\underset {t \in T_a} {\operatorname {maxEnt(D)}}-\sum \limits_{\lambda \in \{-,+\}}\frac{|D_t^\lambda|}{|D|}Ent(D^\lambda_t) Gain(D,a)=t∈Ta​maxGain(D,a,t)​=t∈Ta​maxEnt(D)​−λ∈{−,+}∑​∣D∣∣Dtλ​∣​Ent(Dtλ​)
  $Gain(D,a,t)$ 是样本集 $D$ 基于划分点 $t$ 二分后的信息增益。于是，我们可以选择使 $Gain(D,a,t)$ 最大化的划分点。

3. 注意：

• 与离散属性不同，若当前节点划分属性为连续值，则该属性还可能作为后续节点的划分属性（连续属性会被重复使用，只是每次的划分点可能不同）。

4. 连续值处理，代码示例

• 准备数据：

from math import log2
import pandas as pd
import numpy as npdata = pd.DataFrame({'Hours Studied': [1, 2, 3, 4, 5],'Grade': ['合格', '不合格', '不合格', '合格', '合格']
})

• 设计对应的熵计算函数：

# 计算节点的信息熵
def calculate_entropy(data, target):# 记录每个类别出现的次数values, counts = np.unique(data[target], return_counts=True)# 每个类别所占比例probs = counts / len(data)# 计算该样本集的熵entropy = -sum(p * log2(p) for p in probs if p > 0)return entropy

• 计算初始熵，并对样本进行排序：

# 计算初始熵
initial_entropy = calculate_entropy(data, 'Grade')
print(f"初始熵: {initial_entropy}")
"""
初始熵: 0.9709505944546686
"""# 对样本按连续值进行排序
sorted_data = data.sort_values('Hours Studied')
print(sorted_data)
"""Hours Studied Grade
0              1    合格
1              2   不合格
2              3   不合格
3              4    合格
4              5    合格
"""

• 求最佳划分点，以及以该属性的该划分点划分的信息增益：

def calculate_information_gain(data, attribute, target, current_entropy):""":param data: 排序后的样本集:param attribute: 连续属性的标签:param target: 目标值标签:param current_entropy: 当前熵:return: 返回最优划分点 best_split，和以该属性的该划分点划分的信息增益 max_gain"""values = data[attribute].unique() # 获取连续属性的唯一值max_gain = float('-inf') # 最大增益，初始化为负无穷best_split = None # 最优的划分属性for split in values:# 分割数据data_left = data[data[attribute] <= split]data_right = data[data[attribute] > split]# 计算信息增益gain = current_entropy - (len(data_left) / len(data)) * calculate_entropy(data_left, target) - (len(data_right) / len(data)) * calculate_entropy(data_right, target)# 更新最佳分割点if gain > max_gain:max_gain = gainbest_split = splitreturn best_split, max_gain

• 测试：

best_split, ig = calculate_information_gain(sorted_data, 'Hours Studied', 'Grade', initial_entropy)
print(f"最佳分割点: {best_split}, 信息增益: {ig}")
"""
最佳分割点: 3, 信息增益: 0.4199730940219749
"""

---

5.2 缺失值处理

---

1. 缺失值处理的必要性：

• 现实任务中经常会遇到不完整的样本，即样本的某些属性值缺失。尤其是在属性数目较多的情况下，往往会有大量样本出现缺失值，如果简单的放弃不完整样本，显然是对数据的极大浪费。

2. 缺失值处理要解决的两个问题：

• （1）如何在属性值缺失的情况下，进行划分属性选择？
• （2）给定划分属性，若样本在该属性上缺失，该如何对样本进行划分？

3. 解决方法：

• 给定训练集 $D$ 和属性 $a$ ，令 $\tilde{D}$ 表示 $D$ 中在属性 $a$ 上没有缺失值的样本子集。对问题（1），我们可以根据 $\tilde{D}$ 来判断属性 $a$ 的优劣。假定属性 $a$ 有 $V$ 个可能取值 { a 1 , a 2 , . . . , a V } \{a^1,a^2,...,a^V\} {a1,a2,...,aV}，令 ${\tilde{D}}_v$ 表示 $\tilde{D}$ 中在属性 $a$ 上取值为 $a^v$ 的样本子集，${\tilde{D}}_k$ 表示 $\tilde{D}$ 中属于第 $k$ 类（$k=1,2,...,|y|$）的样本子集。
• 假定给每个样本 $\mathbf{x}$ 赋予一个权重 $w_{\mathbf{x}}$，定义：
  • 无缺失样本所占比例：
    $$\rho =\frac{{\sum }_{\mathbf{x}\in \tilde{D}}{w}_{\mathbf{x}}}{{\sum }_{\mathbf{x}\in D}{w}_{\mathbf{x}}}$$
  • 无缺失样本中，第 $k$ 类所占比例：
    $$\widetilde{p_k}=\frac{{\sum }_{\mathbf{x}\in \widetilde{D_k}}{w}_{\mathbf{x}}}{{\sum }_{\mathbf{x}\in \tilde{D}}{w}_{\mathbf{x}}}$$
  • 无缺失样本中，在属性 $a$ 上取值 $a^v$ 的样本所占比例：
    $$\widetilde{r_v}=\frac{{\sum }_{\mathbf{x}\in \widetilde{D^v}}{w}_{\mathbf{x}}}{{\sum }_{\mathbf{x}\in \tilde{D}}{w}_{\mathbf{x}}}$$
• 基于上述定义，我们将信息增益计算式推广为：
  $$Gain(D,a)=\rho \times Gain(\tilde{D},a)=\rho \times (Ent(\tilde{D})-\sum _{v=1}^V\widetilde{r_v}Ent(\widetilde{D^v}))$$
• 其中：
  $$Ent(\tilde{D})=-\sum _{k=1}^{|y|}\widetilde{p_k}lo{g}_2\widetilde{p_k}$$
  根据信息增益就可以确定最优划分属性啦。
• 对于问题（2），若样本 $\mathbf{x}$ 在划分属性 $a$ 上有值（没有缺失），则将 $\mathbf{x}$ 划入与其对应的子节点，且样本权值在子节点中保持为 $w_{\mathbf{x}}$；若样本 $\mathbf{x}$ 在划分属性 $a$ 上值缺失，则将 $\mathbf{x}$ 同时划入所有子节点，且样本权值在与属性值 $a^v$ 对应的子节点中调整为 $\widetilde{r_v}\times w_{\mathbf{x}}$。直观的看，就是让同一个样本以不同的概率划入到不同的子节点中。

4. 缺失值处理代码实现（建议先看ID3完整实现代码）

缺失值处理较为复杂，这里几乎是把整个树又手撕了一遍，建议先理解最基础的ID3，再来看缺失值处理。

• 准备数据，这里我们使用DataFrame类型的数据：

def createDataSet():# 数据data_set = [[0, 0, 0, 0, 'no'],[0, 0, None, 1, 'no'],[0, 1, 0, 1, 'yes'],[None, 1, 1, 0, 'yes'],[0, 0, 0, 0, 'no'],[1, 0, 0, 0, 'no'],[None, 0, 0, 1, 'no'],[1, 1, 1, 1, 'yes'],[1, 0, 1, 2, 'yes'],[1, 0, 1, 2, 'yes'],[2, None, 1, 2, 'yes'],[2, 0, 1, 1, 'yes'],[2, 1, 0, 1, 'yes'],[2, 1, 0, None, 'yes'],[2, 0, 0, 0, 'no'],]# 属性名labels = ['F1-AGE', 'F2-WORK', 'F3-HOME', 'F4-LOAN', 'result']data = pd.DataFrame(data_set, columns=labels)return data

• 预处理：

# 预处理，给数据加一列权重，每一个样本权重初始化为1
def prepareDataSet(data):# 样本数num_data = len(data)weight = [] # 权重列表for _ in range(num_data):weight.append(1)data['weight'] = weightreturn data

• 划分数据集：

# 根据当前属性的某个属性值划分数据集
def splitDataSet(data, attribute, value, flag=False):""":param flag::param data: 数据集:param attribute: 属性标签:param value: 属性值:param flag: 是否更新缺失集权重，并将其按权放入子集(True表示进行该操作):return:"""sub_data = data.loc[data[attribute]==value]del sub_data[attribute]if flag:# 拿到缺失数据集missing_data = data.loc[data[attribute].isna()]# 拿到无缺失的数据集 D~no_missing_data = data.loc[data[attribute].notna()]del missing_data[attribute]# 无缺失样本中，该 sub_data 的比例，课本中的rv~p_sub_data = np.sum(sub_data['weight']) / np.sum(no_missing_data['weight'])# 更新权重weight = missing_data['weight'].unique()missing_data.loc[missing_data.index, 'weight'] = p_sub_data * weight# 将缺失数据集放入分支节点sub_data = pd.concat([sub_data, missing_data])return sub_data

• 计算节点信息熵：
  • 这里和普通的ID3决策树明显不同，每个类别的比例是按权重计算的，不再单单只看数量。

# 计算节点的信息熵
def calculateEntropy(data, target):values = data[target].unique() # 目标值列表weight = []for i in range(len(values)):data_k = data.loc[data[target] == values[i]] # 取出所有第k类的样本weight.append(np.sum(data_k['weight']))# 每个类别的所占比例，以列表形式存储，书中的pk~probs = weight / np.sum(data['weight'])# 计算该样本集的熵entropy = -sum(p * log2(p) for p in probs if p > 0)return entropy

• 计算信息增益：

# 计算选择该属性的信息增益
def calculateInformationGain(data, attribute, target):""":param data: 数据集:param attribute: 属性标签:param target: 目标值标签:return: 返回信息增益 gain，和缺失数据集 missing_data"""# 拿到无缺失的数据集 D~no_missing_data = data.loc[data[attribute].notna()]# 计算无缺失样本集所占比例，书中的roup_no_missing_data = np.sum(no_missing_data['weight']) / np.sum(data['weight'])# D~的信息熵current_entropy = calculateEntropy(no_missing_data, 'result')# 拿到该属性的所有属性值attribute_value_list = no_missing_data[attribute].unique()# 记录划分子集的信息熵带权加和p_sum_sub_data = 0for value in attribute_value_list:# 拿到属性值为 value 的子集sub_data = splitDataSet(no_missing_data.copy(), attribute, value)# 无缺失样本中，该 sub_data 的比例，课本中的rv~p_sub_data = np.sum(sub_data['weight']) / np.sum(no_missing_data['weight'])p_sum_sub_data += p_sub_data * calculateEntropy(sub_data, target)# 选择该属性的信息增益gain = p_no_missing_data * (current_entropy - p_sum_sub_data)return gain

• 选择最优划分属性：

# 选择最优划分属性
def chooseBestSplitAttribute(data, target):# 记录最大信息增益best_info_gain = 0# 记录最佳属性best_attribute = None# 属性列表(注意不是属性值列表)attribute_list = list(data.columns)attribute_list = attribute_list[:-2] # 最后两列不是属性，丢掉for attribute in attribute_list:cur_info_gain = calculateInformationGain(data, attribute, target)if best_info_gain < cur_info_gain:best_info_gain = cur_info_gainbest_attribute = attributereturn best_attribute

• 递归构建节点：

def isEmptyOrSameLabel(data):if data.empty:return Trueelse:data = data.values.tolist()data = [row[:-2] for row in data]for i in range(len(data)):if data[0] == data[i]:continueelse:return Falsereturn Truedef majorityCnt(class_list):return class_list.mode() # 返回最多的值# 递归生成决策树节点
def createTreeNode(data, target):# 取出当前节点的样本类别class_list = data[target]'''递归终止条件'''# 1. 当前节点包含的样本全属于同一类别，无需再分，将其判别结果设为当前类别if len(class_list.unique()) == 1:return class_list.sample().tolist()[0]# 2. 判断属性是否为空(已经按照所有的属性划分完了) + 判断所有样本在属性上的取值是否相同elif isEmptyOrSameLabel(data):return majorityCnt(class_list)  # 返回当前节点样本最多的类别'''属性划分'''# 1. 选择最好的属性进行划分best_attribute = chooseBestSplitAttribute(data, target)  # 以信息增益为划分准测# 3. 根据最优属性，和其属性值生成树(用字典模拟二叉树)my_tree = {best_attribute: {}}# 5. 获取当前最佳属性属性值的唯一值attribute_value_list = data[data[best_attribute].notna()][best_attribute].unique()# 7. 对每一个唯一值进行分支for value in attribute_value_list:# 递归创建树my_tree[best_attribute][value] = createTreeNode(splitDataSet(data.copy(), best_attribute, value, flag=True), target)# 8. 返回return my_tree

---

---

6. 模拟实现ID3决策树

---

1. 准备数据集：

# 创建数据
def createDataSet():# 数据data_set = [[0, 0, 0, 0, 'no'],[0, 0, 0, 1, 'no'],[0, 1, 0, 1, 'yes'],[0, 1, 1, 0, 'yes'],[0, 0, 0, 0, 'no'],[1, 0, 0, 0, 'no'],[1, 0, 0, 1, 'no'],[1, 1, 1, 1, 'yes'],[1, 0, 1, 2, 'yes'],[1, 0, 1, 2, 'yes'],[2, 0, 1, 2, 'yes'],[2, 0, 1, 1, 'yes'],[2, 1, 0, 1, 'yes'],[2, 1, 0, 2, 'yes'],[2, 0, 0, 0, 'no'],]# 属性名labels = ['F1-AGE', 'F2-WORK', 'F3-HOME', 'F4-LOAN']return data_set, labels

2. 信息熵的计算函数：

# 计算熵 -- 节点层面的概念，且只和目标值有关，和属性无关
def calcEntropy(data_set):# 1. 获取所有的样本数example_num = len(data_set)# 2. 计算每个类别出现的次数label_count = {}   # 一个字典for feat_vec in data_set:cur_label = feat_vec[-1] # 当前标签if cur_label in label_count.keys():label_count[cur_label] += 1else:label_count[cur_label] = 1# 3. 计算熵值(对每个类别求熵值求和)entropy = 0   # 熵for key, value in label_count.items():# 每一个类别的概率值，即所占比例p = label_count[key] / example_num# 计算信息熵entropy += (-p * math.log(p, 2))return entropy

3. 根据信息增益，选择最优划分属性

# 选择最优划分属性，返回最优划分属性的索引
def chooseBestFeatureByEntropy(data_set):# 属性可取值数量(也可以把属性称为特征)num_features = len(data_set[0]) - 1# 计算该节点的熵(未划分时的熵值)cur_entropy = calcEntropy(data_set)# 信息增益best_info_gain = 0.0# 最优属性索引best_feature_index = -1# 遍历所有属性for i in range(num_features):# 拿到当前列的属性feat_list = [example[i] for example in data_set]# 获取唯一值unique_val = set(feat_list)# 分支节点信息熵带权加和sum_child_entropy = 0# 计算各分支(不同属性划分)的熵值for val in unique_val:# 根据当前属性划分 data_set, 得到分支节点 sub_data_setsub_data_set = splitDataSet(data_set, i, val)# 计算该节点权重weight = len(sub_data_set) / len(data_set)# 计算分支节点信息熵带权加和sum_child_entropy += (calcEntropy(sub_data_set) * weight)# 计算信息增益info_gain = cur_entropy - sum_child_entropy# 更新最大信息增益if best_info_gain < info_gain:best_info_gain = info_gainbest_feature_index = ireturn best_feature_index

4. 根据当前节点样本最多的类别，获取判别结果：

# 获取判别结果，判别结果为样本数最多的类别
def majorityCnt(class_list):class_count = {}# 统计 class_list 中每个元素出现的次数for vote in class_list:if vote in class_count.keys():class_count[vote] += 1else:class_count[vote] = 1# 根据字典的值，降序排列sorted_class_count = sorted(class_count.items(), key=lambda x: x[1], reverse=True)return sorted_class_count[0][0]

5. 判断节点是否为空+所有样本在属性上取值是否相同：

# 判断属性是否为空 + 判断所有样本在属性上的取值是否相同
def isEmptyOrSameLabel(data_set, labels):# 属性为空，if len(labels) == 0:return Trueelse:for i in range(0, len(data_set)):# 从0开始，避免只有一个样本的情况if data_set[0] == data_set[i]:continueelse:return Falsereturn True

6. 递归生成决策树节点：

def createTreeNode(data_set, labels):# 取出当前节点的样本类别class_list = [example[-1] for example in data_set]'''递归终止条件'''# 1. 当前节点包含的样本全属于同一类别，无需再分，将其判别结果设为当前类别if len(class_list) == class_list.count(class_list[0]):return class_list[0]# 2. 判断属性是否为空(已经按照所有的属性划分完了) + 判断所有样本在属性上的取值是否相同if isEmptyOrSameLabel(data_set, labels):return majorityCnt(class_list)  # 返回当前节点样本最多的类别'''属性划分'''# 1. 选择最好的属性进行划分(返回值为索引)best_feature_index = chooseBestFeatureByEntropy(data_set)# 2. 利用索引获取真实值，找到最优划分属性best_feature_label = labels[best_feature_index]# 3. 根据最优属性，和其属性值生成树(用字典模拟二叉树)my_tree = {best_feature_label: {}}# 4. 删除被选择的属性del labels[best_feature_index]# 5. 获取当前最佳属性的那一列feature_values = [example[best_feature_index] for example in data_set]# 6. 去重unique_feature_values = set(feature_values)# 7. 对每一个唯一值进行分支for value in unique_feature_values:# 递归创建树my_tree[best_feature_label][value] = createTreeNode(splitDataSet(data_set, best_feature_index, value), labels.copy())# 8. 返回return my_tree

• 这里我们用字典模拟了一颗树，非常巧妙。

7. 未知样本的预测：

# 单个未知数据预测
def classifySingle(input_tree, feat_labels, test_data):first_str = next(iter(input_tree)) # 找到决策树在这一层的划分属性second_dict = input_tree[first_str] # 找到子树feat_index = feat_labels.index(first_str) # 找到对应属性所在下标class_label = None # 置空for key in second_dict.keys(): # 根据属性值，遍历子树，key是属性值if test_data[feat_index] == key: # 找到样本该属性的属性值和当前子树分类时的属性值相等的子树，进入该子树进行判断if type(second_dict[key]).__name__ == 'dict': # 如果拿到的值是字典类型，说明还没到叶子节点，接着递归class_label = classifySingle(second_dict[key], feat_labels, test_data)else:# 拿到的值不是字典类型，说明找到叶子节点了，记录分类结果，递归返回class_label = second_dict[key]return class_label# 多个未知数据预测，复用classify_single
def classifyMultiple(input_tree, feat_labels, test_data):result = []# 计算列表维度def list_dimension(lst):if not isinstance(lst, list):return 0  # 不是列表，返回0维if all(not isinstance(item, list) for item in lst):return 1  # 所有元素都不是列表，返回1维return 1 + max(list_dimension(item) for item in lst)  # 递归计算维度if list_dimension(test_data) == 1: # 维度为1，说明只有一个样本result.append(classifySingle(input_tree, feat_labels, test_data))return resultelse:num_example = len(test_data)for i in range(num_example):result.append(classifySingle(input_tree, feat_labels, test_data[i]))return result

8. 计算树的深度和广度（选择性学习）：

# 计算广度，一个树中叶子结点的数量
def getNumLeaves(my_tree):num_leaves = 0first_str = next(iter(my_tree)) # 取出字典第一层中，仅有的一个键，也相当于是根节点second_dict = my_tree[first_str] # 拿到子树，可能不只一颗子树，需要遍历for key in second_dict.keys(): # 遍历子树if type(second_dict[key]).__name__ == 'dict': # 判断该节点是否是字典，如果不是，代表此节点为叶子结点num_leaves += getNumLeaves(second_dict[key])else:num_leaves += 1return num_leaves# 获取树的深度
def _getTreeDepth(my_tree):max_depth = 0 # 初始化决策树深度first_str = next(iter(my_tree))second_dict = my_tree[first_str]for key in second_dict.keys():if type(second_dict[key]).__name__ == 'dict':this_depth = 1 + _getTreeDepth(second_dict[key])else:this_depth = 1if this_depth > max_depth:max_depth = this_depth # 更新层数return max_depthdef getTreeDepth(my_tree):return _getTreeDepth(my_tree) + 1

---

7. ID3决策树完整代码

---

• 决策树实现：

# 开发时间: 2024/6/5 14:11
import math# 创建数据
def createDataSet():# 数据data_set = [[0, 0, 0, 0, 'no'],[0, 0, 0, 1, 'no'],[0, 1, 0, 1, 'yes'],[0, 1, 1, 0, 'yes'],[0, 0, 0, 0, 'no'],[1, 0, 0, 0, 'no'],[1, 0, 0, 1, 'no'],[1, 1, 1, 1, 'yes'],[1, 0, 1, 2, 'yes'],[1, 0, 1, 2, 'yes'],[2, 0, 1, 2, 'yes'],[2, 0, 1, 1, 'yes'],[2, 1, 0, 1, 'yes'],[2, 1, 0, 2, 'yes'],[2, 0, 0, 0, 'no'],]# 属性名labels = ['F1-AGE', 'F2-WORK', 'F3-HOME', 'F4-LOAN']return data_set, labels# 计算熵 -- 节点层面的概念，且只和目标值有关，和属性无关
def calcEntropy(data_set):# 1. 获取所有的样本数example_num = len(data_set)# 2. 计算每个类别出现的次数label_count = {}   # 一个字典for feat_vec in data_set:cur_label = feat_vec[-1] # 当前标签if cur_label in label_count.keys():label_count[cur_label] += 1else:label_count[cur_label] = 1# 3. 计算熵值(对每个类别求熵值求和)entropy = 0   # 熵for key, value in label_count.items():# 每一个类别的概率值，即所占比例p = label_count[key] / example_num# 计算信息熵entropy += (-p * math.log(p, 2))return entropydef Gini(data_set):# 1. 获取所有的样本数example_num = len(data_set)# 2. 计算每个类别出现的次数label_count = {}  # 一个字典for feat_vec in data_set:cur_label = feat_vec[-1]  # 当前标签if cur_label in label_count.keys():label_count[cur_label] += 1else:label_count[cur_label] = 1# 3. 计算基尼值sum_p = 0for key, value in label_count.items():p = label_count[key] / example_num# 计算公式后面和的那一部分sum_p += p**2return 1 - sum_p# 根据当前选中的属性和属性值去划分数据集
def splitDataSet(data_set, feature_index, value):ret_dataset = [] # 二维列表for feat_vec in data_set:if feat_vec[feature_index] == value:# 将 feature_index 那一列删除delete_feat_vec = feat_vec[:feature_index]delete_feat_vec.extend(feat_vec[feature_index + 1:])# 将删除后的样本追加到新的 data_set 中ret_dataset.append(delete_feat_vec)return ret_dataset# 选择最优划分属性，返回最优划分属性的索引
def chooseBestFeatureByGini(data_set):# 属性可取值数量(也可以把属性称为特征)，列数num_features = len(data_set[0]) - 1# 当前基尼指数cur_gini_index = 0# 最优基尼指数best_gini_index = math.inf# 最优属性索引best_feature_index = -1for i in range(num_features):# 拿到当前列的属性feat_list = [example[i] for example in data_set]# 获取唯一值unique_val = set(feat_list)# 计算各分支的基尼值for val in unique_val:# 根据当前属性划分sub_data_set = splitDataSet(data_set, i, val)# 计算权重weight = len(sub_data_set) / len(data_set)# 计算分支基尼值，并加到基尼指数上cur_gini_index += weight * Gini(sub_data_set)# 更新最小的基尼指数if best_gini_index > cur_gini_index:best_gini_index = cur_gini_indexbest_feature_index = ireturn best_feature_index# 选择最优划分属性，返回最优划分属性的索引
def chooseBestFeatureByEntropy(data_set):# 属性可取值数量(也可以把属性称为特征)num_features = len(data_set[0]) - 1# 计算该节点的熵(未划分时的熵值)cur_entropy = calcEntropy(data_set)# 信息增益best_info_gain = 0.0# 最优属性索引best_feature_index = -1# 遍历所有属性for i in range(num_features):# 拿到当前列的属性feat_list = [example[i] for example in data_set]# 获取唯一值unique_val = set(feat_list)# 分支节点信息熵带权加和sum_child_entropy = 0# 计算各分支(不同属性划分)的熵值for val in unique_val:# 根据当前属性划分 data_set, 得到分支节点 sub_data_setsub_data_set = splitDataSet(data_set, i, val)# 计算该节点权重weight = len(sub_data_set) / len(data_set)# 计算分支节点信息熵带权加和sum_child_entropy += (calcEntropy(sub_data_set) * weight)# 计算信息增益info_gain = cur_entropy - sum_child_entropy# 更新最大信息增益if best_info_gain < info_gain:best_info_gain = info_gainbest_feature_index = ireturn best_feature_index# 获取判别结果，判别结果为样本数最多的类别
def majorityCnt(class_list):class_count = {}# 统计 class_list 中每个元素出现的次数for vote in class_list:if vote in class_count.keys():class_count[vote] += 1else:class_count[vote] = 1# 根据字典的值，降序排列sorted_class_count = sorted(class_count.items(), key=lambda x: x[1], reverse=True)return sorted_class_count[0][0]# 判断属性是否为空 + 判断所有样本在属性上的取值是否相同
def isEmptyOrSameLabel(data_set, labels):# 属性为空，if len(labels) == 0:return Trueelse:for i in range(0, len(data_set)):# 从0开始，避免只有一个样本的情况if data_set[0] == data_set[i]:continueelse:return Falsereturn True# 递归生成决策树节点
def createTreeNode(data_set, labels):# 取出当前节点的样本类别class_list = [example[-1] for example in data_set]'''递归终止条件'''# 1. 当前节点包含的样本全属于同一类别，无需再分，将其判别结果设为当前类别if len(class_list) == class_list.count(class_list[0]):return class_list[0]# 2. 判断属性是否为空(已经按照所有的属性划分完了) + 判断所有样本在属性上的取值是否相同if isEmptyOrSameLabel(data_set, labels):return majorityCnt(class_list)  # 返回当前节点样本最多的类别'''属性划分'''# 1. 选择最好的属性进行划分(返回值为索引)best_feature_index = chooseBestFeatureByEntropy(data_set) # 以信息增益为划分准测# best_feature_index = chooseBestFeatureByGini(data_set) # 以基尼指数为划分准则# 2. 利用索引获取真实值，找到最优划分属性best_feature_label = labels[best_feature_index]# 3. 根据最优属性的类别生成树(用字典模拟二叉树)my_tree = {best_feature_label: {}}# 4. 删除被选择的属性del labels[best_feature_index]# 5. 获取当前最佳属性的那一列feature_values = [example[best_feature_index] for example in data_set]# 6. 去重unique_feature_values = set(feature_values)# 7. 对每一个唯一值进行分支for value in unique_feature_values:# 递归创建树my_tree[best_feature_label][value] = createTreeNode(splitDataSet(data_set, best_feature_index, value), labels.copy())# 8. 返回return my_tree# 计算广度，一个树中叶子结点的数量
def getNumLeaves(my_tree):num_leaves = 0first_str = next(iter(my_tree)) # 取出字典第一层中，仅有的一个键，也相当于是根节点second_dict = my_tree[first_str] # 拿到子树，可能不只一颗子树，需要遍历for key in second_dict.keys(): # 遍历子树if type(second_dict[key]).__name__ == 'dict': # 判断该节点是否是字典，如果不是，代表此节点为叶子结点num_leaves += getNumLeaves(second_dict[key])else:num_leaves += 1return num_leaves# 获取树的深度
def _getTreeDepth(my_tree):max_depth = 0 # 初始化决策树深度first_str = next(iter(my_tree))second_dict = my_tree[first_str]for key in second_dict.keys():if type(second_dict[key]).__name__ == 'dict':this_depth = 1 + _getTreeDepth(second_dict[key])else:this_depth = 1if this_depth > max_depth:max_depth = this_depth # 更新层数return max_depthdef getTreeDepth(my_tree):return _getTreeDepth(my_tree) + 1# 单个未知数据预测
def classifySingle(input_tree, feat_labels, test_data):first_str = next(iter(input_tree)) # 找到决策树在这一层的划分属性second_dict = input_tree[first_str] # 找到子树feat_index = feat_labels.index(first_str) # 找到对应属性所在下标class_label = None # 置空for key in second_dict.keys(): # 根据属性值，遍历子树，key是属性值if test_data[feat_index] == key: # 找到样本该属性的属性值和当前子树分类时的属性值相等的子树，进入该子树进行判断if type(second_dict[key]).__name__ == 'dict': # 如果拿到的值是字典类型，说明还没到叶子节点，接着递归class_label = classifySingle(second_dict[key], feat_labels, test_data)else:# 拿到的值不是字典类型，说明找到叶子节点了，记录分类结果，递归返回class_label = second_dict[key]return class_label# 多个位置数据预测，复用classify_single
def classifyMultiple(input_tree, feat_labels, test_data):result = []# 计算列表维度def list_dimension(lst):if not isinstance(lst, list):return 0  # 不是列表，返回0维if all(not isinstance(item, list) for item in lst):return 1  # 所有元素都不是列表，返回1维return 1 + max(list_dimension(item) for item in lst)  # 递归计算维度n = list_dimension(test_data)if list_dimension(test_data) == 1: # 维度为1，说明只有一个样本result.append(classifySingle(input_tree, feat_labels, test_data))return resultelse:num_example = len(test_data)for i in range(num_example):result.append(classifySingle(input_tree, feat_labels, test_data[i]))return result

• 测试代码：

if __name__ == '__main__':# 准备数据data_set, labels = createDataSet()## 创建+训练树my_ID3_decision_tree = createTreeNode(data_set.copy(), labels.copy()) # 可能有对数据集和labels做修改的操作，这里传拷贝# 打印树print(my_ID3_decision_tree)# 打印树的高度和广度print(getTreeDepth(my_ID3_decision_tree)) # 高度print(getNumLeaves(my_ID3_decision_tree)) # 广度# 预测单个数据test_data = [1, 0, 0, 0]result = classifySingle(my_ID3_decision_tree, labels, test_data)print(result)# 或result = classifyMultiple(my_ID3_decision_tree, labels, test_data)print(result)# 测试多个数据test_data = [[1, 0, 0, 0], [0, 0, 1, 0], [2, 1, 0, 1]]result = classifyMultiple(my_ID3_decision_tree, labels, test_data)print(result)

• 输出结果：

{'F3-HOME': {0: {'F2-WORK': {0: 'no', 1: 'yes'}}, 1: 'yes'}}
3
3
no
['no']
['no', 'yes', 'yes']

• 通过上面的数据集，生成的决策树如下图所示：

---

8. 添加缺失值处理功能的ID3决策树

---

import pandas as pd
import numpy as np
from math import log2def createDataSet():# 数据data_set = [[0, 0, 0, 0, 'no'],[0, 0, None, 1, 'no'],[0, 1, 0, 1, 'yes'],[None, 1, 1, 0, 'yes'],[0, 0, 0, 0, 'no'],[1, 0, 0, 0, 'no'],[None, 0, 0, 1, 'no'],[1, 1, 1, 1, 'yes'],[1, 0, 1, 2, 'yes'],[1, 0, 1, 2, 'yes'],[2, None, 1, 2, 'yes'],[2, 0, 1, 1, 'yes'],[2, 1, 0, 1, 'yes'],[2, 1, 0, None, 'yes'],[2, 0, 0, 0, 'no'],]# 属性名labels = ['F1-AGE', 'F2-WORK', 'F3-HOME', 'F4-LOAN', 'result']data = pd.DataFrame(data_set, columns=labels)return data# 预处理，给数据加一列权重，每一个样本权重初始化为1
def prepareDataSet(data):# 样本数num_data = len(data)weight = [] # 权重列表for _ in range(num_data):weight.append(1)data['weight'] = weightreturn data# 根据当前属性的某个属性值划分数据集
def splitDataSet(data, attribute, value, flag=False):""":param flag::param data: 数据集:param attribute: 属性标签:param value: 属性值:param flag: 是否更新缺失集权重，并将其按权放入子集(True表示进行该操作):return:"""sub_data = data.loc[data[attribute]==value]del sub_data[attribute]if flag:# 拿到缺失数据集missing_data = data.loc[data[attribute].isna()]# 拿到无缺失的数据集 D~no_missing_data = data.loc[data[attribute].notna()]del missing_data[attribute]# 无缺失样本中，该 sub_data 的比例，课本中的rv~p_sub_data = np.sum(sub_data['weight']) / np.sum(no_missing_data['weight'])# 更新权重weight = missing_data['weight'].unique()missing_data.loc[missing_data.index, 'weight'] = p_sub_data * weight# 将缺失数据集放入分支节点sub_data = pd.concat([sub_data, missing_data])return sub_data# 计算节点的信息熵
def calculateEntropy(data, target):values = data[target].unique() # 目标值列表weight = []for i in range(len(values)):data_k = data.loc[data[target] == values[i]] # 取出所有第k类的样本weight.append(np.sum(data_k['weight']))# 每个类别的所占比例，以列表形式存储，书中的pk~probs = weight / np.sum(data['weight'])# 计算该样本集的熵entropy = -sum(p * log2(p) for p in probs if p > 0)return entropy# 计算选择该属性的信息增益
def calculateInformationGain(data, attribute, target):""":param data: 数据集:param attribute: 属性标签:param target: 目标值标签:return: 返回信息增益 gain，和缺失数据集 missing_data"""# 拿到无缺失的数据集 D~no_missing_data = data.loc[data[attribute].notna()]# 计算无缺失样本集所占比例，书中的roup_no_missing_data = np.sum(no_missing_data['weight']) / np.sum(data['weight'])# D~的信息熵current_entropy = calculateEntropy(no_missing_data, 'result')# 拿到该属性的所有属性值attribute_value_list = no_missing_data[attribute].unique()# 记录划分子集的信息熵带权加和p_sum_sub_data = 0for value in attribute_value_list:# 拿到属性值为 value 的子集sub_data = splitDataSet(no_missing_data.copy(), attribute, value)# 无缺失样本中，该 sub_data 的比例，课本中的rv~p_sub_data = np.sum(sub_data['weight']) / np.sum(no_missing_data['weight'])p_sum_sub_data += p_sub_data * calculateEntropy(sub_data, target)# 选择该属性的信息增益gain = p_no_missing_data * (current_entropy - p_sum_sub_data)return gain# 选择最优划分属性
def chooseBestSplitAttribute(data, target):# 记录最大信息增益best_info_gain = 0# 记录最佳属性best_attribute = None# 属性列表(注意不是属性值列表)attribute_list = list(data.columns)attribute_list = attribute_list[:-2] # 最后两列不是属性，丢掉for attribute in attribute_list:cur_info_gain = calculateInformationGain(data, attribute, target)if best_info_gain < cur_info_gain:best_info_gain = cur_info_gainbest_attribute = attributereturn best_attributedef isEmptyOrSameLabel(data):if data.empty:return Trueelse:data = data.values.tolist()data = [row[:-2] for row in data]for i in range(len(data)):if data[0] == data[i]:continueelse:return Falsereturn Truedef majorityCnt(class_list):return class_list.mode() # 返回最多的值# 递归生成决策树节点
def createTreeNode(data, target):# 取出当前节点的样本类别class_list = data[target]'''递归终止条件'''# 1. 当前节点包含的样本全属于同一类别，无需再分，将其判别结果设为当前类别if len(class_list.unique()) == 1:return class_list.sample().tolist()[0]# 2. 判断属性是否为空(已经按照所有的属性划分完了) + 判断所有样本在属性上的取值是否相同elif isEmptyOrSameLabel(data):return majorityCnt(class_list)  # 返回当前节点样本最多的类别'''属性划分'''# 1. 选择最好的属性进行划分best_attribute = chooseBestSplitAttribute(data, target)  # 以信息增益为划分准测# 3. 根据最优属性，和其属性值生成树(用字典模拟二叉树)my_tree = {best_attribute: {}}# 5. 获取当前最佳属性属性值的唯一值attribute_value_list = data[data[best_attribute].notna()][best_attribute].unique()# 7. 对每一个唯一值进行分支for value in attribute_value_list:# 递归创建树my_tree[best_attribute][value] = createTreeNode(splitDataSet(data.copy(), best_attribute, value, flag=True), target)# 8. 返回return my_tree

• 测试代码：

if __name__ == '__main__':data = createDataSet()  # 获取数据data = prepareDataSet(data) # 预处理my_tree = createTreeNode(data, 'result') # 构建树print(my_tree) # 打印树

• 输出：

{'F2-WORK': {0.0: {'F3-HOME': {0.0: 'no', 1.0: {'F1-AGE': {1.0: 'yes', 2.0: 'yes', 0.0: 'no'}}}}, 1.0: 'yes'}}

---