C45决策树实战：从算法原理到生产环境优化

1次阅读

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

在机器学习项目中，决策树因其直观易懂的特性广受欢迎。但当我们面对高维稀疏数据时，传统的 ID3 算法会暴露出明显缺陷：

特征选择时偏向取值多的属性（如用户 ID 这种无意义字段）
无法处理连续型特征
完全没有过拟合控制机制

而 CART 树虽然支持回归任务，但其基尼指数计算方式：

def gini(y: np.ndarray) -> float:
    _, counts = np.unique(y, return_counts=True)
    return 1 - np.sum((counts/len(y))**2)  # O(c) c 为类别数

在处理类别型特征时又不如信息增益比直观。这就是为什么在金融风控、医疗诊断等需要可解释性的场景中，C45 仍是黄金标准。

标准的信息增益公式容易偏向多值属性，C45 引入分裂信息量进行修正：

def gain_ratio(X: np.ndarray, y: np.ndarray, feature_idx: int) -> float:
    """时间复杂度 O(n·logc) n 为样本数"""
    # 计算固有值（Intrinsic Value）values, counts = np.unique(X[:, feature_idx], return_counts=True)
    iv = np.sum(-(counts/len(X)) * np.log2(counts/len(X)))  # 防止除零

    # 计算信息增益
    entropy_parent = entropy(y)  # 父节点熵
    entropy_children = 0
    for v in values:
        mask = X[:, feature_idx] == v
        entropy_children += (np.sum(mask)/len(X)) * entropy(y[mask])

    return (entropy_parent - entropy_children) / iv if iv > 1e-10 else 0

通过 numpy 的广播机制避免循环：

def vectorized_entropy(y: np.ndarray) -> float:
    """向量化熵计算 比循环快 5 倍"""
    _, counts = np.unique(y, return_counts=True)
    p = counts / len(y)
    return -np.sum(p * np.log2(p + 1e-10))  # 防止 log(0)

方法	优点	缺点
二分法	分裂点具有明确业务意义	计算成本高（需排序）
等频分桶	计算效率 O(n)	可能合并重要决策边界

推荐实现：

def find_best_split(X_col: np.ndarray, y: np.ndarray) -> Tuple[float, float]:
    """时间复杂度 O(nlogn)"""
    sorted_idx = np.argsort(X_col)
    best_gain = -1
    split_value = None

    for i in range(1, len(X_col)):
        if X_col[sorted_idx[i]] == X_col[sorted_idx[i-1]]:
            continue
        current_split = (X_col[sorted_idx[i]] + X_col[sorted_idx[i-1]]) / 2
        gain = information_gain(X_col, y, current_split)
        if gain > best_gain:
            best_gain = gain
            split_value = current_split

    return best_gain, split_value

sklearn 的 ccp_alpha 参数与 REP（Reduced Error Pruning）对比：

from sklearn.tree import DecisionTreeClassifier

# 代价复杂度剪枝
tree_ccp = DecisionTreeClassifier(ccp_alpha=0.02)  # 越大剪枝越激进

# 对比 REP 剪枝（需保留验证集）def rep_prune(tree, X_val, y_val):
    original_acc = tree.score(X_val, y_val)
    for node in reversed(post_order_traversal(tree)):  # 自底向上
        if node.is_leaf: continue

        old_left, old_right = node.left, node.right
        node.left = node.right = None  # 临时剪枝
        new_acc = tree.score(X_val, y_val)

        if new_acc >= original_acc:  # REP 核心逻辑
            original_acc = new_acc
        else:
            node.left, node.right = old_left, old_right

当正负样本比例超过 1:10 时：

改用 Gini 指数作为分裂标准
对少数类样本进行加权
设置 min_samples_leaf 参数

# 样本加权示例
sample_weight = np.where(y == 1, 10, 1)  # 正样本权重 *10
tree.fit(X, y, sample_weight=sample_weight)

class EarlyStopping:
    def __init__(self, patience=3):
        self.best_loss = float('inf')
        self.patience = patience
        self.counter = 0

    def __call__(self, current_loss) -> bool:
        if current_loss < self.best_loss:
            self.best_loss = current_loss
            self.counter = 0
        else:
            self.counter += 1
            if self.counter >= self.patience:
                return True
        return False

stopper = EarlyStopping()
while depth < max_depth:
    # ... 训练过程...
    if stopper(val_loss):
        break

验证分裂质量的关键检查点：

子节点纯度应高于父节点
分裂后的信息增益必须非负
连续特征分裂后的左右区间不应有重叠

def test_split_quality():
    X = np.array([[1, 0], [1, 1], [0, 1], [0, 0]])
    y = np.array([1, 0, 1, 0])

    gain, _ = find_best_split(X[:, 0], y)
    assert gain > 0, "分裂后信息增益应为正"

    left_mask = X[:, 0] < 0.5
    assert entropy(y[left_mask]) < entropy(y), "子节点纯度应提升"

当面对超大规模数据时，传统 C45 需要全量数据计算信息增益，这显然不现实。可能的突破方向：