4、最优分箱之卡方分箱：卡方检验原理、卡方分箱算法步骤、Python实现卡方分箱

卡方分箱（Chi-Merge）是一种基于统计检验的监督式分箱方法。它通过卡方检验衡量相邻区间内目标变量分布的相似性，逐步合并最相似的区间，直到满足预设的停止条件。该方法能自动发现特征与目标变量之间的非线性关系，是信用评分、风险建模等场景中最常用的分箱技术之一。

4.1 卡方检验原理

卡方检验（Chi-Square Test）用于判断两个分类变量之间是否独立。在分箱场景中，我们检验“特征区间”与“目标变量（如好坏客户）”是否独立。若两个相邻区间在目标分布上无显著差异（即独立），则合并它们不会损失预测信息。

卡方统计量计算公式：

χ² = Σ ( (O_ij - E_ij)² / E_ij )

其中：
- O_ij：第 i 行第 j 列的观测频数
- E_ij：第 i 行第 j 列的期望频数（假设独立）
- E_ij = (行合计 × 列合计) / 总样本数

卡方检验在分箱中的核心逻辑：

概念 说明
原假设 H₀ 特征区间与目标变量独立（分布无差异）
备择假设 H₁ 特征区间与目标变量不独立（分布有差异）
卡方值 衡量观测频数与期望频数的偏离程度
p 值 在原假设成立下，观察到当前卡方值或更极端值的概率
决策规则 若 p 值 > 显著性水平 α（如 0.05），则接受 H₀，合并区间
关键理解：卡方值越小，说明观测频数与期望频数越接近，两个区间在目标分布上越相似，越应该合并。卡方分箱本质上是“自底向上”的贪心合并算法。
卡方检验原理与分箱决策流程 2×2 列联表（示例） 区间 A 区间 B 好客户 坏客户 O₁₁ O₁₂ O₂₁ O₂₂ 卡方统计量 χ² = Σ (O-E)²/E O：观测频数 E：期望频数 决策规则 p 值 > α → 接受 H₀ → 合并区间 （分布无显著差异） 卡方分箱合并流程 计算相邻区间 χ² 找最小 χ² 对 合并该区间对 … 重复直到满足停止条件（区间数 ≤ 预设值 或 所有 χ² > 阈值）

4.2 卡方分箱算法步骤

卡方分箱（Chi-Merge）是一种自底向上的合并算法，从最细粒度的初始分箱开始，逐步合并最相似的相邻区间。

算法完整步骤：

  1. 初始化：将特征值排序后，每个唯一值作为一个初始区间（分箱）。
  2. 构建列联表：对每个相邻区间对 (i, i+1)，构建 2×2 列联表，统计好/坏客户频数。
  3. 计算卡方值：对每个相邻区间对计算 χ² 统计量。
  4. 合并：找到 χ² 值最小的相邻区间对（即最相似），合并它们。
  5. 重复：重新计算受影响区间的卡方值，重复步骤 3-4。
  6. 停止：当满足以下任一条件时停止：
    • 当前区间数 ≤ 预设最大分箱数（如 5 箱）
    • 所有相邻区间对的 χ² 值均大于卡方阈值（如 α=0.05 对应的临界值）
算法特点：卡方分箱是全局最优的贪心近似，每次合并都选择对整体信息损失最小的操作。它天然处理缺失值（可将缺失单独作为一箱参与合并），且对异常值不敏感。

4.3 Python实现卡方分箱

以下代码实现完整的卡方分箱类，包含卡方计算、合并逻辑和自动分箱功能。

import pandas as pd
import numpy as np
from scipy.stats import chi2_contingency

