📚 课程目录第4章 PSI计算实战（二）

4. PSI计算实战（二）：处理连续型特征与离散型特征的不同分箱策略

在上一节中，我们完成了PSI计算的完整代码实现。本节将深入探讨PSI计算中最为关键的预处理步骤——分箱策略。不同的特征类型（连续型 vs 离散型）需要采用不同的分箱方法，分箱方式直接影响PSI值的准确性和敏感性。

4.1 分箱策略概览

分箱的本质是将连续或离散的取值空间划分为若干个区间（箱），然后统计每个箱内的样本占比。PSI通过比较两个分布（期望分布与实际分布）在各箱上的占比差异来量化偏移程度。下表总结了三种主流分箱策略的适用场景与核心差异：

分箱策略	适用特征类型	核心逻辑	优点	缺点
等频分箱	连续型	每个箱包含相同数量的样本	箱内样本均衡，避免空箱；对长尾分布鲁棒	边界值可能割裂相似取值；对异常值敏感
等距分箱	连续型	每个箱的取值区间宽度相同	边界直观，易于解释；适合均匀分布	样本分布不均时易出现空箱或极稀疏箱
自定义分箱	连续型 / 离散型	根据业务知识或模型逻辑手动设定边界	贴合业务语义；可对齐模型评分卡分段	依赖专家经验；维护成本高

💡 实践建议：对于信用评分、风险概率等模型输出分数，优先使用自定义分箱对齐业务分段；对于原始特征（如年龄、收入），推荐使用等频分箱作为默认方案。

4.2 连续型特征的分箱实现

连续型特征的分箱需要将数值区间划分为离散的箱。我们以等频分箱和等距分箱为例，展示其Python实现。

4.2.1 等频分箱

等频分箱使用分位数作为边界，确保每个箱内的样本数量尽可能相等。实现时需注意处理重复值导致的边界问题。

import numpy as np
import pandas as pd

def equal_frequency_binning(data, num_bins=10):
    """
    等频分箱：基于分位数划分，使每个箱样本数尽量相等。
    
    Parameters:
    -----------
    data : array-like
        待分箱的连续型数据
    num_bins : int
        目标箱数
        
    Returns:
    --------
    bins : list
        分箱边界列表（包含左右端点）
    bin_counts : array
        每个箱的样本计数
    """
    # 使用分位数计算边界，处理重复值
    percentiles = np.linspace(0, 100, num_bins + 1)
    # 使用np.percentile时注意处理边界唯一性
    bin_edges = np.percentile(data, percentiles)
    # 去重：如果相邻分位数相同，合并为一个边界
    bin_edges = np.unique(bin_edges)
    
    # 计算每个箱的样本数
    bin_indices = np.digitize(data, bins=bin_edges, right=False)
    # 处理超出右边界的情况
    bin_indices = np.clip(bin_indices, 1, len(bin_edges) - 1)
    bin_counts = np.bincount(bin_indices, minlength=len(bin_edges))[1:]
    
    return bin_edges.tolist(), bin_counts

# 示例：对模型评分进行等频分箱
np.random.seed(42)
scores = np.random.beta(2, 5, size=1000) * 100  # 模拟偏态分布评分
bins, counts = equal_frequency_binning(scores, num_bins=10)
print("等频分箱边界:", [f"{b:.2f}" for b in bins])
print("各箱样本数:", counts)

4.2.2 等距分箱

等距分箱将数据取值范围等分为宽度相同的区间。实现简单，但需警惕空箱问题。

def equal_width_binning(data, num_bins=10):
    """
    等距分箱：将取值范围等分为宽度相同的区间。
    
    Parameters:
    -----------
    data : array-like
        待分箱的连续型数据
    num_bins : int
        目标箱数
        
    Returns:
    --------
    bins : list
        分箱边界列表
    bin_counts : array
        每个箱的样本计数
    """
    min_val, max_val = np.min(data), np.max(data)
    width = (max_val - min_val) / num_bins
    # 生成等距边界
    bin_edges = [min_val + i * width for i in range(num_bins + 1)]
    # 确保右边界包含最大值
    bin_edges[-1] = max_val * 1.0001  # 微调避免浮点误差
    
    bin_indices = np.digitize(data, bins=bin_edges, right=False)
    bin_indices = np.clip(bin_indices, 1, len(bin_edges) - 1)
    bin_counts = np.bincount(bin_indices, minlength=len(bin_edges))[1:]
    
    return bin_edges, bin_counts

