CART决策树算法实战：从理论到CSDN最佳实践

1次阅读

共计 3640 个字符，预计需要花费 10 分钟才能阅读完成。

决策树算法在机器学习中被广泛应用，但工程化过程中常遇到以下典型问题：

连续特征离散化 ：CART 算法需要将连续特征转换为离散的分裂点，如何选择最佳分裂点直接影响模型性能。
模型解释性 ：虽然决策树本身具有较好的可解释性，但在复杂场景下，树的深度过大会降低解释性。
过拟合问题 ：决策树容易过拟合，尤其是在数据量较小或特征较多的情况下。

算法	时间复杂度	缺失值处理	适用场景
ID3	O(n^2)	不支持	分类问题
C4.5	O(n^2)	支持	分类问题
CART	O(n log n)	支持	分类 / 回归

基尼系数是 CART 算法中用于衡量数据不纯度的指标，计算公式如下：

$$
Gini(D) = 1 – \sum_{k=1}^{K} p_k^2
$$

以下是 Python 实现代码：

import numpy as np

def gini_index(y):
    _, counts = np.unique(y, return_counts=True)
    probabilities = counts / len(y)
    return 1 - np.sum(probabilities ** 2)

以下是递归构建决策树的代码框架：

class DecisionNode:
    def __init__(self, feature_idx=None, threshold=None, left=None, right=None, value=None):
        self.feature_idx = feature_idx  # 分裂特征索引
        self.threshold = threshold      # 分裂阈值
        self.left = left                # 左子树
        self.right = right              # 右子树
        self.value = value              # 叶子节点值

def build_tree(X, y, max_depth=None, min_samples_split=2):
    # 终止条件 1：所有样本属于同一类别
    if len(np.unique(y)) == 1:
        return DecisionNode(value=y[0])

    # 终止条件 2：样本数小于最小分裂样本数
    if len(y) < min_samples_split:
        return DecisionNode(value=np.argmax(np.bincount(y)))

    # 终止条件 3：达到最大深度
    if max_depth is not None and max_depth <= 0:
        return DecisionNode(value=np.argmax(np.bincount(y)))

    # 寻找最佳分裂特征和阈值
    best_gini = float('inf')
    best_feature_idx = None
    best_threshold = None

    for feature_idx in range(X.shape[1]):
        thresholds = np.unique(X[:, feature_idx])
        for threshold in thresholds:
            left_mask = X[:, feature_idx] <= threshold
            right_mask = ~left_mask

            if np.sum(left_mask) == 0 or np.sum(right_mask) == 0:
                continue

            gini_left = gini_index(y[left_mask])
            gini_right = gini_index(y[right_mask])
            weighted_gini = (np.sum(left_mask) * gini_left + np.sum(right_mask) * gini_right) / len(y)

            if weighted_gini < best_gini:
                best_gini = weighted_gini
                best_feature_idx = feature_idx
                best_threshold = threshold

    # 递归构建子树
    left_mask = X[:, best_feature_idx] <= best_threshold
    right_mask = ~left_mask

    left_subtree = build_tree(X[left_mask], y[left_mask], max_depth - 1 if max_depth is not None else None, min_samples_split)
    right_subtree = build_tree(X[right_mask], y[right_mask], max_depth - 1 if max_depth is not None else None, min_samples_split)

    return DecisionNode(feature_idx=best_feature_idx, threshold=best_threshold, left=left_subtree, right=right_subtree)

CCP（Cost-Complexity Pruning）：适用于需要平衡模型复杂度和准确率的场景。
REP（Reduced-Error Pruning）：适用于验证集数据量较大的场景。
PEP（Pessimistic Error Pruning）：适用于数据量较小或需要保守剪枝的场景。

特征相关性 ：高相关性的特征可能导致重要性评估偏差。
数据分布 ：特征重要性依赖于数据分布，不同数据集上的评估结果可能不一致。

对于连续特征，可以采用等宽分箱或等频分箱来减少计算量：

from sklearn.preprocessing import KBinsDiscretizer

# 等宽分箱
discretizer = KBinsDiscretizer(n_bins=10, encode='ordinal', strategy='uniform')
X_binned = discretizer.fit_transform(X)

# 等频分箱
discretizer = KBinsDiscretizer(n_bins=10, encode='ordinal', strategy='quantile')
X_binned = discretizer.fit_transform(X)

可以通过多线程或多进程并行化构建子树：

from concurrent.futures import ThreadPoolExecutor

def parallel_build_tree(X, y, max_depth=None, min_samples_split=2):
    if max_depth is not None and max_depth <= 0:
        return DecisionNode(value=np.argmax(np.bincount(y)))

    best_gini = float('inf')
    best_feature_idx = None
    best_threshold = None

    with ThreadPoolExecutor() as executor:
        futures = []
        for feature_idx in range(X.shape[1]):
            thresholds = np.unique(X[:, feature_idx])
            for threshold in thresholds:
                futures.append(executor.submit(calculate_gini, X, y, feature_idx, threshold))

        for future in futures:
            feature_idx, threshold, gini = future.result()
            if gini < best_gini:
                best_gini = gini
                best_feature_idx = feature_idx
                best_threshold = threshold

    left_mask = X[:, best_feature_idx] <= best_threshold
    right_mask = ~left_mask

    left_subtree = parallel_build_tree(X[left_mask], y[left_mask], max_depth - 1 if max_depth is not None else None, min_samples_split)
    right_subtree = parallel_build_tree(X[right_mask], y[right_mask], max_depth - 1 if max_depth is not None else None, min_samples_split)

    return DecisionNode(feature_idx=best_feature_idx, threshold=best_threshold, left=left_subtree, right=right_subtree)