第 15 篇：波动率：ATR 与布林带

c = day["close"]
band = I.bollinger(c)
year_low = band["bandwidth"].rolling(365).min()
t = pd.Timestamp("2023-08-14", tz="UTC")
table = pd.concat([day[["high", "low", "close"]], I.atr(day["high"], day["low"], c).rename("atr"),
                   I.natr(day["high"], day["low"], c).rename("natr"), band[["upper", "lower", "bandwidth"]],
                   year_low.rename("365 天最低带宽")], axis=1)
print(table.loc["2023-08-01":"2023-08-14"].round({"high": 0, "low": 0, "close": 0, "atr": 0, "natr": 2, "upper": 0, "lower": 0,
                                                  "bandwidth": 4, "365 天最低带宽": 4}).to_string())
natr = I.natr(day["high"], day["low"], c)
before = natr.loc[:"2023-08-11"]
print(f"8 月 14 日 NATR {natr[t]:.2f}%，上一次不高于它是 {before[before <= natr[t]].index[-1].date()}；"
      f"带宽 {band['bandwidth'][t]:.4f}，8 月 11 日之前 365 天里最窄的是 {band['bandwidth'].loc[:'2023-08-11'].iloc[-365:].idxmin().date()}"
      f"（{band['bandwidth'].loc[:'2023-08-11'].iloc[-365:].min():.4f}）；2017 年 8 月以来最窄的是 "
      f"{band['bandwidth'].loc[:'2023-08-11'].idxmin().date()}（{band['bandwidth'].loc[:'2023-08-11'].min():.4f}）")
print(f"过去 20 天收盘价最高 {c.loc[:t].iloc[-20:].max():,.2f}，最低 {c.loc[:t].iloc[-20:].min():,.2f}，"
      f"相差 {c.loc[:t].iloc[-20:].max() / c.loc[:t].iloc[-20:].min() - 1:.2%}")

                              high      low    close    atr  natr    upper    lower  bandwidth  365 天最低带宽
time                                                                                                     
2023-08-01 00:00:00+00:00  29739.0  28586.0  29706.0  663.0  2.23  30871.0  28694.0     0.0731     0.0305
2023-08-02 00:00:00+00:00  30048.0  28928.0  29186.0  695.0  2.38  30472.0  28865.0     0.0541     0.0305
2023-08-03 00:00:00+00:00  29433.0  28968.0  29194.0  679.0  2.33  30384.0  28841.0     0.0521     0.0305
2023-08-04 00:00:00+00:00  29333.0  28808.0  29114.0  668.0  2.29  30288.0  28820.0     0.0497     0.0305
2023-08-05 00:00:00+00:00  29152.0  28979.0  29072.0  633.0  2.18  30189.0  28803.0     0.0470     0.0305
2023-08-06 00:00:00+00:00  29205.0  28992.0  29088.0  603.0  2.07  30092.0  28795.0     0.0440     0.0305
2023-08-07 00:00:00+00:00  29277.0  28701.0  29211.0  601.0  2.06  30037.0  28785.0     0.0426     0.0305
2023-08-08 00:00:00+00:00  30244.0  29146.0  29770.0  636.0  2.14  30011.0  28797.0     0.0413     0.0305
2023-08-09 00:00:00+00:00  30160.0  29377.0  29582.0  647.0  2.19  29978.0  28808.0     0.0398     0.0305
2023-08-10 00:00:00+00:00  29738.0  29320.0  29456.0  630.0  2.14  29909.0  28833.0     0.0366     0.0305
2023-08-11 00:00:00+00:00  29565.0  29252.0  29426.0  608.0  2.07  29856.0  28849.0     0.0343     0.0305
2023-08-12 00:00:00+00:00  29481.0  29382.0  29430.0  571.0  1.94  29698.0  28942.0     0.0258     0.0258
2023-08-13 00:00:00+00:00  29475.0  29272.0  29304.0  545.0  1.86  29699.0  28954.0     0.0254     0.0254
2023-08-14 00:00:00+00:00  29695.0  29102.0  29431.0  548.0  1.86  29709.0  28964.0     0.0254     0.0254
8 月 14 日 NATR 1.86%，上一次不高于它是 2023-01-11；带宽 0.0254，8 月 11 日之前 365 天里最窄的是 2023-01-04（0.0305）；2017 年 8 月以来最窄的是 2023-01-04（0.0305）
过去 20 天收盘价最高 29,770.42，最低 29,072.13，相差 2.40%

small = pd.DataFrame({"open": [100.0, 104, 97, 99], "high": [102.0, 106, 98, 101], "low": [99.0, 103, 94, 98],
                      "close": [101.0, 105, 95, 100]})
small["最高 - 最低"] = small["high"] - small["low"]
small["|最高 - 昨收|"] = (small["high"] - small["close"].shift(1)).abs()
small["|最低 - 昨收|"] = (small["low"] - small["close"].shift(1)).abs()
small["真实波幅"] = X.true_range(small["high"], small["low"], small["close"])
small["ATR(2)"] = I.atr(small["high"], small["low"], small["close"], 2)
print(small.to_string())

    open   high    low  close  最高 - 最低  |最高 - 昨收|  |最低 - 昨收|  真实波幅  ATR(2)
0  100.0  102.0   99.0  101.0      3.0        NaN        NaN   NaN     NaN
1  104.0  106.0  103.0  105.0      3.0        5.0        2.0   5.0     NaN
2   97.0   98.0   94.0   95.0      4.0        7.0       11.0  11.0     8.0
3   99.0  101.0   98.0  100.0      3.0        6.0        3.0   6.0     7.0

for name, df in markets.items():
    tr = X.true_range(df["high"], df["low"], df["close"])
    gap = (tr - (df["high"] - df["low"]))[tr.notna()]
    share = gap / tr[tr.notna()]
    print(f"{name}：真实波幅大于「最高 - 最低」的日子占 {(gap > 1e-9).mean():.1%}，跳空部分占全部真实波幅的 {gap.sum() / tr.sum():.1%}；"
          f"跳空占比最大的三天 " + "、".join(f"{d.date()}（{v:.0%}）" for d, v in share.nlargest(3).items()))

SPY：真实波幅大于「最高 - 最低」的日子占 38.7%，跳空部分占全部真实波幅的 11.0%；跳空占比最大的三天 2017-12-18（70%）、2026-04-08（69%）、2021-06-24（68%）
AAPL：真实波幅大于「最高 - 最低」的日子占 34.5%，跳空部分占全部真实波幅的 9.4%；跳空占比最大的三天 2019-01-03（77%）、2025-04-03（72%）、2024-05-03（69%）
BTC：真实波幅大于「最高 - 最低」的日子占 0.4%，跳空部分占全部真实波幅的 0.0%；跳空占比最大的三天 2017-10-08（1%）、2018-02-09（1%）、2019-06-21（0%）

print(f"没有跳空的连续随机游走：一天的 最高 - 最低 平均是标准差的 2√(2/π) = {2 * np.sqrt(2 / np.pi):.3f} 倍")
for name, df in markets.items():
    close = df["close"]
    sd = np.log(close).diff().rolling(20).std()
    hl = ((df["high"] - df["low"]) / close).rolling(20).mean()
    ratio = (I.natr(df["high"], df["low"], close) / 100 / sd).dropna()
    print(f"{name}：NATR ÷ 20 天收益率标准差 中位数 {ratio.median():.2f}（25% ~ 75%：{ratio.quantile(0.25):.2f} ~ {ratio.quantile(0.75):.2f}）；"
          f"不含跳空的 (最高 - 最低)/收盘 ÷ 标准差 中位数 {(hl / sd).median():.2f}")