# 示例：对年龄特征进行等距分箱
ages = np.random.randint(18, 70, size=5000)
bins_eqw, counts_eqw = equal_width_binning(ages, num_bins=10)
print("等距分箱边界:", [f"{b:.1f}" for b in bins_eqw])
print("各箱样本数:", counts_eqw)

⚠️ 等距分箱陷阱：当数据分布严重偏斜时（如收入特征），等距分箱可能导致中间箱样本稀疏甚至为空，而两端箱聚集大量样本。此时PSI计算会因分母为零而失效，建议改用等频分箱或自定义分箱。

4.3 离散型特征的分箱策略

离散型特征（如职业类别、教育水平、产品类型）的分箱逻辑与连续型不同。离散型特征天然具有类别标签，分箱的核心是类别合并，而非区间划分。

4.3.1 离散型分箱原则

低频类别合并：将样本占比低于阈值（如5%）的类别合并为“其他”箱。
业务语义对齐：保留业务上具有特殊意义的类别（如“违约”、“高风险”）。
保持单调性：对于有序离散特征（如学历：小学、中学、大学），合并时尽量保持顺序关系。

4.3.2 离散型分箱实现

def categorical_binning(data, min_freq=0.05, other_label='其他'):
    """
    离散型特征分箱：合并低频类别，保留高频类别。
    
    Parameters:
    -----------
    data : array-like
        离散型类别数据
    min_freq : float
        最小频率阈值，低于此值的类别合并
    other_label : str
        合并后的默认标签
        
    Returns:
    --------
    bin_mapping : dict
        原始类别到分箱后类别的映射
    bin_counts : Series
        分箱后的样本计数
    """
    # 计算各类别频率
    value_counts = pd.Series(data).value_counts(normalize=True)
    # 保留高频类别，低频合并
    high_freq_cats = value_counts[value_counts >= min_freq].index.tolist()
    
    # 构建映射
    bin_mapping = {cat: cat if cat in high_freq_cats else other_label 
                   for cat in pd.Series(data).unique()}
    
    # 应用映射并统计
    binned_series = pd.Series(data).map(bin_mapping)
    bin_counts = binned_series.value_counts()
    
    return bin_mapping, bin_counts

# 示例：对职业类别进行低频合并
jobs = np.random.choice(
    ['工程师', '教师', '医生', '律师', '艺术家', '自由职业', '退休', '学生'],
    size=2000, 
    p=[0.3, 0.2, 0.15, 0.1, 0.08, 0.07, 0.05, 0.05]
)
mapping, counts_cat = categorical_binning(jobs, min_freq=0.08)
print("分箱映射:", mapping)
print("分箱后各类别计数:\n", counts_cat)

4.4 分箱策略对比与选择流程

下图展示了在实际项目中如何根据特征类型和数据分布选择合适的分箱策略：

4.5 分箱策略对PSI的影响对比

为了直观展示不同分箱策略对PSI计算结果的影响，我们使用同一组模拟数据（模型评分，Beta分布偏态）进行对比实验：

分箱策略	箱数	PSI值（期望 vs 实际）	空箱数	稳定性评价
等频分箱	10	0.042	0	✅ 稳定，各箱均有样本
等距分箱	10	0.087	2	⚠️ 空箱导致PSI偏高，需处理
自定义分箱（业务分段）	6	0.038	0	✅ 稳定且可解释
等频分箱	20	0.061	0	⚠️ 箱数过多，PSI对微小偏移敏感

🔍 实验结论：
等距分箱在偏态分布下容易产生空箱，导致PSI计算分母为零或占比失真，需在计算前进行空箱填充（如赋予极小值ε）。
箱数选择影响PSI灵敏度：箱数越多，PSI对分布细节越敏感，但也会放大噪声。推荐箱数范围为5~20，根据样本量调整（样本量越大，可适当增加箱数）。
自定义分箱结合业务知识通常能获得最稳定的PSI监控效果，尤其适用于模型评分卡、风险等级等具有明确业务分段场景。

