第 15 讲：分层回测与单调性检验

本讲核心目标：
如果说 IC 是因子预测力的“宏观温度计”，那么分层回测（Quantile Portfolio Analysis）就是穿透因子微观收益结构的“CT 断层扫描”。本讲将彻底拆解买方机构如何将全市场股票截面划分为 5 组或 10 组，计算各组净值分层；推导严格的斯皮尔曼单调性得分（Monotonicity Score）与台阶递增率（Step Ratio）；深入剖析 A 股做空受限背景下“空头陷阱（Short-Side Trap）”对多头策略的致命欺骗，实现**多空对冲利差（Long-Short Spread）与多头纯超额（Long-Only Active Excess）**的完全分离评估。

1. 为什么高 IC 因子在实盘中依然可能完全不赚钱？

很多初涉量化的研究员经常困惑：“我的因子 Rank IC 均值有 0.045，年化 ICIR 高达 3.5，为什么实盘构建指增策略时完全跑不赢基准？”

答案就在于：IC 是全市场 5,000 只股票整体相关性的宏观统计量，它无法告诉你超额收益究竟来自多头还是空头！

可放大查看细节，点击百分比恢复原始大小。

1.1 A 股做空受限的残酷制度现实

在境外成熟市场，对冲基金可以通过低成本融券直接做空 Group 1（劣质组），同时做多 Group 10（优质组），享受完整的长短多空利差（Long-Short Spread）。但在 A 股市场：

融券券源极度稀缺且利息高昂（年化利息 6%~10%）；
公募基金与绝大多数私募指增产品合同强制规定只能单边做多（股票仓位 95%~100%）。
结论：如果一个因子的所有超额都来自“能够成功识别出即将退市暴跌的垃圾股”，而推荐出的 Top 优质股票根本跑不赢大盘等权基准，那么这个因子在单边做多的指数增强体系中商业价值为零！

2. 分层回测（Quantile Analysis）的标准工业流程

在每个调仓日截面 $t$ （通常为日频或周频）：

可放大查看细节，点击百分比恢复原始大小。

2.1 截面等分分组规则

设股票池截面有效股票数为 $N$ ，目标分为 $K$ 组（业界通用 $K = 5$ 或 $K = 10$ ）：

将股票按因子值从低到高升序排列；
第 $k$ 组（ $k \in {1, 2, \dots, K}$ ）包含排名在区间 $[\frac{k - 1}{K} \cdot N, \frac{k}{K} \cdot N)$ 内的股票；
若因子为正向因子，Group $K$ 为多头推荐组（Top），Group 1 为空头规避组（Bottom）。

2.2 各组日度等权收益与净值复利计算

在持有期（ $t \to t + 1$ ），第 $k$ 组的等权收益率为：

R_{k, t + 1} = \frac{1}{N_{k}} \sum_{i \in {Group}_{k}} r_{i, t + 1}

全市场候选池等权基准（Benchmark）收益率为：

R_{bench, t + 1} = \frac{1}{N} \sum_{i = 1}^{N} r_{i, t + 1}

各组的累积复利净值资产曲线（Wealth Index）为：

{NAV}_{k, T} = \prod_{t = 1}^{T} (1 + R_{k, t})

3. 单调性检验（Monotonicity Score）数学量化模型

观察分层净值走势图是定性排查，在全自动量化研发流水线（AutoML Alpha Pipeline）中，必须有一套无偏的数学评分算法来量化单调性。

3.1 斯皮尔曼秩单调性得分（Rank Monotonicity Score）

设 $K$ 个分位组按因子排序后的全期平均日收益率向量为 $g = [g_{1}, g_{2}, \dots, g_{K}]$ ，组别自然序数向量为 $k = [1, 2, \dots, K]$ 。

定义因子的截面单调性得分为两者的斯皮尔曼等级相关系数：

MonoScore = SpearmanCorr ([1, 2, \dots, K], [g_{1}, g_{2}, \dots, g_{K}])

完美单调： $MonoScore = + 1.0$ （正向因子，组别越高收益严格递增）或 $- 1.0$ （反向因子）；
健康门槛：买方量化通常要求 $| MonoScore | \geq 0.80$ ；
严重病态：如果 $| MonoScore | < 0.60$ ，说明中间组别收益颠倒混乱（如 Group 3 跑赢了 Group 5），因子底层逻辑缺乏全市场连续性。

3.2 步长单调性比率（Monotonic Step Ratio）

统计相邻组别满足严格收益递增的台阶比例：