没有跳空的连续随机游走：一天的 最高 - 最低 平均是标准差的 2√(2/π) = 1.596 倍
SPY：NATR ÷ 20 天收益率标准差 中位数 1.40（25% ~ 75%：1.25 ~ 1.55）；不含跳空的 (最高 - 最低)/收盘 ÷ 标准差 中位数 1.24
AAPL：NATR ÷ 20 天收益率标准差 中位数 1.41（25% ~ 75%：1.27 ~ 1.60）；不含跳空的 (最高 - 最低)/收盘 ÷ 标准差 中位数 1.29
BTC：NATR ÷ 20 天收益率标准差 中位数 1.59（25% ~ 75%：1.38 ~ 1.81）；不含跳空的 (最高 - 最低)/收盘 ÷ 标准差 中位数 1.58

x = pd.Series([10.0, 11, 12, 11, 13, 14])
small = I.bollinger(x, n=4, k=2)
small.insert(0, "收盘价", x)
small.insert(2, "总体标准差", x.rolling(4).std(ddof=0))
print(small.round(4).to_string())
wrong = c.rolling(20).mean() + 2 * c.rolling(20).std()
print(f"BTC 日线：用 pandas 默认的 rolling().std()（除以 n - 1），上轨最多高出 {(wrong - band['upper']).max():,.2f}，"
      f"带宽平均宽了 {((wrong - c.rolling(20).mean()) / (band['upper'] - band['middle'])).mean() - 1:.2%}")

    收盘价  middle   总体标准差    upper    lower  percent_b  bandwidth
0  10.0     NaN     NaN      NaN      NaN        NaN        NaN
1  11.0     NaN     NaN      NaN      NaN        NaN        NaN
2  12.0     NaN     NaN      NaN      NaN        NaN        NaN
3  11.0   11.00  0.7071  12.4142   9.5858     0.5000     0.2571
4  13.0   11.75  0.8292  13.4083  10.0917     0.8769     0.2823
5  14.0   12.50  1.1180  14.7361  10.2639     0.8354     0.3578
BTC 日线：用 pandas 默认的 rolling().std()（除以 n - 1），上轨最多高出 495.66，带宽平均宽了 2.60%

for name, df in markets.items():
    close = df["close"]
    b = I.bollinger(close)
    ok = b["upper"].notna()
    identity = (b["percent_b"] - 0.5) * b["bandwidth"] - I.bias(close, b["middle"])
    print(f"{name}：收盘价在上轨上方 {(close > b['upper'])[ok].mean():.1%}，在下轨下方 {(close < b['lower'])[ok].mean():.1%}"
          f"（正态分布的说法是各 2.3%）；乖离率 = (%b - 0.5) × 带宽 的最大误差 {identity.abs().max():.1e}")

walk = pd.Series(100 * np.exp(np.cumsum(np.random.default_rng(5).normal(0, 0.01, 200_000))))
b = I.bollinger(walk)
print(f"正态随机游走（20 万步）：收盘价在上轨上方 {(walk > b['upper'])[b['upper'].notna()].mean():.1%}，"
      f"在下轨下方 {(walk < b['lower'])[b['upper'].notna()].mean():.1%}")

SPY：收盘价在上轨上方 6.1%，在下轨下方 4.9%（正态分布的说法是各 2.3%）；乖离率 = (%b - 0.5) × 带宽 的最大误差 1.6e-16
AAPL：收盘价在上轨上方 8.6%，在下轨下方 4.5%（正态分布的说法是各 2.3%）；乖离率 = (%b - 0.5) × 带宽 的最大误差 1.7e-16
BTC：收盘价在上轨上方 7.4%，在下轨下方 4.9%（正态分布的说法是各 2.3%）；乖离率 = (%b - 0.5) × 带宽 的最大误差 2.2e-16
正态随机游走（20 万步）：收盘价在上轨上方 6.2%，在下轨下方 5.5%

def shuffle_bars(df, rng):
    """打乱 K 线顺序：保留每根 K 线相对前一根收盘价的形状，重新拼成价格（第 9 篇）。"""
    rel = np.log(df[cols].div(df["close"].shift(1), axis=0))
    rel = rel.dropna().iloc[rng.permutation(len(df) - 1)].reset_index(drop=True)
    prev_close = df["close"].iloc[0] * np.exp(rel["close"].cumsum().shift(1, fill_value=0))
    out = np.exp(rel[cols]).mul(prev_close, axis=0)
    out.index = df.index[1:]
    return out


def future_mean(x, n=20):
    """之后 n 根（不含这一根）的平均值。"""
    return x[::-1].rolling(n).mean()[::-1].shift(-1)


def forecast_power(df):
    """这一根的 NATR 和之后 20 根的平均真实波幅（占前一根收盘价的比例），两者取对数后的相关系数。"""
    tr_pct = X.true_range(df["high"], df["low"], df["close"]) / df["close"].shift(1)
    now, later = np.log(I.natr(df["high"], df["low"], df["close"])), np.log(future_mean(tr_pct))
    ok = now.notna() & later.notna()
    return np.corrcoef(now[ok], later[ok])[0, 1]


datasets = [("SPY 日线", spy), ("AAPL 日线", aapl), ("BTC 日线", day), ("BTC 4 小时线", h4), ("BTC 1 小时线", h1)]
for name, df in datasets:
    sims = np.array([forecast_power(shuffle_bars(df, rng)) for _ in range(100)])
    print(f"{name}：现在的 NATR 和之后 20 根的平均真实波幅，相关系数 {forecast_power(df):.2f}；"
          f"打乱 K 线顺序 100 次：平均 {sims.mean():+.2f}，最大 {sims.max():+.2f}")

SPY 日线：现在的 NATR 和之后 20 根的平均真实波幅，相关系数 0.69；打乱 K 线顺序 100 次：平均 -0.01，最大 +0.14
AAPL 日线：现在的 NATR 和之后 20 根的平均真实波幅，相关系数 0.66；打乱 K 线顺序 100 次：平均 -0.02，最大 +0.18
BTC 日线：现在的 NATR 和之后 20 根的平均真实波幅，相关系数 0.67；打乱 K 线顺序 100 次：平均 -0.01，最大 +0.15
BTC 4 小时线：现在的 NATR 和之后 20 根的平均真实波幅，相关系数 0.79；打乱 K 线顺序 100 次：平均 +0.00，最大 +0.06
BTC 1 小时线：现在的 NATR 和之后 20 根的平均真实波幅，相关系数 0.82；打乱 K 线顺序 100 次：平均 +0.00，最大 +0.03

rows = []
for name, df in markets.items():
    close = df["close"]
    look = periods_per_year[name]
    n = I.natr(df["high"], df["low"], close)
    rank = n.rolling(look).rank(pct=True)                        # 在过去一年里排第几（只用到这一根为止）
    tr_pct = 100 * X.true_range(df["high"], df["low"], close) / close.shift(1)
    later = pd.DataFrame({"组": pd.cut(rank, [0, 0.2, 0.4, 0.6, 0.8, 1.0], labels=["最低 20%", "20-40%", "40-60%", "60-80%", "最高 20%"]),
                          "现在 NATR": n, "之后 20 根平均真实波幅": future_mean(tr_pct),
                          "之后 20 根收益": close.shift(-20) / close - 1}).dropna()
    later["之后 ÷ 现在"] = later["之后 20 根平均真实波幅"] / later["现在 NATR"]
    g = later.groupby("组", observed=True)
    rows.append(pd.DataFrame({"标的": name, "K 线": g.size(), "现在 NATR": g["现在 NATR"].mean(),
                              "之后 20 根平均真实波幅": g["之后 20 根平均真实波幅"].mean(),
                              "之后 ÷ 现在": g["之后 ÷ 现在"].mean(), "之后 20 根收益": g["之后 20 根收益"].mean(),
                              "上涨比例": g["之后 20 根收益"].apply(lambda v: (v > 0).mean())}).reset_index())
