第 30 篇：过拟合与样本外检验

BTC 2021-01-01 到 2022-12-31：价格 29,332 → 16,542，买入持有年化 -24.93%
在 10,130 组（快线 5–50 × 慢线 20–250）里选年化最高的一组 = (31, 36)
根数         731.0
累计收益      6.4935
年化收益      1.7374
年化波动      0.7271
夏普比率      1.7463
索提诺比率     2.7964
最大回撤     -0.3918
卡玛比率      4.4344
在水下的比例    0.9207
最长水下根数     196.0
最好的一根     0.1954
最差的一根    -0.1302

table = pd.DataFrame({"快线": [f for f, s in PAIRS], "慢线": [s for f, s in PAIRS], "年化": in_annual})
grid = V.surface(table, "快线", "慢线", "年化")
print(grid.loc[29:33, 34:38].round(4).to_string())
print()
print(V.neighbourhood(grid, PAIRS[best]).round(4).to_string())
print(f"\n全网格 {len(PAIRS):,} 格：年化中位 {np.median(in_annual):.2%}，"
      f"90% 分位 {np.percentile(in_annual, 90):.2%}，亏钱的占 {(in_annual < 0).mean():.1%}")

慢线      34      35      36      37      38
快线                                        
29 -0.0150  0.1036  0.2979  0.5080  0.2802
30 -0.0769  0.1970  0.5162  0.4273  0.7164
31  0.4837  0.9717  1.7374  0.6749  0.3978
32  0.6932  1.1423  0.9502  0.5814  0.2443
33  1.2247  1.1393  0.7853  0.7495  0.5765

这一格     1.7374
邻居数     8.0000
邻居中位    0.6282
邻居最差    0.1970
邻居最好    1.1423
落差      1.1092

全网格 10,130 格：年化中位 13.38%，90% 分位 36.74%，亏钱的占 26.8%

rows = []
for years in [1, 2, 3, 5, 9]:
    piece = DAILY[:, -int(365 * years):]
    values = annual(piece)
    top = int(values.argmax())
    fast, slow = PAIRS[top]
    neighbours = [values[WHERE[(f, s)]] for f in (fast - 1, fast, fast + 1)
                  for s in (slow - 1, slow, slow + 1) if (f, s) != (fast, slow) and (f, s) in WHERE]
    rows.append({"回测几年": years, "最好的一格": f"({fast},{slow})", "年化": values[top],
                 "邻居中位": np.median(neighbours), "落差": values[top] - np.median(neighbours),
                 "落差占年化": (values[top] - np.median(neighbours)) / values[top],
                 "全网格中位": np.median(values)})
print(pd.DataFrame(rows).round(4).to_string(index=False))

 回测几年   最好的一格     年化   邻居中位     落差  落差占年化   全网格中位
    1 (50,55) 0.9023 0.3073 0.5950 0.6594 -0.1355
    2 (21,23) 0.5767 0.0621 0.5146 0.8923 -0.0404
    3 (40,41) 0.5554 0.3134 0.2419 0.4356  0.0471
    5 (26,27) 0.4690 0.1577 0.3113 0.6638  0.0197
    9  (8,43) 0.5930 0.5199 0.0731 0.1233  0.0422

plateau = V.plateaus(grid)
rows = []
for label, cell in [("按「这一格最高」挑", PAIRS[best]),
                    ("按「邻域中位最高」挑", (int(plateau.iloc[0]["快线"]), int(plateau.iloc[0]["慢线"]))),
                    ("课程默认（第 12 篇的 50/200，没挑过）", (50, 200))]:
    k = WHERE[cell]
    rows.append({"挑法": label, "参数": f"({cell[0]},{cell[1]})", "样本内年化": in_annual[k],
                 "样本外年化": out_annual[k], "样本外排名": int((out_annual > out_annual[k]).sum()) + 1})
