diff --git a/docs/research_stats.zh-CN.md b/docs/research_stats.zh-CN.md index cc6ceb9..6e9104e 100644 --- a/docs/research_stats.zh-CN.md +++ b/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,6 +24,7 @@ - 偏度/峰度用总体矩估计，峰度为非超额（正态 = 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`）。 @@ -30,4 +32,4 @@ - 不修改 `strategy_lifecycle/performance_metrics.py` 等现有共享指标函数的语义。 - 不接入任何消费者，不升级任何下游 QPK pin，不改 `QPK_PIN`。 -- PBO（CSCV）、与 UES trial journal 的接线留给后续 RS-02。 +- 与 UES trial journal 的接线在 UES 侧（RS-02）完成，QPK 不感知具体策略。 diff --git a/src/quant_platform_kit/research_stats/__init__.py b/src/quant_platform_kit/research_stats/__init__.py index 5cfc379..f475bb2 100644 --- a/src/quant_platform_kit/research_stats/__init__.py +++ b/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", ] diff --git a/src/quant_platform_kit/research_stats/overfitting.py b/src/quant_platform_kit/research_stats/overfitting.py new file mode 100644 index 0000000..aa3147a --- /dev/null +++ b/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, + }, + ) diff --git a/tests/test_research_stats_pbo.py b/tests/test_research_stats_pbo.py new file mode 100644 index 0000000..2a7715a --- /dev/null +++ b/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"]