1
2
3
4
5
p0 = np.linspace(0, 1, 20)
p1 = 1 - p0
entropy = -p0 * np.log2(p0 + 1e-9) - p1 * np.log2(p1 + 1e-9)
plt.plot(p0, entropy)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import numpy as np
from math import log
def entropy(ele):
    """
    输入:
    	->ele:包含类别取值的列表
    输出:
    	->entropy:信息熵值
    """
    # 计算列表中取值的概率分布
    probs = [ele.count(i)/len(ele) for i in set(ele)]
    # 计算信息熵
    entropy = -sum(prob*log(prob,2) for prob in probs)
    return entropy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import pandas as pd
# 以数据集框架读取高尔夫数据集
df = pd.read_csv("./golf_data.csv")
# 计算数据集的经验熵
# "play"为目标变量，即是否打高尔夫
entropy_D = entropy(df["play"].tolist())
# 计算天气特征的经验条件熵
# 其中 subset1,subset2,subset3 为根据天气特征三个取值划分之后的子集
subset1 = df[df["outlook"]=="sunny"]
subset2 = df[df["outlook"]=="overcast"]
subset3 = df[df["outlook"]=="rainy"]
entropy_DA = len(subset1)/len(df)*entropy(subset1["play"].tolist()) + 
             len(subset2)/len(df)*entropy(subset2["play"].tolist()) + 
             len(subset3)/len(df)*entropy(subset3["play"].tolist())
# 计算天气特征的信息增益
info_gain = entropy_D - entropy_DA
print(info_gain)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import numpy as np
from math import log
def entropy(ele):
    """
    输入:
        ->ele:包含类别取值的列表
    输出:
        ->entropy:信息熵值
    """
    # 计算列表中取值的概率分布
    probs = [ele.count(i)/len(ele) for i in set(ele)]
    # 计算信息熵
    entropy = -sum(prob*log(prob,2) for prob in probs)
    return entropy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import pandas as pd
# 以数据集框架读取高尔夫数据集
df = pd.read_csv("./golf_data.csv")
# 计算数据集的经验熵
# "play"为目标变量，即是否打高尔夫
entropy_D = entropy(df["play"].tolist())
# 计算天气特征的经验条件熵
# 其中 subset1,subset2,subset3 为根据天气特征三个取值划分之后的子集
subset1 = df[df["outlook"]=="sunny"]
subset2 = df[df["outlook"]=="overcast"]
subset3 = df[df["outlook"]=="rainy"]
entropy_DA = len(subset1)/len(df)*entropy(subset1["play"].tolist()) + 
             len(subset2)/len(df)*entropy(subset2["play"].tolist()) + 
             len(subset3)/len(df)*entropy(subset3["play"].tolist())
# 计算天气特征的信息增益
info_gain = entropy_D - entropy_DA

# 计算天气特征的信息增益比
entropy_A = entropy(df["outlook"].tolist())

info_gain_rate =  info_gain/entropy_A

print(info_gain)
print(info_gain_rate)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import numpy as np
def gini(nums):
    """
    输入:
        ->nums:包含类别取值的列表
    输出:
        ->gini:基尼指数
    """
    # 获取列表类别的概率分布
    probs = [nums.count(i)/len(nums) for i in set(nums)]
    # 计算基尼指数
    gini = sum([p*(1-p) for p in probs])
    return gini

1
2
3
4
5
6
7
8
9
10
import pandas as pd
df = pd.read_csv("./golf_data.csv")
# 其中 subset1,subset2 为根据天气特征晴和非晴划分之后的子集
subset1 = df[df["outlook"] == "sunny"]
subset2 = df[df["outlook"] != "sunny"]
# 计算基尼指数
gini_DA = len(subset1)/len(df)*gini(subset1["play"].tolist()) + 
          len(subset2)/len(df)*gini(subset2["play"].tolist())
