深入解析CART决策树分池案例：原理、实现与性能优化

1次阅读

没有评论

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

在机器学习领域，决策树因其直观易懂、解释性强等优点被广泛应用。然而在实际业务场景中，特别是面对分池（如用户分层、流量分配）问题时，我们常常遇到两个主要痛点：

数据不均衡问题 ：分池场景下不同类别的样本数量往往差异显著，导致模型容易偏向多数类
特征选择效率低下 ：传统决策树在特征选择时需要遍历所有可能的划分点，计算复杂度高

针对分池问题，我们对比了几种常见的决策树算法：

ID3：基于信息增益，但只能处理离散特征且容易过拟合
C4.5：改进信息增益率，能处理连续特征，但计算复杂度较高
CART：使用基尼系数，能同时处理离散和连续特征，计算效率较高

在分池场景下，CART 因其高效的二元分割特性和对连续特征的良好支持成为最优选择。

CART 决策树使用基尼系数作为分裂标准，其计算公式为：

def gini_index(groups, classes):
    n_instances = sum([len(group) for group in groups])
    gini = 0.0
    for group in groups:
        size = len(group)
        if size == 0:
            continue
        score = 0.0
        for class_val in classes:
            p = [row[-1] for row in group].count(class_val) / size
            score += p * p
        gini += (1.0 - score) * (size / n_instances)
    return gini

针对数据不均衡问题，我们采用以下策略：

在训练前对少数类样本进行过采样
在计算基尼系数时引入类别权重
设置最小样本分裂阈值防止过拟合

以下是一个简化的 CART 决策树实现，重点关注分池场景下的关键优化点：

from collections import Counter

class DecisionTree:
    def __init__(self, max_depth=5, min_samples_split=2):
        self.max_depth = max_depth
        self.min_samples_split = min_samples_split

    def _best_split(self, X, y):
        best_gini = 1.0
        best_index, best_value = None, None

        n_features = X.shape[1]
        for feature_index in range(n_features):
            # 对连续特征寻找最优切分点
            feature_values = sorted(set(X[:, feature_index]))
            for value in feature_values:
                left_indices = X[:, feature_index] <= value
                right_indices = ~left_indices

                groups = [y[left_indices], y[right_indices]]
                gini = gini_index(groups, set(y))

                if gini < best_gini:
                    best_gini = gini
                    best_index = feature_index
                    best_value = value

        return {'index': best_index, 'value': best_value, 'groups': groups}

    def fit(self, X, y):
        self.root = self._grow_tree(X, y)

    def _grow_tree(self, X, y, depth=0):
        # 终止条件判断
        if depth >= self.max_depth or len(y) < self.min_samples_split:
            return Counter(y).most_common(1)[0][0]

        # 寻找最佳分裂
        best = self._best_split(X, y)

        # 递归构建子树
        left_indices = X[:, best['index']] <= best['value']
        right_indices = ~left_indices

        node = {'index': best['index'],
            'value': best['value'],
            'left': self._grow_tree(X[left_indices], y[left_indices], depth+1),
            'right': self._grow_tree(X[right_indices], y[right_indices], depth+1)
        }

        return node

    def predict(self, X):
        return [self._predict_one(x, self.root) for x in X]

    def _predict_one(self, x, node):
        if isinstance(node, dict):
            if x[node['index']] <= node['value']:
                return self._predict_one(x, node['left'])
            else:
                return self._predict_one(x, node['right'])
        else:
            return node

我们在三个不同规模的数据集上测试了优化后的 CART 决策树性能：