rows.append({"挑法": "全网格中位（什么都不挑）", "参数": f"{len(PAIRS):,} 格",
             "样本内年化": np.median(in_annual), "样本外年化": np.median(out_annual),
             "样本外排名": len(PAIRS) // 2})
print(pd.DataFrame(rows).round(4).to_string(index=False))
order = np.argsort(-in_annual)
ladder = [{"样本内前几名": n, "样本外年化中位": np.median(out_annual[order[:n]])}
          for n in [10, 50, 100, 500, 1000, 5000]]
ladder.append({"样本内前几名": len(PAIRS), "样本外年化中位": np.median(out_annual)})
print(pd.DataFrame(ladder).round(4).to_string(index=False))
print(f"\n样本内排名和样本外排名的秩相关："
      f"{pd.Series(in_annual).rank().corr(pd.Series(out_annual).rank()):.4f}")

                      挑法       参数  样本内年化   样本外年化  样本外排名
               按「这一格最高」挑  (31,36) 1.7374 -0.0766   7083
              按「邻域中位最高」挑  (32,34) 0.6932 -0.0077   4551
课程默认（第 12 篇的 50/200，没挑过） (50,200) 0.1203 -0.1439   8927
            全网格中位（什么都不挑） 10,130 格 0.1338 -0.0211   5065
 样本内前几名  样本外年化中位
     10  -0.0155
     50  -0.0075
    100  -0.0052
    500   0.0022
   1000   0.0032
   5000  -0.0392
  10130  -0.0211

样本内排名和样本外排名的秩相关：-0.1597

 样本内前几名  样本外年化中位
     10  -0.0155
     50  -0.0075
    100  -0.0052
    500   0.0022
   1000   0.0032
   5000  -0.0392
  10130  -0.0211

start = int(np.argmax(DATES >= pd.Timestamp("2018-01-01", tz="UTC")))   # 前 250 根留给均线预热
rolling = DAILY[:, start:]
rows = []
for anchored in (False, True):
    for train_bars, test_bars in [(365, 182), (730, 365), (1095, 365)]:
        splits = V.walk_forward(rolling.shape[1], train_bars, test_bars, anchored=anchored)
        picked = V.walk_forward_run(rolling, splits, names=PAIRS)
        stitched = V.stitch(rolling, splits, [PAIRS.index(p) for p in picked["选了谁"]])
        equity = RP.to_curve(pd.Series(stitched, index=pd.RangeIndex(len(stitched))))
        rows.append({"训练窗口": "锚定（越考越长）" if anchored else "滚动（固定长度）",
                     "训练根数": train_bars, "考卷根数": test_bars, "几段": len(splits),
                     "训练期夏普中位": picked["训练期夏普"].median() * np.sqrt(365),
                     "考卷夏普中位": picked["考卷夏普"].median() * np.sqrt(365),
                     "样本外年化": RP.annual_return(equity, 365),
                     "样本外最大回撤": RP.max_drawdown(equity)})
print(pd.DataFrame(rows).round(4).to_string(index=False))
splits = V.walk_forward(rolling.shape[1], 730, 365, anchored=False)
picked = V.walk_forward_run(rolling, splits, names=PAIRS)
picked["训练期夏普"] *= np.sqrt(365)
picked["考卷夏普"] *= np.sqrt(365)
print()
print(picked.round(4).to_string(index=False))
first_test = DATES[start:][splits[0][1].start]
tail = window(str(first_test.date()))
print(f"\n同一段（{first_test.date()} 起）的对照：")
for label, pair in [("课程默认 (50,200)", (50, 200)), ("九年全样本最好的 (8,43)（这是作弊）", (8, 43)),
                    ("决策点那一组 (31,36)", (31, 36))]:
    piece = DAILY[WHERE[pair], tail]
    equity = RP.to_curve(pd.Series(piece, index=pd.RangeIndex(len(piece))))
    print(f"  {label:<34} 年化 {RP.annual_return(equity, 365):>8.2%}   "
          f"夏普 {RP.sharpe(equity.pct_change().dropna(), 365):.4f}")
print(f"  {'买入持有':<34} 年化 {RP.annual_return(close.loc[first_test:], 365):>8.2%}   "
      f"夏普 {RP.sharpe(close.loc[first_test:].pct_change().dropna(), 365):.4f}")

    训练窗口  训练根数  考卷根数  几段  训练期夏普中位  考卷夏普中位   样本外年化  样本外最大回撤
滚动（固定长度）   365   182  15   2.2397  0.1065 -0.0379  -0.7897
滚动（固定长度）   730   365   6   1.6923  0.2493  0.0875  -0.8473
滚动（固定长度）  1095   365   5   1.5992 -0.1388 -0.1634  -0.8042
锚定（越考越长）   365   182  15   1.2128 -0.0799  0.0547  -0.8909
锚定（越考越长）   730   365   6   1.1574  0.6620  0.1806  -0.8826
锚定（越考越长）  1095   365   5   1.0290  0.3004 -0.1398  -0.8826

 第几段      选了谁  训练期夏普    考卷夏普    考卷收益  训练根数  考卷根数
   1 (18, 20) 1.5455  2.6860  4.7503   730   365
   2 (16, 20) 2.4000 -1.0929 -0.7029   730   365
   3 (29, 30) 2.0341  0.5620  0.1685   730   365
   4 (31, 36) 1.7514  1.0642  0.4533   730   365
   5 (50, 56) 1.6333 -0.5748 -0.3615   730   365
   6 (44, 46) 1.5577 -0.0634 -0.1072   730   365

同一段（2020-01-01 起）的对照：
  课程默认 (50,200)                      年化   -5.65%   夏普 0.2123
  九年全样本最好的 (8,43)（这是作弊）              年化   43.01%   夏普 0.8905
  决策点那一组 (31,36)                     年化   42.76%   夏普 0.8865
  买入持有                               年化   43.10%   夏普 0.9015

warm = close.loc["2020-11-01":"2022-12-31"]          # 多留 250 根给均线预热
real_annual, real_sharpe = in_annual[best], in_sharpe[best]
rows = []
for block, label in [(1, "打散（连波动率聚集也没了）"), (20, "20 天的块（波动率聚集还在）")]:
    tops_annual, tops_sharpe = [], []
    for path in V.synthetic_close(warm, n=N_PATHS, block=block, seed=30):
        piece = grid_returns(path, PAIRS)[:, -730:]
        tops_annual.append(annual(piece).max())
        tops_sharpe.append(sharpe(piece).max())
    tops_annual, tops_sharpe = np.array(tops_annual), np.array(tops_sharpe)
    beaten = float((tops_annual >= real_annual).mean())
    rows.append({"假数据怎么造的": label, "「最好一组」年化中位": np.median(tops_annual),
                 "90% 分位": np.percentile(tops_annual, 90), "最大": tops_annual.max(),
                 "比真实还好的比例": beaten, "夏普中位": np.median(tops_sharpe),
                 "夏普比真实还好的比例": float((tops_sharpe >= real_sharpe).mean())})
print(f"真实数据上「最好的一组」：年化 {real_annual:.2%}，夏普 {real_sharpe:.4f}")
monte_carlo = pd.DataFrame(rows)
print(monte_carlo.round(4).to_string(index=False))

真实数据上「最好的一组」：年化 173.74%，夏普 1.7463
        假数据怎么造的  「最好一组」年化中位  90% 分位     最大  比真实还好的比例   夏普中位  夏普比真实还好的比例
  打散（连波动率聚集也没了）      0.9359  2.0401 5.9279     0.185 1.2862       0.200
20 天的块（波动率聚集还在）      1.1552  2.6750 6.7283     0.270 1.4275       0.265

print(f"{len(PAIRS):,} 组参数的夏普比率：中位 {np.median(in_sharpe):.4f}，"
      f"标准差 {in_sharpe.std():.4f}，最大 {in_sharpe.max():.4f}")
print(f"完全没本事时，试 {len(PAIRS):,} 次「最好的那次」期望是 "
      f"{V.expected_max_sharpe(len(PAIRS), in_sharpe.std()):.4f}")
rows = [V.deflated_sharpe(real_sharpe, n, train.shape[1], in_sharpe.std(),
                          skew=float(pd.Series(train[best]).skew()),
                          kurtosis=float(pd.Series(train[best]).kurt()) + 3,
                          periods_per_year=365).rename(f"试了 {n:,} 次")
        for n in [2, 100, 1000, len(PAIRS)]]
print(pd.concat(rows, axis=1).round(4).to_string())

10,130 组参数的夏普比率：中位 0.5357，标准差 0.2829，最大 1.7463
完全没本事时，试 10,130 次「最好的那次」期望是 1.0931
            试了 2 次  试了 100 次  试了 1,000 次  试了 10,130 次
年化夏普        1.7463    1.7463      1.7463       1.7463
试验次数        2.0000  100.0000   1000.0000   10130.0000
白捡的门槛（年化）   0.1470    0.7159      0.9209       1.0931
超过门槛多少      1.5992    1.0303      0.8254       0.6532
t 值         2.2922    1.4768      1.1830       0.9362
打过折的夏普 DSR  0.9891    0.9301      0.8816       0.8254

rows = [V.pbo(train, chunks).rename(f"切成 {chunks} 块") for chunks in (8, 12, 16)]
print(pd.concat(rows, axis=1).round(4).to_string())

               切成 8 块     切成 12 块     切成 16 块
试验次数       10130.0000  10130.0000  10130.0000
切成几块           8.0000     12.0000     16.0000
一共几种分法        70.0000    924.0000  12870.0000
样本外分位中位数       0.3222      0.4332      0.5209
过拟合概率 PBO      0.6143      0.5357      0.4886
logit 中位数     -0.7444     -0.2687      0.0838

"""talab.validate：过拟合与样本外检验。第 30 篇。

第 27 篇保证引擎对、第 28 篇保证数据干净、第 29 篇保证读数不被自己骗。
这个模块回答最后一个也是最难的问题：**这条策略是学会了，还是把答案背下来了。**

难在哪里：**样本内的成绩永远可以做到任意好。**参数多试几组、规则多加几条，
回测收益一定上升——上升本身不携带任何信息。所以这里的每一个函数都不是在「算成绩」，
而是在**给成绩打折**：

1. **参数曲面**（`surface`、`neighbourhood`、`plateaus`）：这一格好，是因为它周围一片都好，
   还是因为它自己恰好踩中了？后者叫尖峰，它在样本外几乎必然塌。
2. **切分**（`split_index`、`walk_forward`、`walk_forward_run`）：拿一段没看过的数据当考卷。
3. **多重检验**（`expected_max_sharpe`、`deflated_sharpe`）：你试了一万次，
   「最好的那次」本来就该很好看——先把这个白捡的部分减掉。
4. **回测过拟合概率**（`pbo`）：把时间切成块，反复交换样本内外，
   数一数「样本内第一名在样本外落到中位数以下」的比例。
"""
from __future__ import annotations

import itertools
import math
from statistics import NormalDist

import numpy as np
import pandas as pd

NORMAL = NormalDist()
EULER = 0.5772156649015329                       # 欧拉常数，`expected_max_sharpe` 要用

def surface(table: pd.DataFrame, index: str, columns: str, values: str) -> pd.DataFrame:
    """把「一行一组参数」的长表摊成二维曲面，行列都是参数值。

    摊成二维之后才能问「这一格的邻居怎么样」——而那正是这一篇的核心问题。
    """
    return table.pivot(index=index, columns=columns, values=values).sort_index().sort_index(axis=1)


def neighbourhood(grid: pd.DataFrame, cell: tuple, radius: int = 1) -> pd.Series:
    """一格和它周围一圈的成绩：本格、邻居的中位数 / 最差 / 最好，以及落差。

    `落差` = 本格 − 邻居中位数。它是这一篇最省事的过拟合警报：
    **落差越大，说明这一格越像是踩中的，而不是踩在一片实地上。**

    ⚠️ 邻居按**网格上的位置**算，不按参数值相差多少。参数网格常常不是等距的
    （5、10、20、40、60 这种），这时候「走错一步」的正确含义是「换成表里相邻的那个值」，
    不是「加一减一」。
    """
    row, column = cell
    i, j = grid.index.get_loc(row), grid.columns.get_loc(column)
    block = grid.iloc[max(i - radius, 0):i + radius + 1, max(j - radius, 0):j + radius + 1]
    here = float(grid.iloc[i, j])
    others = block.to_numpy(float).ravel()
    others = np.delete(others, list(block.index).index(row) * block.shape[1]
                       + list(block.columns).index(column))
    others = others[~np.isnan(others)]
    if not len(others):
        raise ValueError("这一格周围没有有效的邻居")
    return pd.Series({"这一格": here, "邻居数": float(len(others)),
                      "邻居中位": float(np.median(others)), "邻居最差": float(others.min()),
                      "邻居最好": float(others.max()), "落差": here - float(np.median(others))})


def plateaus(grid: pd.DataFrame, radius: int = 1) -> pd.DataFrame:
    """给每一格配上它邻域的中位数，按**邻域中位数**从高到低排序。

    这是「不挑尖峰」的选参数方法：不要问「哪一格最高」，要问
    **「哪一片地最高」**——站在一片高地上，走错一步还在高地上。
    """
    rows = []
    for row in grid.index:
        for column in grid.columns:
            if np.isnan(grid.loc[row, column]):
                continue
            stats = neighbourhood(grid, (row, column), radius)
            rows.append({grid.index.name or "行": row, grid.columns.name or "列": column,
                         "这一格": stats["这一格"], "邻域中位": stats["邻居中位"],
                         "落差": stats["落差"]})
    return pd.DataFrame(rows).sort_values("邻域中位", ascending=False).reset_index(drop=True)

def split_index(n: int, fraction: float = 0.5) -> tuple[slice, slice]:
    """把 n 根 K 线按时间顺序切成前后两段：前面选参数，后面当考卷。

    ⚠️ 只能按**时间顺序**切。随机抽一半当样本外是错的——
    相邻两天的行情高度相关，随机抽等于让考卷和复习资料互相抄。
    """
    if not 0 < fraction < 1:
        raise ValueError("fraction 要在 0 和 1 之间")
    cut = int(n * fraction)
    if cut < 1 or cut >= n:
        raise ValueError("切出来的两段都要至少有一根")
    return slice(0, cut), slice(cut, n)


def walk_forward(n: int, train: int, test: int, anchored: bool = False) -> list[tuple[slice, slice]]:
    """滚动检验的切分：每一段用前 `train` 根挑参数，紧接着的 `test` 根拿来考试。

    - `anchored=False`（默认）：训练窗口长度固定，整段往前滚
    - `anchored=True`：训练窗口从第一根开始，越考越长（更像真实的积累过程）

    各段的考卷首尾相接、互不重叠，把它们拼起来就是一条**每一根都没被看过**的资金曲线。
    """
    if train < 1 or test < 1:
        raise ValueError("train 和 test 都要至少是 1")
    out, start = [], 0
    while start + train + test <= n:
        out.append((slice(0 if anchored else start, start + train),
                    slice(start + train, start + train + test)))
        start += test
    if not out:
        raise ValueError(f"{n} 根放不下一次 {train}+{test} 的切分")
    return out


def walk_forward_run(returns: np.ndarray, splits, names=None, min_bars: int = 2) -> pd.DataFrame:
    """在每一段上选出训练期里夏普最高的那一列，记下它在考卷上的表现。

    `returns` 是 `(试验数, 根数)` 的每根收益矩阵——一行一组参数。
    返回一张表：每一段选了谁、训练期分数、考卷分数，以及**这一段的考卷收益**。
    """
    rows = []
    for k, (train, test) in enumerate(splits):
        inside, outside = returns[:, train], returns[:, test]
        if inside.shape[1] < min_bars or outside.shape[1] < min_bars:
            raise ValueError("某一段太短，算不出标准差")
        score = inside.mean(axis=1) / np.where(inside.std(axis=1) > 0, inside.std(axis=1), np.nan)
        pick = int(np.nanargmax(score))
        rows.append({"第几段": k + 1, "选了谁": names[pick] if names is not None else pick,
                     "训练期夏普": float(score[pick]),
                     "考卷夏普": float(outside[pick].mean() / outside[pick].std()),
                     "考卷收益": float(np.prod(1 + outside[pick]) - 1),
                     "训练根数": inside.shape[1], "考卷根数": outside.shape[1]})
    return pd.DataFrame(rows)


def stitch(returns: np.ndarray, splits, picks) -> np.ndarray:
    """把各段考卷上被选中那一列的收益首尾接起来，得到一条完整的样本外收益序列。"""
    return np.concatenate([returns[int(p), test] for (_, test), p in zip(splits, picks)])

def sma_matrix(values: np.ndarray, windows) -> np.ndarray:
    """一次算出所有窗口长度的简单移动平均，返回 `(窗口个数, 根数)`。"""
    cumulative = np.concatenate([[0.0], np.cumsum(values)])
    out = np.full((len(windows), len(values)), np.nan)
    for k, n in enumerate(windows):
        out[k, n - 1:] = (cumulative[n:] - cumulative[:-n]) / n
    return out

def expected_max_sharpe(n_trials: int, sharpe_std: float) -> float:
    """**完全没有本事**的情况下，试 `n_trials` 次，「最好的那次」夏普期望是多少
    （Bailey & López de Prado 2014）：

    > E[最大值] ≈ 标准差 × [(1 − γ)·Φ⁻¹(1 − 1/N) + γ·Φ⁻¹(1 − 1/(N·e))]

    γ 是欧拉常数 0.5772。`sharpe_std` 是这一万次试验的夏普比率**彼此之间**的标准差。
    直觉：N 越大，最大值越往右跑——**「我试了一万组，最好的那组夏普 1.5」这句话里，
    「一万组」和「1.5」一样重要。**
    """
    if n_trials < 2:
        raise ValueError("至少要有两次试验")
    high = NORMAL.inv_cdf(1 - 1 / n_trials)
    low = NORMAL.inv_cdf(1 - 1 / (n_trials * math.e))
    return sharpe_std * ((1 - EULER) * high + EULER * low)


def deflated_sharpe(sharpe: float, n_trials: int, n_obs: int, sharpe_std: float,
                    skew: float = 0.0, kurtosis: float = 3.0,
                    periods_per_year: int = 252) -> pd.Series:
    """打过折的夏普比率（DSR）：把「试了很多次」和「收益率不是正态」都扣掉之后，
    这条策略的夏普**真的大于 0** 的概率。

    两步：先用 `expected_max_sharpe` 算出「没本事时白捡的那一截」当作门槛，
    再问「观察到的夏普超过这个门槛」有多显著：

    > DSR = Φ( (夏普 − 门槛) × √(根数 − 1) ÷ √(1 − 偏度×夏普 + (峰度−1)/4 × 夏普²) )

    式子里的夏普都是**单根**的（不年化）。`kurtosis` 传的是原始峰度（正态是 3），
    不是超额峰度。DSR 低于 0.95 就该当成「还没排除运气」。
    """
    per_period = sharpe / math.sqrt(periods_per_year)
    threshold = expected_max_sharpe(n_trials, sharpe_std / math.sqrt(periods_per_year))
    spread = math.sqrt(max(1 - skew * per_period + (kurtosis - 1) / 4 * per_period ** 2, 1e-12))
    z = (per_period - threshold) * math.sqrt(max(n_obs - 1, 1)) / spread
    return pd.Series({"年化夏普": sharpe, "试验次数": float(n_trials),
                      "白捡的门槛（年化）": threshold * math.sqrt(periods_per_year),
                      "超过门槛多少": sharpe - threshold * math.sqrt(periods_per_year),
                      "t 值": z, "打过折的夏普 DSR": NORMAL.cdf(z)})


# ---------------------------------------------------------------------------
# 四、回测过拟合概率（CSCV）
# ---------------------------------------------------------------------------

def pbo(returns: np.ndarray, n_chunks: int = 8) -> pd.Series:
    """回测过拟合概率（Bailey 等 2017 的 CSCV）。

    做法：把时间切成 `n_chunks` 块，穷举所有「一半当样本内、另一半当样本外」的分法；
    每一种分法里选出样本内夏普最高的那一列，看它**在样本外排第几**。
    如果这个第一名在样本外经常掉到中位数以下，说明「挑第一名」这个动作本身就是在挑噪声。

    > PBO = 样本内第一名在样本外掉到中位数以下的分法比例

    ⚠️ 它衡量的是**挑选流程**，不是某一组参数。PBO 高不代表这些参数都没用，
    代表「用样本内最优去选参数」这件事在这批数据上不管用。
    n_chunks 要是偶数；16 块有 12,870 种分法，8 块只有 70 种，慢和稳之间自己选。
    """
    if n_chunks % 2 or n_chunks < 4:
        raise ValueError("n_chunks 要是不小于 4 的偶数")
    n_trials, n_bars = returns.shape
    if n_trials < 2:
        raise ValueError("至少要有两列试验")
    edges = np.linspace(0, n_bars, n_chunks + 1).astype(int)
    counts = np.diff(edges).astype(float)
    total = np.array([returns[:, edges[i]:edges[i + 1]].sum(axis=1) for i in range(n_chunks)])
    square = np.array([(returns[:, edges[i]:edges[i + 1]] ** 2).sum(axis=1) for i in range(n_chunks)])

    def score(keys) -> np.ndarray:                 # 任意几块合起来的夏普，O(试验数)
        n = counts[list(keys)].sum()
        mean = total[list(keys)].sum(axis=0) / n
        var = square[list(keys)].sum(axis=0) / n - mean ** 2
        return mean / np.sqrt(np.maximum(var, 1e-300))

    logits, ranks = [], []
    everything = range(n_chunks)
    for inside in itertools.combinations(everything, n_chunks // 2):
        outside = tuple(k for k in everything if k not in inside)
        best = int(np.argmax(score(inside)))
        outside_score = score(outside)
        rank = int((outside_score < outside_score[best]).sum()) + 1      # 1 最差，n_trials 最好
        omega = rank / (n_trials + 1)
        ranks.append(omega)
        logits.append(math.log(omega / (1 - omega)))
    logits = np.array(logits)
    return pd.Series({"试验次数": float(n_trials), "切成几块": float(n_chunks),
                      "一共几种分法": float(len(logits)),
                      "样本外分位中位数": float(np.median(ranks)),
                      "过拟合概率 PBO": float((logits <= 0).mean()),
                      "logit 中位数": float(np.median(logits))})

def synthetic_close(close: pd.Series, n: int = 100, block: int = 1, seed: int = 0) -> np.ndarray:
    """造 n 条「和原序列同分布、但没有原来那段走势」的假价格，返回 `(n, 根数)`。

    `block=1` 是把日收益率完全打散（波动率聚集也一起没了）；
    `block > 1` 用块自助法，保留每一小段内部的顺序，**波动率聚集还在，趋势没了**。
    拿这些假价格重跑一遍同样的参数扫描，就知道「最好的一组」里有多少是白捡的。
    """
    values = close.pct_change().dropna().to_numpy(float)
    m = len(values)
    if block < 1 or block > m:
        raise ValueError("block 要在 1 和样本长度之间")
    rng = np.random.default_rng(seed)
    out = np.empty((n, m + 1))
    for i in range(n):
        if block == 1:
            sample = rng.choice(values, size=m, replace=True)
        else:
            starts = rng.integers(0, m - block + 1, size=int(np.ceil(m / block)))
            sample = np.concatenate([values[s:s + block] for s in starts])[:m]
        out[i] = float(close.iloc[0]) * np.concatenate([[1.0], np.cumprod(1 + sample)])
    return out

def test_expected_max_sharpe_matches_a_simulation():
    """这是整个模块最该钉住的一条：公式算出来的「白捡的最大值」要对得上真的抽一万次。"""
    rng = np.random.default_rng(30)
    for n_trials in (50, 1000, 10_000):
        draws = rng.normal(0.0, 0.3, size=(400, n_trials)).max(axis=1)
        assert V.expected_max_sharpe(n_trials, 0.3) == pytest.approx(draws.mean(), rel=0.05)

def test_pbo_is_a_coin_flip_on_pure_noise():
    """一堆互相没区别的随机策略，样本内第一名在样本外就是随机的——PBO 该在 0.5 附近。"""
    returns = np.random.default_rng(1).normal(0.0, 0.01, (100, 800))
    out = V.pbo(returns, 8)
    assert out["一共几种分法"] == 70
    assert out["过拟合概率 PBO"] == pytest.approx(0.5, abs=0.2)


def test_pbo_is_near_zero_when_one_strategy_is_genuinely_better():
    returns = np.random.default_rng(2).normal(0.0, 0.01, (40, 800))
    returns[7] += 0.01                                      # 第 7 列全程真的更好
    out = V.pbo(returns, 8)
    assert out["过拟合概率 PBO"] < 0.05
    assert out["样本外分位中位数"] > 0.9
    with pytest.raises(ValueError):
        V.pbo(returns, 7)                                   # 块数要是偶数

verdict = pd.DataFrame([
    {"检查": "① 邻居（落差 ÷ 年化）", "这条策略": f"{V.neighbourhood(grid, PAIRS[best])['落差'] / real_annual:.1%}",
     "该是多少": "越接近 0 越好", "过了吗": "否"},
    {"检查": "② 样本外年化", "这条策略": f"{out_annual[best]:.2%}",
     "该是多少": f"网格中位 {np.median(out_annual):.2%}", "过了吗": "否"},
    {"检查": "③ walk-forward 考卷夏普中位", "这条策略": f"{picked['考卷夏普'].median():.4f}",
     "该是多少": f"训练期是 {picked['训练期夏普'].median():.4f}", "过了吗": "否"},
    {"检查": "④ 蒙特卡洛：没有趋势的假数据也能做到",
     "这条策略": f"{monte_carlo['比真实还好的比例'].max():.1%}", "该是多少": "低于 5%", "过了吗": "否"},
    {"检查": "⑤ 多重检验：白捡的门槛（年化夏普）",
     "这条策略": f"{real_sharpe:.4f}", "该是多少": f"门槛 {V.expected_max_sharpe(len(PAIRS), in_sharpe.std()):.4f}",
     "过了吗": "勉强"},
    {"检查": "⑥ 回测过拟合概率 PBO", "这条策略": f"{V.pbo(train, 16)['过拟合概率 PBO']:.4f}",
     "该是多少": "低于 0.5 才有意义", "过了吗": "否"},
])
print(verdict.to_string(index=False))

                   检查   这条策略        该是多少 过了吗
        ① 邻居（落差 ÷ 年化）  63.8%    越接近 0 越好   否
              ② 样本外年化 -7.66% 网格中位 -2.11%   否
③ walk-forward 考卷夏普中位 0.2493 训练期是 1.6923   否
  ④ 蒙特卡洛：没有趋势的假数据也能做到  27.0%       低于 5%   否
   ⑤ 多重检验：白捡的门槛（年化夏普） 1.7463   门槛 1.0931  勉强
        ⑥ 回测过拟合概率 PBO 0.4886 低于 0.5 才有意义   否

spy = D.load_nasdaq_daily("data/nasdaq/SPY_historical.json").drop(pd.Timestamp("2026-04-20"))
aapl = D.adjust_total_return(D.unadjust_splits(D.load_nasdaq_daily("data/nasdaq/AAPL_historical.json"),
                                               D.SPLITS["AAPL"]), D.SPLITS["AAPL"],
                             D.load_nasdaq_dividends("data/nasdaq/AAPL_dividends.json"))
MAIN = [(f, s, b, k) for f in [20, 30, 40, 50, 60, 80] for s in [100, 150, 200, 250, 300]
        for b in [5, 10, 20, 40, 60] for k in [2.0, 2.5, 3.0, 3.5, 4.0]]
DEFAULT = (50, 200, 20, 3.0)
print(f"主线网格 {len(MAIN)} 组（快线 × 慢线 × 突破窗口 × 吊灯倍数），默认那一组是 {DEFAULT}")
rows, surfaces = [], {}
for name, frame, periods in [("SPY", spy, 252), ("AAPL", aapl, 252), ("BTC", btc, 365)]:
    price = frame["close"]
    curves = []
    for fast, slow, breakout, k in MAIN:
        plan = BT.Plan(entry=((I.sma(price, fast) > I.sma(price, slow))
                              & (price >= price.rolling(breakout).max())).fillna(False),
                       exit=I.cross_below(I.sma(price, fast), I.sma(price, slow)).fillna(False),
                       stop="chandelier", k=k, trigger="close", sizing="risk", risk_per_trade=0.10)
        curves.append(BT.run(frame, plan, 100_000.0)["资金曲线"].pct_change().fillna(0).to_numpy())
    matrix = np.array(curves)
    scores = sharpe(matrix, periods)
    here = MAIN.index(DEFAULT)
    slice_table = pd.DataFrame([{"快线": f, "慢线": s, "夏普": scores[k]}
                                for k, (f, s, b, kk) in enumerate(MAIN)
                                if (b, kk) == (DEFAULT[2], DEFAULT[3])])
    face = V.surface(slice_table, "快线", "慢线", "夏普")
    surfaces[name] = face
    around = V.neighbourhood(face, (DEFAULT[0], DEFAULT[1]))
    rows.append({"标的": name, "默认那一组的夏普": scores[here],
                 "排第几": f"{int((scores > scores[here]).sum()) + 1}/{len(MAIN)}",
                 "邻居中位": around["邻居中位"], "落差": around["落差"],
                 "网格最好": scores.max(), "白捡的门槛": V.expected_max_sharpe(len(MAIN), scores.std()),
                 "最好的 DSR": V.deflated_sharpe(scores.max(), len(MAIN), matrix.shape[1],
                                               scores.std(), periods_per_year=periods)["打过折的夏普 DSR"],
                 "PBO": V.pbo(matrix, 16)["过拟合概率 PBO"]})
print(pd.DataFrame(rows).round(4).to_string(index=False))

主线网格 750 组（快线 × 慢线 × 突破窗口 × 吊灯倍数），默认那一组是 (50, 200, 20, 3.0)
  标的  默认那一组的夏普     排第几   邻居中位      落差   网格最好  白捡的门槛  最好的 DSR    PBO
 SPY    0.4261 518/750 0.4534 -0.0273 0.8275 0.4834   0.8604 0.7476
AAPL    0.7011 416/750 0.6089  0.0922 1.1732 0.5979   0.9647 0.4324
 BTC    0.8903 282/750 0.9088 -0.0186 1.1908 0.5977   0.9623 0.5211

BTC 上固定「突破 20 日新高、吊灯 3 ATR」时的夏普曲面（行是快线，列是慢线）：
慢线    100    150    200    250    300
快线                                   
20  1.159  1.004  0.890  0.950  1.094
30  1.174  1.023  0.858  0.970  1.043
40  1.131  0.976  0.894  0.944  0.960
50  1.020  0.943  0.890  0.908  0.973
60  0.988  0.910  0.875  0.894  0.993
80  0.915  0.940  0.828  0.787  1.058

FAST3 = [10, 20, 30, 40, 50, 60, 80, 100]
SLOW3 = [100, 120, 150, 180, 200, 220, 250, 300]
COMBOS = [(f, s, h, l) for f in FAST3 for s in SLOW3 if s > f
          for h in [10, 20, 40, 60, 90] for l in [10, 20, 40, 60, 90]]
values = close.to_numpy()
averages = {n: close.rolling(n).mean().to_numpy() for n in set(FAST3) | set(SLOW3)}
highs = {n: close.rolling(n).max().to_numpy() for n in [10, 20, 40, 60, 90]}
lows = {n: close.rolling(n).min().to_numpy() for n in [10, 20, 40, 60, 90]}
step = np.diff(values) / values[:-1]
EXP3 = np.empty((len(COMBOS), len(step)))
for k, (fast, slow, high, low) in enumerate(COMBOS):
    state = np.full(len(values), np.nan)
    state[values >= highs[high]] = 1.0                 # 创新高就进
    state[values <= lows[low]] = 0.0                   # 破新低就走
    holding = pd.Series(state).ffill().fillna(0.0).to_numpy() * (averages[fast] > averages[slow])
    holding[np.isnan(averages[fast]) | np.isnan(averages[slow])] = 0.0
    EXP3[k] = holding[:-1] * step
LOW3, HIGH3 = "2018-01-01", "2021-12-31"
inside3, outside3 = window(LOW3, HIGH3), window("2022-01-01")
train3, test3 = EXP3[:, inside3], EXP3[:, outside3]
scores3, annual3, out3 = sharpe(train3), annual(train3), annual(test3)
pick = int(scores3.argmax())
curve3 = RP.to_curve(pd.Series(train3[pick], index=close.index[1:][inside3]))
print("「我做了一条 BTC 日线策略：快线在慢线上方（大方向向上）、收盘创 N 日新高就买入，")
print(f"  跌破 M 日新低就卖出。四个参数在 {len(COMBOS):,} 组里选夏普最高的一组 = {COMBOS[pick]}，")
print(f"  {LOW3} 到 {HIGH3} 的成绩是这样，你看有没有问题？」")
print(RP.metrics(curve3, 365).drop(["起", "止"]).round(4).to_string())
print(f"\n同期买入持有：年化 {RP.annual_return(close.loc[LOW3:HIGH3], 365):.2%}、"
      f"夏普 {RP.sharpe(close.loc[LOW3:HIGH3].pct_change().dropna(), 365):.4f}、"
      f"最大回撤 {RP.max_drawdown(close.loc[LOW3:HIGH3]):.2%}")

「我做了一条 BTC 日线策略：快线在慢线上方（大方向向上）、收盘创 N 日新高就买入，
  跌破 M 日新低就卖出。四个参数在 1,575 组里选夏普最高的一组 = (30, 120, 40, 20)，
  2018-01-01 到 2021-12-31 的成绩是这样，你看有没有问题？」
根数        1462.0
累计收益      9.5645
年化收益      0.8021
年化波动       0.399
夏普比率      1.6744
索提诺比率     2.7632
最大回撤     -0.2655
卡玛比率      3.0211
在水下的比例    0.6436
最长水下根数     400.0
最好的一根      0.172
最差的一根    -0.1347

同期买入持有：年化 36.33%、夏普 0.7963、最大回撤 -81.18%

holding3 = (train3[pick] != 0).astype(int)
edges = np.diff(np.concatenate([[0], holding3]))
opens, closes = np.where(edges == 1)[0], np.where(edges == -1)[0]
if len(closes) < len(opens):
    closes = np.append(closes, len(holding3))
path = curve3.to_numpy()
trades = pd.Series([path[b] / path[a] - 1 for a, b in zip(opens, closes)])
slice3 = pd.DataFrame([{"快线": f, "慢线": s, "夏普": scores3[k]} for k, (f, s, h, l) in enumerate(COMBOS)
                       if (h, l) == (COMBOS[pick][2], COMBOS[pick][3])])
face3 = V.surface(slice3, "快线", "慢线", "夏普")
checks = pd.DataFrame([
    {"查什么": "第 27 篇：最大回撤是不是 0", "结果": f"{RP.max_drawdown(curve3):.2%}", "判定": "过"},
    {"查什么": "第 27 篇：持仓日上涨的比例",
     "结果": f"{float((train3[pick][train3[pick] != 0] > 0).mean()):.2%}", "判定": "过"},
    {"查什么": "① 参数曲面的落差",
     "结果": f"{V.neighbourhood(face3, COMBOS[pick][:2])['落差']:.4f}（邻居中位 "
             f"{V.neighbourhood(face3, COMBOS[pick][:2])['邻居中位']:.4f}）", "判定": "过（是平原）"},
    {"查什么": "② 多重检验：白捡的门槛",
     "结果": f"夏普 {scores3[pick]:.4f} vs 门槛 "
             f"{V.expected_max_sharpe(len(COMBOS), scores3.std()):.4f}，"
             f"DSR {V.deflated_sharpe(scores3[pick], len(COMBOS), train3.shape[1], scores3.std(), periods_per_year=365)['打过折的夏普 DSR']:.4f}",
     "判定": "过（勉强）"},
    {"查什么": "③ 回测过拟合概率 PBO",
     "结果": f"{V.pbo(train3, 16)['过拟合概率 PBO']:.4f}", "判定": "过"},
    {"查什么": "④ 整个网格的成绩",
     "结果": f"{len(COMBOS):,} 组里夏普大于 1 的占 {(scores3 > 1).mean():.1%}，亏钱的占 {(annual3 < 0).mean():.1%}",
     "判定": "不过"},
    {"查什么": "⑤ 第 29 篇：逐笔的 t 值",
     "结果": f"{len(trades)} 笔，平均每笔 {trades.mean():.2%}，t 值 "
             f"{RP.trade_metrics(trades)['t 值']:.4f}，95% 区间下界 {RP.trade_metrics(trades)['95% 下界']:.2%}",
     "判定": "不过"},
    {"查什么": "⑥ 样本外（2022-01 到 2026-08）",
     "结果": f"年化 {out3[pick]:.2%}，排 {int((out3 > out3[pick]).sum()) + 1}/{len(COMBOS):,}，"
             f"网格中位 {np.median(out3):.2%}，买入持有 {RP.annual_return(close.loc['2022-01-01':], 365):.2%}",
     "判定": "不过"},
])
print(checks.to_string(index=False))

                     查什么                                            结果     判定
        第 27 篇：最大回撤是不是 0                                       -26.55%      过
         第 27 篇：持仓日上涨的比例                                        57.27%      过
               ① 参数曲面的落差                           0.1732（邻居中位 1.5012） 过（是平原）
            ② 多重检验：白捡的门槛             夏普 1.6744 vs 门槛 0.7893，DSR 0.9614  过（勉强）
           ③ 回测过拟合概率 PBO                                        0.4408      过
               ④ 整个网格的成绩             1,575 组里夏普大于 1 的占 51.4%，亏钱的占 0.1%     不过
        ⑤ 第 29 篇：逐笔的 t 值     9 笔，平均每笔 37.68%，t 值 1.9902，95% 区间下界 0.57%     不过
⑥ 样本外（2022-01 到 2026-08） 年化 8.89%，排 1129/1,575，网格中位 12.83%，买入持有 11.28%     不过

截止日挪一挪，同一组参数：
  截到 2020-12-31：年化  94.60%，夏普 1.9921
  截到 2021-06-30：年化  90.00%，夏普 1.7884
  截到 2021-12-31：年化  80.21%，夏普 1.6744
  截到 2022-06-30：年化  68.00%，夏普 1.5637
  截到 2022-12-31：年化  59.44%，夏普 1.4824

"""talab.validate：过拟合与样本外检验。第 30 篇。

第 27 篇保证引擎对、第 28 篇保证数据干净、第 29 篇保证读数不被自己骗。
这个模块回答最后一个也是最难的问题：**这条策略是学会了，还是把答案背下来了。**

难在哪里：**样本内的成绩永远可以做到任意好。**参数多试几组、规则多加几条，
回测收益一定上升——上升本身不携带任何信息。所以这里的每一个函数都不是在「算成绩」，
而是在**给成绩打折**：

1. **参数曲面**（`surface`、`neighbourhood`、`plateaus`）：这一格好，是因为它周围一片都好，
   还是因为它自己恰好踩中了？后者叫尖峰，它在样本外几乎必然塌。
2. **切分**（`split_index`、`walk_forward`、`walk_forward_run`）：拿一段没看过的数据当考卷。
3. **多重检验**（`expected_max_sharpe`、`deflated_sharpe`）：你试了一万次，
   「最好的那次」本来就该很好看——先把这个白捡的部分减掉。
4. **回测过拟合概率**（`pbo`）：把时间切成块，反复交换样本内外，
   数一数「样本内第一名在样本外落到中位数以下」的比例。
"""
from __future__ import annotations

import itertools
import math
from statistics import NormalDist

import numpy as np
import pandas as pd

NORMAL = NormalDist()
EULER = 0.5772156649015329                       # 欧拉常数，`expected_max_sharpe` 要用


# ---------------------------------------------------------------------------
# 一、参数曲面：平原还是尖峰
# ---------------------------------------------------------------------------

def surface(table: pd.DataFrame, index: str, columns: str, values: str) -> pd.DataFrame:
    """把「一行一组参数」的长表摊成二维曲面，行列都是参数值。

    摊成二维之后才能问「这一格的邻居怎么样」——而那正是这一篇的核心问题。
    """
    return table.pivot(index=index, columns=columns, values=values).sort_index().sort_index(axis=1)


def neighbourhood(grid: pd.DataFrame, cell: tuple, radius: int = 1) -> pd.Series:
    """一格和它周围一圈的成绩：本格、邻居的中位数 / 最差 / 最好，以及落差。

    `落差` = 本格 − 邻居中位数。它是这一篇最省事的过拟合警报：
    **落差越大，说明这一格越像是踩中的，而不是踩在一片实地上。**

    ⚠️ 邻居按**网格上的位置**算，不按参数值相差多少。参数网格常常不是等距的
    （5、10、20、40、60 这种），这时候「走错一步」的正确含义是「换成表里相邻的那个值」，
    不是「加一减一」。
    """
    row, column = cell
    i, j = grid.index.get_loc(row), grid.columns.get_loc(column)
    block = grid.iloc[max(i - radius, 0):i + radius + 1, max(j - radius, 0):j + radius + 1]
    here = float(grid.iloc[i, j])
    others = block.to_numpy(float).ravel()
    others = np.delete(others, list(block.index).index(row) * block.shape[1]
                       + list(block.columns).index(column))
    others = others[~np.isnan(others)]
    if not len(others):
        raise ValueError("这一格周围没有有效的邻居")
    return pd.Series({"这一格": here, "邻居数": float(len(others)),
                      "邻居中位": float(np.median(others)), "邻居最差": float(others.min()),
                      "邻居最好": float(others.max()), "落差": here - float(np.median(others))})


def plateaus(grid: pd.DataFrame, radius: int = 1) -> pd.DataFrame:
    """给每一格配上它邻域的中位数，按**邻域中位数**从高到低排序。

    这是「不挑尖峰」的选参数方法：不要问「哪一格最高」，要问
    **「哪一片地最高」**——站在一片高地上，走错一步还在高地上。
    """
    rows = []
    for row in grid.index:
        for column in grid.columns:
            if np.isnan(grid.loc[row, column]):
                continue
            stats = neighbourhood(grid, (row, column), radius)
            rows.append({grid.index.name or "行": row, grid.columns.name or "列": column,
                         "这一格": stats["这一格"], "邻域中位": stats["邻居中位"],
                         "落差": stats["落差"]})
    return pd.DataFrame(rows).sort_values("邻域中位", ascending=False).reset_index(drop=True)


# ---------------------------------------------------------------------------
# 二、切分：样本内、样本外、walk-forward
# ---------------------------------------------------------------------------

def split_index(n: int, fraction: float = 0.5) -> tuple[slice, slice]:
    """把 n 根 K 线按时间顺序切成前后两段：前面选参数，后面当考卷。

    ⚠️ 只能按**时间顺序**切。随机抽一半当样本外是错的——
    相邻两天的行情高度相关，随机抽等于让考卷和复习资料互相抄。
    """
    if not 0 < fraction < 1:
        raise ValueError("fraction 要在 0 和 1 之间")
    cut = int(n * fraction)
    if cut < 1 or cut >= n:
        raise ValueError("切出来的两段都要至少有一根")
    return slice(0, cut), slice(cut, n)


def walk_forward(n: int, train: int, test: int, anchored: bool = False) -> list[tuple[slice, slice]]:
    """滚动检验的切分：每一段用前 `train` 根挑参数，紧接着的 `test` 根拿来考试。

    - `anchored=False`（默认）：训练窗口长度固定，整段往前滚
    - `anchored=True`：训练窗口从第一根开始，越考越长（更像真实的积累过程）

    各段的考卷首尾相接、互不重叠，把它们拼起来就是一条**每一根都没被看过**的资金曲线。
    """
    if train < 1 or test < 1:
        raise ValueError("train 和 test 都要至少是 1")
    out, start = [], 0
    while start + train + test <= n:
        out.append((slice(0 if anchored else start, start + train),
                    slice(start + train, start + train + test)))
        start += test
    if not out:
        raise ValueError(f"{n} 根放不下一次 {train}+{test} 的切分")
    return out


def walk_forward_run(returns: np.ndarray, splits, names=None, min_bars: int = 2) -> pd.DataFrame:
    """在每一段上选出训练期里夏普最高的那一列，记下它在考卷上的表现。

    `returns` 是 `(试验数, 根数)` 的每根收益矩阵——一行一组参数。
    返回一张表：每一段选了谁、训练期分数、考卷分数，以及**这一段的考卷收益**。
    """
    rows = []
    for k, (train, test) in enumerate(splits):
        inside, outside = returns[:, train], returns[:, test]
        if inside.shape[1] < min_bars or outside.shape[1] < min_bars:
            raise ValueError("某一段太短，算不出标准差")
        score = inside.mean(axis=1) / np.where(inside.std(axis=1) > 0, inside.std(axis=1), np.nan)
        pick = int(np.nanargmax(score))
        rows.append({"第几段": k + 1, "选了谁": names[pick] if names is not None else pick,
                     "训练期夏普": float(score[pick]),
                     "考卷夏普": float(outside[pick].mean() / outside[pick].std()),
                     "考卷收益": float(np.prod(1 + outside[pick]) - 1),
                     "训练根数": inside.shape[1], "考卷根数": outside.shape[1]})
    return pd.DataFrame(rows)


def stitch(returns: np.ndarray, splits, picks) -> np.ndarray:
    """把各段考卷上被选中那一列的收益首尾接起来，得到一条完整的样本外收益序列。"""
    return np.concatenate([returns[int(p), test] for (_, test), p in zip(splits, picks)])


# ---------------------------------------------------------------------------
# 三、多重检验：你试了多少次
# ---------------------------------------------------------------------------

def expected_max_sharpe(n_trials: int, sharpe_std: float) -> float:
    """**完全没有本事**的情况下，试 `n_trials` 次，「最好的那次」夏普期望是多少
    （Bailey & López de Prado 2014）：

    > E[最大值] ≈ 标准差 × [(1 − γ)·Φ⁻¹(1 − 1/N) + γ·Φ⁻¹(1 − 1/(N·e))]

    γ 是欧拉常数 0.5772。`sharpe_std` 是这一万次试验的夏普比率**彼此之间**的标准差。
    直觉：N 越大，最大值越往右跑——**「我试了一万组，最好的那组夏普 1.5」这句话里，
    「一万组」和「1.5」一样重要。**
    """
    if n_trials < 2:
        raise ValueError("至少要有两次试验")
    high = NORMAL.inv_cdf(1 - 1 / n_trials)
    low = NORMAL.inv_cdf(1 - 1 / (n_trials * math.e))
    return sharpe_std * ((1 - EULER) * high + EULER * low)


def deflated_sharpe(sharpe: float, n_trials: int, n_obs: int, sharpe_std: float,
                    skew: float = 0.0, kurtosis: float = 3.0,
                    periods_per_year: int = 252) -> pd.Series:
    """打过折的夏普比率（DSR）：把「试了很多次」和「收益率不是正态」都扣掉之后，
    这条策略的夏普**真的大于 0** 的概率。

    两步：先用 `expected_max_sharpe` 算出「没本事时白捡的那一截」当作门槛，
    再问「观察到的夏普超过这个门槛」有多显著：

    > DSR = Φ( (夏普 − 门槛) × √(根数 − 1) ÷ √(1 − 偏度×夏普 + (峰度−1)/4 × 夏普²) )

    式子里的夏普都是**单根**的（不年化）。`kurtosis` 传的是原始峰度（正态是 3），
    不是超额峰度。DSR 低于 0.95 就该当成「还没排除运气」。
    """
    per_period = sharpe / math.sqrt(periods_per_year)
    threshold = expected_max_sharpe(n_trials, sharpe_std / math.sqrt(periods_per_year))
    spread = math.sqrt(max(1 - skew * per_period + (kurtosis - 1) / 4 * per_period ** 2, 1e-12))
    z = (per_period - threshold) * math.sqrt(max(n_obs - 1, 1)) / spread
    return pd.Series({"年化夏普": sharpe, "试验次数": float(n_trials),
                      "白捡的门槛（年化）": threshold * math.sqrt(periods_per_year),
                      "超过门槛多少": sharpe - threshold * math.sqrt(periods_per_year),
                      "t 值": z, "打过折的夏普 DSR": NORMAL.cdf(z)})


# ---------------------------------------------------------------------------
# 四、回测过拟合概率（CSCV）
# ---------------------------------------------------------------------------

def pbo(returns: np.ndarray, n_chunks: int = 8) -> pd.Series:
    """回测过拟合概率（Bailey 等 2017 的 CSCV）。

    做法：把时间切成 `n_chunks` 块，穷举所有「一半当样本内、另一半当样本外」的分法；
    每一种分法里选出样本内夏普最高的那一列，看它**在样本外排第几**。
    如果这个第一名在样本外经常掉到中位数以下，说明「挑第一名」这个动作本身就是在挑噪声。

    > PBO = 样本内第一名在样本外掉到中位数以下的分法比例

    ⚠️ 它衡量的是**挑选流程**，不是某一组参数。PBO 高不代表这些参数都没用，
    代表「用样本内最优去选参数」这件事在这批数据上不管用。
    n_chunks 要是偶数；16 块有 12,870 种分法，8 块只有 70 种，慢和稳之间自己选。
    """
    if n_chunks % 2 or n_chunks < 4:
        raise ValueError("n_chunks 要是不小于 4 的偶数")
    n_trials, n_bars = returns.shape
    if n_trials < 2:
        raise ValueError("至少要有两列试验")
    edges = np.linspace(0, n_bars, n_chunks + 1).astype(int)
    counts = np.diff(edges).astype(float)
    total = np.array([returns[:, edges[i]:edges[i + 1]].sum(axis=1) for i in range(n_chunks)])
    square = np.array([(returns[:, edges[i]:edges[i + 1]] ** 2).sum(axis=1) for i in range(n_chunks)])

    def score(keys) -> np.ndarray:                 # 任意几块合起来的夏普，O(试验数)
        n = counts[list(keys)].sum()
        mean = total[list(keys)].sum(axis=0) / n
        var = square[list(keys)].sum(axis=0) / n - mean ** 2
        return mean / np.sqrt(np.maximum(var, 1e-300))

    logits, ranks = [], []
    everything = range(n_chunks)
    for inside in itertools.combinations(everything, n_chunks // 2):
        outside = tuple(k for k in everything if k not in inside)
        best = int(np.argmax(score(inside)))
        outside_score = score(outside)
        rank = int((outside_score < outside_score[best]).sum()) + 1      # 1 最差，n_trials 最好
        omega = rank / (n_trials + 1)
        ranks.append(omega)
        logits.append(math.log(omega / (1 - omega)))
    logits = np.array(logits)
    return pd.Series({"试验次数": float(n_trials), "切成几块": float(n_chunks),
                      "一共几种分法": float(len(logits)),
                      "样本外分位中位数": float(np.median(ranks)),
                      "过拟合概率 PBO": float((logits <= 0).mean()),
                      "logit 中位数": float(np.median(logits))})


# ---------------------------------------------------------------------------
# 五、蒙特卡洛：如果这个市场本来就没有规律
# ---------------------------------------------------------------------------

def synthetic_close(close: pd.Series, n: int = 100, block: int = 1, seed: int = 0) -> np.ndarray:
    """造 n 条「和原序列同分布、但没有原来那段走势」的假价格，返回 `(n, 根数)`。

    `block=1` 是把日收益率完全打散（波动率聚集也一起没了）；
    `block > 1` 用块自助法，保留每一小段内部的顺序，**波动率聚集还在，趋势没了**。
    拿这些假价格重跑一遍同样的参数扫描，就知道「最好的一组」里有多少是白捡的。
    """
    values = close.pct_change().dropna().to_numpy(float)
    m = len(values)
    if block < 1 or block > m:
        raise ValueError("block 要在 1 和样本长度之间")
    rng = np.random.default_rng(seed)
    out = np.empty((n, m + 1))
    for i in range(n):
        if block == 1:
            sample = rng.choice(values, size=m, replace=True)
        else:
            starts = rng.integers(0, m - block + 1, size=int(np.ceil(m / block)))
            sample = np.concatenate([values[s:s + block] for s in starts])[:m]
        out[i] = float(close.iloc[0]) * np.concatenate([[1.0], np.cumprod(1 + sample)])
    return out

"""talab.validate 的测试（第 30 篇）。"""
import math

import numpy as np
import pandas as pd
import pytest

from talab import validate as V


def make_surface(values) -> pd.DataFrame:
    frame = pd.DataFrame(values, index=[10, 20, 30], columns=[100, 200, 300])
    frame.index.name, frame.columns.name = "快线", "慢线"
    return frame


def test_surface_spreads_a_long_table_into_two_dimensions():
    table = pd.DataFrame({"快线": [20, 10, 20, 10], "慢线": [200, 200, 100, 100],
                          "年化": [0.4, 0.3, 0.2, 0.1]})
    grid = V.surface(table, "快线", "慢线", "年化")
    assert list(grid.index) == [10, 20] and list(grid.columns) == [100, 200]
    assert grid.loc[20, 200] == 0.4 and grid.loc[10, 100] == 0.1


def test_neighbourhood_is_the_ring_around_one_cell():
    grid = make_surface([[0.1, 0.2, 0.3], [0.4, 1.0, 0.5], [0.6, 0.7, 0.8]])
    out = V.neighbourhood(grid, (20, 200))
    assert out["这一格"] == 1.0 and out["邻居数"] == 8
    assert out["邻居中位"] == pytest.approx(0.45)          # 0.1…0.8 去掉中心的中位数
    assert out["邻居最差"] == pytest.approx(0.1)
    assert out["邻居最好"] == pytest.approx(0.8)
    assert out["落差"] == pytest.approx(0.55)
    corner = V.neighbourhood(grid, (10, 100))              # 角上只有三个邻居
    assert corner["邻居数"] == 3


def test_neighbourhood_counts_steps_on_the_grid_not_parameter_distance():
    """参数网格常常不等距（5、10、20、40、60），「走错一步」指的是换成表里相邻的那个值。"""
    frame = pd.DataFrame([[0.0, 0.0, 0.0], [0.0, 1.0, 0.2], [0.0, 0.0, 0.0]],
                         index=[5, 10, 200], columns=[20, 40, 1000])
    out = V.neighbourhood(frame, (10, 40))
    assert out["邻居数"] == 8                               # 值差 960 的那一列也算邻居
    assert out["邻居最好"] == pytest.approx(0.2)


def test_plateaus_ranks_by_the_neighbourhood_not_by_the_cell():
    """一根针（自己高、周围低）要排在一片高地（自己一般、周围都不错）后面。"""
    grid = make_surface([[0.5, 0.5, 0.0], [0.5, 0.5, 0.0], [0.0, 0.0, 9.0]])
    ranked = V.plateaus(grid)
    top = ranked.iloc[0]
    assert (top["快线"], top["慢线"]) != (30, 300)          # 9.0 那根针不是第一
    needle = ranked[(ranked["快线"] == 30) & (ranked["慢线"] == 300)].iloc[0]
    assert needle["这一格"] == 9.0 and needle["落差"] == pytest.approx(9.0)
    assert ranked["邻域中位"].is_monotonic_decreasing


def test_split_index_cuts_in_time_order():
    inside, outside = V.split_index(100, 0.7)
    assert (inside.start, inside.stop) == (0, 70)
    assert (outside.start, outside.stop) == (70, 100)
    with pytest.raises(ValueError):
        V.split_index(100, 1.0)
    with pytest.raises(ValueError):
        V.split_index(3, 0.01)


def test_walk_forward_windows_do_not_overlap_and_cover_in_order():
    splits = V.walk_forward(1000, train=300, test=100)
    assert len(splits) == 7
    tests = [test for _, test in splits]
    assert tests[0].start == 300 and tests[-1].stop == 1000
    for a, b in zip(tests, tests[1:]):
        assert a.stop == b.start                            # 考卷首尾相接、不重叠
    for train, test in splits:
        assert train.stop == test.start and train.stop - train.start == 300
    with pytest.raises(ValueError):
        V.walk_forward(100, train=300, test=100)


def test_walk_forward_anchored_keeps_growing_the_training_window():
    rolling = V.walk_forward(1000, 300, 100, anchored=False)
    anchored = V.walk_forward(1000, 300, 100, anchored=True)
    assert [t for _, t in rolling] == [t for _, t in anchored]        # 考卷完全一样
    assert all(train.start == 0 for train, _ in anchored)
    lengths = [train.stop - train.start for train, _ in anchored]
    assert lengths == sorted(lengths) and lengths[0] == 300 < lengths[-1]


def test_walk_forward_run_picks_the_best_of_the_training_window():
    rng = np.random.default_rng(0)
    returns = rng.normal(0, 0.01, (4, 400))
    returns[1, :200] += 0.05                                # 第 1 列只在前半段好
    returns[2, 200:] += 0.05                                # 第 2 列只在后半段好
    splits = V.walk_forward(400, 200, 200)
    out = V.walk_forward_run(returns, splits, names=list("甲乙丙丁"))
    assert len(out) == 1 and out.loc[0, "选了谁"] == "乙"      # 训练期挑的是乙
    assert out.loc[0, "训练期夏普"] > out.loc[0, "考卷夏普"]    # 考卷上乙已经不行了


def test_stitch_joins_the_test_windows_end_to_end():
    returns = np.arange(30, dtype=float).reshape(3, 10)
    splits = V.walk_forward(10, 4, 3)
    joined = V.stitch(returns, splits, [0, 2])
    assert len(joined) == sum(test.stop - test.start for _, test in splits)
    assert joined[0] == returns[0, splits[0][1].start]
    assert joined[-1] == returns[2, splits[1][1].stop - 1]


def test_expected_max_sharpe_matches_a_simulation():
    """这是整个模块最该钉住的一条：公式算出来的「白捡的最大值」要对得上真的抽一万次。"""
    rng = np.random.default_rng(30)
    for n_trials in (50, 1000, 10_000):
        draws = rng.normal(0.0, 0.3, size=(400, n_trials)).max(axis=1)
        assert V.expected_max_sharpe(n_trials, 0.3) == pytest.approx(draws.mean(), rel=0.05)


def test_expected_max_sharpe_grows_with_trials_and_scales_with_spread():
    assert V.expected_max_sharpe(10, 1.0) < V.expected_max_sharpe(1000, 1.0)
    assert V.expected_max_sharpe(1000, 2.0) == pytest.approx(2 * V.expected_max_sharpe(1000, 1.0))
    with pytest.raises(ValueError):
        V.expected_max_sharpe(1, 1.0)


def test_deflated_sharpe_falls_as_you_admit_more_trials():
    same = dict(sharpe=1.5, n_obs=1000, sharpe_std=0.4, periods_per_year=252)
    few = V.deflated_sharpe(n_trials=2, **same)
    many = V.deflated_sharpe(n_trials=10_000, **same)
    assert few["年化夏普"] == many["年化夏普"] == 1.5          # 夏普一个字没变
    assert few["白捡的门槛（年化）"] < many["白捡的门槛（年化）"]
    assert few["打过折的夏普 DSR"] > many["打过折的夏普 DSR"]
    assert 0 <= many["打过折的夏普 DSR"] <= 1


def test_pbo_is_a_coin_flip_on_pure_noise():
    """一堆互相没区别的随机策略，样本内第一名在样本外就是随机的——PBO 该在 0.5 附近。"""
    returns = np.random.default_rng(1).normal(0.0, 0.01, (100, 800))
    out = V.pbo(returns, 8)
    assert out["一共几种分法"] == 70
    assert out["过拟合概率 PBO"] == pytest.approx(0.5, abs=0.2)


def test_pbo_is_near_zero_when_one_strategy_is_genuinely_better():
    returns = np.random.default_rng(2).normal(0.0, 0.01, (40, 800))
    returns[7] += 0.01                                      # 第 7 列全程真的更好
    out = V.pbo(returns, 8)
    assert out["过拟合概率 PBO"] < 0.05
    assert out["样本外分位中位数"] > 0.9
    with pytest.raises(ValueError):
        V.pbo(returns, 7)                                   # 块数要是偶数


def test_synthetic_close_keeps_the_length_and_the_starting_price():
    close = pd.Series(np.cumprod(1 + np.random.default_rng(3).normal(0, 0.02, 500)) * 100,
                      index=pd.date_range("2020-01-01", periods=500, freq="D"))
    paths = V.synthetic_close(close, n=5, block=20, seed=3)
    assert paths.shape == (5, len(close))
    assert np.allclose(paths[:, 0], float(close.iloc[0]))
    assert not np.allclose(paths[0], close.to_numpy())       # 是新造的，不是原样抄回来
    fast = V.synthetic_close(close, n=3, block=1, seed=3)
    assert fast.shape == (3, len(close))
    with pytest.raises(ValueError):
        V.synthetic_close(close, block=10_000)

308 passed in 1.07s
276 passed, 32 skipped in 1.15s