别再只会用pandas.cut了！金融风控实战：用决策树和卡方分箱提升模型效果的保姆级教程

巷中人

422人浏览 · 2026-05-18 05:30:44

巷中人 · 2026-05-18 05:30:44 发布

金融风控中的特征分箱进阶：决策树与卡方分箱实战解析

在金融风控建模过程中，特征工程的质量往往直接决定了模型效果的上限。许多数据分析师虽然掌握了基础的等频、等距分箱方法，但在面对真实业务场景时，这些无监督分箱技术常常显得力不从心。本文将带你深入两种有监督分箱方法——决策树分箱和卡方分箱的实战应用，并构建完整的评估流水线。

1. 为什么需要进阶分箱技术？

传统分箱方法如pandas.cut虽然简单易用，但在金融风控场景中存在明显局限。等距分箱对异常值敏感，等频分箱可能破坏特征与目标变量的相关性。而有监督分箱能够充分利用标签信息，使分箱结果更有利于模型学习。

关键优势对比：

分箱类型	是否利用标签	抗异常值能力	保持单调性	计算复杂度
等距分箱	否	弱	无保证	低
等频分箱	否	中	无保证	中
决策树分箱	是	强	有保证	高
卡方分箱	是	中	无保证	中

提示：在金融评分卡开发中，通常要求特征具有单调的WOE趋势，这时决策树分箱的优势尤为明显。

2. 决策树分箱实战

决策树分箱的核心思想是利用树模型的划分点作为分箱边界。这种方法天然具备特征选择能力，且能自动处理非线性关系。

2.1 基于sklearn的实现

from sklearn.tree import DecisionTreeClassifier
import numpy as np

def tree_binning(feature, target, max_leaf_nodes=5):
    # 确保输入为二维数组
    X = np.array(feature).reshape(-1, 1)
    y = np.array(target)
    
    # 训练决策树
    tree = DecisionTreeClassifier(
        max_leaf_nodes=max_leaf_nodes,
        random_state=42
    )
    tree.fit(X, y)
    
    # 提取分箱边界
    thresholds = tree.tree_.threshold[tree.tree_.threshold != -2]
    bins = np.sort(thresholds)
    
    # 添加最小最大值
    return np.concatenate([[-np.inf], bins, [np.inf]])

参数调优要点：

max_leaf_nodes：控制分箱数量，通常3-5个箱足够
min_samples_leaf：防止过拟合，建议设置为总样本的5%
class_weight：不平衡数据时设为"balanced"

2.2 分箱效果评估

决策树分箱后，我们需要检查WOE(Weight of Evidence)的单调性：

import pandas as pd

def calculate_woe(df, feature, target):
    df_grouped = df.groupby(feature)[target].agg(['count', 'sum'])
    df_grouped.columns = ['total', 'bad']
    df_grouped['good'] = df_grouped['total'] - df_grouped['bad']
    
    # 计算WOE和IV
    bad_rate_total = df_grouped['bad'].sum() / df_grouped['total'].sum()
    good_rate_total = 1 - bad_rate_total
    
    df_grouped['bad_rate'] = df_grouped['bad'] / df_grouped['total']
    df_grouped['good_rate'] = df_grouped['good'] / df_grouped['total']
    
    df_grouped['woe'] = np.log(
        (df_grouped['good'] / df_grouped['good'].sum()) / 
        (df_grouped['bad'] / df_grouped['bad'].sum())
    )
    
    df_grouped['iv'] = (
        (df_grouped['good'] / df_grouped['good'].sum()) - 
        (df_grouped['bad'] / df_grouped['bad'].sum())
    ) * df_grouped['woe']
    
    return df_grouped

3. 卡方分箱技术详解

卡方分箱基于相邻区间的卡方统计量进行合并，适用于分类变量和连续变量。与决策树分箱相比，它不强制要求单调性，但能更好地保留原始分布特征。

3.1 卡方分箱实现步骤

初始化：将连续变量排序后，每个值作为一个单独区间
计算卡方值：对每对相邻区间计算卡方统计量
合并区间：合并卡方值最小的相邻区间
停止条件：直到满足预设的箱数或卡方阈值

Python实现核心代码：

from scipy.stats import chi2_contingency

def chi2_merge(data, target, max_bins=5):
    # 初始化为每个唯一值一个箱
    unique_values = np.sort(data.unique())
    bins = [[val] for val in unique_values]
    
    while len(bins) > max_bins:
        min_chi2 = float('inf')
        best_idx = -1
        
        # 寻找最佳合并对
        for i in range(len(bins)-1):
            # 构建列联表
            contigency = pd.crosstab(
                data.isin(bins[i]+bins[i+1]),
                target
            )
            
            # 计算卡方值
            chi2 = chi2_contingency(contigency)[0]
            
            if chi2 < min_chi2:
                min_chi2 = chi2
                best_idx = i
                
        # 执行合并
        bins[best_idx] = bins[best_idx] + bins[best_idx+1]
        del bins[best_idx+1]
    
    # 转换为分箱边界
    boundaries = [min(bin_) for bin_ in bins]
    boundaries.append(max(bins[-1]))
    
    return sorted(boundaries)

3.2 卡方分箱的调优技巧

初始分箱：对于连续变量，可先进行等频预分箱(50-100箱)再合并
稀疏箱处理：合并样本量小于5%的箱
单调性修正：对WOE非单调的箱进行人工调整

4. 构建完整的分箱评估流水线

将分箱、WOE转换和IV评估整合为可复用的Pipeline，可以大幅提升特征工程效率。

4.1 自定义转换器实现

from sklearn.base import BaseEstimator, TransformerMixin

class WOETransformer(BaseEstimator, TransformerMixin):
    def __init__(self, binning_method='tree', n_bins=5):
        self.binning_method = binning_method
        self.n_bins = n_bins
        self.bin_dict = {}
        self.woe_dict = {}
        
    def fit(self, X, y):
        for col in X.columns:
            if self.binning_method == 'tree':
                bins = tree_binning(X[col], y, self.n_bins)
            else:
                bins = chi2_merge(X[col], y, self.n_bins)
                
            # 存储分箱边界
            self.bin_dict[col] = bins
            
            # 计算WOE
            binned = pd.cut(X[col], bins=bins, labels=False)
            woe_df = calculate_woe(
                pd.DataFrame({col: binned, 'target': y}),
                col, 'target'
            )
            self.woe_dict[col] = woe_df['woe'].to_dict()
            
        return self
    
    def transform(self, X):
        X_woe = X.copy()
        for col in X.columns:
            bins = self.bin_dict[col]
            woe_map = self.woe_dict[col]
            
            # 分箱并映射WOE值
            binned = pd.cut(
                X[col], 
                bins=bins, 
                labels=False,
                include_lowest=True
            )
            X_woe[col] = binned.map(woe_map)
            
        return X_woe

4.2 流水线集成示例

from sklearn.pipeline import Pipeline
from sklearn.linear_model import LogisticRegression

# 构建完整Pipeline
pipeline = Pipeline([
    ('woe_transform', WOETransformer(binning_method='chi2', n_bins=4)),
    ('model', LogisticRegression())
])

# 使用示例
pipeline.fit(X_train, y_train)
score = pipeline.score(X_test, y_test)