print("按「现在的 NATR 在过去一年里的排位」分成五组：")
print(pd.concat(rows).to_string(index=False, formatters={"现在 NATR": "{:.2f}".format, "之后 20 根平均真实波幅": "{:.2f}".format,
                                                          "之后 ÷ 现在": "{:.2f}".format, "之后 20 根收益": "{:+.2%}".format,
                                                          "上涨比例": "{:.1%}".format}))

按「现在的 NATR 在过去一年里的排位」分成五组：
     组   标的  K 线 现在 NATR 之后 20 根平均真实波幅 之后 ÷ 现在 之后 20 根收益  上涨比例
最低 20%  SPY  580    0.90          1.02    1.17    +1.07% 71.6%
20-40%  SPY  346    1.07          1.19    1.20    +0.34% 59.8%
40-60%  SPY  411    1.20          1.24    1.07    +0.81% 71.3%
60-80%  SPY  402    1.42          1.32    0.97    +1.87% 73.9%
最高 20%  SPY  488    2.10          1.98    1.01    +1.31% 61.7%
最低 20% AAPL  435    1.71          1.90    1.13    +2.56% 68.5%
20-40% AAPL  412    1.92          2.14    1.13    +2.87% 68.9%
40-60% AAPL  355    2.09          2.10    1.02    +3.54% 68.2%
60-80% AAPL  459    2.56          2.63    1.06    +2.05% 61.4%
最高 20% AAPL  567    3.17          2.93    0.96    +1.02% 55.7%
最低 20%  BTC  779    3.14          3.58    1.19    +1.40% 50.6%
20-40%  BTC  585    3.83          4.27    1.13    +3.02% 54.0%
40-60%  BTC  516    4.47          4.67    1.06    +2.04% 57.0%
60-80%  BTC  564    5.22          5.04    0.99    +3.37% 54.1%
最高 20%  BTC  460    7.11          5.81    0.84    +5.23% 59.1%

i = c.index.get_loc(t)
for n in [1, 2, 3, 5, 10, 20]:
    print(f"{n} 天后（{c.index[i + n].date()}）：收盘 {c.iloc[i + n]:,.2f}（{c.iloc[i + n] / c[t] - 1:+.2%}），"
          f"带宽 {band['bandwidth'].iloc[i + n]:.4f}，NATR {natr.iloc[i + n]:.2f}")
crash = pd.Timestamp("2023-08-17", tz="UTC")
row = day.loc[crash]
print(f"8 月 17 日：开盘 {row['open']:,.2f}，最低 {row['low']:,.2f}（{row['low'] / row['open'] - 1:+.2%}），收盘 {row['close']:,.2f}，"
      f"真实波幅是 8 月 16 日 ATR 的 {X.true_range(day['high'], day['low'], c)[crash] / I.atr(day['high'], day['low'], c).iloc[i + 2]:.1f} 倍")
after = c.iloc[i + 1:i + 61]
print(f"之后 60 天：最高收盘 {after.max():,.2f}（{after.idxmax().date()}），最低收盘 {after.min():,.2f}（{after.idxmin().date()}）")

1 天后（2023-08-15）：收盘 29,200.00（-0.78%），带宽 0.0257，NATR 1.85
2 天后（2023-08-16）：收盘 28,730.51（-2.38%），带宽 0.0312，NATR 1.88
3 天后（2023-08-17）：收盘 26,623.41（-9.54%），带宽 0.0860，NATR 2.86
5 天后（2023-08-19）：收盘 26,100.01（-11.32%），带宽 0.1546，NATR 2.95
10 天后（2023-08-24）：收盘 26,180.05（-11.05%），带宽 0.2181，NATR 2.84
20 天后（2023-09-03）：收盘 25,971.21（-11.76%），带宽 0.1410，NATR 2.91
8 月 17 日：开盘 28,730.51，最低 25,166.00（-12.41%），收盘 26,623.41，真实波幅是 8 月 16 日 ATR 的 6.7 倍
之后 60 天：最高收盘 29,200.00（2023-08-15），最低收盘 25,162.52（2023-09-11）

def label_test(values, flag, n=2000):
    """flag 为真的组减去其余的平均值；把标签随机打乱 n 次，看差距不小于实际的比例（第 10 篇）。"""
    v, f = np.asarray(values, float), np.asarray(flag, bool)
    ok = ~np.isnan(v)
    v, f = v[ok], f[ok]
    observed = v[f].mean() - v[~f].mean()
    sims = np.array([v[p].mean() - v[~p].mean() for p in (rng.permutation(f) for _ in range(n))])
    return observed, (np.abs(sims) >= abs(observed)).mean()


def squeeze_study(df, look):
    """带宽创出 look 根新低的第一根（收口），和带宽排在过去 look 根最低 10% 的全部 K 线、全部 K 线比较。"""
    close = df["close"]
    b = I.bollinger(close)
    a = I.atr(df["high"], df["low"], close)
    tr = X.true_range(df["high"], df["low"], close)
    lowest = b["bandwidth"].rolling(look).min()
    new_low = (b["bandwidth"] <= lowest) & lowest.notna()
    squeeze = new_low & ~new_low.shift(1, fill_value=False)
    rank = b["bandwidth"].rolling(look).rank(pct=True)
    high_later = df["high"][::-1].rolling(20).max()[::-1].shift(-1)
    low_later = df["low"][::-1].rolling(20).min()[::-1].shift(-1)
    frame = pd.DataFrame({"收口": squeeze, "最低 10%": rank <= 0.1, "之后 ÷ 现在 ATR": future_mean(tr) / a,
                          "之后 20 根高低点距离（ATR）": (high_later - low_later) / a,
                          "之后 20 根平均真实波幅（%）": 100 * future_mean(tr) / close,
                          "之后 20 根上涨": (close.shift(-20) > close).astype(float)}).loc[rank.notna() & high_later.notna()]
    return frame, b, a


rows = []
for (name, df), look in zip(datasets, [252, 252, 365, 365 * 6, 365 * 24]):
    frame, b, a = squeeze_study(df, look)
    low = frame[frame["最低 10%"]]
    _, p = label_test(low["之后 ÷ 现在 ATR"], low["收口"])
    for label, part in [("收口", frame[frame["收口"]]), ("带宽最低 10%", low), ("全部", frame)]:
        rows.append({"数据": name, "组": label, "K 线": len(part)} | part.iloc[:, 2:].mean().to_dict()
                    | {"p（收口 vs 最低 10%）": p if label == "收口" else np.nan})
print(pd.DataFrame(rows).to_string(index=False, float_format=lambda v: "" if pd.isna(v) else f"{v:.2f}"))

print("同样的计算放在打乱 K 线顺序的价格上（没有波动率聚集），各打乱 20 次取平均：")
rows = []
for (name, df), look in list(zip(datasets, [252, 252, 365, 365 * 6]))[:4]:
    sims = []
    for _ in range(20):
        frame, _, _ = squeeze_study(shuffle_bars(df, rng), look)
        sims.append({"收口 次数": frame["收口"].sum(), "收口": frame.loc[frame["收口"], "之后 ÷ 现在 ATR"].mean(),
                     "带宽最低 10%": frame.loc[frame["最低 10%"], "之后 ÷ 现在 ATR"].mean(), "全部": frame["之后 ÷ 现在 ATR"].mean()})
    rows.append({"数据": name} | pd.DataFrame(sims).mean().to_dict())
