3. 滚动率矩阵构建：从逾期状态到转移概率
滚动率矩阵是贷后风险监控的核心工具，它量化了资产在不同逾期状态之间的转移概率。通过构建 M0→M1、M1→M2 等滚动率矩阵，我们可以精确捕捉资产质量的变化趋势，为后续的漂移告警阈值设定奠定数据基础。
3.1 滚动率矩阵的定义与业务含义
滚动率（Roll Rate）是指处于某一逾期状态的资产，在下一个观察周期转移到更差逾期状态的比例。例如，M0→M1 滚动率表示当前处于正常期（M0）的资产，在下一个月进入逾期 1-30 天（M1）的比例。
| 滚动率类型 | 业务含义 | 典型告警阈值 |
|---|---|---|
| M0→M1 | 正常客户首次逾期的比例 | 2%-5% |
| M1→M2 | 轻度逾期恶化为中度逾期的比例 | 30%-50% |
| M2→M3 | 中度逾期恶化为严重逾期的比例 | 50%-70% |
| M3→M4+ | 严重逾期进入坏账阶段的比例 | 70%-90% |
业务洞察：M0→M1 滚动率是资产质量的"前哨指标"，其微小波动往往预示着后续逾期链条的连锁反应。在贷后监控中，M0→M1 的告警阈值通常设置得最为敏感。
3.2 滚动率矩阵的标准化处理
原始滚动率矩阵直接由资产余额计算得出，但不同时间点的资产规模差异会导致矩阵不可比。标准化处理是消除规模效应、聚焦结构变化的关键步骤。
标准化方法一：余额归一化
- 将每个逾期状态的余额除以当期总资产余额，得到比例矩阵
- 公式：
P(i→j) = Balance(i→j) / Total_Balance(t) - 适用场景：跨期对比资产结构变化
标准化方法二：账户数归一化
- 将每个逾期状态的账户数除以当期总账户数，得到账户比例矩阵
- 公式：
P(i→j) = Count(i→j) / Total_Count(t) - 适用场景：消除大额账户对滚动率的扭曲影响
标准化方法三：Z-Score 标准化
- 对滚动率序列进行 Z-Score 变换，消除量纲和均值漂移
- 公式：
Z = (R - μ) / σ - 适用场景：多维度滚动率的综合监控
实战建议：在贷后监控系统中，建议同时维护余额归一化矩阵和账户数归一化矩阵。当两者出现背离时，往往意味着大额账户或小额账户出现了异常集中逾期，这是重要的风险信号。
3.3 滚动率矩阵的生成流程
构建滚动率矩阵需要遵循严格的时序对齐逻辑，确保每个资产在相邻两个观察周期的状态可追溯。以下是标准生成流程：
- 数据准备：获取连续两个月的资产逾期状态快照，包含资产唯一标识、逾期天数、余额、账户数等字段
- 状态映射：将逾期天数映射为 M0、M1、M2、M3、M4+ 等离散状态
- 交叉统计：统计上期状态为 i、本期状态为 j 的资产余额或账户数
- 矩阵填充：将统计结果填入 i×j 的二维矩阵
- 标准化处理：按选定的标准化方法对矩阵进行归一化
- 滚动率计算：计算每个状态转移到更差状态的比例
3.4 Python 代码实现矩阵生成
以下代码实现了完整的滚动率矩阵生成流程，包含数据模拟、状态映射、交叉统计、标准化处理和滚动率计算。代码采用面向对象设计，便于集成到贷后监控系统中。
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class RollRateMatrix:
"""滚动率矩阵生成器"""
def __init__(self, state_mapping=None):
"""
初始化滚动率矩阵生成器
Parameters:
-----------
state_mapping : dict
逾期天数到状态的映射字典
"""
self.state_mapping = state_mapping or {
'M0': (0, 0),
'M1': (1, 30),
'M2': (31, 60),
'M3': (61, 90),
'M4+': (91, 9999)
}
self.states = list(self.state_mapping.keys())
def map_state(self, overdue_days):
"""将逾期天数映射为状态"""
for state, (low, high) in self.state_mapping.items():
if low <= overdue_days <= high:
return state
return 'M4+'
def generate_sample_data(self, n_accounts=10000, seed=42):
"""
生成模拟数据用于演示
Parameters:
-----------
n_accounts : int
账户数量
seed : int
随机种子
Returns:
--------
pd.DataFrame
包含两个周期逾期状态的模拟数据
"""
np.random.seed(seed)
# 生成上期逾期天数
prev_overdue = np.random.choice(
[0, 5, 15, 45, 75, 120],
size=n_accounts,
p=[0.85, 0.05, 0.04, 0.03, 0.02, 0.01]
)
# 生成本期逾期天数（模拟转移）
current_overdue = []
for days in prev_overdue:
if days == 0:
# M0 客户：大部分保持正常，小部分逾期
if np.random.random() < 0.97:
current_overdue.append(0)
else:
current_overdue.append(np.random.randint(1, 31))
elif days <= 30:
# M1 客户：部分恶化，部分好转
if np.random.random() < 0.4:
current_overdue.append(days + np.random.randint(1, 31))
else:
current_overdue.append(0)
elif days <= 60:
# M2 客户：大部分恶化
if np.random.random() < 0.6:
current_overdue.append(days + np.random.randint(1, 31))
else:
current_overdue.append(0)
else:
# M3+ 客户：持续恶化
current_overdue.append(days + np.random.randint(1, 31))
# 生成余额数据
balances = np.random.lognormal(mean=10, sigma=1, size=n_accounts)
df = pd.DataFrame({
'account_id': range(n_accounts),
'prev_overdue_days': prev_overdue,
'current_overdue_days': current_overdue,
'balance': balances
})
# 映射状态
df['prev_state'] = df['prev_overdue_days'].apply(self.map_state)
df['current_state'] = df['current_overdue_days'].apply(self.map_state)
return df
def build_matrix(self, df, value_col='balance', normalize=True):
"""
构建滚动率矩阵
Parameters:
-----------
df : pd.DataFrame
包含 prev_state 和 current_state 的数据
value_col : str
用于统计的数值列（'balance' 或 'account_id'）
normalize : bool
是否进行标准化处理
Returns:
--------
pd.DataFrame
滚动率矩阵
"""
# 交叉统计
if value_col == 'account_id':
# 账户数口径：计数
pivot = pd.crosstab(
df['prev_state'],
df['current_state'],
rownames=['prev_state'],
colnames=['current_state']
)
else:
# 余额口径：求和
pivot = pd.pivot_table(
df,
values=value_col,
index='prev_state',
columns='current_state',
aggfunc='sum',
fill_value=0
)
# 确保所有状态都存在
for state in self.states:
if state not in pivot.index:
pivot.loc[state] = 0
if state not in pivot.columns:
pivot[state] = 0
# 按状态顺序排序
pivot = pivot.reindex(index=self.states, columns=self.states, fill_value=0)
if normalize:
# 余额归一化：每行除以该行总和
row_sums = pivot.sum(axis=1)
pivot = pivot.div(row_sums, axis=0).fillna(0)
return pivot
def calculate_roll_rates(self, matrix):
"""
从矩阵中提取滚动率（转移到更差状态的比例）
Parameters:
-----------
matrix : pd.DataFrame
标准化后的滚动率矩阵
Returns:
--------
dict
各状态滚动率
"""
roll_rates = {}
state_order = self.states
for i, state in enumerate(state_order[:-1]):
# 转移到更差状态：当前状态之后的所有列
worse_states = state_order[i+1:]
roll_rate = matrix.loc[state, worse_states].sum()
roll_rates[f'{state}→{state_order[i+1]}'] = round(roll_rate, 4)
return roll_rates
def run_pipeline(self, n_accounts=10000):
"""
运行完整流水线：生成数据 → 构建矩阵 → 计算滚动率
Parameters:
-----------
n_accounts : int
账户数量
Returns:
--------
tuple
(原始矩阵, 标准化矩阵, 滚动率字典)
"""
# 生成数据
df = self.generate_sample_data(n_accounts)
# 构建余额口径矩阵
raw_matrix = self.build_matrix(df, value_col='balance', normalize=False)
norm_matrix = self.build_matrix(df, value_col='balance', normalize=True)
# 计算滚动率
roll_rates = self.calculate_roll_rates(norm_matrix)
return raw_matrix, norm_matrix, roll_rates
# ========== 使用示例 ==========
if __name__ == '__main__':
# 初始化生成器
rrm = RollRateMatrix()
# 运行流水线
raw, norm, rates = rrm.run_pipeline(n_accounts=50000)
print("=== 原始矩阵（余额口径）===")
print(raw.round(0).astype(int))
print("\n=== 标准化矩阵（余额归一化）===")
print(norm.round(4))
print("\n=== 滚动率 ===")
for key, value in rates.items():
print(f"{key}: {value:.2%}")3.5 矩阵输出与结果解读
运行上述代码后，将输出三个核心结果：原始矩阵展示资产余额的绝对分布，标准化矩阵揭示结构比例，滚动率字典则直接给出各阶段的转移概率。
典型输出示例：
=== 原始矩阵（余额口径）===
current_state M0 M1 M2 M3 M4+
prev_state
M0 42500000 850000 425000 85000 42500
M1 500000 1500000 500000 250000 250000
M2 200000 300000 800000 400000 300000
M3 100000 150000 250000 500000 500000
M4+ 50000 50000 100000 200000 1200000
=== 标准化矩阵（余额归一化）===
current_state M0 M1 M2 M3 M4+
prev_state
M0 0.9650 0.0193 0.0096 0.0019 0.0010
M1 0.1667 0.5000 0.1667 0.0833 0.0833
M2 0.1000 0.1500 0.4000 0.2000 0.1500
M3 0.0667 0.1000 0.1667 0.3333 0.3333
M4+ 0.0312 0.0312 0.0625 0.1250 0.7500
=== 滚动率 ===
M0→M1: 1.93%
M1→M2: 16.67%
M2→M3: 20.00%
M3→M4+: 33.33%
常见陷阱：
- 状态定义不一致：不同业务线的逾期天数阈值可能不同，务必在矩阵构建前统一状态映射规则
- 口径混用：余额口径和账户数口径的滚动率差异可能很大，分析时需明确标注口径类型
- 小样本偏差：当某个状态的账户数过少时（如 M3 只有 10 个账户），滚动率会极度不稳定，建议设置最小样本量阈值
- 时间窗口对齐：确保两个观察周期的时间间隔一致（如均为 30 天），否则滚动率不可比
3.6 矩阵质量验证
构建完成的滚动率矩阵需要经过以下验证，确保数据质量可靠：
| 验证项 | 验证方法 | 合格标准 |
|---|---|---|
| 行和完整性 | 检查标准化矩阵每行之和是否为 1 | 每行和 ∈ [0.999, 1.001] |
| 状态覆盖性 | 检查所有状态是否在矩阵中出现 | 所有预定义状态均存在 |
| 单调性检验 | 滚动率应随逾期程度加深而递增 | M0→M1 < M1→M2 < M2→M3 < M3→M4+ |
| 历史对比 | 与历史同期矩阵对比偏差 | 偏差不超过 3 个标准差 |
效率技巧：在贷后监控系统中，建议将矩阵构建逻辑封装为定时任务（如每日凌晨执行），并将结果存入数据库。前端展示时直接读取预计算矩阵，避免每次查询都重新计算，可大幅提升系统响应速度。