机器学习算法原理实现——cart决策树

发布时间 2023-09-10 17:39:31作者: bonelee

 

 

cart决策树示例:

 

本文目标,仿照sklearn写一个cart树,但是仅仅使用max_depth作为剪枝依据。

 

 

 

我们本次实现cart分类,因此用到gini指数:

 为了帮助理解:

 

 

好了,理解了基尼指数。我们看下cart树的构建步骤:

注意还有几个细节:

 

cart树每个treenode存储了哪些数据?

在CART决策树中,每个节点(TreeNode)通常存储以下数据:

  1. 划分特征:这是用于根据某种条件划分数据集的特征。例如,如果一个节点用"年龄 > 30"作为分割条件,那么"年龄"就是这个节点的划分特征。

  2. 划分阈值:与划分特征配合使用,定义了数据应如何分割。在上面的例子中,阈值是30。

  3. 左子节点:满足划分条件的数据子集的节点。例如,在上面的"年龄 > 30"例子中,大于30岁的数据会被划分到左子节点。

  4. 右子节点:不满足划分条件的数据子集的节点。在上面的例子中,30岁及以下的数据会被划分到右子节点。

  5. 类标签:只在叶节点中有效。表示该节点所代表的数据子集中最常见的类别。当新数据通过决策树进行预测时,最终到达的叶节点的类标签就是其预测结果。

  6. 数据子集:节点当前代表的数据子集。在许多实际实现中,为了节省内存,节点可能不直接存储数据子集,而是存储数据索引或其他引用。

  7. 基尼不纯度或其他不纯度指标:代表当前数据子集的不纯度。在构建树的过程中,这个指标用于判断是否应该继续划分当前节点。

  8. 其他可选信息:如节点深度、父节点引用、数据点的数量等。

这些数据允许决策树在训练过程中进行递归分割,以及在预测过程中导航通过树结构。

 

 好了,实现代码如下:

import numpy as np

class TreeNode:
    def __init__(self, gini, num_samples, num_samples_per_class, predicted_class):
        self.gini = gini
        self.num_samples = num_samples
        self.num_samples_per_class = num_samples_per_class
        self.predicted_class = predicted_class
        self.feature_index = 0
        self.threshold = 0
        self.left = None
        self.right = None

def gini(y):
    m = len(y)
    return 1.0 - sum([(np.sum(y == c) / m) ** 2 for c in np.unique(y)])

def grow_tree(X, y, depth=0, max_depth=None):
    classes = np.unique(y)
    num_samples_per_class = [np.sum(y == c) for c in classes]
    predicted_class = classes[np.argmax(num_samples_per_class)]
    node = TreeNode(
        gini=gini(y),
        num_samples=len(y),
        num_samples_per_class=num_samples_per_class,
        predicted_class=predicted_class,
    )

    if depth < max_depth:
        idx, thr = best_split(X, y)
        if idx is not None:
            indices_left = X[:, idx] < thr
            X_left, y_left = X[indices_left], y[indices_left]
            X_right, y_right = X[~indices_left], y[~indices_left]
            node.feature_index = idx
            node.threshold = thr
            node.left = grow_tree(X_left, y_left, depth + 1, max_depth)
            node.right = grow_tree(X_right, y_right, depth + 1, max_depth)
    return node


def best_split(X, y):
    """
    用numpy实现best_split,见下,可以先看不用numpy的实现
    """
    n_samples, n_features = X.shape
    
    if len(np.unique(y)) == 1:
        return None, None
    
    best = {}
    min_gini = float('inf')
    
    for feature_idx in range(n_features):
        thresholds = np.unique(X[:, feature_idx])
        for threshold in thresholds:
            left_mask = X[:, feature_idx] < threshold
            right_mask = ~left_mask
            
            gini_left = gini(y[left_mask])
            gini_right = gini(y[right_mask])
            
            weighted_gini = len(y[left_mask]) / n_samples * gini_left + len(y[right_mask]) / n_samples * gini_right
            if weighted_gini < min_gini:
                best = {
                    'feature_index': feature_idx,
                    'threshold': threshold,
                    'left_labels': y[left_mask],
                    'right_labels': y[right_mask],
                    'gini': weighted_gini
                }
                min_gini = weighted_gini
                
    return best['feature_index'], best['threshold']


