feat(research_stats): RS-02 PBO via CSCV (research-only, fail-closed)

docs/research_stats.zh-CN.md

@@ -2,7 +2,7 @@
 
 `quant_platform_kit.research_stats` 是回测证据的研究统计原语。它**不**读取账户、不调用券商、不下单、不写风险政策，也不授予 paper/shadow/live 权限。运行时、执行内核和通知模块不得 import 它（`tests/test_research_stats_import_boundary.py` 强制）。
 
-安装：`pip install 'quant-platform-kit[research]'`。numpy/pandas 只在 `research` extra（以及原有 `dev` extra）里，核心 `dependencies` 仍为空。PSR/DSR 与成本压力只用标准库；只有 bootstrap 延迟 import numpy。
+安装：`pip install 'quant-platform-kit[research]'`。numpy/pandas 只在 `research` extra（以及原有 `dev` extra）里，核心 `dependencies` 仍为空。PSR/DSR、PBO 与成本压力只用标准库；只有 bootstrap 延迟 import numpy。
 
 本模块与测试只使用合成数据，不包含任何真实回测数字。
 
@@ -13,6 +13,7 @@
 | 概率 Sharpe（PSR，Bailey & López de Prado 2012） | `probabilistic_sharpe_ratio(returns, sr_benchmark=0, min_observations=30)` | 样本不足、零方差、非有限、≤−1、bool → `UNCOMPUTABLE`，`value=None` |
 | 去膨胀 Sharpe（DSR，2014） | `deflated_sharpe_ratio(returns, trial_sharpes, n_effective_trials=None)` | 试验数 < 2、有效试验数 > 已登记数、试验 Sharpe 方差为 0 → `UNCOMPUTABLE` |
 | 期望最大 Sharpe | `expected_max_sharpe(variance, n_trials)` | 同上 |
+| 回测过拟合概率（PBO，CSCV，Bailey 等 2017） | `probability_of_backtest_overfitting(returns_by_trial, n_splits=16, metric="sharpe", min_rows_per_block=2)` | 试验数 < 2、各试验长度不一致、块数为奇数或 > 16、每块行数不足、任一块指标无定义（如零方差）→ `UNCOMPUTABLE` |
 | 平稳块自助法（Politis & Romano 1994） | `stationary_bootstrap_ci(returns, *, seed, mean_block_length, n_resamples=1000, confidence=0.95)` | seed 必填；块长 > 样本、重抽样 < 100 → `UNCOMPUTABLE`；某个统计量无定义时只有该项 `UNCOMPUTABLE` |
 | 成本压力 | `cost_stress_recompute(gross, turnover, *, cost_bps_per_side, multipliers=(1,2,3))` | 长度不等、负换手、非有限、重复倍数 → `PARKED` |
 | 报告模板 | `BacktestReportV1` + `schemas/backtest-report.v1.schema.json` | `live_ready=False`、`no_order=True` 恒定；非 COMPUTED 指标必须 `value=None` + `reason_code`；development 数据不得标 `OOS_EVALUATED` |
@@ -23,11 +24,12 @@
 - 偏度/峰度用总体矩估计，峰度为非超额（正态 = 3）。
 - `min_observations=30` 只是沿用 Kelly 契约 v2 的结构下限，不是质量阈值。
 - bootstrap 统计量：单期对数增长 `mean(log1p r)`、单期 Sharpe、含初始净值的最大回撤；百分位区间，不做自相关以外的校正，不构成未来风险保证，也不替代多重试验校正。
