4. PSI计算工具与代码实现：使用Python（pandas、numpy）实现PSI计算函数、自动化脚本框架

在催收触达监控体系中，PSI（Population Stability Index）的计算需要高效、可复用的工具支撑。本节将提供基于Python的PSI计算函数实现，并构建一套适用于催收场景的自动化监控脚本框架，帮助团队快速部署稳定性监控。

4.1 PSI计算核心函数

PSI的计算本质是衡量两个分布（实际分布 vs 预期分布）在各分箱上的偏差。以下函数支持等频分箱与等距分箱两种模式，并内置了分箱合并与零值处理逻辑。

import pandas as pd
import numpy as np

def calculate_psi(actual, expected, bins=10, method='quantile'):
    """
    计算群体稳定性指数 (PSI)
    
    参数:
        actual (pd.Series): 实际分布数据（如当前周期触达率）
        expected (pd.Series): 预期分布数据（如基准周期触达率）
        bins (int): 分箱数量，默认10
        method (str): 分箱方法，'quantile' 等频分箱，'uniform' 等距分箱
    
    返回:
        float: PSI值
        pd.DataFrame: 分箱明细表
    """
    # 合并数据，确定分箱边界
    all_data = pd.concat([actual, expected], ignore_index=True)
    
    if method == 'quantile':
        # 等频分箱：基于预期数据的分位数
        bin_edges = np.percentile(expected.dropna(), np.linspace(0, 100, bins + 1))
    elif method == 'uniform':
        # 等距分箱：基于预期数据的范围
        bin_edges = np.linspace(expected.min(), expected.max(), bins + 1)
    else:
        raise ValueError("method 参数仅支持 'quantile' 或 'uniform'")
    
    # 确保边界唯一，避免分箱失败
    bin_edges = np.unique(bin_edges)
    if len(bin_edges) < 2:
        return 0.0, pd.DataFrame()
    
    # 分箱标签
    labels = [f'Bin_{i+1}' for i in range(len(bin_edges) - 1)]
    
    # 对实际和预期数据分别分箱
    actual_bins = pd.cut(actual, bins=bin_edges, labels=labels, include_lowest=True, right=False)
    expected_bins = pd.cut(expected, bins=bin_edges, labels=labels, include_lowest=True, right=False)
    
    # 计算各箱占比
    actual_pct = actual_bins.value_counts(normalize=True).reindex(labels, fill_value=0)
    expected_pct = expected_bins.value_counts(normalize=True).reindex(labels, fill_value=0)
    
    # 零值平滑处理：将0替换为极小值，避免除零
    epsilon = 1e-6
    actual_pct = actual_pct.replace(0, epsilon)
    expected_pct = expected_pct.replace(0, epsilon)
    
    # 计算各箱PSI贡献
    psi_components = (actual_pct - expected_pct) * np.log(actual_pct / expected_pct)
    psi_total = psi_components.sum()
    
    # 构建明细表
    detail_df = pd.DataFrame({
        '分箱': labels,
        '实际占比': actual_pct.values,
        '预期占比': expected_pct.values,
        'PSI贡献': psi_components.values
    })
    
    return round(psi_total, 6), detail_df
函数设计要点：
  • 分箱方法选择：催收触达率通常呈偏态分布，推荐使用 quantile 等频分箱，确保每箱样本量均衡。
  • 零值平滑：当某箱实际占比为0时，替换为 1e-6，避免对数计算报错。
  • 边界唯一性：当数据高度集中时，分位数可能重复，使用 np.unique 自动合并重复边界。

4.2 自动化监控脚本框架

