4. 基础回测框架：搭建简单的量化策略回测引擎，计算每日/每笔坏账率

坏账率哨兵告警阈值优化的前提，是拥有一个能够准确、稳定地模拟策略历史表现的回测引擎。本节将搭建一个轻量级但功能完整的回测框架，重点实现每日坏账率与每笔坏账率的精确计算，为后续告警阈值分析提供可靠的数据基础。

4.1 回测引擎核心架构

一个基础的量化策略回测引擎，通常包含以下四个核心模块。它们按顺序协作，完成从原始数据到坏账率指标的完整计算链路。

4.2 坏账率计算口径定义

在回测引擎中，坏账率并非单一指标。根据计算粒度和业务含义，我们至少需要区分以下两种口径：

4.3 回测引擎代码实现

以下是一个基于 Python 的轻量级回测引擎核心代码。它实现了每日坏账率和每笔坏账率的计算逻辑，并保留了扩展接口。

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

class BacktestEngine:
    """量化策略回测引擎——坏账率计算核心"""

    def __init__(self, loan_book: pd.DataFrame, repayment_schedule: pd.DataFrame):
        """
        初始化回测引擎
        :param loan_book: 放款记录表，字段包括 [loan_id, principal, issue_date, term]
        :param repayment_schedule: 还款计划表，字段包括 [loan_id, due_date, due_amount, status]
        """
        self.loan_book = loan_book.copy()
        self.repayment_schedule = repayment_schedule.copy()
        self.daily_metrics = None
        self.loan_metrics = None

    def _compute_daily_bad_rate(self) -> pd.DataFrame:
        """计算每日坏账率"""
        # 1. 生成每日资产池状态
        date_range = pd.date_range(
            start=self.loan_book['issue_date'].min(),
            end=self.repayment_schedule['due_date'].max(),
            freq='D'
        )
        daily_records = []
        for current_date in date_range:
            # 期初在贷余额：所有已放款且未结清的本金
            active_loans = self.loan_book[
                (self.loan_book['issue_date'] <= current_date)
            ]
            # 简化处理：假设到期未还即进入坏账
            bad_loans = self.repayment_schedule[
                (self.repayment_schedule['due_date'] == current_date) &
                (self.repayment_schedule['status'] == 'overdue')
            ]
            beginning_balance = active_loans['principal'].sum()
            new_bad_amount = bad_loans['due_amount'].sum()
            daily_bad_rate = new_bad_amount / beginning_balance if beginning_balance > 0 else 0.0
            daily_records.append({
                'date': current_date,
                'beginning_balance': beginning_balance,
                'new_bad_amount': new_bad_amount,
                'daily_bad_rate': daily_bad_rate
            })
        return pd.DataFrame(daily_records)

    def _compute_loan_bad_rate(self) -> pd.DataFrame:
        """计算每笔坏账率"""
        # 按 loan_id 聚合还款状态
        loan_performance = self.repayment_schedule.groupby('loan_id').agg(
            total_due=('due_amount', 'sum'),
            total_overdue=('due_amount', lambda x: x[self.repayment_schedule.loc[x.index, 'status'] == 'overdue'].sum())
        ).reset_index()
        loan_performance = loan_performance.merge(
            self.loan_book[['loan_id', 'principal']], on='loan_id', how='left'
        )
        loan_performance['bad_rate_per_loan'] = loan_performance['total_overdue'] / loan_performance['principal']
        return loan_performance

    def run(self):
        """执行完整回测"""
        self.daily_metrics = self._compute_daily_bad_rate()
        self.loan_metrics = self._compute_loan_bad_rate()
        print(f"回测完成：共 {len(self.daily_metrics)} 个交易日，{len(self.loan_metrics)} 笔贷款。")
        return self.daily_metrics, self.loan_metrics

4.4 回测输出与可视化

运行上述引擎后，我们将得到两个核心数据集。下表展示了每日坏账率输出的前5行示例：

每笔坏账率输出则包含每个 loan_id 的最终损失比例，可用于后续的策略分组对比和坏账分布分析。

4.5 回测引擎验证要点

在将回测引擎投入哨兵告警阈值优化之前，必须完成以下验证，确保计算结果的可靠性：

• 数据完整性验证： 检查 loan_id 是否在放款表和还款计划表中一一对应，无缺失、无重复。
• 坏账定义一致性： 确认逾期天数阈值（如 M3+）与业务定义一致，避免因口径差异导致坏账率失真。
• 边界条件处理： 验证期初在贷余额为 0 或分母极小时的每日坏账率计算是否返回 0 而非 NaN 或无穷大。
• 滚动窗口平滑： 对每日坏账率序列计算 7 日移动平均，观察是否存在异常尖峰，排查数据或逻辑错误。