print("之后 20 根平均真实波幅 ÷ 现在的 ATR：")
print(pd.DataFrame(rows).to_string(index=False, float_format=lambda v: f"{v:.2f}"))

       数据        组   K 线  之后 ÷ 现在 ATR  之后 20 根高低点距离（ATR）  之后 20 根平均真实波幅（%）  之后 20 根上涨  p（收口 vs 最低 10%）
   SPY 日线       收口     9         1.07               5.75              1.16       0.78             0.62
   SPY 日线 带宽最低 10%   256         1.14               5.59              1.12       0.70              NaN
   SPY 日线       全部  2222         1.08               5.55              1.34       0.68              NaN
  AAPL 日线       收口    17         1.02               6.25              2.00       0.82             0.42
  AAPL 日线 带宽最低 10%   251         1.08               6.09              1.99       0.69              NaN
  AAPL 日线       全部  2223         1.06               5.59              2.40       0.64              NaN
   BTC 日线       收口    18         1.35               8.45              3.95       0.33             0.13
   BTC 日线 带宽最低 10%   438         1.16               6.83              3.56       0.45              NaN
   BTC 日线       全部  2899         1.07               5.35              4.59       0.55              NaN
BTC 4 小时线       收口    19         1.32               7.24              1.16       0.63             0.34
BTC 4 小时线 带宽最低 10%  2270         1.22               6.57              1.26       0.53              NaN
BTC 4 小时线       全部 17566         1.05               5.25              1.77       0.53              NaN
BTC 1 小时线       收口    21         1.45               7.92              0.43       0.52             0.07
BTC 1 小时线 带宽最低 10%  8847         1.23               6.52              0.52       0.51              NaN
BTC 1 小时线       全部 70315         1.05               5.21              0.86       0.52              NaN
同样的计算放在打乱 K 线顺序的价格上（没有波动率聚集），各打乱 20 次取平均：
之后 20 根平均真实波幅 ÷ 现在的 ATR：
       数据  收口 次数   收口  带宽最低 10%   全部
   SPY 日线  11.20 1.15      1.13 1.04
  AAPL 日线  10.70 1.15      1.11 1.04
   BTC 日线   8.25 1.20      1.16 1.05
BTC 4 小时线   9.75 1.28      1.18 1.03

print("突破方向：收口之后 20 根以内，收盘价第一次越过上轨（做多）或下轨（做空）；对照是所有「前一根还在带内、这一根收在带外」的 K 线")
rows = []
for (name, df), look in zip(datasets, [252, 252, 365, 365 * 6, 365 * 24]):
    frame, b, a = squeeze_study(df, look)
    close = df["close"]
    side = np.sign((close > b["upper"]).astype(int) - (close < b["lower"]).astype(int))
    breakout = (side != 0) & (side.shift(1) == 0) & a.notna()
    after_squeeze = pd.Series(False, index=close.index)
    for when in frame.index[frame["收口"]]:
        k = close.index.get_loc(when)
        later = breakout.iloc[k + 1:k + 21]
        if later.any():
            after_squeeze[later.idxmax()] = True
    times = close.index[breakout]
    hit = pd.Series(X.first_passage(close, df["high"], df["low"], a, times, side[times].to_numpy()), index=times)
    _, p = label_test(hit, after_squeeze[times])
    rows.append({"数据": name, "收口后的突破": int(hit[after_squeeze[times]].notna().sum()),
                 "顺向": hit[after_squeeze[times]].mean(), "全部突破": int(hit.notna().sum()), "全部突破 顺向": hit.mean(), "p": p})
print(pd.DataFrame(rows).to_string(index=False, float_format=lambda v: f"{v:.3f}"))

突破方向：收口之后 20 根以内，收盘价第一次越过上轨（做多）或下轨（做空）；对照是所有「前一根还在带内、这一根收在带外」的 K 线
       数据  收口后的突破    顺向  全部突破  全部突破 顺向     p
   SPY 日线       7 0.571   148    0.486 0.710
  AAPL 日线       9 0.889   152    0.572 0.083
   BTC 日线      13 0.692   167    0.551 0.384
BTC 4 小时线      15 0.333  1035    0.555 0.119
BTC 1 小时线      12 0.583  4048    0.523 0.758

def stop_hit(df, distance, horizon=20):
    """在每根 K 线收盘价买入，止损放在收盘价下方 distance（价格单位），之后 horizon 根里最低价碰到算被打掉。"""
    close, low, dist = df["close"].to_numpy(), df["low"].to_numpy(), distance.to_numpy()
    out = np.full(len(close), np.nan)
    for k in range(len(close) - horizon):
        if not np.isnan(dist[k]):
            out[k] = float((low[k + 1:k + 1 + horizon] <= close[k] - dist[k]).any())
    return pd.Series(out, index=df.index)


rows = []
for name, df in markets.items():
    close = df["close"]
    a = I.atr(df["high"], df["low"], close)
    percent = float((3 * a / close).median())                     # 固定百分比取 3 ATR 的历史中位数，两种止损平均距离相同
    rank = (a / close).rolling(periods_per_year[name]).rank(pct=True)
    frame = pd.DataFrame({"组": pd.cut(rank, [0, 0.2, 0.4, 0.6, 0.8, 1.0], labels=["最低 20%", "20-40%", "40-60%", "60-80%", "最高 20%"]),
                          f"固定 {percent:.2%}": stop_hit(df, close * percent), "3 ATR": stop_hit(df, 3 * a)}).dropna()
    table = frame.groupby("组", observed=True).mean().T
    table.insert(0, "标的", name)
    rows.append(table)
print("20 根以内被打掉的比例，按「现在的 NATR 在过去一年里的排位」分组：")
print(pd.concat(rows).to_string(float_format=lambda v: f"{v:.1%}"))

20 根以内被打掉的比例，按「现在的 NATR 在过去一年里的排位」分组：
组            标的  最低 20%  20-40%  40-60%  60-80%  最高 20%
固定 3.26%    SPY   25.7%   32.4%   35.5%   32.8%   53.3%
3 ATR       SPY   32.2%   31.2%   33.1%   26.9%   33.8%
固定 6.24%   AAPL   27.1%   23.1%   20.8%   37.7%   47.8%
3 ATR      AAPL   34.0%   25.5%   20.8%   32.2%   34.2%
固定 13.19%   BTC   23.5%   26.3%   26.2%   24.5%   26.5%
3 ATR       BTC   29.3%   29.9%   26.0%   15.8%    9.6%

rows = []
for name, df in markets.items():
    close = df["close"]
    r = close.pct_change()
    n = I.natr(df["high"], df["low"], close)
    weight = (1.0 / n.shift(1)).clip(upper=4.0 / n.median())       # 前一根的 NATR 越大，这一根拿得越少；最多 4 倍
    weight = weight / weight.mean()                                 # 平均仓位和固定仓位一样
    scaled = (weight * r).dropna()
    fixed = r.loc[scaled.index]
    by_year = pd.DataFrame({"固定金额": fixed, "按 ATR 调整": scaled}).groupby(scaled.index.year).std() * np.sqrt(periods_per_year[name])
    rows.append({"标的": name, "固定金额 最低年份波动": by_year["固定金额"].min(), "最高": by_year["固定金额"].max(),
                 "最高 ÷ 最低": by_year["固定金额"].max() / by_year["固定金额"].min(),
                 "按 ATR 调整 最低": by_year["按 ATR 调整"].min(), "最高 ": by_year["按 ATR 调整"].max(),
                 "最高 ÷ 最低 ": by_year["按 ATR 调整"].max() / by_year["按 ATR 调整"].min()})
print(pd.DataFrame(rows).to_string(index=False, float_format=lambda v: f"{v:.2f}"))

  标的  固定金额 最低年份波动   最高  最高 ÷ 最低  按 ATR 调整 最低  最高   最高 ÷ 最低 
 SPY         0.07 0.34     4.95         0.11 0.15      1.29
AAPL         0.16 0.47     2.93         0.19 0.28      1.46
 BTC         0.42 1.16     2.78         0.51 0.67      1.30