print(gini_DA)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 导入 numpy 库
import numpy as np
# 导入对数计算模块
from math import log
def entropy(ele):
    """
    输入：包含类别取值的列表
    输出：信息熵值
    """
    # 计算列表中取值的概率分布
    probs = [ele.count(i)/len(ele) for i in set(ele)]
    # 计算信息熵
    entropy = -sum(prob*log(prob,2) for prob in probs)
    return entropy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def df_split(df, col):
    """
    输入:
    df: 待划分的训练数据
    col: 划分数据的依据特征
    输出:
    res_dict: 根据特征取值划分后的不同数据集字典
    """
    # 获取依据特征的不同取值
    unique_col_val = df[col].unique()
    # 创建划分结果的数据框字典
    res_dict = {elem:pd.DataFrame for elem in unique_col_val}
    # 根据特征取值进行划分
    for key in res_dict.keys():
        res_dict[key] = df[:][df[col] == key]
    return res_dict

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
# 根据训练集和标签选择信息增益最大的特征作为最优特征
def choose_best_feature(df, label):
    """
    输入:
    df: 待划分的训练数据
    label: 训练标签
    输出:
    mas_value: 最大信息增益值
    best_feature: 最优特征
    max_splited: 根据最优特征划分后的数据字典
    """
    # 计算训练标签的信息熵
    entropy_D = entropy(df[label].tolist())
    # 特征集
    cols = [col for col in df.colums if col not in [label]]
    # 初始化最大信息增益值、最优特征和划分后的数据集
    max_value, best_feature = -999, None
    max_splited = None
    # 遍历特征并根据特征划分后的数据集
    for col in cols:
        # 根据当前特征取值划分后的数据集
        splited_set = df_split(df, col)
        # 初始化经验条件熵
        entropy_DA = 0
        # 对划分后的数据集遍历计算
        for subset_col, subset in splited_set.items():
            # 计算划分后的数据子集的标签信息熵
            entropy_Di = entropy(subset[label].tolist())
            # 计算当前特征的经验条件熵
            entropy_DA += len(subset)/len(df) * entropy_Di
        # 计算当前特征的信息增益
        info_gain = entropu_D - entropy_DA
        # 获取最大信息增益，并保存对应的特征和划分结果
        if info_gain > max_value:
            max_value, best_feature = info_gain, col
            max_splited = splited_set
    return max_value, best_feature, max_splited

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class ID3tree:
    # 定义决策树结点类
    class TreeNode:
        # 定义树节点
        def __init__(self, name):
            self.name = name
            self.connections = {}
        # 定义树连接
        def connect(self, label, node):
            self.connections[label] = node
    # 定义全局变量，包括数据集、特征集、标签和根节点
    def __init__(self, df, label):
        self.columns = df.columns
        self.df = df
        self.label = label
        self.root = self.TreeNode("Root")
    
    # 构建树的调用
    def construct_tree(self):
        self.construct(self.root, "", self.df, self.columns)
        
    # 决策树构建方法
    def construct(self, parent_node, parent_label, sub_df, columns):
        # 选择最优特征
        max_value, best_feature, max_splited = choose_best_feature(sub_df[columns], self.label)
        # 如果选不到最优特征的，则构建单结点树
        if not best_feature:
            node = self.TreeNode(sub_df[self.label].iloc[0])
            parent_node.connect(parent_label, node)
            return 
        # 根据最优特征以及子结点构建树
        node = self.TreeNode(best_feature)
        parent_node.connect(parent_label, node)
        # 以 A-Ag 为新的特征集
        new_columns = [col for col in columns if col != best_feature]
        # 递归地构建决策树
        for splited_value, splited_data in max_splited.items():
            self.construct(node, splited_value, splited_data, new_columns)
    # 打印决策树
    def print_tree(self, node, tabs):
        print(tabs + node.name)
        for connection, child_node in node.connections.items():
            print(str(tabs) + "t" + "(" + str(connection) + ")")
            self.print_tree(child_node, tabs + "tt")

1
2
3
4
5
6
7
8
9
10
import pandas as pd
# 载入数据
df = pd.read_csv("./golf_data.csv")
# 创建 ID3 实例模型
id3_tree = ID3tree(df, "play")
# 构建 ID3 决策树
id3_tree.construct_tree()
# 打印决策树
id3_tree.print_tree(id3_tree.root, "")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import numpy as np
import pandas as pd
from sklearn import preprocessing
from sklearn.model_selection import train_test_split
from sklearn import tree
# 载入数据
df = pd.read_csv('golf_data.csv')

