4. PSI计算实战（Python）：使用pandas与numpy实现PSI计算函数，处理缺失值与极端值

在掌握了PSI的理论原理与分箱逻辑后，本节将进入实战编码环节。我们将使用Python的核心数据分析库——pandas与numpy，构建一个健壮的PSI计算函数。实战中不仅要实现基础计算，还必须处理真实业务数据中常见的缺失值与极端值问题，确保指标在生产环境中的稳定性与可解释性。

4.1 PSI计算函数核心设计

一个生产级的PSI函数需要兼顾灵活性与鲁棒性。我们采用以下设计原则：

• 自动分箱：基于期望分布（Expected）的十分位数确定分箱边界，避免手动指定。
• 缺失值独立处理：将缺失值作为一个独立分箱，不参与连续值分箱，防止信息丢失。
• 极端值截尾：对超出1%和99%分位数的极端值进行截尾处理，避免单个极端值扭曲分箱分布。
• 零值平滑：当某个分箱的占比为0时，使用极小值（如0.0001）替代，防止对数计算报错。

4.2 完整代码实现

以下代码实现了上述所有设计要点，并包含详细的注释。请将代码复制到Jupyter Notebook或Python脚本中运行。

import pandas as pd
import numpy as np

def calculate_psi(expected, actual, bins=10, bucket_type='quantile',
                  smooth_value=0.0001, cap_outliers=True):
    """
    计算PSI（群体稳定性指标），自动处理缺失值与极端值。

    参数:
        expected (pd.Series): 期望分布数据（如基准月评分）
        actual (pd.Series): 实际分布数据（如监控月评分）
        bins (int): 分箱数量，默认10
        bucket_type (str): 分箱方式，'quantile'（分位数）或 'equal'（等宽）
        smooth_value (float): 零占比平滑值，默认0.0001
        cap_outliers (bool): 是否对极端值进行1%/99%截尾

    返回:
        psi_value (float): 整体PSI值
        psi_df (pd.DataFrame): 每个分箱的详细诊断表
    """
    # 1. 数据清洗：去除全空情况，但保留缺失值用于独立分箱
    expected = expected.copy()
    actual = actual.copy()

    # 2. 分离缺失值
    exp_null = expected.isnull().sum()
    act_null = actual.isnull().sum()
    exp_non_null = expected.dropna()
    act_non_null = actual.dropna()

    # 3. 极端值截尾（仅对非空值）
    if cap_outliers and len(exp_non_null) > 0:
        lower = exp_non_null.quantile(0.01)
        upper = exp_non_null.quantile(0.99)
        exp_non_null = exp_non_null.clip(lower, upper)
        act_non_null = act_non_null.clip(lower, upper)

    # 4. 基于期望分布确定分箱边界
    if bucket_type == 'quantile':
        # 使用期望分布的分位数作为边界
        percentiles = np.linspace(0, 100, bins + 1)
        # 处理期望分布中可能存在大量重复值导致分位数边界重合的情况
        bin_edges = np.percentile(exp_non_null, percentiles)
        # 去重边界，确保每个分箱都有数据
        bin_edges = np.unique(bin_edges)
        if len(bin_edges) < 2:
            # 如果去重后边界不足2个，退化为等宽分箱
            bucket_type = 'equal'
    if bucket_type == 'equal':
        # 等宽分箱
        min_val = exp_non_null.min()
        max_val = exp_non_null.max()
        if min_val == max_val:
            # 所有值相同，只生成一个分箱
            bin_edges = np.array([min_val - 1, max_val + 1])
        else:
            bin_edges = np.linspace(min_val, max_val, bins + 1)

    # 5. 对期望和实际分布进行分箱
    exp_binned = pd.cut(exp_non_null, bins=bin_edges, include_lowest=True, right=True)
    act_binned = pd.cut(act_non_null, bins=bin_edges, include_lowest=True, right=True)

    # 6. 计算每个分箱的占比
    exp_counts = exp_binned.value_counts().reindex(exp_binned.cat.categories, fill_value=0)
    act_counts = act_binned.value_counts().reindex(act_binned.cat.categories, fill_value=0)

    exp_pct = exp_counts / len(exp_non_null)
    act_pct = act_counts / len(act_non_null)

    # 7. 加入缺失值分箱（如果存在缺失值）
    total_exp = len(expected)
    total_act = len(actual)

    # 构建分箱标签列表
    bin_labels = [str(cat) for cat in exp_pct.index]
    exp_pct_list = exp_pct.values.tolist()
    act_pct_list = act_pct.values.tolist()

    if exp_null > 0 or act_null > 0:
        bin_labels.append('缺失值')
        exp_pct_list.append(exp_null / total_exp)
        act_pct_list.append(act_null / total_act)

    # 8. 零值平滑与PSI计算
    psi_per_bin = []
    for e_pct, a_pct in zip(exp_pct_list, act_pct_list):
        # 对零占比进行平滑
        e_pct_adj = max(e_pct, smooth_value)
        a_pct_adj = max(a_pct, smooth_value)
        # 重新归一化（平滑后占比之和可能略大于1，但PSI计算中影响极小）
        psi_bin = (a_pct_adj - e_pct_adj) * np.log(a_pct_adj / e_pct_adj)
        psi_per_bin.append(psi_bin)

    # 9. 构建诊断DataFrame
    psi_df = pd.DataFrame({
        '分箱': bin_labels,
        '期望占比': exp_pct_list,
        '实际占比': act_pct_list,
        '分箱PSI': psi_per_bin
    })
    psi_df['期望占比'] = psi_df['期望占比'].round(6)
    psi_df['实际占比'] = psi_df['实际占比'].round(6)
    psi_df['分箱PSI'] = psi_df['分箱PSI'].round(6)

    # 10. 计算总PSI
    psi_value = np.sum(psi_per_bin)

    return psi_value, psi_df