def mainline_v1(df, k=3.0, n_atr=14, breakout=20, fast=50, slow=200):
    """主线策略 v1：v0 加上 ATR 吊灯止损。没有仓位管理、没有成本。

    入场：收盘时快均线在慢均线上方，下一根开盘买入（和 v0 相同）。
    止损：买入那一根的止损价 = 开盘价 - k × 前一根的 ATR。之后每一根开盘前更新：
          止损价 = max(原来的止损价, 买入以来的最高价 - k × 前一根的 ATR)，只上移、不下移。
          开盘价已经不高于止损价，按开盘价卖出（跳空击穿）；否则最低价碰到止损价，按止损价卖出。
    趋势出场：收盘时快均线不在慢均线上方，下一根开盘卖出（和 v0 相同）。
    止损之后：趋势还在，也不马上买回。要等收盘价高于之前 breakout 根的最高收盘价（创新高），下一根开盘再买。
    返回每根 K 线的收益率、是否持有，以及每笔交易。
    """
    o, h, l, c = (df[x].to_numpy(float) for x in cols)
    a = I.atr(df["high"], df["low"], df["close"], n_atr).to_numpy()
    trend = (I.sma(df["close"], fast) > I.sma(df["close"], slow)).to_numpy()
    first = df.index.get_loc(I.sma(df["close"], slow).first_valid_index())
    r, held, trades = np.zeros(len(c)), np.zeros(len(c)), []
    holding = buy = sell = locked = False
    for i in range(first, len(c)):
        if holding:
            stop, base = max(stop, highest - k * a[i - 1]), c[i - 1]
        elif buy:
            holding, buy, entry, entry_i, highest = True, False, o[i], i, o[i]
            stop, base = o[i] - k * a[i - 1], o[i]
        if holding:
            held[i] = 1
            exit_price = reason = None
            if sell:
                exit_price, reason = o[i], "趋势"
            elif o[i] <= stop:
                exit_price, reason = o[i], "止损（跳空）"
            elif l[i] <= stop:
                exit_price, reason = stop, "止损"
            if reason is None:
                r[i], highest = c[i] / base - 1, max(highest, h[i])
            else:
                r[i] = exit_price / base - 1
                trades.append((df.index[entry_i], entry, df.index[i], exit_price, reason))
                holding, sell, locked = False, False, reason.startswith("止损")
        if holding:                                                   # 收盘时决定下一根做什么
            sell = not trend[i]
        elif locked:
            if trend[i] and c[i] > c[i - breakout:i].max():
                buy, locked = True, False
        else:
            buy = bool(trend[i])
    index = df.index[first:]
    trades = pd.DataFrame(trades, columns=["买入日", "买入价", "卖出日", "卖出价", "原因"])
    trades["收益"] = trades["卖出价"] / trades["买入价"] - 1
    return pd.Series(r[first:], index=index), pd.Series(held[first:], index=index), trades

def summary(r, n, held=None, trades=None):
    equity = (1 + r).cumprod()
    out = {"年化": equity.iloc[-1] ** (n / len(r)) - 1, "年化波动": r.std() * np.sqrt(n),
           "最大回撤": (equity / equity.cummax() - 1).min()}
    out["年化 ÷ 波动"] = out["年化"] / out["年化波动"]
    out["年化 ÷ 回撤"] = out["年化"] / -out["最大回撤"]
    if held is not None:
        out["在场时间"] = held.mean()
    if trades is not None:
        out |= {"交易次数": len(trades), "止损出场": int(trades["原因"].str.startswith("止损").sum()),
                "最差一笔": trades["收益"].min()}
    return out


check = []
for name, df in markets.items():
    signal = mainline_v0(df)
    start = signal.first_valid_index()
    d = df.loc[start:]
    v0 = backtest_next_open(d, signal.loc[start:])
    r, held, trades = mainline_v1(df, k=1e12)
    check.append(f"{name} {np.abs(r - v0).max():.1e}")
    rows = {"买入持有": summary(d["close"].pct_change().fillna(0.0), periods_per_year[name]),
            "主线 v0": summary(v0, periods_per_year[name], signal.loc[start:].shift(1).fillna(0.0))}
    for k in [3.0, 2.0, 4.0, 5.0]:
        r, held, trades = mainline_v1(df, k=k)
        rows[f"主线 v1（{k:.0f} ATR）"] = summary(r, periods_per_year[name], held, trades)
        if k == 3.0:
            example = trades
    print(f"{name}：{start.date()} 至 {d.index[-1].date()}")
    count = lambda v: "" if pd.isna(v) else f"{v:.0f}"
    print(pd.DataFrame(rows).T.to_string(float_format=lambda v: "" if pd.isna(v) else f"{v:.3f}",
                                         formatters={"交易次数": count, "止损出场": count}))
    if name == "SPY":
        print("SPY 主线 v1（3 ATR）2020 年和 2025 年的交易：")
        print(example[example["买入日"].dt.year.isin([2020, 2025])].round({"买入价": 2, "卖出价": 2, "收益": 4}).to_string(index=False))
print("止损距离设成无穷大时，v1 和 v0 每根 K 线收益率的最大差：", "，".join(check))

SPY：2017-06-30 至 2026-09-15
                年化  年化波动   最大回撤  年化 ÷ 波动  年化 ÷ 回撤  在场时间 交易次数 止损出场   最差一笔
买入持有         0.132 0.186 -0.341    0.711    0.388   NaN  NaN  NaN    NaN
主线 v0        0.085 0.157 -0.341    0.545    0.251 0.816  NaN  NaN    NaN
主线 v1（3 ATR） 0.029 0.071 -0.140    0.414    0.210 0.495   49   49 -0.044
主线 v1（2 ATR） 0.017 0.059 -0.150    0.284    0.113 0.408   76   76 -0.028
主线 v1（4 ATR） 0.048 0.081 -0.126    0.598    0.383 0.561   33   33 -0.061
主线 v1（5 ATR） 0.052 0.089 -0.142    0.587    0.368 0.614   26   25 -0.061
SPY 主线 v1（3 ATR）2020 年和 2025 年的交易：
       买入日    买入价        卖出日    卖出价     原因      收益
2020-02-06 333.91 2020-02-24 323.14 止损（跳空） -0.0323
2020-07-13 320.13 2020-09-03 348.19     止损  0.0877
2020-10-08 342.85 2020-10-26 337.94     止损 -0.0143
2020-11-10 353.49 2021-01-04 366.19     止损  0.0359
2025-01-22 605.92 2025-02-03 590.67     止损 -0.0252
2025-02-19 610.08 2025-02-25 595.55     止损 -0.0238
2025-07-03 622.45 2025-08-01 624.94     止损  0.0040
2025-08-11 637.46 2025-10-10 659.54     止损  0.0346
2025-10-27 682.73 2025-11-06 669.18     止损 -0.0198
2025-12-01 678.81 2026-01-20 679.35     止损  0.0008
AAPL：2017-06-30 至 2026-09-15
                年化  年化波动   最大回撤  年化 ÷ 波动  年化 ÷ 回撤  在场时间 交易次数 止损出场   最差一笔
买入持有         0.284 0.300 -0.385    0.948    0.737   NaN  NaN  NaN    NaN
主线 v0        0.155 0.262 -0.459    0.592    0.338 0.803  NaN  NaN    NaN
主线 v1（3 ATR） 0.092 0.149 -0.213    0.616    0.430 0.430   45   45 -0.078
主线 v1（2 ATR） 0.059 0.127 -0.164    0.465    0.359 0.335   75   75 -0.045
主线 v1（4 ATR） 0.102 0.167 -0.299    0.610    0.340 0.497   33   33 -0.124
主线 v1（5 ATR） 0.106 0.183 -0.379    0.580    0.279 0.553   27   27 -0.124
BTC：2018-03-04 至 2026-08-31
                年化  年化波动   最大回撤  年化 ÷ 波动  年化 ÷ 回撤  在场时间 交易次数 止损出场   最差一笔