StepRatio = \frac{1}{K - 1} \sum_{j = 1}^{K - 1} I (g_{j + 1} > g_{j})

当 $K = 5$ 时存在 4 个台阶。若 4 个台阶全增， $StepRatio = 100 %$ ；若有 1 处倒挂，则降为 $75 %$ 。

4. 多空对冲利差 vs 多头超额分离度量

4.1 多空对冲利差组合（Long-Short Spread）

R_{L/S, t} = R_{Top, t} - R_{Bottom, t}

衡量因子的最大理论套利上限。
输出指标：多空年化复合收益（CAGR）、多空年化夏普比率（L/S Sharpe）与多空最大回撤（L/S MDD）。

4.2 多头超额组合（Long-Only Active Excess）

R_{Active, t} = R_{Top, t} - R_{bench, t}

衡量因子直接用于构建单边做多指数增强时的真实超额能力。
输出指标：多头超额年化收益与纯多头信息比率（Active IR）。

4.3 多头/空头收益贡献分离度（Alpha Asymmetry Ratio）

买方最核心的“照妖镜”指标：

Long Contribution Ratio = \frac{{\overset{―}{R}}_{Top} - {\overset{―}{R}}_{bench}}{{\overset{―}{R}}_{Top} - {\overset{―}{R}}_{Bottom}}

多头收益贡献度判定标准
├── 1. 均衡健康型 (40% ~ 60%): 多头能打出超额，空头能避开暴跌，属于完美全能因子
├── 2. 多头驱动型 (> 60%): 极受公募指增欢迎，选出的股票爆发力强，能大幅跑赢宽基
└── 3. 空头陷阱型 (< 20%): 致命警报！绝大部分收益由做空劣质股贡献，多头几乎没有超额！

5. 动手实操：工业级 Polars 分层回测引擎实现

下面的 Python 代码严格遵循机构级买方规范，利用 Polars 原生分箱表达式，高吞吐计算 5 分位/10 分位各组净值曲线、多空利差、多头超额与单调性打分。

在项目根目录运行 PYTHONPATH=tutorial python3 -m examples.lecture_15_quantile_and_monotonicity。依赖安装、共享模块和数据契约见项目说明。

"""
A 股教学示例：因子分层回测引擎 (AshareQuantileBacktestEngine)
支持：5/10 分层收益计算、净值曲线追踪、单调性得分与多头超额解耦
"""

from dataclasses import dataclass
from typing import Dict, List, Tuple
import numpy as np
import polars as pl
from scipy import stats
from examples.factor_utils import validate_panel, annualized_ratio


@dataclass
class QuantileSummaryReport:
    """分层回测评估报告"""
    factor_name: str
    n_quantiles: int
    group_annual_returns: Dict[int, float]
    group_sharpe_ratios: Dict[int, float]
    group_max_drawdowns: Dict[int, float]
    long_short_annual_ret: float
    long_short_sharpe: float
    long_short_mdd: float
    long_only_excess_annual_ret: float
    long_only_excess_sharpe: float
    long_contribution_ratio: float
    monotonicity_score: float
    step_ratio: float

    def print_summary(self):
        """格式化输出报告"""
        print("=" * 70)
        print(f"📈 因子分层回测深度剖析报告: [{self.factor_name}] ({self.n_quantiles} 分位)")
        print("=" * 70)
        print(f"{'组别 (Quantile)':<18} | {'年化复合收益':<14} | {'年化夏普比率':<14} | {'历史最大回撤':<14}")
        print("-" * 70)
        for q in range(1, self.n_quantiles + 1):
            ann_ret = self.group_annual_returns[q]
            shp = self.group_sharpe_ratios[q]
            mdd = self.group_max_drawdowns[q]
            tag = " (Bottom 劣质组)" if q == 1 else (" (Top 优质多头组)" if q == self.n_quantiles else "")
            print(f"Group {q:<2}{tag:<10} | {ann_ret * 100:10.2f}%   | {shp:12.3f}   | {mdd * 100:10.2f}%")
        print("-" * 70)
        print("【多空对冲与多头超额分离度量】")
        print(f"  • 多空对冲年化利差 (L/S Spread) : {self.long_short_annual_ret * 100:8.2f}%")
        print(f"  • 多空夏普比率 (L/S Sharpe)     : {self.long_short_sharpe:8.3f}")
        print(f"  • 多空最大回撤 (L/S MDD)        : {self.long_short_mdd * 100:8.2f}%")
        print(f"  • 多头超额算术年化 (Top-Bench): {self.long_only_excess_annual_ret * 100:8.2f}%")
        print(f"  • 纯多头信息比率 (Active IR)    : {self.long_only_excess_sharpe:8.3f}")
        print(f"  • 多头收益贡献占比              : {self.long_contribution_ratio * 100:8.2f}%")
        print("-" * 70)
        print("【单调性稳健性检验 (Monotonicity)】")
        print(f"  • 斯皮尔曼单调性得分 (MonoScore) : {self.monotonicity_score:8.3f}")
        print(f"  • 台阶递增满足率 (StepRatio)    : {self.step_ratio * 100:8.2f}%")
        print("=" * 70)


