CatBoost自定义损失函数实战：解决非对称业务场景的模型优化难题

1次阅读

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

在金融风控或医疗诊断等场景中，不同类型的误判代价差异显著。例如将高风险用户误判为低风险（假阴性）可能造成巨额损失，而将低风险误判为高风险（假阳性）仅增加人工审核成本。标准的对数损失函数平等对待所有错误类型，导致模型无法适配业务需求。

常见的替代方案及其局限性：

class_weight 参数：
仅能全局调整类别权重
无法区分同一类别中不同样本的误判代价
不支持样本级精细控制
后处理阈值调整：
割裂模型训练与决策过程
可能破坏概率校准性
自定义损失函数：
支持样本级代价定义
保持端到端优化一致性
需处理梯度计算等底层细节

CatBoost 通过 ObjectiveFunction 基类提供扩展接口，需实现三个核心方法：

from typing import List, Tuple
import numpy as np
from catboost import ObjectiveFunction

class WeightedLogLoss(ObjectiveFunction):
    def __init__(self, false_positive_cost: float = 1.0, 
                 false_negative_cost: float = 5.0):
        self.fp_cost = false_positive_cost
        self.fn_cost = false_negative_cost

    def calc_ders_range(self, approxes: List[float], 
                       targets: List[float], 
                       weights: List[float]) -> List[Tuple[float, float]]:
        """计算一阶导数和 Hessian 矩阵"""
        assert len(approxes) == len(targets)
        if weights is None:
            weights = [1.0] * len(targets)

        result = []
        for approx, target, weight in zip(approxes, targets, weights):
            p = 1.0 / (1.0 + np.exp(-approx))

            # 根据样本真实标签选择代价权重
            cost = self.fn_cost if target > 0.5 else self.fp_cost

            # 梯度计算增加数值稳定性保护
            eps = 1e-16
            p_clip = np.clip(p, eps, 1.0 - eps)

            der1 = cost * weight * (p_clip - target)
            der2 = cost * weight * p_clip * (1.0 - p_clip)
            result.append((der1, der2))
        return result

from catboost import CatBoostClassifier, Pool
import pandas as pd

# 加载数据
data = pd.read_csv("risk_data.csv")
train_pool = Pool(data=data.drop("label", axis=1),
    label=data["label"],
    weight=data.get("sample_weight")
)

# 定义自定义损失
custom_loss = WeightedLogLoss(false_negative_cost=10.0)

# 模型训练
model = CatBoostClassifier(
    iterations=1000,
    learning_rate=0.05,
    custom_loss=[custom_loss],
    eval_metric="AUC",
    early_stopping_rounds=50,
    task_type="GPU"  # 启用 GPU 加速
)

model.fit(
    train_pool,
    verbose=100,
    plot=True
)