买入持有         0.253 0.624 -0.766    0.406    0.331   NaN  NaN  NaN    NaN
主线 v0        0.191 0.476 -0.667    0.401    0.286 0.517  NaN  NaN    NaN
主线 v1（3 ATR） 0.082 0.291 -0.565    0.283    0.146 0.205   41   39 -0.209
主线 v1（2 ATR） 0.176 0.247 -0.351    0.711    0.500 0.155   53   52 -0.140
主线 v1（4 ATR） 0.116 0.324 -0.591    0.359    0.197 0.267   30   28 -0.284
主线 v1（5 ATR） 0.131 0.356 -0.571    0.369    0.230 0.344   24   22 -0.359
止损距离设成无穷大时，v1 和 v0 每根 K 线收益率的最大差： SPY 0.0e+00，AAPL 0.0e+00，BTC 0.0e+00

import talib

rows = []
for name, df in [("BTC 日线", day), ("SPY 日线", spy), ("BTC 1 小时线", h1)]:
    high, low, close = df["high"], df["low"], df["close"]
    H, L, C = (s.to_numpy(float) for s in (high, low, close))
    stoch, band, m = I.stochastic(high, low, close), I.bollinger(close), I.macd(close)
    upper, middle, lower = talib.BBANDS(C, 20, 2, 2, 0)
    pairs = [("SMA(20)", I.sma(close, 20), talib.SMA(C, 20)), ("EMA(20)", I.ema(close, 20), talib.EMA(C, 20)),
             ("WMA(20)", I.wma(close, 20), talib.WMA(C, 20)), ("MACD 线", m["macd"], talib.MACD(C)[0]),
             ("MACD 信号线", m["signal"], talib.MACD(C)[1]), ("MACD 柱", m["hist"], talib.MACD(C)[2]),
             ("RSI(14)", I.rsi(close), talib.RSI(C, 14)), ("Stochastic %K", stoch["k"], talib.STOCH(H, L, C, 14, 3, 0, 3, 0)[0]),
             ("Stochastic %D", stoch["d"], talib.STOCH(H, L, C, 14, 3, 0, 3, 0)[1]),
             ("真实波幅", X.true_range(high, low, close), talib.TRANGE(H, L, C)), ("ATR(14)", I.atr(high, low, close), talib.ATR(H, L, C, 14)),
             ("NATR(14)", I.natr(high, low, close), talib.NATR(H, L, C, 14)),
             ("布林上轨", band["upper"], upper), ("布林中轨", band["middle"], middle), ("布林下轨", band["lower"], lower),
             ("+DI(14)", X.adx(high, low, close)["plus_di"], talib.PLUS_DI(H, L, C, 14)),
             ("-DI(14)", X.adx(high, low, close)["minus_di"], talib.MINUS_DI(H, L, C, 14)),
             ("ADX(14)", X.adx(high, low, close)["adx"], talib.ADX(H, L, C, 14))]
    for label, ours, theirs in pairs:
        ours = ours.to_numpy(float)
        both = ~np.isnan(ours) & ~np.isnan(theirs)
        error = np.abs(ours[both] - theirs[both])
        rows.append({"数据": name, "指标": label, "NaN 位置一致": bool((np.isnan(ours) == np.isnan(theirs)).all()),
                     "最大绝对误差": error.max(), "最大相对误差": (error / np.maximum(np.abs(theirs[both]), 1e-12)).max()})
table = pd.DataFrame(rows)
table["通过（相对误差 ≤ 1e-8）"] = table["NaN 位置一致"] & (table["最大相对误差"] <= 1e-8)
print(table.to_string(index=False, formatters={"最大绝对误差": "{:.1e}".format, "最大相对误差": "{:.1e}".format}))

       数据            指标  NaN 位置一致  最大绝对误差  最大相对误差  通过（相对误差 ≤ 1e-8）
   BTC 日线       SMA(20)      True 1.5e-10 2.4e-15             True
   BTC 日线       EMA(20)      True 0.0e+00 0.0e+00             True
   BTC 日线       WMA(20)      True 4.5e-10 1.3e-14             True
   BTC 日线        MACD 线      True 1.5e-11 7.6e-14             True
   BTC 日线      MACD 信号线      True 5.2e-12 5.3e-14             True
   BTC 日线        MACD 柱      True 1.2e-11 5.1e-13             True
   BTC 日线       RSI(14)      True 4.3e-14 1.6e-15             True
   BTC 日线 Stochastic %K      True 2.1e-13 6.2e-14             True
   BTC 日线 Stochastic %D      True 2.2e-13 3.1e-14             True
   BTC 日线          真实波幅      True 0.0e+00 0.0e+00             True
   BTC 日线       ATR(14)      True 1.4e-12 5.5e-16             True
   BTC 日线      NATR(14)      True 7.1e-15 5.7e-16             True
   BTC 日线          布林上轨      True 2.4e-08 8.6e-13             True
   BTC 日线          布林中轨      True 1.5e-10 2.4e-15             True
   BTC 日线          布林下轨      True 2.4e-08 8.8e-13             True
   BTC 日线       +DI(14)      True 2.8e-01 1.7e-02            False
   BTC 日线       -DI(14)      True 2.3e-01 8.2e-03            False
   BTC 日线       ADX(14)      True 2.5e-01 5.7e-03            False
   SPY 日线       SMA(20)      True 2.2e-12 2.9e-15             True
   SPY 日线       EMA(20)      True 0.0e+00 0.0e+00             True
   SPY 日线       WMA(20)      True 6.5e-12 1.6e-14             True
   SPY 日线        MACD 线      True 0.0e+00 0.0e+00             True
   SPY 日线      MACD 信号线      True 3.2e-15 9.3e-15             True
   SPY 日线        MACD 柱      True 3.2e-15 5.7e-14             True
   SPY 日线       RSI(14)      True 4.3e-14 1.2e-15             True
   SPY 日线 Stochastic %K      True 3.3e-13 1.1e-13             True
   SPY 日线 Stochastic %D      True 3.7e-13 4.7e-14             True
   SPY 日线          真实波幅      True 0.0e+00 0.0e+00             True
   SPY 日线       ATR(14)      True 3.6e-15 4.5e-16             True
   SPY 日线      NATR(14)      True 1.8e-15 5.8e-16             True
   SPY 日线          布林上轨      True 3.3e-10 4.2e-13             True
   SPY 日线          布林中轨      True 2.2e-12 2.9e-15             True
   SPY 日线          布林下轨      True 3.3e-10 4.3e-13             True
   SPY 日线       +DI(14)      True 3.6e-01 2.5e-02            False
   SPY 日线       -DI(14)      True 5.3e-01 1.7e-02            False
   SPY 日线       ADX(14)      True 8.3e-01 4.5e-02            False
BTC 1 小时线       SMA(20)      True 2.6e-10 8.7e-15             True
BTC 1 小时线       EMA(20)      True 1.5e-11 2.2e-16             True
BTC 1 小时线       WMA(20)      True 1.6e-09 2.8e-14             True
BTC 1 小时线        MACD 线      True 9.1e-13 1.5e-14             True
BTC 1 小时线      MACD 信号线      True 2.3e-13 8.0e-12             True
BTC 1 小时线        MACD 柱      True 7.3e-13 3.8e-13             True
BTC 1 小时线       RSI(14)      True 4.3e-14 1.5e-15             True
BTC 1 小时线 Stochastic %K      True 2.3e-12 7.4e-12             True
BTC 1 小时线 Stochastic %D      True 3.3e-12 1.1e-12             True
BTC 1 小时线          真实波幅      True 0.0e+00 0.0e+00             True
BTC 1 小时线       ATR(14)      True 4.5e-13 6.3e-16             True
BTC 1 小时线      NATR(14)      True 3.6e-15 7.9e-16             True
BTC 1 小时线          布林上轨      True 4.5e-06 1.3e-10             True
BTC 1 小时线          布林中轨      True 2.6e-10 8.7e-15             True
BTC 1 小时线          布林下轨      True 4.5e-06 1.3e-10             True
BTC 1 小时线       +DI(14)      True 2.8e-01 1.9e-02            False
BTC 1 小时线       -DI(14)      True 4.0e-01 2.2e-02            False
BTC 1 小时线       ADX(14)      True 3.3e-01 2.9e-02            False

