XGBoost极限梯度提升树之推导

import numpy as np

# 1. 模拟XGBoost打分函数
print("=== XGBoost打分函数模拟 ===")

def calculate_score(G, H, lambda_=1.0):
    """计算叶节点的打分"""
    return -0.5 * (G ** 2) / (H + lambda_)

def calculate_tree_score(G_list, H_list, lambda_=1.0, gamma=0.5):
    """计算整棵树的打分"""
    score = 0
    for G, H in zip(G_list, H_list):
        score += calculate_score(G, H, lambda_)
    score -= gamma * len(G_list)  # 减去叶子节点数惩罚
    return score

def calculate_gain(G_L, H_L, G_R, H_R, lambda_=1.0, gamma=0.5):
    """计算分裂增益"""
    # 分裂前的打分（合并为一个节点）
    score_before = calculate_score(G_L + G_R, H_L + H_R, lambda_)
    
    # 分裂后的打分（两个节点）
    score_after = calculate_score(G_L, H_L, lambda_) + calculate_score(G_R, H_R, lambda_)
    
    # 增益 = 分裂后 - 分裂前 - gamma
    gain = score_after - score_before - gamma
    return gain

# 示例
print("\n--- 示例1：应该分裂 ---")
G_L, H_L = 10, 5
G_R, H_R = 8, 4
lambda_ = 1.0
gamma = 0.5

gain = calculate_gain(G_L, H_L, G_R, H_R, lambda_, gamma)
print(f"左子树: G={G_L}, H={H_L}")
print(f"右子树: G={G_R}, H={H_R}")
print(f"增益: {gain:.4f}")
print(f"决策: {'分裂' if gain > 0 else '不分裂'}")

print("\n--- 示例2：不应该分裂 ---")
G_L, H_L = 2, 5
G_R, H_R = 1, 4

gain = calculate_gain(G_L, H_L, G_R, H_R, lambda_, gamma)
print(f"左子树: G={G_L}, H={H_L}")
print(f"右子树: G={G_R}, H={H_R}")
print(f"增益: {gain:.4f}")
print(f"决策: {'分裂' if gain > 0 else '不分裂'}")

# 2. 模拟XGBoost训练过程
print("\n=== 模拟XGBoost训练过程 ===")

class SimpleXGBoostNode:
    """简单的XGBoost节点"""
    def __init__(self):
        self.is_leaf = True
        self.value = 0
        self.left = None
        self.right = None
        self.split_feature = None
        self.split_value = None

class SimpleXGBoost:
    """简单的XGBoost实现"""
    
    def __init__(self, n_estimators=5, learning_rate=0.1, max_depth=3, 
                 lambda_=1.0, gamma=0.5):
        self.n_estimators = n_estimators
        self.learning_rate = learning_rate
        self.max_depth = max_depth
        self.lambda_ = lambda_
        self.gamma = gamma
        self.trees = []
        self.initial_pred = 0
    
    def fit(self, X, y):
        # 初始化预测值（均值）
        self.initial_pred = np.mean(y)
        current_pred = np.full(len(y), self.initial_pred)
        
        for i in range(self.n_estimators):
            # 计算一阶导数和二阶导数（平方损失）
            g = current_pred - y  # 一阶导数
            h = np.ones(len(y))    # 二阶导数（平方损失的二阶导数为1）
            
            # 构建树
            tree = self._build_tree(X, g, h, depth=0)
            self.trees.append(tree)
            
            # 更新预测值
            leaf_values = self._predict_tree(X, tree)
            current_pred += self.learning_rate * leaf_values
            
            # 计算损失
            loss = 0.5 * np.sum((y - current_pred) ** 2)
            print(f"第{i+1}轮: 损失={loss:.4f}")
    
    def _build_tree(self, X, g, h, depth):
        """构建树"""
        node = SimpleXGBoostNode()
        
        if depth >= self.max_depth or len(X) < 2:
            # 叶节点
            G = np.sum(g)
            H = np.sum(h)
            node.value = -G / (H + self.lambda_)
            return node
        
        # 找最佳分裂点
        best_gain = 0
        best_feature = None
        best_value = None
        best_left_idx = None
        
        n_features = X.shape[1]
        for feature in range(n_features):
            unique_values = np.unique(X[:, feature])
            for value in unique_values:
                left_idx = X[:, feature] <= value
                right_idx = ~left_idx
                
                if np.sum(left_idx) == 0 or np.sum(right_idx) == 0:
                    continue
                
                G_L = np.sum(g[left_idx])
                H_L = np.sum(h[left_idx])
                G_R = np.sum(g[right_idx])
                H_R = np.sum(h[right_idx])
                
                gain = calculate_gain(G_L, H_L, G_R, H_R, self.lambda_, self.gamma)
                
                if gain > best_gain:
                    best_gain = gain
                    best_feature = feature
                    best_value = value
                    best_left_idx = left_idx
        
        if best_gain > 0:
            node.is_leaf = False
            node.split_feature = best_feature
            node.split_value = best_value
            node.left = self._build_tree(X[best_left_idx], g[best_left_idx], 
                                          h[best_left_idx], depth + 1)
            node.right = self._build_tree(X[~best_left_idx], g[~best_left_idx], 
                                           h[~best_left_idx], depth + 1)
        else:
            G = np.sum(g)
            H = np.sum(h)
            node.value = -G / (H + self.lambda_)
        
        return node
    
    def _predict_tree(self, X, tree):
        """单棵树预测"""
        predictions = np.zeros(len(X))
        for i in range(len(X)):
            node = tree
            while not node.is_leaf:
                if X[i, node.split_feature] <= node.split_value:
                    node = node.left
                else:
                    node = node.right
            predictions[i] = node.value
        return predictions
    
    def predict(self, X):
        """预测"""
        pred = np.full(len(X), self.initial_pred)
        for tree in self.trees:
            pred += self.learning_rate * self._predict_tree(X, tree)
        return pred

# 生成数据
from sklearn.datasets import make_regression
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error

X, y = make_regression(
    n_samples=200,
    n_features=5,
    noise=0.1,
    random_state=42
)

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# 训练
xgb = SimpleXGBoost(
    n_estimators=10,
    learning_rate=0.1,
    max_depth=3,
    lambda_=1.0,
    gamma=0.5
)
xgb.fit(X_train, y_train)

# 预测
y_pred = xgb.predict(X_test)
mse = mean_squared_error(y_test, y_pred)
print(f"\n测试集MSE: {mse:.4f}")

# 3. 不同gamma值的影响
print("\n=== 不同gamma值的影响 ===")
for gamma in [0, 0.5, 1.0, 2.0]:
    xgb = SimpleXGBoost(
        n_estimators=10,
        learning_rate=0.1,
        max_depth=3,
        lambda_=1.0,
        gamma=gamma
    )
    xgb.fit(X_train, y_train)
    y_pred = xgb.predict(X_test)
    mse = mean_squared_error(y_test, y_pred)
    print(f"gamma={gamma}: MSE={mse:.4f}")