最直观的机器学习算法,从if-else到智能决策
决策树通过递归地将特征空间划分为矩形区域来学习决策规则。每个内部节点是一个特征判断,每个叶节点是一个预测值。
| 准则 | 公式 | 算法 | 特点 |
|---|---|---|---|
| 信息增益 | IG = H(parent) - Σ(nⱼ/n)H(childⱼ) | ID3 | 偏向多值特征 |
| 信息增益率 | GR = IG / IV (内在信息) | C4.5 | 修正多值偏向 |
| 基尼指数 | Gini = 1 - Σpᵢ² | CART | 计算快,sklearn默认 |
信息熵度量集合的不确定性,信息增益衡量分裂后不确定性的减少量。
import numpy as np
from collections import Counter
def entropy(y):
"""计算信息熵"""
counts = Counter(y)
total = len(y)
return -sum((c/total) * np.log2(c/total) for c in counts.values() if c > 0)
def info_gain(y, left_mask, right_mask):
"""计算信息增益"""
n = len(y)
n_left, n_right = left_mask.sum(), right_mask.sum()
if n_left == 0 or n_right == 0:
return 0
H_parent = entropy(y)
H_left = entropy(y[left_mask])
H_right = entropy(y[right_mask])
return H_parent - (n_left/n * H_left + n_right/n * H_right)
# 示例: 200样本,2类
np.random.seed(42)
X_class0 = np.random.multivariate_normal([2, 3], [[0.8, 0.2], [0.2, 0.8]], 100)
X_class1 = np.random.multivariate_normal([6, 7], [[0.8, 0.2], [0.2, 0.8]], 100)
X = np.vstack([X_class0, X_class1])
y = np.array([0]*100 + [1]*100)
print(f"初始熵: {entropy(y):.4f}") # 1.0000 (最不确定)
# 特征1在阈值5的信息增益
left = X[:, 1] <= 5
right = ~left
gain = info_gain(y, left, right)
print(f"特征1≤5的信息增益: {gain:.4f}") # 0.8888
class DecisionNode:
def __init__(self, feature=None, threshold=None,
left=None, right=None, value=None):
self.feature = feature
self.threshold = threshold
self.left = left
self.right = right
self.value = value # 叶节点的预测值
def build_tree(X, y, depth=0, max_depth=5, min_samples=5):
"""递归构建决策树"""
# 终止条件
if len(set(y)) == 1: # 纯节点
return DecisionNode(value=y[0])
if depth >= max_depth: # 达到最大深度
return DecisionNode(value=Counter(y).most_common(1)[0][0])
if len(y) < min_samples: # 样本太少
return DecisionNode(value=Counter(y).most_common(1)[0][0])
# 寻找最佳分裂
best_gain, best_feat, best_thr = 0, 0, 0
for feat in range(X.shape[1]):
thresholds = np.unique(np.percentile(X[:, feat],
np.arange(10, 100, 10)))
for thr in thresholds:
left_mask = X[:, feat] <= thr
right_mask = ~left_mask
gain = info_gain(y, left_mask, right_mask)
if gain > best_gain:
best_gain, best_feat, best_thr = gain, feat, thr
if best_gain <= 0: # 无法改善
return DecisionNode(value=Counter(y).most_common(1)[0][0])
# 递归构建子树
left_mask = X[:, best_feat] <= best_thr
right_mask = ~left_mask
return DecisionNode(
feature=best_feat, threshold=best_thr,
left=build_tree(X[left_mask], y[left_mask], depth+1, max_depth, min_samples),
right=build_tree(X[right_mask], y[right_mask], depth+1, max_depth, min_samples)
)
def predict_tree(node, x):
"""单样本预测"""
if node.value is not None:
return node.value
if x[node.feature] <= node.threshold:
return predict_tree(node.left, x)
return predict_tree(node.right, x)
# 训练与评估
split = int(0.7 * len(X))
tree = build_tree(X[:split], y[:split], max_depth=5)
predictions = np.array([predict_tree(tree, x) for x in X[split:]])
accuracy = np.mean(predictions == y[split:])
print(f"决策树准确率: {accuracy:.4f}") # 1.0000
决策树容易过拟合——树越深,训练集拟合越好,但泛化越差。剪枝是控制过拟合的关键技术。
# 剪枝效果对比
print("max_depth vs 准确率:")
for max_d in [1, 2, 3, 5, 8]:
tree_d = build_tree(X[:split], y[:split], max_depth=max_d)
pred_train = np.array([predict_tree(tree_d, x) for x in X[:split]])
pred_test = np.array([predict_tree(tree_d, x) for x in X[split:]])
acc_train = np.mean(pred_train == y[:split])
acc_test = np.mean(pred_test == y[split:])
gap = acc_train - acc_test
flag = '⚠️ 过拟合' if gap > 0.05 else '✅ 良好'
print(f" depth={max_d}: 训练={acc_train:.3f}, 测试={acc_test:.3f}, 差距={gap:.3f} {flag}")
# 输出示例:
# depth=1: 训练=1.000, 测试=1.000, 差距=0.000 ✅ 良好
# depth=8: 训练=1.000, 测试=0.983, 差距=0.017 ✅ 良好
def gini(y):
"""计算基尼指数"""
counts = Counter(y)
total = len(y)
return 1 - sum((c/total)**2 for c in counts.values())
# 对比信息熵和基尼
for distribution in [[50,50], [70,30], [90,10], [99,1]]:
y_demo = [0]*distribution[0] + [1]*distribution[1]
print(f"[{distribution[0]}, {distribution[1]}]: "
f"熵={entropy(y_demo):.4f}, 基尼={gini(y_demo):.4f}")
| 分布 | 熵 | 基尼 | 差异 |
|---|---|---|---|
| [50, 50] | 1.0000 | 0.5000 | 最大不确定性 |
| [70, 30] | 0.8813 | 0.4200 | 有所偏向 |
| [90, 10] | 0.4690 | 0.1800 | 明显偏向 |
| [99, 1] | 0.0808 | 0.0198 | 几乎纯 |
决策树也可用于回归——叶节点输出均值,分裂准则改为MSE最小化:
def mse(y):
"""均方误差"""
return np.mean((y - y.mean())**2) if len(y) > 0 else 0
def build_regression_tree(X, y, depth=0, max_depth=4, min_samples=10):
if depth >= max_depth or len(y) < min_samples:
return DecisionNode(value=y.mean() if len(y) > 0 else 0)
best_score, best_feat, best_thr = float('inf'), 0, 0
for feat in range(X.shape[1]):
for thr in np.percentile(X[:, feat], np.arange(10, 100, 10)):
left = X[:, feat] <= thr
right = ~left
if left.sum() < 5 or right.sum() < 5:
continue
score = mse(y[left]) * left.sum() + mse(y[right]) * right.sum()
if score < best_score:
best_score, best_feat, best_thr = score, feat, thr
if best_score == float('inf'):
return DecisionNode(value=y.mean())
left_mask = X[:, best_feat] <= best_thr
return DecisionNode(
feature=best_feat, threshold=best_thr,
left=build_regression_tree(X[left_mask], y[left_mask], depth+1, max_depth, min_samples),
right=build_regression_tree(X[~left_mask], y[~left_mask], depth+1, max_depth, min_samples)
)