high, low, close = day["high"], day["low"], day["close"]
H, L, C = (s.to_numpy(float) for s in (high, low, close))
plus_dm = X.directional_movement(high, low)["plus_dm"].to_numpy()
n = 14
running = np.nansum(plus_dm[1:n])                                  # TA-Lib：先把前 n - 1 个 +DM 加起来
talib_style = np.full(len(plus_dm), np.nan)
for k in range(n, len(plus_dm)):
    running = running - running / n + plus_dm[k]
    talib_style[k] = running
tr = X.true_range(high, low, close).to_numpy()
running = np.nansum(tr[1:n])
talib_tr = np.full(len(tr), np.nan)
for k in range(n, len(tr)):
    running = running - running / n + tr[k]
    talib_tr[k] = running
print(f"按 TA-Lib 的初始值重算 +DI，和 talib.PLUS_DI 的最大差：{np.nanmax(np.abs(100 * talib_style / talib_tr - talib.PLUS_DI(H, L, C, 14))):.1e}")
ours = X.adx(high, low, close)
gap = pd.DataFrame({"+DI": (ours["plus_di"] - talib.PLUS_DI(H, L, C, 14)).abs(), "ADX": (ours["adx"] - talib.ADX(H, L, C, 14)).abs()})
start = gap["+DI"].first_valid_index()
k0 = day.index.get_loc(start)
print(pd.DataFrame({f"第 {k} 根": gap.iloc[k0 + k] for k in [0, 14, 50, 100, 200, 300, 400]}).T
      .to_string(formatters={"+DI": "{:.1e}".format, "ADX": "{:.1e}".format}))
print(f"(13/14)^300 = {(13 / 14) ** 300:.1e}")

按 TA-Lib 的初始值重算 +DI，和 talib.PLUS_DI 的最大差：7.1e-15
            +DI     ADX
第 0 根   1.2e-01     NaN
第 14 根  1.2e-01 1.9e-01
第 50 根  2.2e-02 9.5e-02
第 100 根 1.5e-04 7.7e-03
第 200 根 2.7e-08 5.3e-06
第 300 根 1.1e-10 4.7e-09
第 400 根 4.6e-14 3.7e-12
(13/14)^300 = 2.2e-10

def indicator_panel(df):
    """同一组 K 线上的 14 个指标读数，每一列都是「这根 K 线收盘时」的值。"""
    close, high, low = df["close"], df["high"], df["low"]
    m, band, stoch, dmi = I.macd(close), I.bollinger(close), I.stochastic(high, low, close), X.adx(high, low, close)
    change = close.diff()
    return pd.DataFrame({
        "RSI": I.rsi(close), "%K": stoch["k"], "%b": band["percent_b"], "乖离率": I.bias(close, I.sma(close, 20)),
        "20 日涨幅": close.pct_change(20), "效率比": change.rolling(20).sum() / change.abs().rolling(20).sum(),
        "+DI 减 -DI": dmi["plus_di"] - dmi["minus_di"], "MACD 线": m["macd"] / close, "MACD 柱": m["hist"] / close,
        "ADX": dmi["adx"], "NATR": I.natr(high, low, close), "带宽": band["bandwidth"],
        "20 日波动": np.log(close).diff().rolling(20).std(), "相对成交量": I.relative_volume(df["volume"], 20)}).dropna()


for name, df in [("BTC 日线", day), ("SPY 日线", spy)]:
    panel = indicator_panel(df)
    corr = panel.corr(method="spearman")
    print(f"{name}（{len(panel)} 根，Spearman 秩相关）：")
    print(corr.round(2).to_string())
    linked = corr.abs() >= 0.8
    seen, groups = set(), []
    for column in corr.columns:
        if column in seen:
            continue
        group, todo = [], [column]
        while todo:
            item = todo.pop()
            if item not in seen:
                seen.add(item)
                group.append(item)
                todo += [other for other in corr.columns if linked.loc[item, other] and other not in seen]
        groups.append(group)
    print("   |相关系数| ≥ 0.8 连在一起的组：", " ｜ ".join("、".join(g) for g in groups))

BTC 日线（3269 根，Spearman 秩相关）：
            RSI    %K    %b   乖离率  20 日涨幅   效率比  +DI 减 -DI  MACD 线  MACD 柱   ADX  NATR    带宽  20 日波动  相对成交量
RSI        1.00  0.84  0.91  0.93    0.88  0.90       0.94    0.86    0.55  0.08 -0.27  0.08   -0.07   0.05
%K         0.84  1.00  0.89  0.88    0.71  0.74       0.77    0.63    0.76  0.09 -0.17  0.12   -0.01   0.06
%b         0.91  0.89  1.00  0.94    0.77  0.80       0.85    0.63    0.74  0.05 -0.22  0.07   -0.05   0.06
乖离率        0.93  0.88  0.94  1.00    0.85  0.85       0.88    0.72    0.76  0.07 -0.22  0.09   -0.04   0.06
20 日涨幅     0.88  0.71  0.77  0.85    1.00  0.97       0.86    0.86    0.52  0.08 -0.22  0.09   -0.05   0.06
效率比        0.90  0.74  0.80  0.85    0.97  1.00       0.87    0.84    0.53  0.09 -0.24  0.08   -0.08   0.07
+DI 减 -DI  0.94  0.77  0.85  0.88    0.86  0.87       1.00    0.84    0.54  0.05 -0.32  0.04   -0.13   0.05
MACD 线     0.86  0.63  0.63  0.72    0.86  0.84       0.84    1.00    0.26  0.07 -0.23  0.08   -0.07   0.05
MACD 柱     0.55  0.76  0.74  0.76    0.52  0.53       0.54    0.26    1.00  0.07 -0.10  0.10    0.02   0.03
ADX        0.08  0.09  0.05  0.07    0.08  0.09       0.05    0.07    0.07  1.00  0.30  0.53    0.36   0.06
NATR      -0.27 -0.17 -0.22 -0.22   -0.22 -0.24      -0.32   -0.23   -0.10  0.30  1.00  0.69    0.90   0.06
带宽         0.08  0.12  0.07  0.09    0.09  0.08       0.04    0.08    0.10  0.53  0.69  1.00    0.75   0.12
20 日波动    -0.07 -0.01 -0.05 -0.04   -0.05 -0.08      -0.13   -0.07    0.02  0.36  0.90  0.75    1.00   0.02
相对成交量      0.05  0.06  0.06  0.06    0.06  0.07       0.05    0.05    0.03  0.06  0.06  0.12    0.02   1.00
   |相关系数| ≥ 0.8 连在一起的组： RSI、MACD 线、+DI 减 -DI、效率比、20 日涨幅、乖离率、%b、%K ｜ MACD 柱 ｜ ADX ｜ NATR、20 日波动 ｜ 带宽 ｜ 相对成交量

# ---------------------------------------------------------------------------
# 七、波动率：ATR 和布林带（第 15 篇）
# ---------------------------------------------------------------------------

def atr(high: pd.Series, low: pd.Series, close: pd.Series, n: int = 14) -> pd.Series:
    """平均真实波幅（Wilder）：第 8 篇的真实波幅 true_range，用 Wilder 平滑取 n 根的平均。

    真实波幅从第 2 根开始才有（要用前一根收盘价），所以第一个 ATR 出现在第 n + 1 根，是前 n 个真实波幅的平均。
    单位和价格相同。和 TA-Lib 的 ATR 一致。
    """
    _check_period(n)
    return wilder_smooth(true_range(high, low, close), n)