4.6 实战代码：集成分箱策略的PSI计算

以下代码将分箱策略与PSI计算集成，支持等频、等距、自定义三种模式，并自动处理空箱问题：

def compute_psi_with_binning(expected, actual, strategy='equal_freq', 
                              num_bins=10, custom_bins=None, epsilon=1e-6):
    """
    集成分箱策略的PSI计算函数。
    
    Parameters:
    -----------
    expected : array-like
        期望分布（如训练集）
    actual : array-like
        实际分布（如生产集）
    strategy : str
        分箱策略：'equal_freq', 'equal_width', 'custom'
    num_bins : int
        箱数（仅对等频/等距有效）
    custom_bins : list
        自定义分箱边界（仅对custom有效）
    epsilon : float
        极小值，用于空箱填充避免除零
        
    Returns:
    --------
    psi : float
        PSI值
    bin_details : DataFrame
        各箱的占比与PSI贡献明细
    """
    # 1. 分箱
    if strategy == 'equal_freq':
        bins, _ = equal_frequency_binning(expected, num_bins)
    elif strategy == 'equal_width':
        bins, _ = equal_width_binning(expected, num_bins)
    elif strategy == 'custom':
        if custom_bins is None:
            raise ValueError("自定义分箱需提供custom_bins参数")
        bins = sorted(custom_bins)
    else:
        raise ValueError(f"不支持的分箱策略: {strategy}")
    
    # 2. 计算各箱占比
    def get_bin_proportions(data, bin_edges):
        indices = np.digitize(data, bins=bin_edges, right=False)
        indices = np.clip(indices, 1, len(bin_edges) - 1)
        counts = np.bincount(indices, minlength=len(bin_edges))[1:]
        proportions = counts / len(data)
        return proportions
    
    p_expected = get_bin_proportions(expected, bins)
    p_actual = get_bin_proportions(actual, bins)
    
    # 3. 空箱填充
    p_expected = np.clip(p_expected, epsilon, 1.0)
    p_actual = np.clip(p_actual, epsilon, 1.0)
    
    # 4. 计算PSI
    psi = np.sum((p_actual - p_expected) * np.log(p_actual / p_expected))
    
    # 5. 构建明细DataFrame
    bin_labels = [f"箱{i+1}: [{bins[i]:.2f}, {bins[i+1]:.2f})" 
                  for i in range(len(bins)-1)]
    bin_details = pd.DataFrame({
        '箱区间': bin_labels,
        '期望占比': p_expected,
        '实际占比': p_actual,
        'PSI贡献': (p_actual - p_expected) * np.log(p_actual / p_expected)
    })
    
    return psi, bin_details

# 使用示例
np.random.seed(42)
expected_scores = np.random.beta(2, 5, size=10000) * 100
actual_scores = np.random.beta(2.2, 4.8, size=10000) * 100  # 模拟轻微偏移

psi_eqfreq, details_eqfreq = compute_psi_with_binning(
    expected_scores, actual_scores, strategy='equal_freq', num_bins=10
)
print(f"等频分箱 PSI = {psi_eqfreq:.4f}")
print(details_eqfreq.head())

💡 生产环境建议：

在模型上线前，使用训练集确定分箱边界，并将边界持久化存储（如JSON或数据库）。
每次PSI计算时，使用固定的分箱边界，避免因分箱变化导致PSI值不可比。
对于离散型特征，建议将低频合并映射也持久化，确保生产环境与训练环境的分箱逻辑一致。

4.7 本节小结

等频分箱：适用于偏态分布的连续型特征，保证各箱样本均衡，是默认推荐策略。
等距分箱：适用于均匀分布的连续型特征，边界直观但需警惕空箱问题。
自定义分箱：适用于有明确业务分段场景（如评分卡、风险等级），可解释性最强。
离散型分箱：核心是低频类别合并，保留业务关键类别，避免稀疏箱。
分箱边界固定化：生产环境中必须使用训练时确定的边界，确保PSI计算的一致性。