滚动率计算逻辑

滚动率（Roll Rate）是贷后风险监控的核心指标，它衡量资产从当前逾期状态向更严重逾期状态迁移的比例。准确计算滚动率是设定漂移告警阈值的前提。本节将详细讲解两种主流的滚动率计算方法：基于账龄（Vintage）和基于观测点（Observation Point），并给出可直接复用的 Python 实现。

一、基于账龄（Vintage）的滚动率计算

Vintage 分析以“放款月份”为分组维度，追踪同一批客户在后续各月中的逾期表现。其滚动率计算关注的是：同一 Vintage 的资产，在特定账龄（如第 6 个月）时，从 M0 滚动到 M1 的比例。

计算步骤：

1. 分组：按放款月份（Vintage）对资产进行分组。
2. 确定观测账龄：选定一个固定的账龄月数（例如，观察第 6 个月的表现）。
3. 统计状态：统计该 Vintage 在账龄第 N 个月时，处于 M0（正常）和 M1（逾期 1-30 天）的资产数量。
4. 计算迁移：统计该 Vintage 在账龄第 N+1 个月时，从 M0 变为 M1 的资产数量。
5. 计算比率：滚动率 = 从 M0 迁移至 M1 的资产数 / 账龄第 N 个月时 M0 的资产总数。

二、基于观测点（Observation Point）的滚动率计算

观测点法以“固定日历时间”为锚点，计算在特定时间点处于某一逾期状态的资产，在下一个月末的迁移情况。这是贷后监控中最常用的实时滚动率计算方法。

计算步骤：

1. 确定观测点：选定一个具体的月末日期（例如 2024-01-31）。
2. 统计期初状态：统计观测点当天，所有资产所处的逾期状态（M0, M1, M2...）。
3. 统计期末状态：统计观测点下一个月末（例如 2024-02-29），同一批资产所处的逾期状态。
4. 构建迁移矩阵：交叉统计期初状态与期末状态，形成 N x N 的迁移矩阵。
5. 计算比率：例如，M0→M1 滚动率 = 期初为 M0 且期末变为 M1 的资产数 / 期初 M0 的总资产数。

三、Python 实现滚动率计算函数

下面提供一个通用的 Python 函数，支持上述两种计算模式。该函数基于 Pandas 实现，适用于生产环境中的批量计算。

import pandas as pd
import numpy as np

def calculate_roll_rate(df, method='observation_point', 
                        vintage_col='vintage_month',
                        obs_date_col='obs_date',
                        current_state_col='current_state',
                        next_state_col='next_state',
                        target_transition=('M0', 'M1')):
    """
    计算滚动率（支持 Vintage 和观测点两种模式）
    
    Parameters
    ----------
    df : pd.DataFrame
        输入数据，必须包含期初状态和期末状态
    method : str
        'vintage' 或 'observation_point'
    vintage_col : str
        放款月份列名（仅 vintage 模式需要）
    obs_date_col : str
        观测日期列名（仅 observation_point 模式需要）
    current_state_col : str
        期初逾期状态列名
    next_state_col : str
        期末逾期状态列名
    target_transition : tuple
        目标迁移路径，如 ('M0', 'M1')
    
    Returns
    -------
    pd.DataFrame
        包含滚动率计算结果的数据框
    """
    from_state, to_state = target_transition
    
    if method == 'observation_point':
        # 按观测点分组计算
        result = []
        for obs_date, group in df.groupby(obs_date_col):
            # 期初处于 from_state 的资产
            initial_count = len(group[group[current_state_col] == from_state])
            if initial_count == 0:
                continue
            # 期末迁移到 to_state 的资产
            transition_count = len(group[
                (group[current_state_col] == from_state) & 
                (group[next_state_col] == to_state)
            ])
            roll_rate = transition_count / initial_count
            result.append({
                'obs_date': obs_date,
                'from_state': from_state,
                'to_state': to_state,
                'initial_count': initial_count,
                'transition_count': transition_count,
                'roll_rate': round(roll_rate, 4)
            })
        return pd.DataFrame(result)
    
    elif method == 'vintage':
        # 按 Vintage 分组计算
        result = []
        for vintage, group in df.groupby(vintage_col):
            initial_count = len(group[group[current_state_col] == from_state])
            if initial_count == 0:
                continue
            transition_count = len(group[
                (group[current_state_col] == from_state) & 
                (group[next_state_col] == to_state)
            ])
            roll_rate = transition_count / initial_count
            result.append({
                'vintage_month': vintage,
                'from_state': from_state,
                'to_state': to_state,
                'initial_count': initial_count,
                'transition_count': transition_count,
                'roll_rate': round(roll_rate, 4)
            })
        return pd.DataFrame(result)
    
    else:
        raise ValueError("method 参数必须为 'vintage' 或 'observation_point'")

# 示例用法
if __name__ == "__main__":
    # 构造示例数据
    sample_data = pd.DataFrame({
        'obs_date': ['2024-01-31'] * 1000 + ['2024-02-29'] * 1000,
        'current_state': ['M0'] * 800 + ['M1'] * 200 + ['M0'] * 750 + ['M1'] * 250,
        'next_state': ['M0'] * 700 + ['M1'] * 100 + ['M0'] * 50 + ['M1'] * 150 + 
                      ['M0'] * 650 + ['M1'] * 100 + ['M0'] * 60 + ['M1'] * 190
    })
    
    # 计算观测点滚动率
    obs_result = calculate_roll_rate(
        sample_data, 
        method='observation_point',
        target_transition=('M0', 'M1')
    )
    print("观测点滚动率结果：")
    print(obs_result)

四、滚动率计算的关键参数与输出

无论采用哪种方法，滚动率计算的核心输出都是一个 迁移矩阵。下表展示了 M0→M1 滚动率的典型输出结构：

在实际的贷后监控系统中，滚动率计算函数通常作为数据管道的一部分，每日或每月自动执行，输出结果直接用于后续的漂移检测和告警阈值判定。