def natr(high: pd.Series, low: pd.Series, close: pd.Series, n: int = 14) -> pd.Series:
    """归一化的 ATR：ATR 占收盘价的百分比，100 × ATR ÷ 收盘价。不同价位、不同标的之间可以比较。和 TA-Lib 的 NATR 一致。"""
    return 100 * atr(high, low, close, n) / close


def bollinger(close: pd.Series, n: int = 20, k: float = 2.0) -> pd.DataFrame:
    """布林带：中轨是 n 根简单平均，上下轨是中轨加减 k 倍标准差。

    ⚠️ 标准差是这 n 个收盘价的总体标准差（除以 n），和 Bollinger 本人、TA-Lib 的 BBANDS 一致；
    pandas 的 rolling().std() 默认除以 n - 1，算出来的带会宽一些。
    返回五列：
    middle、upper、lower
    percent_b：收盘价在带里的位置，(收盘价 - 下轨) ÷ (上轨 - 下轨)，下轨是 0、上轨是 1，可以小于 0 或大于 1
    bandwidth：带宽，(上轨 - 下轨) ÷ 中轨，等于 2k 倍标准差占中轨的比例
    前 n - 1 根为 NaN；n 个价格完全相同时，上下轨重合，percent_b 为 NaN。
    """
    _check_period(n)
    middle = sma(close, n)
    width = k * close.rolling(n).std(ddof=0)
    upper, lower = middle + width, middle - width
    spread = upper - lower
    return pd.DataFrame({"middle": middle, "upper": upper, "lower": lower,
                         "percent_b": ((close - lower) / spread).where(spread != 0),
                         "bandwidth": spread / middle})

def test_atr_by_hand():
    close = hourly([101, 105, 95, 100])
    high, low = close + [1, 1, 3, 1], close - [2, 2, 1, 2]
    # 真实波幅：第 2 根 max(3, |106 - 101|, |103 - 101|) = 5；第 3 根 max(4, 7, 11) = 11；第 4 根 max(3, 6, 3) = 6
    # ATR(2)：第 3 根 (5 + 11) / 2 = 8；第 4 根 8 + (6 - 8) / 2 = 7
    assert I.atr(high, low, close, 2).tolist()[2:] == [8, 7] and I.atr(high, low, close, 2).iloc[:2].isna().all()
    assert I.natr(high, low, close, 2).iloc[3] == pytest.approx(100 * 7 / 100)


def test_bollinger_by_hand():
    b = I.bollinger(hourly([10, 11, 12, 11, 13, 14]), n=4, k=2)
    # 第 4 根：10、11、12、11，平均 11，总体方差 (1 + 0 + 1 + 0) / 4 = 0.5，上下轨 11 ± 2√0.5
    assert b.iloc[:3].isna().all().all()
    assert b["upper"].iloc[3:].tolist() == pytest.approx([12.41421356, 13.40831239, 14.73606798])
    assert b["lower"].iloc[3:].tolist() == pytest.approx([9.58578644, 10.09168761, 10.26393202])
    assert b["percent_b"].iloc[3:].tolist() == pytest.approx([0.5, 0.87688918, 0.83541020])
    assert b["bandwidth"].iloc[3:].tolist() == pytest.approx([0.25712974, 0.28226594, 0.35777088])


def test_bias_equals_percent_b_times_bandwidth():
    rng = np.random.default_rng(15)
    close = hourly(100 * np.exp(np.cumsum(rng.normal(0, 0.02, 300))))
    b = I.bollinger(close)
    # %b - 0.5 = (收盘价 - 中轨) ÷ (4 倍标准差)，带宽 = 4 倍标准差 ÷ 中轨，两者相乘就是乖离率
    np.testing.assert_allclose((b["percent_b"] - 0.5) * b["bandwidth"], I.bias(close, b["middle"]), rtol=1e-10, atol=1e-14)


def test_bollinger_on_a_flat_line():
    b = I.bollinger(hourly(np.full(30, 5.0)))
    assert (b["upper"].dropna() == 5).all() and (b["bandwidth"].dropna() == 0).all() and b["percent_b"].isna().all()


def test_volatility_indicators_never_use_the_future():
    rng = np.random.default_rng(150)
    close = hourly(100 * np.exp(np.cumsum(rng.normal(0, 0.02, 400))))
    high, low = close * (1 + rng.uniform(0, 0.01, 400)), close * (1 - rng.uniform(0, 0.01, 400))
    full_atr, full_band = I.atr(high, low, close), I.bollinger(close)
    for k in [30, 200, 399]:
        pd.testing.assert_series_equal(I.atr(high.iloc[:k], low.iloc[:k], close.iloc[:k]), full_atr.iloc[:k])
        pd.testing.assert_frame_equal(I.bollinger(close.iloc[:k]), full_band.iloc[:k])


@pytest.mark.parametrize("n", [2, 14, 50])
def test_atr_and_natr_match_talib(n):
    talib = pytest.importorskip("talib")
    rng = np.random.default_rng(n + 15)
    close = hourly(50_000 * np.exp(np.cumsum(rng.normal(0, 0.03, 2000))))
    high, low = close * (1 + rng.uniform(0, 0.02, 2000)), close * (1 - rng.uniform(0, 0.02, 2000))
    for ours, theirs in [(I.atr(high, low, close, n), talib.ATR(high.to_numpy(), low.to_numpy(), close.to_numpy(), n)),
                         (I.natr(high, low, close, n), talib.NATR(high.to_numpy(), low.to_numpy(), close.to_numpy(), n))]:
        assert (ours.isna().to_numpy() == np.isnan(theirs)).all()
        np.testing.assert_allclose(ours.to_numpy(), theirs, rtol=1e-10, atol=1e-8)


@pytest.mark.parametrize("n,k", [(20, 2.0), (10, 1.5), (50, 2.5)])
def test_bollinger_matches_talib(n, k):
    talib = pytest.importorskip("talib")
    rng = np.random.default_rng(n)
    close = hourly(50_000 * np.exp(np.cumsum(rng.normal(0, 0.03, 2000))))
    ours = I.bollinger(close, n, k)
    for column, theirs in zip(["upper", "middle", "lower"], talib.BBANDS(close.to_numpy(), n, k, k, 0)):
        assert (ours[column].isna().to_numpy() == np.isnan(theirs)).all()
        # TA-Lib 用累加的平方和算方差，价格从 5 万跌到几百时舍入误差会放大到 1e-10 左右，所以这里放宽到 1e-8
        np.testing.assert_allclose(ours[column].to_numpy(), theirs, rtol=1e-8)

def test_adx_converges_to_talib():
    """talab 的 Wilder 平滑用前 n 个值的平均做初始值；TA-Lib 的 +DM、-DM、TR 只用前 n - 1 个（第 15 篇实验二）。
    两者的差每根乘以 (n - 1) / n，开头最多差零点几，几百根之后一致。"""
    talib = pytest.importorskip("talib")
    rng = np.random.default_rng(8)
    close = pd.Series(100 * np.exp(np.cumsum(rng.normal(0, 0.02, 1000))))
    high, low = close * (1 + rng.uniform(0, 0.02, 1000)), close * (1 - rng.uniform(0, 0.02, 1000))
    ours = X.adx(high, low, close)
    H, L, C = high.to_numpy(), low.to_numpy(), close.to_numpy()
    for column, theirs in [("plus_di", talib.PLUS_DI(H, L, C, 14)), ("minus_di", talib.MINUS_DI(H, L, C, 14)),
                           ("adx", talib.ADX(H, L, C, 14))]:
        assert (ours[column].isna().to_numpy() == np.isnan(theirs)).all()
        assert np.nanmax(np.abs(ours[column].to_numpy() - theirs)) > 1e-3             # 开头确实不一样
        np.testing.assert_allclose(ours[column].to_numpy()[500:], theirs[500:], rtol=1e-8)

pytest -q

........................................................................ [ 55%]
..........................................................               [100%]
130 passed in 0.46s