class AshareQuantileBacktestEngine:
    """
    A 股分层回测引擎
    """

    def __init__(
        self,
        n_quantiles: int = 5,
        date_col: str = "trade_date",
        symbol_col: str = "symbol",
        annual_trading_days: float = 250.0
    ):
        if n_quantiles < 2:
            raise ValueError("至少需要两个分组")
        self.n_quantiles = n_quantiles
        self.date_col = date_col
        self.symbol_col = symbol_col
        self.annual_days = annual_trading_days

    def _calc_mdd(self, net_values: np.ndarray) -> float:
        """计算净值序列的最大回撤"""
        net_values = np.r_[1.0, np.asarray(net_values, dtype=float)]
        peak = np.maximum.accumulate(net_values)
        drawdown = (net_values - peak) / (peak + 1e-12)
        return float(np.min(drawdown))

    def run_quantile_backtest(
        self,
        df: pl.DataFrame,
        factor_col: str,
        forward_ret_col: str
    ) -> Tuple[pl.DataFrame, QuantileSummaryReport]:
        """
        执行分层回测核心运算
        """
        validate_panel(df, self.date_col, self.symbol_col)
        # 只用当时已知的因子决定候选池；标签缺失不能反过来改变选股。
        valid_df = df.filter(pl.col(factor_col).is_finite()).sort([self.date_col, self.symbol_col])
        if valid_df.is_empty() or valid_df[self.date_col].n_unique() != df[self.date_col].n_unique():
            raise ValueError("存在没有有效因子的日期")
        if valid_df.select((~pl.col(forward_ret_col).is_finite()).fill_null(True).any()).item():
            raise ValueError("候选池前瞻收益缺失；请补齐标签或显式裁剪完整评估区间")
        if valid_df.filter(pl.col(forward_ret_col) < -1).height:
            raise ValueError("股票简单收益率不能小于 -100%")
        counts = valid_df.group_by(self.date_col).len()
        if counts["len"].min() < self.n_quantiles:
            raise ValueError("截面股票数少于分组数")
        # 同分按 symbol 稳定破同分；此时组间差异不代表因子能区分同分股票。
        binned_df = valid_df.with_columns(
            (((pl.col(factor_col).rank(method="ordinal").over(self.date_col) - 1) /
               pl.len().over(self.date_col) * self.n_quantiles).floor().cast(pl.Int32) + 1)
            .alias("quantile_group")
        )

        # 限制边界防止浮点溢出
        binned_df = binned_df.with_columns(
            pl.col("quantile_group").clip(1, self.n_quantiles)
        )

        # 2. 计算每日各组等权收益率
        group_daily_returns = (
            binned_df.group_by([self.date_col, "quantile_group"])
            .agg(pl.col(forward_ret_col).mean().alias("daily_ret"))
            .sort([self.date_col, "quantile_group"])
        )

        # 3. 计算每日全市场等权基准收益率
        bench_daily = (
            valid_df.group_by(self.date_col)
            .agg(pl.col(forward_ret_col).mean().alias("bench_ret"))
            .sort(self.date_col)
        )

        # 4. 透视转换为宽表格式 (每列一个 Group 收益)
        pivot_df = group_daily_returns.pivot(
            on="quantile_group",
            index=self.date_col,
            values="daily_ret"
        ).sort(self.date_col)

        pivot_df = pivot_df.join(bench_daily, on=self.date_col, how="left")

        # 重命名分组列为 G1, G2 ...
        for q in range(1, self.n_quantiles + 1):
            pivot_df = pivot_df.rename({str(q): f"G{q}"})

        # 5. 添加多空利差列与多头超额列
        top_col = f"G{self.n_quantiles}"
        bottom_col = "G1"
        pivot_df = pivot_df.with_columns([
            (pl.col(top_col) - pl.col(bottom_col)).alias("long_short_spread"),
            (pl.col(top_col) - pl.col("bench_ret")).alias("top_excess_ret")
        ])

        # 6. 计算各组统计量
        T = pivot_df.height
        ann_rets = {}
        sharpes = {}
        mdds = {}
        raw_ann_rets = []

        for q in range(1, self.n_quantiles + 1):
            g_col = f"G{q}"
            ret_series = pivot_df[g_col].to_numpy()
            
            # 年化几何复合收益率
            wealth = np.cumprod(1.0 + ret_series)
            ann_ret = float(wealth[-1] ** (self.annual_days / T) - 1.0)
            ann_vol = float(np.std(ret_series, ddof=1) * np.sqrt(self.annual_days))
            sharpe = annualized_ratio(ret_series, self.annual_days, annual_rf=.02)
            mdd = self._calc_mdd(wealth)

            ann_rets[q] = ann_ret
            sharpes[q] = sharpe
            mdds[q] = mdd
            raw_ann_rets.append(float(np.mean(ret_series)))

        # 多空对冲统计
        ls_series = pivot_df["long_short_spread"].to_numpy()
        ls_wealth = np.cumprod(1.0 + ls_series)
        ls_ann_ret = float(ls_wealth[-1] ** (self.annual_days / T) - 1.0) if np.all(ls_series > -1) else float("nan")
        ls_ann_vol = float(np.std(ls_series, ddof=1) * np.sqrt(self.annual_days))
        ls_sharpe = annualized_ratio(ls_series, self.annual_days)
        ls_mdd = self._calc_mdd(ls_wealth) if np.all(ls_series > -1) else float("nan")

        # 多头超额统计
        excess_series = pivot_df["top_excess_ret"].to_numpy()
        excess_wealth = np.cumprod(1.0 + excess_series)
        excess_ann_ret = float(np.mean(excess_series) * self.annual_days) # 算术年化主动收益
        excess_ann_vol = float(np.std(excess_series, ddof=1) * np.sqrt(self.annual_days))
        excess_ir = annualized_ratio(excess_series, self.annual_days)

        # 同一算术均值口径，避免用 CAGR 差值冒充收益贡献。
        spread_mean = float(np.mean(ls_series))
        long_contrib = float(np.mean(excess_series) / spread_mean) if abs(spread_mean) > 1e-12 else float("nan")

        # 单调性打分
        group_ids = np.arange(1, self.n_quantiles + 1)
        mono_score, _ = stats.spearmanr(group_ids, raw_ann_rets)
        
        # 台阶递增满足率
        step_diffs = np.diff(raw_ann_rets)
        step_ratio = float(np.mean(step_diffs > 0))

        report = QuantileSummaryReport(
            factor_name=factor_col,
            n_quantiles=self.n_quantiles,
            group_annual_returns=ann_rets,
            group_sharpe_ratios=sharpes,
            group_max_drawdowns=mdds,
            long_short_annual_ret=ls_ann_ret,
            long_short_sharpe=ls_sharpe,
            long_short_mdd=ls_mdd,
            long_only_excess_annual_ret=excess_ann_ret,
            long_only_excess_sharpe=excess_ir,
            long_contribution_ratio=long_contrib,
            monotonicity_score=float(mono_score),
            step_ratio=step_ratio
        )

        return pivot_df, report