+- PBO：T×N 矩阵（同一日期上的全部试验，含失败/落选者的收益）按时间切成 S 个等长连续块，不能整除时丢弃最早的行并在 `rows_dropped_leading` 报告；枚举 C(S,S/2) 种 IS/OOS 组合，λ = logit(IS 最优者的 OOS 相对名次 / (N+1))，PBO = λ ≤ 0 的比例。纯噪声下 PBO 通常偏高（IS/OOS 互补），它是过拟合风险描述，不是及格线。
 - 成本：`turnover[t]` = 期初买卖成交额合计 / 交易前净值，每单位成交额付 `cost_bps_per_side × 倍数`；`net = (1 − τ·c)(1 + gross) − 1`。半 L1 口径（现金腿不计费）须由调用方先换算。
 - 报告模板只验结构与诚实性约束，不判断经济有效性，不设任何 Sharpe/回撤及格线；否决项阈值必须由候选在读取留出结果前冻结（`gates.veto_items[].frozen_at/frozen_by`）。
 
 ## 非目标
 
 - 不修改 `strategy_lifecycle/performance_metrics.py` 等现有共享指标函数的语义。
 - 不接入任何消费者，不升级任何下游 QPK pin，不改 `QPK_PIN`。
-- PBO（CSCV）、与 UES trial journal 的接线留给后续 RS-02。
+- 与 UES trial journal 的接线在 UES 侧（RS-02）完成，QPK 不感知具体策略。

src/quant_platform_kit/research_stats/__init__.py

@@ -5,7 +5,7 @@
 Runtime, execution and notification modules must not import it (enforced by
 ``tests/test_research_stats_import_boundary.py``).
 