在催收运营中，PSI监控需要按日/周周期运行，并自动输出告警。以下框架封装了数据加载、PSI计算、阈值判断与结果存储四个环节。

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class PSIMonitor:
    """催收触达PSI自动化监控框架"""
    
    def __init__(self, config):
        self.config = config
        self.psi_threshold = config.get('psi_threshold', 0.1)
        self.warning_threshold = config.get('warning_threshold', 0.25)
        self.bins = config.get('bins', 10)
        self.method = config.get('method', 'quantile')
    
    def load_data(self, date_str):
        """模拟加载指定日期的催收触达数据"""
        # 实际场景中替换为数据库查询或文件读取
        # 此处生成示例数据
        np.random.seed(hash(date_str) % 2**32)
        n = 10000
        data = pd.DataFrame({
            'contact_rate': np.random.beta(2, 5, n),  # 模拟触达率分布
            'customer_id': range(n)
        })
        return data
    
    def get_expected_distribution(self):
        """获取基准分布（如上一周或上月同期）"""
        # 实际场景中从基准表读取
        expected_data = self.load_data('2025-01-01')
        return expected_data['contact_rate']
    
    def run_daily_check(self, date_str):
        """执行单日PSI检查"""
        # 加载当日数据
        actual_data = self.load_data(date_str)
        actual_series = actual_data['contact_rate']
        
        # 获取基准分布
        expected_series = self.get_expected_distribution()
        
        # 计算PSI
        psi_value, detail_df = calculate_psi(
            actual_series, expected_series,
            bins=self.bins, method=self.method
        )
        
        # 阈值判断
        status = '正常'
        if psi_value > self.warning_threshold:
            status = '严重漂移'
        elif psi_value > self.psi_threshold:
            status = '轻度漂移'
        
        # 记录结果
        result = {
            'date': date_str,
            'psi': psi_value,
            'status': status,
            'sample_size': len(actual_series),
            'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        }
        
        return result, detail_df
    
    def run_batch(self, start_date, end_date):
        """批量运行多日监控"""
        date_range = pd.date_range(start=start_date, end=end_date, freq='D')
        results = []
        
        for single_date in date_range:
            date_str = single_date.strftime('%Y-%m-%d')
            result, _ = self.run_daily_check(date_str)
            results.append(result)
            print(f"[{date_str}] PSI = {result['psi']:.4f}, 状态: {result['status']}")
        
        results_df = pd.DataFrame(results)
        return results_df

# 使用示例
if __name__ == '__main__':
    config = {
        'psi_threshold': 0.1,
        'warning_threshold': 0.25,
        'bins': 10,
        'method': 'quantile'
    }
    monitor = PSIMonitor(config)
    # 单日检查
    result, detail = monitor.run_daily_check('2025-03-15')
    print(f"单日PSI: {result['psi']}, 状态: {result['status']}")
    # 批量检查
    batch_results = monitor.run_batch('2025-03-01', '2025-03-15')
框架扩展建议：
  • 数据源接入：将 load_data 方法替换为数据库查询（如MySQL、ClickHouse），支持按催收策略分组计算。
  • 告警推送：在 run_daily_check 中集成钉钉/企业微信机器人，当状态为“严重漂移”时自动发送告警。
  • 结果持久化：将 results_df 写入监控结果表，用于长期趋势分析。

4.3 PSI计算流程与数据流转

下图展示了从原始触达数据到PSI监控结果的完整数据流转与计算流程，帮助理解各模块之间的依赖关系。

PSI计算流程与数据流转图 PSI计算流程与数据流转 原始触达数据 数据预处理 分箱计算 基准分布 实际分布 PSI计算 阈值判断 告警输出 监控看板 说明：分箱计算同时输出基准分布与实际分布的分箱结果，PSI计算基于两分布在各箱的占比差异

4.4 参数配置与阈值调整

PSI监控框架的灵活性体现在参数可配置化。以下表格列出了核心参数及其推荐配置，团队可根据催收策略的敏感度进行调整。

参数名称 默认值 推荐范围 说明
psi_threshold 0.1 0.05 ~ 0.15 轻度漂移阈值，超过此值触发黄色告警
warning_threshold 0.25 0.20 ~ 0.35 严重漂移阈值，超过此值触发红色告警
bins 10 5 ~ 20 分箱数量，数据量大时可适当增加
method quantile quantile / uniform 分箱方法，偏态分布推荐 quantile
epsilon 1e-6 1e-8 ~ 1e-4 零值平滑系数，避免除零错误
阈值调整注意事项：
  • 策略敏感度：对于高价值客户催收策略，建议降低 psi_threshold 至 0.05，提前发现分布偏移。
  • 分箱数量：分箱过少会掩盖局部偏移，分箱过多则对噪声敏感。建议根据样本量动态调整：样本量 > 10万时使用 20 箱，样本量 < 1万时使用 5 箱。
  • 基准更新频率：基准分布建议每周或每月更新一次，避免因业务周期性波动导致误告警。

4.5 批量运行与结果输出

以下代码展示了如何调用 PSIMonitor 框架进行批量监控，并将结果保存为CSV文件，便于后续分析。

# 批量运行并保存结果
monitor = PSIMonitor(config)
batch_results = monitor.run_batch('2025-03-01', '2025-03-15')

# 保存监控结果
batch_results.to_csv('psi_monitor_results.csv', index=False)
print("监控结果已保存至 psi_monitor_results.csv")

# 查看漂移日期
drifted_days = batch_results[batch_results['status'] != '正常']
print(f"漂移天数: {len(drifted_days)}")
print(drifted_days[['date', 'psi', 'status']])
结果分析建议：
  • 趋势分析：将每日PSI值绘制成折线图，观察漂移是否呈持续上升趋势。
  • 归因分析：当PSI超过阈值时，查看 detail_df 中各分箱的PSI贡献，定位具体偏移区间。
  • 策略联动：将PSI监控结果与催收策略调整记录关联，评估策略变更对分布稳定性的影响。

4.6 代码实现常见问题与避坑

常见问题 原因 解决方案
PSI计算结果为 NaN 分箱后某箱占比为0，对数计算报错 使用 epsilon 平滑零值，或合并空箱
分箱边界重复导致分箱失败 数据高度集中，分位数重复 使用 np.unique 去重，或增加分箱数量
PSI值异常大（>1.0） 实际分布与基准分布差异极大，或分箱方法不当 检查数据源是否异常，改用等频分箱
批量运行速度慢 每次运行都重新加载基准分布 将基准分布缓存为全局变量，或从数据库一次性读取
← 上一章 📖 返回目录 下一章 →