如何在CatBoost中实现非连续自定义损失函数？

损失函数规则拆解 #

设目标值为y，预测值为p，固定惩罚常数为C（可自定义）：

• 当p >= y时，损失 = p - y，一阶导数为1，二阶导数为0
• 当p < y时，损失 = C，一阶导数为0，二阶导数为0

CatBoost要求自定义损失函数返回一阶导数（梯度）和二阶导数（海森矩阵对角线元素），以下是具体实现方案：

类式实现（推荐，支持参数配置） #

通过继承ICustomLoss基类实现，便于灵活调整固定惩罚常数：

from catboost import CatBoostRegressor, Pool
from catboost.losses import ICustomLoss

class CustomAsymmetricLoss(ICustomLoss):
    def __init__(self, constant_loss=10.0):
        # 初始化固定惩罚常数，可根据业务需求修改
        self.constant_loss = constant_loss

    def calc_ders_range(self, y_true, y_pred, weight=None):
        # 计算一阶、二阶导数
        ders1, ders2 = [], []
        for y, p in zip(y_true, y_pred):
            if p >= y:
                ders1.append(1.0)
                ders2.append(0.0)
            else:
                ders1.append(0.0)
                ders2.append(0.0)
        
        # 处理样本权重（若有）
        if weight is not None:
            ders1 = [d * w for d, w in zip(ders1, weight)]
            ders2 = [d * w for d, w in zip(ders2, weight)]
        return ders1, ders2

    def get_final_error(self, error, weight):
        # 计算全局损失值（用于模型评估）
        total_loss = sum(e * w for e, w in zip(error, weight))
        total_weight = sum(weight)
        return total_loss / total_weight if total_weight != 0 else 0.0

# 使用示例
# 1. 准备数据集
X = [[1,2], [3,4], [5,6], [7,8]]
y = [10, 20, 15, 25]
train_pool = Pool(X, y)

# 2. 初始化模型并使用自定义损失
model = CatBoostRegressor(
    loss_function=CustomAsymmetricLoss(constant_loss=10.0),
    iterations=100,
    learning_rate=0.1,
    verbose=False
)

# 3. 训练与预测
model.fit(train_pool)
preds = model.predict(X)
print("预测结果:", preds)

函数式实现（轻量版） #

若无需参数配置，也可直接传入损失函数和导数函数：

from catboost import CatBoostRegressor, Pool

def custom_loss(y_true, y_pred):
    loss = []
    CONSTANT_LOSS = 10.0
    for y, p in zip(y_true, y_pred):
        loss.append(p - y if p >= y else CONSTANT_LOSS)
    return loss

def custom_loss_deriv(y_true, y_pred):
    ders1, ders2 = [], []
    for y, p in zip(y_true, y_pred):
        if p >= y:
            ders1.append(1.0)
            ders2.append(0.0)
        else:
            ders1.append(0.0)
            ders2.append(0.0)
    return ders1, ders2

# 使用示例
model = CatBoostRegressor(
    loss_function=(custom_loss, custom_loss_deriv),
    iterations=100,
    learning_rate=0.1,
    verbose=False
)
model.fit(Pool(X, y))

关键说明 #

1. 二阶导数均为0是因为损失函数在两段都是线性或常数，导数的导数为0
2. 固定惩罚常数constant_loss可根据业务场景调整，比如目标值为10时可设为9（或其他符合需求的数值）
3. 若使用带权重的样本，代码已处理权重对导数的影响

内容的提问来源于stack exchange，提问作者Ron Bigman