-``sharpe_inference`` and ``cost_stress`` use only the standard library.
+``sharpe_inference``, ``overfitting`` (PBO/CSCV) and ``cost_stress`` use only the standard library.
 ``bootstrap`` lazily imports numpy from the optional ``research`` extra
 (``pip install 'quant-platform-kit[research]'``).
 """
@@ -38,6 +38,10 @@
     CostStressScenario,
     cost_stress_recompute,
 )
+from quant_platform_kit.research_stats.overfitting import (
+    PBO_METRICS,
+    probability_of_backtest_overfitting,
+)
 from quant_platform_kit.research_stats.sharpe_inference import (
     STATUS_COMPUTED,
     STATUS_UNCOMPUTABLE,
@@ -58,6 +62,7 @@
     "CostStressScenario",
     "DataSpec",
     "MetricEntry",
+    "PBO_METRICS",
     "ReportGates",
     "STATUS_COMPUTED",
     "STATUS_UNCOMPUTABLE",
@@ -73,5 +78,6 @@
     "expected_max_sharpe",
     "load_backtest_report_schema",
     "probabilistic_sharpe_ratio",
+    "probability_of_backtest_overfitting",
     "stationary_bootstrap_ci",
 ]

src/quant_platform_kit/research_stats/overfitting.py

@@ -0,0 +1,172 @@
+"""Probability of Backtest Overfitting via CSCV (research-only, stdlib only).
+
+Bailey, D. H., Borwein, J. M., López de Prado, M. and Zhu, Q. J. (2017),
+"The Probability of Backtest Overfitting", Journal of Computational Finance
+20(4). Combinatorially Symmetric Cross-Validation (CSCV):
+
+1. Arrange the per-period returns of **all N trials** on the **same dates**
+   as a T × N matrix and split the rows into S contiguous, equal blocks
+   (S even). Leading rows that do not fill a block are dropped and reported.
+2. For each of the C(S, S/2) ways to choose S/2 blocks as in-sample (IS),
+   the remaining blocks are out-of-sample (OOS).
+3. Select the IS-best trial n* by the performance metric; compute its
+   relative OOS rank ω = rank / (N + 1) (average ranks for ties) and the
+   logit λ = ln(ω / (1 − ω)).
+4. PBO = share of combinations with λ ≤ 0 (the IS winner is at or below the
+   OOS median).
+
+Performance metric: ``sharpe`` (per-period mean / sample std, ddof=1) or
+``mean_log_growth`` (mean of log(1 + r)). IS-best ties pick the first trial in
+input order and are counted in ``details["is_best_ties"]``.
+
+Fail-closed: fewer than two trials, ragged/misaligned columns, odd or too few
+blocks, blocks shorter than ``min_rows_per_block``, or any undefined block
+metric (e.g. zero variance) return ``UNCOMPUTABLE`` with ``value=None``.
+"""
+
+from __future__ import annotations
+
+import itertools
+import math
+from typing import Mapping, Sequence
+
+from quant_platform_kit.research_stats._validation import (
+    ResearchInputError,
+    positive_int,
+    simple_returns,
+)
+from quant_platform_kit.research_stats.sharpe_inference import (
+    STATUS_COMPUTED,
+    StatResult,
+    _uncomputable,
+)
+
+PBO_METRICS = ("sharpe", "mean_log_growth")
+MAX_SPLITS = 16  # C(16, 8) = 12,870 combinations bounds runtime.
+_MIN_TRIALS = 2
+
+
+def _block_sums(column: Sequence[float], rows: int, n_splits: int) -> list[tuple[float, float, float]]:
+    """(sum, sum of squares, sum of log1p) per block."""
+    out = []
+    for b in range(n_splits):
+        chunk = column[b * rows:(b + 1) * rows]
+        out.append((math.fsum(chunk), math.fsum(x * x for x in chunk), math.fsum(math.log1p(x) for x in chunk)))
+    return out
+
+
+def _performance(stats: Sequence[tuple[float, float, float]], count: int, metric: str) -> float | None:
+    total = math.fsum(s[0] for s in stats)
+    if metric == "mean_log_growth":
+        return math.fsum(s[2] for s in stats) / count
+    squares = math.fsum(s[1] for s in stats)
+    mean = total / count
+    variance = (squares - count * mean * mean) / (count - 1)
+    # Guard against round-off: treat relatively tiny variance as zero.
+    if not math.isfinite(variance) or variance <= 1e-15 * max(1.0, squares / count):
+        return None
+    return mean / math.sqrt(variance)
+
+
+def _average_rank(values: Sequence[float], index: int) -> float:
+    """1-based ascending rank of values[index], averaging ties."""
+    target = values[index]
+    below = sum(1 for v in values if v < target)
+    equal = sum(1 for v in values if v == target)
+    return below + (equal + 1) / 2.0
+
+
+def probability_of_backtest_overfitting(
+    returns_by_trial: Mapping[str, Sequence[object]] | Sequence[Sequence[object]],
+    *,
+    n_splits: int = 16,
+    metric: str = "sharpe",
+    min_rows_per_block: int = 2,
+) -> StatResult:
+    """CSCV PBO over aligned per-period trial returns (all trials, incl. losers)."""
+    name = "probability_of_backtest_overfitting"
+    try:
+        if metric not in PBO_METRICS:
+            raise ResearchInputError(f"metric must be one of {PBO_METRICS}")
+        splits = positive_int(n_splits, "n_splits", minimum=2)
+        if splits % 2 or splits > MAX_SPLITS:
+            raise ResearchInputError(f"n_splits must be even and <= {MAX_SPLITS}")
+        min_rows = positive_int(min_rows_per_block, "min_rows_per_block", minimum=2)
+        if isinstance(returns_by_trial, Mapping):
+            labels = [str(k) for k in returns_by_trial.keys()]
+            raw = list(returns_by_trial.values())
+        elif isinstance(returns_by_trial, (str, bytes)):
+            raise ResearchInputError("returns_by_trial must be a mapping or sequence of sequences")
+        else:
+            raw = list(returns_by_trial)
+            labels = [str(i) for i in range(len(raw))]
+        columns = [simple_returns(col, f"returns_by_trial[{labels[i]}]") for i, col in enumerate(raw)]
+    except (ResearchInputError, TypeError) as exc:
+        return _uncomputable(name, "invalid_input", message=str(exc))
+
+    n_trials = len(columns)
+    if n_trials < _MIN_TRIALS:
+        return _uncomputable(name, "insufficient_trials", n_trials=n_trials, min_trials=_MIN_TRIALS)
+    lengths = {len(c) for c in columns}
+    if len(lengths) != 1:
+        return _uncomputable(name, "trial_returns_not_aligned", lengths=sorted(lengths))
+    total_rows = lengths.pop()
+    rows = total_rows // splits
+    if rows < min_rows:
+        return _uncomputable(
+            name, "insufficient_observations", observations=total_rows, n_splits=splits,
+            min_rows_per_block=min_rows,
+        )
+    dropped = total_rows - rows * splits
+    trimmed = [c[dropped:] for c in columns]  # drop oldest leading rows only
+    blocks = [_block_sums(c, rows, splits) for c in trimmed]
+
+    # Every single block must have a defined metric; otherwise some
+    # combination could silently select from undefined values.
+    for t, per_block in enumerate(blocks):
+        for b in range(splits):
+            if _performance([per_block[b]], rows, metric) is None:
+                return _uncomputable(name, "undefined_performance", trial=labels[t], block=b)
+
+    half = splits // 2
+    lambdas: list[float] = []
+    is_best_ties = 0
+    for train in itertools.combinations(range(splits), half):
+        test = [b for b in range(splits) if b not in train]
+        is_perf: list[float] = []
+        oos_perf: list[float] = []
+        for per_block in blocks:
+            p_is = _performance([per_block[b] for b in train], rows * half, metric)
+            p_oos = _performance([per_block[b] for b in test], rows * half, metric)
+            if p_is is None or p_oos is None:
+                return _uncomputable(name, "undefined_performance", combination=list(train))
+            is_perf.append(p_is)
+            oos_perf.append(p_oos)
+        best_value = max(is_perf)
+        best = is_perf.index(best_value)
+        if is_perf.count(best_value) > 1:
+            is_best_ties += 1
+        omega = _average_rank(oos_perf, best) / (n_trials + 1)
+        lambdas.append(math.log(omega / (1.0 - omega)))
+
+    overfit = sum(1 for lam in lambdas if lam <= 0.0)
+    ordered = sorted(lambdas)
+    mid = len(ordered) // 2
+    median = ordered[mid] if len(ordered) % 2 else (ordered[mid - 1] + ordered[mid]) / 2.0
+    return StatResult(
+        name=name,
+        status=STATUS_COMPUTED,
+        value=overfit / len(lambdas),
+        details={
+            "method": "cscv",
+            "metric": metric,
+            "n_trials": n_trials,
+            "n_splits": splits,
+            "rows_per_block": rows,
+            "rows_dropped_leading": dropped,
+            "n_combinations": len(lambdas),
+            "logit_median": median,
+            "is_best_ties": is_best_ties,
+            "trial_labels": labels,
+        },
+    )

tests/test_research_stats_pbo.py

@@ -0,0 +1,109 @@
+"""Synthetic-only tests for PBO (CSCV). No real market data."""
+
+from __future__ import annotations
+
+import itertools
+import math
+import random
+import statistics
+
+import pytest
+
+from quant_platform_kit.research_stats import (
+    STATUS_COMPUTED,
+    STATUS_UNCOMPUTABLE,
+    probability_of_backtest_overfitting,
+)
+
+
+def _noise(n_trials: int, rows: int, seed: int) -> dict[str, list[float]]:
+    rng = random.Random(seed)
+    return {f"t{k}": [rng.gauss(0.0, 0.01) for _ in range(rows)] for k in range(n_trials)}
+
+
+def test_persistent_skill_gives_zero_pbo() -> None:
+    rng = random.Random(1)
+    trials = {f"t{k}": [0.001 * k + rng.gauss(0.0, 0.0002) for _ in range(320)] for k in range(6)}
+    result = probability_of_backtest_overfitting(trials, n_splits=8)
+    assert result.status == STATUS_COMPUTED
+    assert result.value == 0.0
+    assert result.details["n_combinations"] == math.comb(8, 4)
+
+
+def test_reversing_skill_gives_full_pbo() -> None:
+    # Two blocks; A wins block 0 and loses block 1, B the opposite.
+    a = [0.02, 0.01, 0.03, -0.02, -0.01, -0.03]
+    b = [-0.02, -0.01, -0.03, 0.02, 0.01, 0.03]
+    result = probability_of_backtest_overfitting({"A": a, "B": b}, n_splits=2, min_rows_per_block=3)
+    assert result.status == STATUS_COMPUTED
+    assert result.value == 1.0
+
+
+def test_pure_noise_is_deterministic_and_far_from_zero() -> None:
+    # With complementary IS/OOS halves, IS luck implies OOS give-back, so
+    # pure-noise PBO sits well above zero (often above one half).
+    trials = _noise(20, 400, seed=7)
+    first = probability_of_backtest_overfitting(trials, n_splits=10)
+    second = probability_of_backtest_overfitting(trials, n_splits=10)
+    assert first.to_dict() == second.to_dict()
+    assert 0.3 < first.value <= 1.0
+
+
+def test_matches_independent_bruteforce() -> None:
+    trials = _noise(5, 60, seed=3)
+    cols = list(trials.values())
+    S, rows = 6, 10
+    lambdas = []
+    for train in itertools.combinations(range(S), S // 2):
+        test = [b for b in range(S) if b not in train]
+
+        def perf(col, blocks):
+            xs = [x for b in blocks for x in col[b * rows:(b + 1) * rows]]
+            return statistics.mean(xs) / statistics.stdev(xs)
+
+        is_p = [perf(c, train) for c in cols]
+        oos_p = [perf(c, test) for c in cols]
+        best = is_p.index(max(is_p))
+        rank = sorted(oos_p).index(oos_p[best]) + 1
+        omega = rank / (len(cols) + 1)
+        lambdas.append(math.log(omega / (1 - omega)))
+    expected = sum(1 for lam in lambdas if lam <= 0) / len(lambdas)
+    result = probability_of_backtest_overfitting(trials, n_splits=S)
+    assert result.value == pytest.approx(expected)
+
+
+def test_log_growth_metric_and_leading_rows_dropped() -> None:
+    trials = _noise(4, 103, seed=9)
+    result = probability_of_backtest_overfitting(trials, n_splits=10, metric="mean_log_growth")
+    assert result.status == STATUS_COMPUTED
+    assert result.details["rows_dropped_leading"] == 3
+    assert result.details["rows_per_block"] == 10
+
+
+@pytest.mark.parametrize(
+    ("data", "kwargs", "reason"),
+    [
+        ({"only": [0.01, -0.01] * 50}, {}, "insufficient_trials"),
+        ({"a": [0.01, -0.01] * 50, "b": [0.01, -0.01] * 49}, {}, "trial_returns_not_aligned"),
+        (_noise(3, 20, 1), {"n_splits": 16}, "insufficient_observations"),
+        (_noise(3, 100, 1), {"n_splits": 7}, "invalid_input"),
+        (_noise(3, 100, 1), {"n_splits": 18}, "invalid_input"),
+        (_noise(3, 100, 1), {"metric": "cagr"}, "invalid_input"),
+        ({"a": [0.01] * 64, "b": [0.01, -0.01] * 32}, {"n_splits": 4}, "undefined_performance"),
+        ({"a": [math.nan] * 64, "b": [0.0] * 64}, {"n_splits": 4}, "invalid_input"),
+        ({"a": [-1.0] * 64, "b": [0.0] * 64}, {"n_splits": 4}, "invalid_input"),
+    ],
+)
+def test_fails_closed_without_a_number(data, kwargs, reason) -> None:
+    result = probability_of_backtest_overfitting(data, **kwargs)
+    assert result.status == STATUS_UNCOMPUTABLE
+    assert result.value is None
+    assert result.reason_code == reason
+
+
+def test_sequence_input_and_tie_counting() -> None:
+    col = [0.01, -0.005, 0.002, -0.001] * 10
+    result = probability_of_backtest_overfitting([col, list(col)], n_splits=4)
+    assert result.status == STATUS_COMPUTED
+    assert result.details["is_best_ties"] == result.details["n_combinations"]
+    assert result.details["trial_labels"] == ["0", "1"]