# 对属性进行编码
encoder = preprocessing.LabelEncoder()
for i in range(0,df.shape[1]):
    df[df.columns[i]] = encoder.fit_transform(df[df.columns[i]])
X = np.array(df[["humility","outlook","temp","windy"]])
y = df["play"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=1)

# 定义 ID3 模型（模型参数为："entropy"）
id3_tree = tree.DecisionTreeClassifier(criterion = "entropy", random_state = 30)
id3_tree.fit(X_train, y_train)
pre = id3_tree.predict(X_test)
print(y_test.values)
print(pre)

# 评分
score = id3_tree.score(X_test, y_test)
print(score)

# 画树
import graphviz
feature_name = ["湿度","天气","温度","是否有风"]
dot_data = tree.export_graphviz(id3_tree
                                ,feature_names = feature_name
                                ,class_names = ["去", "不去"]
                                ,filled = True
                                ,rounded = True
                               )
graph = graphviz.Source(dot_data)
graph

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
# 根据训练集和标签选择信息增益最大的特征作为最优特征
def choose_best_feature(df, label):
    """
    输入:
        df->待划分的训练数据
        label->训练标签
    输出:
        mas_value->最大信息增益值
        best_feature->最优特征
        max_splited->根据最优特征划分后的数据字典
    """
    # 计算训练标签的信息熵
    entropy_D = entropy(df[label].tolist())
    # 特征集
    cols = [col for col in df.columns if col not in [label]]
    # 初始化最大信息增益值、最优特征和划分后的数据集
    max_value, best_feature = -999, None
    max_splited = None
    # 遍历特征并根据特征划分后的数据集
    for col in cols:
        # 根据当前特征取值划分后的数据集
        splited_set = df_split(df, col)
        # 初始化经验条件熵
        entropy_DA = 0
        # 对划分后的数据集遍历计算
        for subset_col, subset in splited_set.items():
            # 计算划分后的数据子集的标签信息熵
            entropy_Di = entropy(subset[label].tolist())
            # 计算当前特征的经验条件熵
            entropy_DA += len(subset)/len(df) * entropy_Di
        # 计算当前特征的信息增益
        info_gain = entropy_D - entropy_DA
        
        # 信息增益率
        
        entropy_A = entropy(df[label].tolist())
        if(entropy_A != 0):
            info_gain_rate =  info_gain/entropy_A
        else :
            info_gain_rate = 0;
        
        # 获取最大信息增益，并保存对应的特征和划分结果
        #if info_gain > max_value:
        #    max_value, best_feature = info_gain, col
        #    max_splited = splited_set
        
        # 获取最大信息增益比，并保存对应的特征和划分结果
        if info_gain_rate >max_value:
            max_value, best_feature = info_gain_rate, col
            max_splited = splited_set
    return max_value, best_feature, max_splited

1
2
3
4
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, mean_squared_error

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class TreeNode():
    def __init__(self
                 , feature_idx = None
                 , threshold = None
                 , leaf_value = None
                 , left_branch = None
                 , right_branch = None
                ):
        # 特征索引
        self.feature_idx = feature_idx
        # 特征划分阈值
        self.threshold = threshold
        # 叶子节点取值
        self.leaf_value = leaf_value
        # 左子树
        self.left_branch = left_branch
        # 右子树
        self.right_branch = right_branch

1
2
3
4
5
6
7
8
9
10
11
12
13
# 定义二叉特征划分函数
def feature_split(X, feature_idx, threshold):
    split_func = None
    if isinstance(threshold, int) or isinstance(threshold, float):
        split_func = lambda sample: sample[feature_idx] >= threshold
    else:
        split_func = lambda sample: sample[feature_idx] == threshold

    X_left = np.array([sample for sample in X if split_func(sample)])
    X_right = np.array([sample for sample in X if not split_func(sample)])

    return np.array([X_left, X_right])