def best_split2(X, y):
    """
    不用numpy实现best_split
    """
    n_samples, n_features = len(X), len(X[0])
    
    # 如果样本中只有一种输出标签或样本为空,则返回None
    if len(set(y)) == 1:
        return None, None
    
    # 初始化最佳分割的信息
    best = {}
    min_gini = float('inf')
    
    # 遍历每个特征
    for feature_idx in range(n_features):
        # 获取当前特征的所有唯一值,并排序
        unique_values = sorted(set(row[feature_idx] for row in X))
        
        # 遍历每个唯一值,考虑将其作为分割阈值
        for value in unique_values:
            left_y, right_y = [], []
            
            # 对于每个样本,根据其特征值与阈值的关系分到左子集或右子集
            for i, row in enumerate(X):
                if row[feature_idx] < value:
                    left_y.append(y[i])
                else:
                    right_y.append(y[i])
            
            # 计算左子集和右子集的基尼指数
            gini_left = 1.0 - sum([(left_y.count(label) / len(left_y)) ** 2 for label in set(left_y)])
            gini_right = 1.0 - sum([(right_y.count(label) / len(right_y)) ** 2 for label in set(right_y)])
            
            # 计算加权基尼指数
            weighted_gini = len(left_y) / len(y) * gini_left + len(right_y) / len(y) * gini_right
            
            # 如果当前基尼值小于已知的最小基尼值,更新最佳分割
            if weighted_gini < min_gini:
                best = {
                    'feature_index': feature_idx,
                    'threshold': value,
                    'left_labels': left_y,
                    'right_labels': right_y,
                    'gini': weighted_gini
                }
                min_gini = weighted_gini
                
    return best['feature_index'], best['threshold']

def predict_tree(node, X):
    if node.left is None and node.right is None:
        return node.predicted_class
    if X[node.feature_index] < node.threshold:
        return predict_tree(node.left, X)
    else:
        return predict_tree(node.right, X)

def predict_tree2(node, X):
    if node.left is None and node.right is None:
        return node.predicted_class
    if X[node.feature_index] < node.threshold:
        return predict_tree(node.left, X)
    else:
        return predict_tree(node.right, X)

class CARTClassifier:
    def __init__(self, max_depth=None):
        self.max_depth = max_depth

    def fit(self, X, y):
        self.tree_ = grow_tree(X, y, max_depth=self.max_depth)

    def predict(self, X):
        return [predict_tree(self.tree_, x) for x in X]

# 使用示例
if __name__ == "__main__":
    """
    # 好好理解下这个分割的函数
    X = np.array([[2.5], [3.5], [1], [1.5], [2], [3], [0]])
    y = np.array([1, 1, 0, 0, 1, 0, 2])
    best_idx, best_thr = best_split(X, y)
    """

    from sklearn.datasets import load_iris

    data = load_iris()
    X, y = data.data, data.target

    clf = CARTClassifier(max_depth=4)
    clf.fit(X, y)
    preds = clf.predict(X)

    accuracy = sum(preds == y) / len(y)
    print(f"Accuracy: {accuracy:.4f}")

    from sklearn.tree import DecisionTreeClassifier
    # 创建分类树实例
    clf = DecisionTreeClassifier(max_depth=4)
    # 分类树训练
    clf.fit(X, y)
    preds = clf.predict(X)
    accuracy = sum(preds == y) / len(y)
    print(f"sklearn Accuracy: {accuracy:.4f}")

  

输出:

Accuracy: 0.9933
sklearn Accuracy: 0.9933