4. 回测框架搭建：基于Python的回测引擎基础架构与数据流设计

一个健壮的回测引擎是量化策略验证的基石，尤其当我们需要在回测过程中嵌入坏账率哨兵PSI监控时，引擎的数据流设计必须支持时间切片、样本外滚动以及实时指标计算。本节将拆解回测引擎的核心模块与数据流动路径。

4.1 回测引擎核心模块架构

我们将回测引擎划分为五个独立且可组合的模块，每个模块负责单一职责，便于后续扩展PSI监控逻辑。

4.2 数据流设计：时间轴滚动与分布快照

回测数据流遵循严格的时间顺序，PSI监控要求我们在每个时间窗口（如每月）结束时，对当前批次的评分分布与历史参考分布进行对比。

4.3 核心代码骨架：回测引擎主循环

以下代码展示了嵌入PSI监控的回测主循环结构。注意 psi_monitor 对象在每次时间窗口结束时被调用。

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

class BacktestEngine:
    def __init__(self, data: pd.DataFrame, strategy, psi_monitor, window='M'):
        self.data = data.sort_values('apply_date')
        self.strategy = strategy
        self.psi_monitor = psi_monitor
        self.window = window
        self.results = []

    def run(self):
        # 按时间窗口分组
        for window_id, group in self.data.groupby(pd.Grouper(key='apply_date', freq=self.window)):
            if len(group) == 0:
                continue

            # 1. 策略执行：生成授信决策
            decisions = self.strategy.decide(group)

            # 2. 计算当前窗口的评分分布
            current_scores = decisions['score'].values

            # 3. PSI哨兵监控：对比参考分布
            psi_value = self.psi_monitor.calculate(current_scores)
            is_alarm = psi_value > self.psi_monitor.threshold

            # 4. 风控指标计算
            bad_rate = decisions['is_bad'].mean()
            pass_rate = decisions['is_approved'].mean()

            # 5. 记录结果
            self.results.append({
                'window_id': window_id,
                'psi': psi_value,
                'psi_alarm': is_alarm,
                'bad_rate': bad_rate,
                'pass_rate': pass_rate,
                'sample_size': len(group)
            })

            # 6. 若触发预警，记录详细日志
            if is_alarm:
                self.psi_monitor.log_alarm(window_id, psi_value, current_scores)

        return pd.DataFrame(self.results)

4.4 数据流中的关键字段设计

为了支持PSI监控与坏账率回溯，回测数据流中的每个样本必须包含以下核心字段：

4.5 回测引擎初始化与PSI哨兵注入

在实际使用中，我们通过依赖注入的方式将PSI哨兵模块挂载到回测引擎中，保持引擎核心逻辑的纯净性。

# 示例：初始化PSI哨兵
from psi_sentinel import PSISentinel

# 参考分布：使用训练集评分
reference_distribution = np.histogram(train_scores, bins=20, range=(0,1))[0]

psi_sentinel = PSISentinel(
    reference_dist=reference_distribution,
    threshold=0.25,          # PSI预警阈值
    bins=20,
    alarm_callback=send_alert  # 可选：预警回调函数
)

# 注入回测引擎
engine = BacktestEngine(
    data=full_data,
    strategy=my_strategy,
    psi_monitor=psi_sentinel,
    window='M'               # 按月监控
)

result_df = engine.run()

4.6 回测输出结构：含PSI预警的绩效表

回测完成后，引擎返回的DataFrame应包含以下列，便于后续分析PSI预警与坏账率的关系：