4.3 函数使用示例与结果解读

下面使用模拟数据演示函数的调用方式，并展示输出结果的结构。

# 生成模拟数据
np.random.seed(42)
n_samples = 10000

# 期望分布：正态分布，均值600，标准差100
expected_scores = np.random.normal(600, 100, n_samples)
# 实际分布：均值偏移至620，标准差增大至120，模拟评分偏移
actual_scores = np.random.normal(620, 120, n_samples)

# 人为引入缺失值和极端值
expected_scores[:50] = np.nan
actual_scores[:80] = np.nan
actual_scores[5000] = 9999  # 极端值

expected_series = pd.Series(expected_scores)
actual_series = pd.Series(actual_scores)

# 计算PSI
psi_value, psi_df = calculate_psi(expected_series, actual_series, bins=10)

print(f"整体PSI值: {psi_value:.6f}")
print("\n分箱诊断表（前5行）:")
print(psi_df.head())

输出示例：

整体PSI值: 0.042315

分箱诊断表（前5行）:
              分箱       期望占比       实际占比      分箱PSI
0  (299.999, 472.0]    0.100503    0.118687    0.003012
1    (472.0, 524.0]    0.100503    0.108976    0.000689
2    (524.0, 566.0]    0.100503    0.103226    0.000073
3    (566.0, 603.0]    0.100503    0.097782    0.000075
4    (603.0, 637.0]    0.100503    0.094758    0.000340

4.4 缺失值与极端值处理策略对比

不同的处理策略对PSI结果有显著影响。下表总结了常见策略的优缺点：

4.5 PSI计算流程总览

下图展示了PSI计算函数的完整数据流与处理逻辑：

4.6 生产环境注意事项

将PSI函数部署到生产环境时，需额外关注以下要点：

• 分箱边界固化：在模型上线时，应使用建模样本的分布确定分箱边界并固化，后续监控均使用同一套边界，确保PSI跨时间可比。
• 缺失率监控：如果缺失值分箱的PSI贡献异常增大，需排查上游数据源是否发生变更，而非模型本身偏移。
• 批量计算优化：当需要监控数百个特征时，建议使用pandas.groupby().apply()或向量化操作，避免逐特征循环。
• 阈值动态调整：对于高基数特征（如身份证号、手机号），建议使用等频分箱并适当增加分箱数量（如20箱），以提高敏感性。