1
2
3
4
5
6
7
8
# 计算基尼指数
def calculate_gini(y):
    # 将数组转化为列表
    y = y.tolist()
    probs = [y.count(i)/len(y) for i in np.unique(y)]
    gini = sum([p*(1-p) for p in probs])
    return gini

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class BinaryDecisionTree(object):
    # 决策树初始化
    def __init__(self
                 , min_samples_split = 2
                 , min_gini_impurity = 999
                 , max_depth = float("inf")
                 , loss = None
                ):
        # 根结点
        self.root = None  
        # 节点最小分裂样本数
        self.min_samples_split = min_samples_split
        # 节点初始化基尼不纯度
        self.mini_gini_impurity = min_gini_impurity
        # 树最大深度
        self.max_depth = max_depth
        # 基尼不纯度计算函数
        self.gini_impurity_calculation = None
        # 叶子节点值预测函数
        self._leaf_value_calculation = None
        # 损失函数
        self.loss = loss

    # 决策树训练
    def fit(self, X, y, loss=None):
        # 递归构建决策树
        self.root = self._build_tree(X, y)
        self.loss = None

    # 决策树构建函数
    def _build_tree(self, X, y, current_depth = 0):
        # 初始化最小基尼不纯度
        init_gini_impurity = 999
        # 初始化最佳特征索引和阈值
        best_criteria = None    
        # 初始化数据子集
        best_sets = None        

        # 合并输入和标签
        Xy = np.concatenate((X, y), axis=1)
        # 获取样本数和特征数
        n_samples, n_features = X.shape
        # 设定决策树构建条件
        # 训练样本数量大于节点最小分裂样本数且当前树深度小于最大深度
        if n_samples >= self.min_samples_split and current_depth <= self.max_depth:
            # 遍历计算每个特征的基尼不纯度
            for feature_idx in range(n_features):
                # 获取第i特征的所有取值
                feature_values = np.expand_dims(X[:, feature_idx], axis=1)
                # 获取第i个特征的唯一取值
                unique_values = np.unique(feature_values)

                # 遍历取值并寻找最佳特征分裂阈值
                for threshold in unique_values:
                    # 特征节点二叉分裂
                    Xy1, Xy2 = feature_split(Xy, feature_idx, threshold)
                    # 如果分裂后的子集大小都不为0
                    if len(Xy1) > 0 and len(Xy2) > 0:
                        # 获取两个子集的标签值
                        y1 = Xy1[:, n_features:]
                        y2 = Xy2[:, n_features:]

                        # 计算基尼不纯度
                        impurity = self.impurity_calculation(y, y1, y2)

                        # 获取最小基尼不纯度
                        # 最佳特征索引和分裂阈值
                        if impurity < init_gini_impurity:
                            init_gini_impurity = impurity
                            best_criteria = {"feature_idx": feature_idx, "threshold": threshold}
                            best_sets = {
                                "leftX": Xy1[:, :n_features],   
                                "lefty": Xy1[:, n_features:],   
                                "rightX": Xy2[:, :n_features],  
                                "righty": Xy2[:, n_features:]   
                                }
        
        # 如果计算的最小不纯度小于设定的最小不纯度
        if init_gini_impurity < self.mini_gini_impurity:
            # 分别构建左右子树
            left_branch = self._build_tree(best_sets["leftX"], best_sets["lefty"], current_depth + 1)
            right_branch = self._build_tree(best_sets["rightX"], best_sets["righty"], current_depth + 1)
            return TreeNode(feature_idx = best_criteria["feature_idx"]
                            , threshold = best_criteria["threshold"]
                            , left_branch = left_branch
                            , right_branch = right_branch
                           )
        # 计算叶子计算取值
        leaf_value = self._leaf_value_calculation(y)

        return TreeNode(leaf_value=leaf_value)

    # 定义二叉树值预测函数
    def predict_value(self, x, tree=None):
        if tree is None:
            tree = self.root

        # 如果叶子节点已有值，则直接返回已有值
        if tree.leaf_value is not None:
            return tree.leaf_value

        # 选择特征并获取特征值
        feature_value = x[tree.feature_idx]

        # 判断落入左子树还是右子树
        branch = tree.right_branch
        if isinstance(feature_value, int) or isinstance(feature_value, float):
            if feature_value >= tree.threshold:
                branch = tree.left_branch
        elif feature_value == tree.threshold:
            branch = tree.left_branch

        # 测试子集
        return self.predict_value(x, branch)

    # 数据集预测函数
    def predict(self, X):
        y_pred = [self.predict_value(sample) for sample in X]
        return y_pred

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class ClassificationTree(BinaryDecisionTree):
    # 定义基尼不纯度计算过程
    def _calculate_gini_impurity(self, y, y1, y2):
        p = len(y1) / len(y)
        gini = calculate_gini(y)
        gini_impurity = p * calculate_gini(y1) + (1-p) * calculate_gini(y2)
        return gini_impurity
    
    # 多数投票
    def _majority_vote(self, y):
        most_common = None
        max_count = 0
        for label in np.unique(y):
            # 统计多数
            count = len(y[y == label])
            if count > max_count:
                most_common = label
                max_count = count
        return most_common
    
    # 分类树拟合
    def fit(self, X, y):
        self.impurity_calculation = self._calculate_gini_impurity
        self._leaf_value_calculation = self._majority_vote
        super(ClassificationTree, self).fit(X, y)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from sklearn import datasets
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
data = datasets.load_iris()
X, y = data.data, data.target
y = y.reshape((-1, 1))
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size = 0.3)
# 创建分类树模型实例
clf = ClassificationTree()
# 分类树训练
clf.fit(X_train, y_train)
# 分类树预测
y_pred = clf.predict(X_test)
print(y_test.T)
print(y_pred)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
class RegressionTree(BinaryDecisionTree):
    def _calculate_variance_reduction(self, y, y1, y2):
        var_tot = np.var(y, axis=0)
        var_y1 = np.var(y1, axis=0)
        var_y2 = np.var(y2, axis=0)
        frac_1 = len(y1) / len(y)
        frac_2 = len(y2) / len(y)
        # 计算方差减少量
        variance_reduction = var_tot - (frac_1 * var_y1 + frac_2 * var_y2)
        
        return sum(variance_reduction)

    # 节点值取平均
    def _mean_of_y(self, y):
        value = np.mean(y, axis=0)
        return value if len(value) > 1 else value[0]

    def fit(self, X, y):
        self.impurity_calculation = self._calculate_variance_reduction
        self._leaf_value_calculation = self._mean_of_y
        super(RegressionTree, self).fit(X, y)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 波士顿房价