# ================= 合成数据演示（统计结果不是实盘收益证明） =================
if __name__ == "__main__":
    np.random.seed(42)
    n_days = 250
    n_stocks = 600

    import pandas as pd
    dates = pd.bdate_range("2025-01-01", periods=n_days).date.tolist()
    symbols = [f"{s:06d}.SH" for s in range(n_stocks)]

    records = []
    # 模拟真实市场中的高单调性 Alpha 因子
    for d in dates:
        # 市场大盘 Beta 收益
        market_beta = np.random.normal(0.0004, 0.012)
        # 真实的个股纯 Alpha 打分
        true_latent_alpha = np.random.normal(0, 1.0, n_stocks)
        
        # 收益与因子呈现高单调性对应：Top 组收益明显抬高，Bottom 组收益明显拉低
        stock_returns = market_beta + true_latent_alpha * 0.006 + np.random.normal(0, 0.015, n_stocks)
        
        # 观测到的因子值 (包含测量噪声)
        observed_factor = true_latent_alpha + np.random.normal(0, 0.5, n_stocks)

        for s_idx, s in enumerate(symbols):
            records.append({
                "trade_date": d,
                "symbol": s,
                "factor_score": observed_factor[s_idx],
                "ret_fwd_1d": stock_returns[s_idx]
            })

    sim_df = pl.DataFrame(records)

    # 运行 5 分位分层回测
    engine = AshareQuantileBacktestEngine(n_quantiles=5)
    daily_res, report = engine.run_quantile_backtest(
        sim_df,
        factor_col="factor_score",
        forward_ret_col="ret_fwd_1d"
    )

    print(">>> 各组每日收益率透视表 (前 5 行):")
    print(daily_res.head(5))

    # 打印买方标准分层报告
    report.print_summary()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268