class ChiMerge:
    """卡方分箱实现"""
    
    def __init__(self, max_bins=5, alpha=0.05):
        self.max_bins = max_bins      # 最大分箱数
        self.alpha = alpha            # 显著性水平
        self.bins_ = None             # 分箱边界
        self.chi2_threshold_ = None   # 卡方阈值
        
    def _chi2_test(self, crosstab):
        """计算两个区间的卡方统计量"""
        chi2, p, _, _ = chi2_contingency(crosstab, correction=False)
        return chi2, p
    
    def fit(self, X, y):
        """执行卡方分箱"""
        df = pd.DataFrame({'feature': X, 'target': y})
        df = df.sort_values('feature').reset_index(drop=True)
        
        # 步骤1：初始化——每个唯一值作为一个区间
        unique_vals = sorted(df['feature'].unique())
        bins = unique_vals.copy()
        
        # 步骤2-5：迭代合并
        while len(bins) > self.max_bins:
            chi2_pairs = []
            
            # 计算所有相邻区间的卡方值
            for i in range(len(bins) - 1):
                left_val = bins[i]
                right_val = bins[i+1]
                
                # 获取两个区间的数据
                if i == 0:
                    left_mask = df['feature'] <= left_val
                else:
                    left_mask = (df['feature'] > bins[i-1]) & (df['feature'] <= left_val)
                
                right_mask = (df['feature'] > left_val) & (df['feature'] <= right_val)
                
                left_target = df.loc[left_mask, 'target']
                right_target = df.loc[right_mask, 'target']
                
                # 构建2x2列联表
                left_good = (left_target == 0).sum()
                left_bad = (left_target == 1).sum()
                right_good = (right_target == 0).sum()
                right_bad = (right_target == 1).sum()
                
                crosstab = np.array([[left_good, left_bad],
                                     [right_good, right_bad]])
                
                chi2, p = self._chi2_test(crosstab)
                chi2_pairs.append((chi2, p, i))
            
            # 找到最小卡方值的区间对
            chi2_pairs.sort(key=lambda x: x[0])
            min_chi2, min_p, merge_idx = chi2_pairs[0]
            
            # 检查停止条件：最小卡方值是否超过阈值
            if min_chi2 > self.chi2_threshold_:
                break
            
            # 合并区间
            bins.pop(merge_idx + 1)
        
        # 保存分箱边界
        self.bins_ = bins
        self.chi2_threshold_ = chi2_contingency(
            np.array([[1, 1], [1, 1]]), correction=False
        )[1]  # 使用p值阈值
        
        return self
    
    def transform(self, X):
        """将特征映射到分箱"""
        return pd.cut(X, bins=[-np.inf] + self.bins_ + [np.inf], 
                      labels=False, right=True)
    
    def fit_transform(self, X, y):
        """拟合并转换"""
        self.fit(X, y)
        return self.transform(X)

# 使用示例
if __name__ == "__main__":
    # 生成示例数据
    np.random.seed(42)
    n = 1000
    X = np.random.randn(n) * 10 + 50  # 年龄特征
    # 模拟好坏客户：年龄与坏账率的U型关系
    prob = 0.1 + 0.3 * (np.abs(X - 50) / 20)
    y = np.random.binomial(1, prob)
    
    # 执行卡方分箱
    chi_merger = ChiMerge(max_bins=5, alpha=0.05)
    bins = chi_merger.fit_transform(X, y)
    
    print("分箱边界:", chi_merger.bins_)
    print("分箱结果分布:")
    print(pd.Series(bins).value_counts().sort_index())
注意事项：
  • 卡方分箱要求目标变量为二分类（0/1），且每个区间内应同时包含两类样本，否则卡方检验失效。
  • 初始分箱时，若某个唯一值对应的样本量过少（如 < 5），建议先合并到相邻区间，避免列联表出现零频数。
  • 实际应用中常结合业务约束：如每箱至少包含 5% 样本、单调性校验等，需在合并循环中额外判断。

4.4 卡方分箱 vs 其他分箱方法

对比维度 卡方分箱 等距分箱 等频分箱 决策树分箱
监督/无监督 监督 无监督 无监督 监督
目标变量 二分类 不需要 不需要 分类/回归
分箱依据 卡方统计量 等宽 等样本量 信息增益/基尼
处理异常值 鲁棒 敏感 较鲁棒 鲁棒
单调性保证 需额外约束 不保证 不保证 需额外约束
计算复杂度 中等 低 低 高
卡方分箱合并过程可视化（年龄特征示例） 初始分箱（10箱） 20-25 25-30 30-35 35-40 40-45 45-50 50-55 55-60 60-65 65+ 合并最小 χ² 对 中间分箱（6箱） 20-30（合并） 30-35 35-45（合并） 45-50 50-55 55+（合并） 继续合并 最终分箱（5箱） 20-30 30-45 45-50 50-55 55+ 卡方分箱自动发现年龄与风险的非线性关系：低龄和高龄区间被合并

4.5 卡方分箱的工程实践要点

生产环境建议：
  • 缺失值处理：将缺失值单独作为一个区间参与合并，或先填充再分箱。
  • 单调性约束：在合并循环中增加坏账率单调性检查，若不单调则跳过当前合并。
  • 最小样本量：每箱至少包含总样本的 5% 或至少 30 个样本，避免过拟合。
  • 分箱数选择：信用评分通常 4-6 箱，风控模型 5-8 箱，根据业务可解释性需求调整。
  • 性能优化：对大数据量（百万级），可先等频分箱到 100-200 箱作为初始分箱，再执行卡方合并。

卡方分箱是监督分箱中最经典的方法之一，它通过统计检验自动发现特征与目标的最优分段，在信用评分卡、客户流失预测、医疗诊断等二分类场景中广泛应用。掌握卡方分箱的原理和实现，是构建可解释性机器学习模型的重要基础。

← 上一章 📖 返回目录 下一章 →