from sklearn.datasets import load_boston
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error
# 获取输入和标签
X, y = load_boston(return_X_y = True)
y = y.reshape((-1, 1))
# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size = 0.3)
# 定义回归模型
reg = RegressionTree()
# 模型训练
reg.fit(X_train, y_train)
# 模型预测
y_pred = reg.predict(X_test)
# 均方误差
mse = mean_squared_error(y_test, y_pred)
print("MES of CRAT regression tree based on Numpy: ", mse)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import numpy as np
import pandas as pd
import graphviz
from sklearn import datasets
from sklearn.model_selection import train_test_split
from sklearn import tree
from sklearn.metrics import accuracy_score

# 按 7:3 划分训练集，测试集
data = datasets.load_iris()
X_train, X_test, y_train, y_test = train_test_split(data.data, data.target, test_size = 0.3)

# 定义模型
clf = tree.DecisionTreeClassifier()
# 模型训练
clf.fit(X_train, y_train)
# 模型预测
y_pred = clf.predict(X_test)

# accuracy
print(accuracy_score(y_test, y_pred))

# 画图
dot_data = tree.export_graphviz(clf
                                , feature_names = data.feature_names
                                , class_names = data.target_names
                                , filled = True
                                , rounded = True
                               )  
graph = graphviz.Source(dot_data)  
graph

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from sklearn.datasets import load_boston
from sklearn.model_selection import train_test_split
from sklearn.tree import DecisionTreeRegressor
from sklearn.metrics import mean_squared_error
import matplotlib.pyplot as plt
# 按 7:3 划分训练集，测试集
data = load_boston()
X_train, X_test, y_train, y_test = train_test_split(data.data, data.target, test_size = 0.3)

# 定义模型
reg = DecisionTreeRegressor()
# 模型训练
reg.fit(X_train, y_train)
# 模型预测
y_pred = reg.predict(X_test)

# mse
mse = mean_squared_error(y_test, y_pred)
print("Mean Squared Error:", mse)

# 画图
plt.plot(y_test, label = 'target')
plt.plot(y_pred, label = 'pred')
plt.legend()
plt.show()