核心运行输出与分层断层扫描解读：

======================================================================
📈 因子分层回测深度剖析报告: [factor_score] (5 分位)
======================================================================
组别 (Quantile)      | 年化复合收益         | 年化夏普比率         | 历史最大回撤        
----------------------------------------------------------------------
Group 1  (Bottom 劣质组) |     -86.00%   |       -9.345   |     -86.12%
Group 2            |     -55.12%   |       -3.824   |     -59.22%
Group 3            |     -11.47%   |       -0.570   |     -28.74%
Group 4            |      88.58%   |        3.036   |      -7.95%
Group 5  (Top 优质多头组) |     490.76%   |        8.478   |      -5.21%
----------------------------------------------------------------------
【多空对冲与多头超额分离度量】
  • 多空对冲年化利差 (L/S Spread) :  3996.09%
  • 多空夏普比率 (L/S Sharpe)     :  122.566
  • 多空最大回撤 (L/S MDD)        :     0.00%
  • 多头超额算术年化 (Top-Bench):   187.50%
  • 纯多头信息比率 (Active IR)    :   95.418
  • 多头收益贡献占比              :    50.12%
----------------------------------------------------------------------
【单调性稳健性检验 (Monotonicity)】
  • 斯皮尔曼单调性得分 (MonoScore) :    1.000
  • 台阶递增满足率 (StepRatio)    :   100.00%
======================================================================

完美的台阶单调递增：
- 从 Group 1 到 Group 5，年化复合收益率呈现严密的单调递增（-86.0% $\to$ -55.1% $\to$ -11.5% $\to$ +88.6% $\to$ +490.8%），MonoScore = 1.000，StepRatio = 100%；
多头与空头贡献完全对称：
- 多头收益贡献占比为 50.12%，说明多头选股能力与空头排雷能力完全势均力敌，没有陷入空头陷阱。

6. 本讲小结与课后动手作业

核心知识图谱梳理：

分层回测穿透微观结构：分层是验证因子在不同分位是否具备单调排序能力的唯一利器；
警惕空头陷阱：在 A 股做空受限下，必须单独考察 Top 组相对 Benchmark 的超额贡献占比；
数学量化单调性：使用斯皮尔曼秩单调性得分与台阶递增率，实现流水线自动筛选因子。

课后动手实训：

实操任务：
打开 tutorial/examples/lecture_15_quantile_and_monotonicity.py。
将初始化参数中的 n_quantiles 从 5 调整为 10（进行更细腻的 10 分位回测），重新运行观察各组的收益阶梯。
思考题：当分位数从 5 组切分为 10 组甚至 20 组时，Top 组的年化收益通常会上升还是下降？为什么？这与组合换手率存在怎样的权衡取舍？

第 15 讲：分层回测与单调性检验 ​

1. 为什么高 IC 因子在实盘中依然可能完全不赚钱？ ​

1.1 A 股做空受限的残酷制度现实 ​

2. 分层回测（Quantile Analysis）的标准工业流程 ​

2.1 截面等分分组规则 ​

2.2 各组日度等权收益与净值复利计算 ​

3. 单调性检验（Monotonicity Score）数学量化模型 ​

3.1 斯皮尔曼秩单调性得分（Rank Monotonicity Score） ​

3.2 步长单调性比率（Monotonic Step Ratio） ​

4. 多空对冲利差 vs 多头超额分离度量 ​

4.1 多空对冲利差组合（Long-Short Spread） ​

4.2 多头超额组合（Long-Only Active Excess） ​

4.3 多头/空头收益贡献分离度（Alpha Asymmetry Ratio） ​

5. 动手实操：工业级 Polars 分层回测引擎实现 ​

核心运行输出与分层断层扫描解读： ​

6. 本讲小结与课后动手作业 ​

核心知识图谱梳理： ​

课后动手实训： ​