2. 分群评分沙盘概念

2.1 什么是分群评分沙盘

分群评分沙盘（Segmentation & Score Sandbox）是授信决策引擎中用于离线模拟、验证和优化策略组合的隔离环境。它将客群按照风险特征划分为多个子群（Segment），并为每个子群分配独立的评分卡与决策规则，在沙盘环境中进行“假设推演”，从而在不影响生产流量的前提下完成策略调优。

沙盘的核心组件包括：

• 客群分群器： 基于规则（如年龄、收入、征信分）或模型（如聚类、决策树）将申请客群划分为若干同质子群。
• 评分卡矩阵： 每个子群对应一张评分卡，评分卡可以是逻辑回归、XGBoost 或集成模型，输出违约概率或信用分数。
• 策略编排层： 在沙盘内定义每个子群的准入规则、额度矩阵、定价策略，支持 Python 脚本动态编排。
• 回测引擎： 使用历史样本数据模拟策略执行，输出通过率、坏账率、ROI 等关键指标。

2.2 沙盘在策略调优与回测中的作用

沙盘是策略调优的“安全实验室”，其核心价值体现在以下三个维度：

回测流程通常包含以下步骤：

1. 数据准备： 选取最近 6-12 个月的历史申请样本，包含特征、标签（逾期表现）、决策结果。
2. 分群映射： 将样本按沙盘定义的分群规则分配到对应子群。
3. 评分计算： 对每个子群使用对应的评分卡计算分数。
4. 策略执行： 按沙盘策略规则（准入、额度、定价）生成模拟决策。
5. 指标汇总： 计算通过率、坏账率、平均额度、风险调整后收益等。
6. 差异分析： 对比沙盘结果与生产实际结果，校准策略参数。

2.3 沙盘与生产环境的差异

沙盘环境与生产环境在数据、时效、风险、迭代方式上存在本质区别，理解这些差异是正确使用沙盘的前提。

2.4 沙盘策略编排的 Python 实现思路

在分群评分沙盘中，Python 策略编排的核心是将分群逻辑、评分计算、规则决策封装为可复用的函数或类，并通过配置驱动的方式实现快速迭代。以下是一个简化的沙盘编排代码示例：

import pandas as pd
import numpy as np
from typing import Dict, List, Callable

class SegmentScoreSandbox:
    """分群评分沙盘策略编排器"""
    
    def __init__(self, segment_rules: Dict[str, Callable], 
                 scorecards: Dict[str, Callable],
                 policy_rules: Dict[str, Dict]):
        self.segment_rules = segment_rules  # 分群规则函数字典
        self.scorecards = scorecards        # 评分卡函数字典
        self.policy_rules = policy_rules    # 策略规则字典
    
    def assign_segment(self, df: pd.DataFrame) -> pd.Series:
        """将样本分配到对应子群"""
        segments = pd.Series(index=df.index, dtype='object')
        for seg_name, rule_func in self.segment_rules.items():
            mask = rule_func(df)
            segments[mask] = seg_name
        return segments
    
    def compute_scores(self, df: pd.DataFrame, 
                       segments: pd.Series) -> pd.Series:
        """按子群计算评分"""
        scores = pd.Series(index=df.index, dtype='float64')
        for seg_name in segments.unique():
            mask = segments == seg_name
            scores[mask] = self.scorecards[seg_name](df[mask])
        return scores
    
    def execute_policy(self, df: pd.DataFrame, 
                       scores: pd.Series,
                       segments: pd.Series) -> pd.DataFrame:
        """执行策略规则，返回决策结果"""
        decisions = []
        for seg_name in segments.unique():
            mask = segments == seg_name
            seg_df = df[mask].copy()
            seg_scores = scores[mask]
            # 应用该子群的策略规则
            seg_policy = self.policy_rules[seg_name]
            seg_df['decision'] = np.where(
                seg_scores >= seg_policy['approve_threshold'],
                'approve', 'reject'
            )
            seg_df['score'] = seg_scores
            decisions.append(seg_df)
        return pd.concat(decisions)
    
    def backtest(self, df: pd.DataFrame) -> Dict:
        """执行完整回测流程"""
        segments = self.assign_segment(df)
        scores = self.compute_scores(df, segments)
        result = self.execute_policy(df, scores, segments)
        # 汇总指标
        summary = {
            'total_count': len(result),
            'approve_rate': (result['decision'] == 'approve').mean(),
            'avg_score': result['score'].mean(),
            'segment_distribution': segments.value_counts().to_dict()
        }
        return summary