Python实战：用代码拆解风控三大核心指标的计算逻辑
信贷风控领域有三个如同"三体运动"般相互关联的核心指标——Vintage分析、滚动率矩阵和迁移率表。许多从业者虽然能背诵它们的定义，但当面对真实的业务数据时，却常常陷入"纸上谈兵"的困境。本文将以一个模拟的信贷数据集为例，手把手带你用Python实现这三个指标的计算与可视化，让你真正掌握从原始数据到业务洞见的完整链条。
1. 环境准备与数据模拟
1.1 工具库导入
工欲善其事，必先利其器。我们需要以下Python库来完成本次分析：
import pandas as pd import numpy as np from datetime import datetime, timedelta import matplotlib.pyplot as plt import seaborn as sns from matplotlib.ticker import PercentFormatter1.2 模拟信贷数据集生成
由于真实业务数据涉及商业机密，我们模拟生成一个包含10,000笔贷款记录的虚拟数据集：
np.random.seed(2023) # 生成放款日期（集中在2022年1月-12月） loan_dates = pd.date_range('2022-01-01', '2022-12-31', freq='D') loan_df = pd.DataFrame({ 'loan_id': [f'L{str(i).zfill(5)}' for i in range(1, 10001)], 'loan_date': np.random.choice(loan_dates, 10000), 'loan_amount': np.random.lognormal(mean=7.5, sigma=0.3, size=10000).round(2), 'term': 12 # 假设都是12期产品 }) # 为每笔贷款生成还款计划表 repayment_records = [] for _, row in loan_df.iterrows(): for i in range(1, row['term']+1): due_date = row['loan_date'] + pd.DateOffset(months=i) repay_date = due_date + pd.DateOffset(days=np.random.choice( [0, 0, 0, 0, 1, 2, 3, 5, 7, 15, 30, 60, 90], p=[0.6,0.1,0.05,0.05,0.05,0.03,0.02,0.02,0.02,0.02,0.02,0.01,0.01] )) repayment_records.append({ 'loan_id': row['loan_id'], 'due_date': due_date, 'repay_date': repay_date, 'term_no': i }) repay_df = pd.DataFrame(repayment_records) repay_df['days_past_due'] = (repay_df['repay_date'] - repay_df['due_date']).dt.days repay_df['dpd_status'] = repay_df['days_past_due'].apply( lambda x: 'M0' if x <=0 else 'M1' if x<=30 else 'M2' if x<=60 else 'M3' if x<=90 else 'M4+' )2. Vintage分析实战
2.1 计算各账龄的逾期率
Vintage分析的核心是按放款月份对齐，观察不同"出生队列"的资产质量随时间的变化：
# 计算每笔贷款在每个MOB的最坏逾期状态 mob_status = repay_df.groupby(['loan_id', 'term_no'])['dpd_status'].max().unstack() # 合并放款信息 vintage_df = loan_df.merge(mob_status, left_on='loan_id', right_index=True) # 计算各MOB的M4+逾期率 def calculate_vintage(df, mob_max=12): result = [] for mob in range(1, mob_max+1): temp = df.groupby(pd.to_datetime(df['loan_date']).dt.to_period('M'))[mob].apply( lambda x: (x=='M4+').mean() ).reset_index() temp['MOB'] = mob result.append(temp) return pd.concat(result) vintage_result = calculate_vintage(vintage_df)2.2 Vintage曲线可视化
将计算结果可视化，可以直观看到不同月份放款资产的质量演变：
plt.figure(figsize=(12, 6)) for cohort in vintage_result['loan_date'].unique(): cohort_data = vintage_result[vintage_result['loan_date']==cohort] plt.plot(cohort_data['MOB'], cohort_data[0]*100, label=cohort.strftime('%Y-%m'), marker='o') plt.title('Vintage Analysis - M4+ Delinquency Rate (%)') plt.xlabel('Month on Book (MOB)') plt.ylabel('M4+ Delinquency Rate (%)') plt.grid(True) plt.legend(bbox_to_anchor=(1.05, 1), loc='upper left') plt.tight_layout() plt.show()关键业务解读：
- 成熟期判断：当曲线趋于平缓时（通常MOB8-9），说明资产质量基本稳定
- 队列对比：2022年下半年放款的资产质量明显优于上半年，可能反映了风控策略优化
- 早期预警：MOB3前的曲线陡峭程度反映了短期风险捕捉能力
3. 滚动率矩阵构建
3.1 滚动率计算逻辑
滚动率分析揭示了客户逾期状态间的转化规律，是定义好坏客户的重要依据：
def calculate_roll_rate(repay_df, observation_date='2023-03-31', window=6): # 确定观察点和窗口期 obs_date = pd.to_datetime(observation_date) lookback_start = obs_date - pd.DateOffset(months=window) lookforward_end = obs_date + pd.DateOffset(months=window) # 获取观察期内的最坏状态 lookback_status = repay_df[ (repay_df['due_date'] > lookback_start) & (repay_df['due_date'] <= obs_date) ].groupby('loan_id')['dpd_status'].max().reset_index() lookback_status.columns = ['loan_id', 'lookback_status'] # 获取表现期内的最坏状态 lookforward_status = repay_df[ (repay_df['due_date'] > obs_date) & (repay_df['due_date'] <= lookforward_end) ].groupby('loan_id')['dpd_status'].max().reset_index() lookforward_status.columns = ['loan_id', 'lookforward_status'] # 合并结果 roll_rate_df = lookback_status.merge(lookforward_status, on='loan_id', how='inner') # 生成滚动率矩阵 roll_matrix = pd.crosstab( index=roll_rate_df['lookback_status'], columns=roll_rate_df['lookforward_status'], normalize='index' ).round(4)*100 return roll_matrix roll_rate_matrix = calculate_roll_rate(repay_df)3.2 滚动率矩阵可视化
用热力图展示状态间的转化概率，业务洞见一目了然：
plt.figure(figsize=(10, 6)) sns.heatmap(roll_rate_matrix, annot=True, fmt='.1f', cmap='Blues', cbar_kws={'format': PercentFormatter()}) plt.title('Roll Rate Matrix (%)') plt.xlabel('Forward Status (6 months after observation)') plt.ylabel('Lookback Status (6 months before observation)') plt.tight_layout() plt.show()关键业务结论：
- M4+客户有92%的概率保持坏账状态，是典型的"坏客户"
- M3客户有65%的概率恶化为M4+，需要重点关注
- M0客户有96%的概率保持正常，说明整体资产质量良好
4. 迁移率分析实现
4.1 迁移率计算步骤
迁移率分析展示了逾期状态的动态流转过程，是预测未来坏账的基础：
def calculate_flow_rate(repay_df): # 按月统计各状态的贷款余额 monthly_status = repay_df.groupby([ pd.to_datetime(repay_df['due_date']).dt.to_period('M'), 'dpd_status' ])['loan_id'].nunique().unstack().fillna(0) # 计算迁移率 flow_rates = {} status_order = ['M0', 'M1', 'M2', 'M3', 'M4+'] for i in range(len(status_order)-1): from_status = status_order[i] to_status = status_order[i+1] flow_rate = monthly_status[to_status].shift(-1) / monthly_status[from_status] flow_rates[f'{from_status}-{to_status}'] = flow_rate.mean() return pd.DataFrame.from_dict(flow_rates, orient='index', columns=['flow_rate']) flow_rates = calculate_flow_rate(repay_df)4.2 迁移路径分析
将迁移率结果可视化，可以清晰看到风险的传导路径：
flow_rates.plot(kind='bar', figsize=(10, 5), legend=False) plt.title('Average Flow Rates Between Statuses') plt.ylabel('Flow Rate') plt.xticks(rotation=45) plt.grid(axis='y') plt.tight_layout() plt.show()关键业务应用：
- 催收策略优化：M1-M2迁移率较高，说明早期催收效果不佳
- 损失预测：结合各阶段迁移率，可以计算预期坏账损失
- 风险预警：特定月份迁移率异常波动可能反映外部经济环境变化
5. 指标联动与业务应用
5.1 三大指标的关系图谱
将这三大指标整合到一个分析框架中，可以形成完整的风险管理闭环：
Vintage分析 → 确定账户成熟期（表现期长度） ↓ 滚动率分析 → 定义好坏客户标准 ↓ 迁移率分析 → 预测未来坏账损失5.2 自动化监控报表实现
将上述分析过程封装成可复用的函数，方便定期生成监控报表：
def generate_risk_report(repay_data, loan_data, report_date): # Vintage分析 vintage_result = calculate_vintage(loan_data.merge( repay_data.groupby(['loan_id', 'term_no'])['dpd_status'].max().unstack(), left_on='loan_id', right_index=True )) # 滚动率分析 roll_matrix = calculate_roll_rate(repay_data, observation_date=report_date) # 迁移率分析 flow_rates = calculate_flow_rate(repay_data) return { 'vintage': vintage_result, 'roll_rate': roll_matrix, 'flow_rate': flow_rates } latest_report = generate_risk_report(repay_df, loan_df, '2023-06-30')在实际项目中，我会将这个报表系统与调度工具（如Airflow）集成，实现风险指标的自动化监控。当M1-M2迁移率超过阈值时自动触发预警，帮助团队及时发现风险苗头。
