"""
排序 AUC 代理损失函数
面向 AUC / 排序质量优化，使用平方铰链（squared hinge）代理，
提供与 OrdinalRankLoss 的 logistic 代理不同的间隔惩罚。
支持硬负例挖掘与固定 margin。
"""
from __future__ import annotations
from typing import Tuple
import numpy as np
from .base import BaseLoss
from ._loss_math import binary_inputs, bce_terms, nonnegative, positive, unit_interval
[文档]
class RankingAUCProxyLoss(BaseLoss):
"""排序 AUC 代理损失，使用平方铰链代理优化 AUC / 排序质量。
作为 :class:`OrdinalRankLoss` 的增强版本，核心改进:
1. **平方铰链代理**: ``max(0, margin - Δ)²`` 比 logistic 代理
``log(1+exp(-Δ))`` 使用不同的间隔惩罚；效果需在验证集比较。
2. **硬负例挖掘**: 优先选择违反 margin 的样本对（``p_pos - p_neg < margin``），
避免在已充分排序的 pair 上浪费梯度。
3. **固定 margin**: 通过参数控制目标间距；未实现训练进度调度。
硬挖掘选中样本对改变、同分或 hinge 边界处不可微；其余区间按 n × 平均损失
计算概率导数和 Hessian 对角项，不返回跨样本二阶项。
数学形式::
Δ_ij = p_i - p_j (i ∈ pos, j ∈ neg)
L_pair = mean(max(0, margin - Δ_ij)²)
Loss = bce_weight × BCE + rank_weight × L_pair
:param rank_weight: 排序损失权重，默认 1.0
:param bce_weight: 交叉熵权重，默认 1.0
:param margin: 正负样本预测间距目标，默认 0.2
- 内部经验: 0.1~0.3 之间；margin 过大可能导致梯度爆炸
:param max_pairs: 最多采样的正负样本对数，默认 20000
:param hard_mining_ratio: 硬负例挖掘比例，默认 1.0（全部使用）
- 设为 0.5 则只保留 50% 违反 margin 最严重的 pair
:param random_state: 随机种子，保证采样可复现，默认 42
:param name: 损失函数名称，默认 "ranking_auc_proxy_loss"
**参考样例**
>>> import numpy as np
>>> from hscredit.core.models.losses import RankingAUCProxyLoss
>>>
>>> loss = RankingAUCProxyLoss(rank_weight=2.0, margin=0.2)
>>> y_true = np.array([0, 0, 1, 1])
>>> y_pred = np.array([0.1, 0.3, 0.7, 0.9])
>>> round(loss(y_true, y_pred), 6) >= 0
True
>>>
>>> # 在 XGBoost 中使用
>>> import xgboost as xgb
>>> dtrain = xgb.DMatrix(X_train, label=y_train)
>>> bst = xgb.train({}, dtrain, obj=loss.to_xgboost(), num_boost_round=200)
**引用**
平方铰链（squared hinge）排序代理与 AUC 优化见 Joachims, T. (2005). *A Support Vector
Method for Multivariate Performance Measures.* ICML 2005；硬负例挖掘思想见
Shrivastava, A. et al. (2016). *Training Region-based Object Detectors with Online
Hard Example Mining (OHEM).* CVPR 2016. https://arxiv.org/abs/1604.03540
"""
def __init__(
self,
rank_weight: float = 1.0,
bce_weight: float = 1.0,
margin: float = 0.2,
max_pairs: int = 20000,
hard_mining_ratio: float = 1.0,
random_state: int = 42,
name: str = "ranking_auc_proxy_loss",
):
super().__init__(name)
nonnegative(rank_weight=rank_weight, bce_weight=bce_weight, margin=margin)
positive(max_pairs=max_pairs, hard_mining_ratio=hard_mining_ratio)
unit_interval(hard_mining_ratio=hard_mining_ratio)
if not isinstance(max_pairs, (int, np.integer)):
raise ValueError("max_pairs 必须为正整数。")
self.rank_weight = rank_weight
self.bce_weight = bce_weight
self.margin = margin
self.max_pairs = max_pairs
self.hard_mining_ratio = hard_mining_ratio
self.random_state = random_state
def _prepare_pairs(
self,
y_true: np.ndarray,
y_pred: np.ndarray,
) -> Tuple[np.ndarray, np.ndarray]:
"""准备正负样本对，支持硬负例挖掘。"""
pos_idx = np.flatnonzero(y_true == 1)
neg_idx = np.flatnonzero(y_true == 0)
if len(pos_idx) == 0 or len(neg_idx) == 0:
return np.array([], dtype=int), np.array([], dtype=int)
rng = np.random.default_rng(self.random_state)
# 全组合或采样
total_pairs = len(pos_idx) * len(neg_idx)
if total_pairs <= self.max_pairs:
pos_grid = np.repeat(pos_idx, len(neg_idx))
neg_grid = np.tile(neg_idx, len(pos_idx))
else:
# 随机采样
sample_size = self.max_pairs
pos_choices = rng.choice(pos_idx, size=sample_size, replace=True)
neg_choices = rng.choice(neg_idx, size=sample_size, replace=True)
pos_grid = pos_choices
neg_grid = neg_choices
# 硬负例挖掘: 只保留违反 margin 最严重的 pair
if self.hard_mining_ratio < 1.0 and len(pos_grid) > 0:
diff = y_pred[pos_grid] - y_pred[neg_grid]
violation = self.margin - diff
# 只保留有违反的 pair（violation > 0）
violated_mask = violation > 0
if np.sum(violated_mask) > 0:
violated_idx = np.flatnonzero(violated_mask)
# 按违反程度排序，取最严重的 ratio%
n_keep = max(
1,
int(len(violated_idx) * self.hard_mining_ratio),
)
violation_scores = violation[violated_idx]
top_k = np.argsort(-violation_scores)[:n_keep]
selected = violated_idx[top_k]
pos_grid = pos_grid[selected]
neg_grid = neg_grid[selected]
return pos_grid, neg_grid
def _squared_hinge_loss(
self,
y_true: np.ndarray,
y_pred: np.ndarray,
) -> float:
"""计算平方铰链排序损失。"""
pos_pairs, neg_pairs = self._prepare_pairs(y_true, y_pred)
if len(pos_pairs) == 0:
return 0.0
diff = y_pred[pos_pairs] - y_pred[neg_pairs]
violation = np.maximum(0, self.margin - diff)
return float(np.mean(violation**2))
def __call__(
self,
y_true: np.ndarray,
y_pred: np.ndarray,
) -> float:
"""计算排序 AUC 代理损失。
:param y_true: 真实标签, shape (n_samples,)
:param y_pred: 预测概率, shape (n_samples,)
:return: 损失值
"""
y_true, y_pred = binary_inputs(y_true, y_pred)
bce = -np.mean(y_true * np.log(y_pred) + (1 - y_true) * np.log(1 - y_pred))
rank_loss = self._squared_hinge_loss(y_true, y_pred)
return float(self.bce_weight * bce + self.rank_weight * rank_loss)
[文档]
def gradient(
self,
y_true: np.ndarray,
y_pred: np.ndarray,
) -> np.ndarray:
"""计算梯度。
平方铰链梯度::
dL/dp_i = -2 × max(0, m - Δ_ij) (对 pos 样本 i)
dL/dp_j = 2 × max(0, m - Δ_ij) (对 neg 样本 j)
:param y_true: 真实标签
:param y_pred: 预测概率
:return: 梯度数组, shape (n_samples,)
"""
y_true, y_pred = binary_inputs(y_true, y_pred)
# BCE 梯度
grad = self.bce_weight * bce_terms(y_true, y_pred)[1]
# 排序梯度
pos_pairs, neg_pairs = self._prepare_pairs(y_true, y_pred)
if len(pos_pairs) == 0 or self.rank_weight == 0:
return grad
diff = y_pred[pos_pairs] - y_pred[neg_pairs]
violation = np.maximum(0, self.margin - diff)
# 平方铰链梯度: d/dp[max(0,m-d)²] = -2*max(0,m-d) for pos, +2*... for neg
pair_grad = 2 * violation * (len(y_true) * self.rank_weight / len(pos_pairs))
np.add.at(grad, pos_pairs, -pair_grad)
np.add.at(grad, neg_pairs, pair_grad)
return grad
[文档]
def hessian(
self,
y_true: np.ndarray,
y_pred: np.ndarray,
) -> np.ndarray:
"""计算二阶导数。
:param y_true: 真实标签
:param y_pred: 预测概率
:return: 二阶导数数组, shape (n_samples,)
"""
y_true, y_pred = binary_inputs(y_true, y_pred)
# BCE 二阶导
hess = self.bce_weight * bce_terms(y_true, y_pred)[2]
# 排序二阶导: d²/dp²[max(0,m-d)²] = 2 where violation > 0
pos_pairs, neg_pairs = self._prepare_pairs(y_true, y_pred)
if len(pos_pairs) > 0 and self.rank_weight != 0:
diff = y_pred[pos_pairs] - y_pred[neg_pairs]
active = (self.margin - diff) > 0
pair_hess = 2.0 * active.astype(float) * (len(y_true) * self.rank_weight / len(pos_pairs))
np.add.at(hess, pos_pairs, pair_hess)
np.add.at(hess, neg_pairs, pair_hess)
return hess