3. 等距分箱：等距分箱原理、在回放场景下的适用性、优缺点分析

等距分箱（Equal-Width Binning）是最直观的连续变量离散化方法之一。它将变量的取值范围均匀划分为 k 个等宽的区间，每个区间内的样本被赋予相同的箱标签或数值。在回放验证场景中，等距分箱常用于对评分卡输出、概率预测值或风险特征进行快速分段，以便观察模型在不同分数段上的表现稳定性。

3.1 等距分箱原理

给定连续变量 X，其最小值为 min，最大值为 max，设定分箱数量 k，则每个箱的宽度为：

width = (max - min) / k

第 i 个箱的区间范围为：

[min + (i-1)*width, min + i*width),   i = 1, 2, ..., k

最后一个箱通常包含右端点，即闭区间 [min + (k-1)*width, max]。所有样本根据其取值落入对应区间，完成分箱。

3.2 在回放场景下的适用性

回放验证（Replay Validation）的核心目标是比较模型在开发样本与回放样本上的表现一致性。等距分箱在此场景中具有以下适用特征：

3.3 优缺点分析

优点

• 实现简单、计算高效： 仅需一次排序和除法运算，适合大规模回放数据的批处理。
• 边界直观、可解释性强： 箱边界为等差数列，业务人员容易理解每个箱代表的数值范围。
• 适合均匀分布变量： 当变量在取值范围内近似均匀分布时，等距分箱能获得良好的分段效果。
• 便于跨模型比较： 固定箱边界后，不同模型的输出可以在同一尺度上直接对比。

缺点

• 对异常值极度敏感： 单个离群点会大幅拉伸箱宽，导致正常样本集中在少数箱中，失去区分能力。
• 样本分布不均： 实际数据往往呈偏态分布，等距分箱会导致部分箱样本过多、部分箱样本过少，影响统计检验的可靠性。
• 无法自适应数据分布： 箱边界固定，无法根据数据密度自动调整，在回放样本分布偏移时表现不稳定。
• 信息损失较大： 对于非线性关系较强的变量，等距分箱可能破坏原始变量与目标变量之间的关联模式。

3.4 回放场景中的使用建议

1. 预处理阶段： 在回放流水线中，先对变量进行截尾处理（如 1% 和 99% 分位数截断），消除极端值对等距分箱的影响。
2. 固定箱边界： 使用开发样本的 min 和 max 确定箱边界，回放样本沿用同一套边界，确保跨时间窗口的可比性。
3. 结合统计检验： 在每个等距箱内计算 PSI（群体稳定性指标）或卡方检验，量化回放样本与开发样本的分布差异。
4. 动态调整 k 值： 根据样本量动态选择分箱数量，经验公式为 k = min(10, sqrt(n)/2)，避免小样本下箱内统计量不可靠。

3.5 代码示例：Python 实现等距分箱与回放验证

import numpy as np
import pandas as pd

def equal_width_binning(series, k=10, lower_bound=None, upper_bound=None):
    """
    等距分箱函数
    :param series: 待分箱的连续变量
    :param k: 分箱数量
    :param lower_bound: 下界截断值（可选）
    :param upper_bound: 上界截断值（可选）
    :return: 分箱标签数组
    """
    if lower_bound is None:
        lower_bound = series.min()
    if upper_bound is None:
        upper_bound = series.max()
    
    bins = np.linspace(lower_bound, upper_bound, k + 1)
    labels = [f'箱{i+1}' for i in range(k)]
    
    binned = pd.cut(series, bins=bins, labels=labels, include_lowest=True)
    return binned

# 回放验证示例：比较开发样本与回放样本的箱内分布
dev_scores = np.random.normal(600, 100, 10000)  # 开发样本
replay_scores = np.random.normal(620, 110, 5000)  # 回放样本（分布偏移）

# 使用开发样本的边界进行等距分箱
dev_bins = equal_width_binning(dev_scores, k=10)
replay_bins = equal_width_binning(replay_scores, k=10, 
                                  lower_bound=dev_scores.min(), 
                                  upper_bound=dev_scores.max())

# 统计每个箱的样本量
dev_counts = pd.Series(dev_bins).value_counts().sort_index()
replay_counts = pd.Series(replay_bins).value_counts().sort_index()

# 计算 PSI（群体稳定性指标）
psi = np.sum((replay_counts / replay_counts.sum() - dev_counts / dev_counts.sum()) * 
             np.log((replay_counts / replay_counts.sum()) / (dev_counts / dev_counts.sum())))
print(f"等距分箱 PSI: {psi:.4f}")