竞赛之王,从原理到实战的完整攻略
XGBoost(eXtreme Gradient Boosting)是陈天奇2016年提出的梯度提升框架,是Kaggle竞赛中使用最广泛的算法。它通过序列地拟合残差,每棵新树修正前面所有树的错误。
最优叶子权重:
import numpy as np
np.random.seed(42)
# 生成回归数据
n = 500
X = np.random.randn(n, 5)
y = 3*X[:,0] + 2*X[:,1] - 1.5*X[:,2] + np.random.randn(n)*0.5
# 回归树
class RegTree:
def __init__(self, max_depth=3, min_samples=10):
self.max_depth = max_depth
self.min_samples = min_samples
self.tree = {}
def fit(self, X, y):
self.tree = self._build(X, y, 0)
def _build(self, X, y, depth):
if depth >= self.max_depth or len(y) < self.min_samples:
return {'value': np.mean(y)}
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 = np.var(y[left])*left.sum() + np.var(y[right])*right.sum()
if score < best_score:
best_score, best_feat, best_thr = score, feat, thr
if best_score == float('inf'):
return {'value': np.mean(y)}
left_mask = X[:, best_feat] <= best_thr
return {
'feature': best_feat, 'threshold': best_thr,
'left': self._build(X[left_mask], y[left_mask], depth+1),
'right': self._build(X[~left_mask], y[~left_mask], depth+1)
}
def predict_one(self, x, node=None):
if node is None: node = self.tree
if 'value' in node: return node['value']
child = node['left'] if x[node['feature']] <= node['threshold'] else node['right']
return self.predict_one(x, child)
def predict(self, X):
return np.array([self.predict_one(x) for x in X])
class GradientBoosting:
def __init__(self, n_estimators=50, lr=0.1, max_depth=3):
self.n_estimators = n_estimators
self.lr = lr
self.max_depth = max_depth
self.trees = []
self.base_pred = 0
def fit(self, X, y):
self.base_pred = np.mean(y)
residuals = y - self.base_pred
for i in range(self.n_estimators):
tree = RegTree(max_depth=self.max_depth)
tree.fit(X, residuals)
preds = tree.predict(X)
residuals -= self.lr * preds
self.trees.append(tree)
if (i+1) % 10 == 0:
y_pred = self.predict(X)
rmse = np.sqrt(np.mean((y - y_pred)**2))
print(f" 第{i+1}轮: RMSE={rmse:.4f}")
def predict(self, X):
pred = np.full(len(X), self.base_pred)
for tree in self.trees:
pred += self.lr * tree.predict(X)
return pred
# 训练
split = int(0.8 * n)
model = GradientBoosting(n_estimators=50, lr=0.1, max_depth=3)
model.fit(X[:split], y[:split])
# 评估
y_pred = model.predict(X[split:])
rmse = np.sqrt(np.mean((y[split:] - y_pred)**2))
r2 = 1 - np.sum((y[split:]-y_pred)**2) / np.sum((y[split:]-y[split:].mean())**2)
print(f"\n测试RMSE: {rmse:.4f}")
print(f"测试R²: {r2:.4f}") # 0.9551
# 基于分裂频率的特征重要性
importances = np.zeros(X.shape[1])
def count_splits(node, feat_counts):
if 'feature' in node:
feat_counts[node['feature']] += 1
count_splits(node['left'], feat_counts)
count_splits(node['right'], feat_counts)
for tree in model.trees:
count_splits(tree.tree, importances)
importances = importances / importances.sum()
print("特征重要性:")
for i, imp in enumerate(importances):
bar = '█' * int(imp * 40)
print(f" 特征{i}: {imp:.4f} {bar}")
# 特征0: 0.3383 █████████████ (真实系数3)
# 特征1: 0.3264 █████████████ (真实系数2)
# 特征2: 0.3056 ████████████ (真实系数-1.5)
# 特征3: 0.0208 (噪声)
# 特征4: 0.0089 (噪声)
| 参数 | 含义 | 推荐范围 | 调参优先级 |
|---|---|---|---|
| n_estimators | 树的数量 | 100-5000 | ⭐⭐⭐ |
| learning_rate (eta) | 学习率 | 0.01-0.3 | ⭐⭐⭐ |
| max_depth | 树深度 | 3-10 | ⭐⭐⭐ |
| min_child_weight | 叶节点最小权重 | 1-10 | ⭐⭐ |
| subsample | 行采样比例 | 0.5-1.0 | ⭐⭐ |
| colsample_bytree | 列采样比例 | 0.5-1.0 | ⭐⭐ |
| reg_alpha (L1) | L1正则化 | 0-10 | ⭐ |
| reg_lambda (L2) | L2正则化 | 0-10 | ⭐ |
| gamma | 分裂最小增益 | 0-5 | ⭐ |
# 早停实现
def train_with_early_stopping(X_train, y_train, X_val, y_val,
max_rounds=500, patience=20):
model = GradientBoosting(n_estimators=1, lr=0.1, max_depth=3)
model.base_pred = np.mean(y_train)
residuals = y_train - model.base_pred
best_val_loss = float('inf')
best_trees = []
wait = 0
for round_num in range(max_rounds):
tree = RegTree(max_depth=3)
tree.fit(X_train, residuals)
model.trees.append(tree)
residuals -= 0.1 * tree.predict(X_train)
# 验证集评估
val_pred = model.predict(X_val)
val_loss = np.mean((y_val - val_pred)**2)
if val_loss < best_val_loss:
best_val_loss = val_loss
best_trees = model.trees.copy()
wait = 0
else:
wait += 1
if wait >= patience:
print(f"早停于第{round_num+1}轮")
break
model.trees = best_trees
return model
| 技巧 | 说明 | 效果 |
|---|---|---|
| 交叉验证 | 5-fold CV而非单次split | 评估更稳定 |
| 特征工程 | 交叉特征、统计特征、时序特征 | 提分最大 |
| 模型融合 | XGBoost+LightGBM+CatBoost加权 | 1-3%提升 |
| Stacking | 用多模型预测作为新特征 | 再提1-2% |
| 伪标签 | 用高置信预测扩充训练集 | 半监督提分 |
| 目标编码 | 类别特征→目标变量均值 | 类别特征提分 |
# Kaggle风格特征工程
def feature_engineering(X):
"""自动生成交互特征"""
X_new = X.copy()
n_features = X.shape[1]
# 二阶交互特征
for i in range(n_features):
for j in range(i+1, n_features):
X_new = np.column_stack([
X_new,
X[:, i] * X[:, j], # 乘法交互
X[:, i] - X[:, j], # 差值
np.abs(X[:, i] - X[:, j]) # 绝对差
])
# 统计特征
X_new = np.column_stack([
X_new,
X.mean(axis=1), # 行均值
X.std(axis=1), # 行标准差
X.max(axis=1) - X.min(axis=1), # 行极差
np.median(X, axis=1) # 行中位数
])
return X_new