第 18 篇：道氏理论与斐波那契

t = pd.Timestamp("2025-03-14")
known = spy.loc[:t]                                           # 只用决策当天收盘为止的数据
swings = X.zigzag(known["close"], 0.05)
print(swings.tail(3).to_string(index=False))
start, end = swings["price"].iloc[-2], swings["price"].iloc[-1]
levels = Pt.retracement_levels(start, end)
print(f"从 {swings['time'].iloc[-2].date()} 的 {start:.2f} 涨到 {swings['time'].iloc[-1].date()} 的 {end:.2f}，涨了 {end / start - 1:.2%}")
print(levels.round(2).to_string())
print(known.loc["2025-03-07":, ["open", "high", "low", "close"]].to_string())
level = levels[0.618]
print(f"61.8% 回撤位 {level:.2f}；3 月 11 日最低 {known['low']['2025-03-11']:.2f}，3 月 13 日最低 {known['low']['2025-03-13']:.2f}，"
      f"3 月 14 日收盘 {known['close'][t]:.2f}，高出回撤位 {known['close'][t] / level - 1:.2%}")

      time  price  kind confirmed_at
2024-07-16 564.86     1   2024-08-02
2024-08-05 517.38    -1   2024-08-14
2025-02-19 612.93     1   2025-03-04
从 2024-08-05 的 517.38 涨到 2025-02-19 的 612.93，涨了 18.47%
0.236    590.38
0.382    576.43
0.500    565.16
0.618    553.88
0.786    537.83
              open     high       low   close
date                                         
2025-03-07  570.90  577.390  565.6300  575.92
2025-03-10  567.59  569.540  555.5900  560.58
2025-03-11  559.40  564.020  552.0200  555.92
2025-03-12  562.17  563.110  553.6900  558.87
2025-03-13  558.49  559.105  549.6800  551.42
2025-03-14  556.11  563.830  551.4917  562.81
61.8% 回撤位 553.88；3 月 11 日最低 552.02，3 月 13 日最低 549.68，3 月 14 日收盘 562.81，高出回撤位 1.61%

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


def ratio_summary(close, threshold):
    """回撤比例和扩展比例的几个统计量。"""
    ratios = Pt.swing_ratios(X.zigzag(close, threshold))
    r, e = ratios["retracement"].dropna().to_numpy(), ratios["extension"].dropna().to_numpy()
    partial = r[r < 1]                                            # 没有把前一段整段吃掉的回撤
    near = lambda x, levels, width: np.mean(np.min(np.abs(x[:, None] - np.array(levels)[None, :]), axis=1) <= width)
    return {"段数": len(r), "回撤 < 1 的比例": np.mean(r < 1), "其中在 1/3 到 2/3": np.mean((partial >= 1 / 3) & (partial <= 2 / 3)),
            "回撤离斐波那契比例 ≤ 0.02": near(partial, Pt.FIB_RETRACEMENTS, 0.02),
            "扩展离斐波那契比例 ≤ 0.05": near(e, Pt.FIB_EXTENSIONS, 0.05)}


rows = []
for name, df in datasets:
    real = ratio_summary(df["close"], thresholds[name])
    shuffled = pd.DataFrame([ratio_summary(shuffle_bars(df, rng)["close"], thresholds[name]) for _ in range(20)]).mean()
    rows.append({"数据": name, "口径": "真实"} | real)
    rows.append({"数据": name, "口径": "打乱 20 次平均"} | shuffled.to_dict())
table = pd.DataFrame(rows)
table["段数"] = table["段数"].round().astype(int)
print(table.to_string(index=False, float_format=lambda v: f"{v:.3f}"))

       数据        口径   段数  回撤 < 1 的比例  其中在 1/3 到 2/3  回撤离斐波那契比例 ≤ 0.02  扩展离斐波那契比例 ≤ 0.05
   SPY 日线        真实  205       0.502          0.524             0.252             0.049
   SPY 日线 打乱 20 次平均  215       0.500          0.441             0.252             0.072
  AAPL 日线        真实  249       0.502          0.440             0.256             0.073
  AAPL 日线 打乱 20 次平均  256       0.496          0.423             0.240             0.072
   BTC 日线        真实  342       0.482          0.382             0.255             0.059
   BTC 日线 打乱 20 次平均  364       0.502          0.423             0.240             0.069
BTC 4 小时线        真实 1927       0.513          0.395             0.235             0.066
BTC 4 小时线 打乱 20 次平均 2012       0.503          0.420             0.244             0.068
BTC 1 小时线        真实 7625       0.508          0.425             0.233             0.071
BTC 1 小时线 打乱 20 次平均 7777       0.499          0.429             0.242             0.067

fib = [1, 1]
while len(fib) < 16:
    fib.append(fib[-1] + fib[-2])
print("斐波那契数列：", fib)
print("相邻两项之比：", [round(fib[k + 1] / fib[k], 4) for k in range(9, 15)])
phi = (1 + 5 ** 0.5) / 2
print(f"黄金比例 φ = {phi:.6f}；1/φ = {1 / phi:.6f}；1/φ² = {1 / phi ** 2:.6f}；1/φ³ = {1 / phi ** 3:.6f}；"
      f"√(1/φ) = {(1 / phi) ** 0.5:.6f}；φ² = {phi ** 2:.6f}；√φ = {phi ** 0.5:.6f}")

斐波那契数列： [1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610, 987]
相邻两项之比： [1.6182, 1.618, 1.6181, 1.618, 1.618, 1.618]
黄金比例 φ = 1.618034；1/φ = 0.618034；1/φ² = 0.381966；1/φ³ = 0.236068；√(1/φ) = 0.786151；φ² = 2.618034；√φ = 1.272020

rows = []
spy_high, spy_low = spy.loc["2025-02-19", "high"], spy.loc["2024-08-05", "low"]
for label, a, b in [("收盘价 ZigZag 5%：2024-08-05 收盘 → 2025-02-19 收盘", start, end),
                    ("用最低价和最高价：2024-08-05 最低 → 2025-02-19 最高", spy_low, spy_high)]:
    rows.append({"起点和终点": label} | Pt.retracement_levels(a, b).round(2).rename(lambda r: f"{r:.1%}").to_dict())
for threshold in [0.03, 0.08]:
    sw = X.zigzag(known["close"], threshold)
    ups = [(sw["price"].iloc[k - 1], sw["price"].iloc[k], sw["time"].iloc[k - 1].date(), sw["time"].iloc[k].date())
           for k in range(1, len(sw)) if sw["kind"].iloc[k] == 1]
    a, b, ta, tb = ups[-1]
    rows.append({"起点和终点": f"收盘价 ZigZag {threshold:.0%}：{ta} → {tb}"}
                | Pt.retracement_levels(a, b).round(2).rename(lambda r: f"{r:.1%}").to_dict())
print(pd.DataFrame(rows).to_string(index=False))
print("假如回撤就停在 61.8%，下一段的扩展目标：")
print(Pt.extension_levels(start, end, levels[0.618]).round(2).to_string())

                                      起点和终点  23.6%  38.2%  50.0%  61.8%  78.6%
收盘价 ZigZag 5%：2024-08-05 收盘 → 2025-02-19 收盘 590.38 576.43 565.16 553.88 537.83
     用最低价和最高价：2024-08-05 最低 → 2025-02-19 最高 588.93 573.90 561.75 549.60 532.30
      收盘价 ZigZag 3%：2025-01-10 → 2025-02-19 605.27 600.54 596.71 592.88 587.43
      收盘价 ZigZag 8%：2024-08-05 → 2025-02-19 590.38 576.43 565.16 553.88 537.83
假如回撤就停在 61.8%，下一段的扩展目标：
1.272    675.42
1.618    708.48
2.618    804.03

after = spy.loc["2025-03-17":"2025-06-30"]
full_levels = Pt.retracement_levels(start, end)
for when in ["2025-03-25", "2025-04-03", "2025-04-08", "2025-04-09", "2025-05-12", "2025-06-27"]:
    close = spy.loc[when, "close"]
    nearest = (full_levels - close).abs().idxmin()
    print(f"{when}：收盘 {close:.2f}，离得最近的回撤位是 {nearest:.1%}（{full_levels[nearest]:.2f}），相差 {close / full_levels[nearest] - 1:+.2%}")
print(f"4 月 8 日收盘 {spy.loc['2025-04-08', 'close']:.2f}，比这一段的起点 {start:.2f} 还低 {spy.loc['2025-04-08', 'close'] / start - 1:.2%}；"
      f"第一次收盘回到 2 月 19 日 {end:.2f} 上方：{after.index[after['close'] > end][0].date()}")

2025-03-25：收盘 575.46，离得最近的回撤位是 38.2%（576.43），相差 -0.17%
2025-04-03：收盘 536.70，离得最近的回撤位是 78.6%（537.83），相差 -0.21%
2025-04-08：收盘 496.48，离得最近的回撤位是 78.6%（537.83），相差 -7.69%
2025-04-09：收盘 548.62，离得最近的回撤位是 61.8%（553.88），相差 -0.95%
2025-05-12：收盘 582.99，离得最近的回撤位是 38.2%（576.43），相差 +1.14%
2025-06-27：收盘 614.91，离得最近的回撤位是 23.6%（590.38），相差 +4.15%
4 月 8 日收盘 496.48，比这一段的起点 517.38 还低 -4.04%；第一次收盘回到 2 月 19 日 612.93 上方：2025-06-27

offsets = (-0.06, -0.03, 0.03, 0.06)
neighbors = {f: np.round(np.array(offsets) + f, 3) for f in Pt.FIB_RETRACEMENTS}          # 每个斐波那契比例左右各两个邻居
grid = np.unique(np.round(np.concatenate([np.arange(0.14, 0.8601, 0.01), Pt.FIB_RETRACEMENTS,
                                          *neighbors.values()]), 3))


def fib_versus_neighbors(tests, n=1000):
    """每个斐波那契比例的「停住」比例减去它四个邻居的平均，五个比例再取平均；按「段」重抽样 n 次给出 95% 区间和 p 值。

    和自己左右对称的邻居比，可以抵消「回撤越浅越容易停住」这种随比例平滑变化的趋势。
    """
    tests = tests.dropna(subset=["held"])
    wins = tests.pivot_table(index="end", columns="ratio", values="held", aggfunc="sum", fill_value=0)
    counts = tests.pivot_table(index="end", columns="ratio", values="held", aggfunc="count", fill_value=0)

    def stat(rows):
        rate = wins.iloc[rows].sum() / counts.iloc[rows].sum()
        return np.mean([rate[f] - rate[neighbors[f]].mean() for f in Pt.FIB_RETRACEMENTS])

    everyone = np.arange(len(wins))
    boot = np.array([stat(rng.integers(0, len(wins), len(wins))) for _ in range(n)])
    return stat(everyone), np.percentile(boot, [2.5, 97.5]), 2 * min((boot <= 0).mean(), (boot >= 0).mean())


rows, curves = [], {}
for name, df in datasets:
    atr = I.atr(df["high"], df["low"], df["close"])
    tests = Pt.retracement_tests(df["high"], df["low"], df["close"], X.zigzag(df["close"], thresholds[name]), atr, grid)
    curves[name] = tests.groupby("ratio")["held"].mean()
    diff, (low, high), p = fib_versus_neighbors(tests)
    by_ratio = tests.dropna(subset=["held"]).groupby("ratio")["held"]
    rows.append({"数据": name, "段数": tests["end"].nunique(), "碰到 61.8% 的次数": int(by_ratio.count().get(0.618, 0))}
                | {f"{r:.1%}": by_ratio.mean().get(r, np.nan) for r in Pt.FIB_RETRACEMENTS}
                | {"斐波那契 - 左右邻居": diff, "95% 区间": f"{low:+.3f} ~ {high:+.3f}", "p": p})
    if name == "BTC 1 小时线":
        hourly_tests = tests
print("回撤位被碰到之后「停住」的比例（先朝原方向走出 1 个 ATR）：")
print(pd.DataFrame(rows).to_string(index=False, float_format=lambda v: f"{v:.3f}"))
print("BTC 1 小时线，逐个比例和它的四个邻居比，再按 2022 年前后分开：")
rows = []
for label, part in [("全部", hourly_tests), ("2022 年前", hourly_tests[hourly_tests["touched_at"] < pd.Timestamp("2022-01-01", tz="UTC")]),
                    ("2022 年起", hourly_tests[hourly_tests["touched_at"] >= pd.Timestamp("2022-01-01", tz="UTC")])]:
    rate = part.dropna(subset=["held"]).groupby("ratio")["held"].mean()
    rows.append({"时段": label} | {f"{f:.1%}": rate[f] - rate[neighbors[f]].mean() for f in Pt.FIB_RETRACEMENTS})
print(pd.DataFrame(rows).to_string(index=False, float_format=lambda v: f"{v:+.3f}"))

回撤位被碰到之后「停住」的比例（先朝原方向走出 1 个 ATR）：
       数据   段数  碰到 61.8% 的次数  23.6%  38.2%  50.0%  61.8%  78.6%  斐波那契 - 左右邻居          95% 区间     p
   SPY 日线  103            53  0.583  0.625  0.472  0.377  0.375        0.083 -0.001 ~ +0.171 0.052
  AAPL 日线  118            62  0.500  0.400  0.500  0.516  0.435       -0.032 -0.116 ~ +0.046 0.454
   BTC 日线  168            73  0.750  0.588  0.460  0.370  0.440       -0.007 -0.094 ~ +0.072 0.832
BTC 4 小时线  849           409  0.667  0.518  0.444  0.482  0.541       -0.007 -0.039 ~ +0.026 0.694
BTC 1 小时线 3166          1455  0.709  0.552  0.513  0.524  0.560        0.015 +0.000 ~ +0.029 0.046
BTC 1 小时线，逐个比例和它的四个邻居比，再按 2022 年前后分开：
     时段  23.6%  38.2%  50.0%  61.8%  78.6%
     全部 +0.060 +0.017 +0.001 -0.002 -0.000
2022 年前 +0.084 +0.019 +0.003 -0.008 -0.001
2022 年起 +0.026 +0.015 -0.004 +0.004 -0.000

def sharpshooter(df, base, level_sets, anchors_per_threshold=2, width=0.25):
    """每个结束了一段回撤的摆动点 C：用 C 之前已经走完的若干段作起点和终点画回撤位，看有没有一条落在 C 的 width 个 ATR 以内。

    阈值取 base 的 0.6、1、1.6、2.5 倍；每个阈值取 C 之前最近的 anchors_per_threshold 段。
    返回「只用最近一段、一个阈值」和「所有段、所有阈值」两种画法下，命中的比例（对每组比例分别算）。
    """
    close = df["close"]
    atr = I.atr(df["high"], df["low"], close)
    turns = X.zigzag(close, base)
    candidates = {m: X.zigzag(close, base * m).sort_values("time").reset_index(drop=True) for m in (0.6, 1.0, 1.6, 2.5)}
    one, many, counted = np.zeros(len(level_sets)), np.zeros(len(level_sets)), 0
    for row in turns.itertuples():
        if np.isnan(atr[row.time]):
            continue
        legs = []
        for m, sw in candidates.items():
            earlier = sw[sw["time"] < row.time]
            for k in range(len(earlier) - 1, max(len(earlier) - 1 - anchors_per_threshold, 0), -1):
                a, b = earlier["price"].iloc[k - 1], earlier["price"].iloc[k]
                if (b - a) * (row.price - b) < 0:                  # 只要方向和这次回撤相反的那些段
                    legs.append((m, a, b))
        if not legs:
            continue
        counted += 1
        natural = [leg for leg in legs if leg[0] == 1.0][:1]
        for s, ratios in enumerate(level_sets):
            ratios = np.asarray(ratios)
            hit = lambda group: any(np.min(np.abs(b - ratios * (b - a) - row.price)) <= width * atr[row.time] for _, a, b in group)
            one[s] += hit(natural)
            many[s] += hit(legs)
    return one / counted, many / counted, counted


shifts = [d for d in np.round(np.arange(-0.06, 0.0601, 0.01), 2) if d != 0]
shifted_sets = [np.array(Pt.FIB_RETRACEMENTS) + d for d in shifts]          # 同样的间距，整体平移
rows = []
for name, df in datasets[:4]:
    one, many, counted = sharpshooter(df, thresholds[name], [Pt.FIB_RETRACEMENTS] + shifted_sets)
    rows.append({"数据": name, "转折点": counted, "只画一次：斐波那契": one[0], "只画一次：平移后的比例": one[1:].mean(),
                 "随便挑起点终点：斐波那契": many[0], "随便挑起点终点：平移后的比例": many[1:].mean(),
                 "平移的 12 组里不低于斐波那契的": int((many[1:] >= many[0]).sum())})
print("转折点离某条回撤位 0.25 个 ATR 以内的比例：")
print(pd.DataFrame(rows).to_string(index=False, float_format=lambda v: f"{v:.3f}"))

转折点离某条回撤位 0.25 个 ATR 以内的比例：
       数据  转折点  只画一次：斐波那契  只画一次：平移后的比例  随便挑起点终点：斐波那契  随便挑起点终点：平移后的比例  平移的 12 组里不低于斐波那契的
   SPY 日线  206      0.340        0.337         0.534           0.524                  1
  AAPL 日线  250      0.356        0.350         0.564           0.546                  3
   BTC 日线  342      0.383        0.382         0.558           0.564                  9
BTC 4 小时线 1927      0.389        0.392         0.593           0.604                 12

# ---------------------------------------------------------------------------
# 五、斐波那契回撤与扩展（第 18 篇）
# ---------------------------------------------------------------------------

FIB_RETRACEMENTS = (0.236, 0.382, 0.5, 0.618, 0.786)
FIB_EXTENSIONS = (1.272, 1.618, 2.618)


def retracement_levels(start: float, end: float, ratios=FIB_RETRACEMENTS) -> pd.Series:
    """一段行情从 start 走到 end，回撤 ratio 的价位 = end - ratio × (end - start)。

    上涨段（end > start）的回撤位在 end 下方，下跌段在 end 上方，公式相同。返回以 ratio 为索引的价位。
    """
    ratios = np.asarray(ratios, dtype=float)
    return pd.Series(end - ratios * (end - start), index=ratios, name="level")


def extension_levels(a: float, b: float, c: float, ratios=FIB_EXTENSIONS) -> pd.Series:
    """A 走到 B、回撤到 C 之后，下一段的扩展目标 = C + ratio × (B - A)。返回以 ratio 为索引的价位。"""
    ratios = np.asarray(ratios, dtype=float)
    return pd.Series(c + ratios * (b - a), index=ratios, name="level")


def swing_ratios(swings: pd.DataFrame) -> pd.DataFrame:
    """每个摆动点结束的那一段，相对前面几段的比例。

    retracement：这一段的长度 ÷ 前一段的长度（前一段被回撤了多少，1 表示回撤到起点）
    extension：这一段的长度 ÷ 前面隔一段的那一段的长度（A→B、B→C、C→D 里的 CD ÷ AB）
    比例都用摆动点的价格算，第一、二个摆动点没有前一段，为 NaN。
    """
    ordered = swings.sort_values("time").reset_index(drop=True)
    p = ordered["price"].to_numpy(dtype=float)
    leg = np.abs(np.diff(p, prepend=np.nan))                 # 第 n 个摆动点结束的那一段的长度
    retracement = leg / np.roll(leg, 1)
    extension = leg / np.roll(leg, 2)
    retracement[:2], extension[:3] = np.nan, np.nan
    return pd.DataFrame({"time": ordered["time"], "kind": ordered["kind"], "retracement": retracement,
                         "extension": extension})


def retracement_tests(high: pd.Series, low: pd.Series, close: pd.Series, swings: pd.DataFrame, atr: pd.Series,
                      ratios, distance: float = 1.0, horizon: int = 20) -> pd.DataFrame:
    """实时地检验回撤位：每一段 A→B 被确认之后，价格第一次回撤到某个比例的价位时，看它是「停住」还是「穿过」。

    对每两个相邻的摆动点 A、B（B 在 confirmed_at 被确认）：
    1. 各个比例的价位 = B - ratio × (B - A)
    2. 从 B 到 confirmed_at 之间已经碰过的价位不算：那时还不知道这一段存在
    3. 从 confirmed_at 的下一根开始往后看，收盘价越过 B（新高，上涨段）或越过 A（整段被吃掉）就停止
    4. 某个价位第一次被碰到（上涨段看最低价 ≤ 价位）的那一根，记为 touched_at
    5. 从这个价位出发，上下各放 distance 个 ATR（用 confirmed_at 那一根的 ATR）。
       之后 horizon 根里，先朝原来方向走出 distance 个 ATR 记 1（停住了），先反向走出记 0（穿过了），都没有记 NaN。
       碰到价位的那一根自己就走过了反向的线，记 0
    返回每一次「碰到」一行：start、end、confirmed_at、ratio、level、touched_at、held。
    """
    h, l, c, a = (s.to_numpy(dtype=float) for s in (high, low, close, atr))
    pos = {t: i for i, t in enumerate(close.index)}
    ordered = swings.sort_values("time").reset_index(drop=True)
    ratios = np.asarray(ratios, dtype=float)
    rows = []
    for n in range(1, len(ordered)):
        start, end = ordered["price"].iloc[n - 1], ordered["price"].iloc[n]
        sign = 1 if end > start else -1                        # 1：上涨段，回撤向下
        seen = pos[ordered["confirmed_at"].iloc[n]]
        if np.isnan(a[seen]):
            continue
        levels = end - ratios * (end - start)
        width = distance * a[seen]
        k0 = pos[ordered["time"].iloc[n]]
        reached = (l[k0:seen + 1].min() if sign == 1 else h[k0:seen + 1].max())
        done = sign * (reached - levels) <= 0
        for j in range(seen + 1, len(c)):
            if sign * (c[j] - end) > 0 or sign * (c[j] - start) < 0 or done.all():
                break
            extreme = l[j] if sign == 1 else h[j]
            hit = ~done & (sign * (extreme - levels) <= 0)
            for g in np.flatnonzero(hit):
                level, held = levels[g], np.nan
                if sign * (extreme - (level - sign * width)) <= 0:
                    held = 0.0
                else:
                    for m in range(j + 1, min(j + 1 + horizon, len(c))):
                        if sign * ((l[m] if sign == 1 else h[m]) - (level - sign * width)) <= 0:
                            held = 0.0
                            break
                        if sign * ((h[m] if sign == 1 else l[m]) - (level + sign * width)) >= 0:
                            held = 1.0
                            break
                rows.append((ordered["time"].iloc[n - 1], ordered["time"].iloc[n], ordered["confirmed_at"].iloc[n],
                             ratios[g], level, close.index[j], held))
            done |= hit
    return pd.DataFrame(rows, columns=["start", "end", "confirmed_at", "ratio", "level", "touched_at", "held"])

# ---------------------------------------------------------------------------
# 斐波那契回撤与扩展（第 18 篇）
# ---------------------------------------------------------------------------

def test_retracement_and_extension_levels_by_hand():
    up = Pt.retracement_levels(100, 200)
    assert up.tolist() == pytest.approx([176.4, 161.8, 150, 138.2, 121.4])
    down = Pt.retracement_levels(200, 100)                                    # 下跌段的回撤位在上方
    assert down.tolist() == pytest.approx([123.6, 138.2, 150, 161.8, 178.6])
    assert Pt.extension_levels(100, 200, 150).tolist() == pytest.approx([277.2, 311.8, 411.8])


def test_swing_ratios_by_hand():
    close = pd.Series([100.0, 200, 150, 300, 225, 240], index=pd.date_range("2024-01-01", periods=6, freq="D"))
    swings = manual_swings(close, [(0, -1), (1, 1), (2, -1), (3, 1), (4, -1)])
    ratios = Pt.swing_ratios(swings)
    # 段长：100、50、150、75。回撤：50/100、150/50、75/150；扩展：150/100、75/50
    assert ratios["retracement"].tolist()[2:] == pytest.approx([0.5, 3.0, 0.5])
    assert ratios["extension"].tolist()[3:] == pytest.approx([1.5, 1.5])
    assert ratios["retracement"].iloc[:2].isna().all() and ratios["extension"].iloc[:3].isna().all()


def retracement_bars(after):
    """0 到 10 根从 100 涨到 200；第 11、12 根回落（第 12 根最低 175），之后按 after 给出的 (最高, 最低, 收盘)。"""
    rows = [(p, p, p) for p in np.linspace(100, 200, 11)] + [(200, 184, 185), (186, 175, 180)] + list(after)
    index = pd.date_range("2024-01-01", periods=len(rows), freq="D")
    frame = pd.DataFrame(rows, columns=["high", "low", "close"], index=index)
    swings = pd.DataFrame([(index[0], 100.0, -1, index[1]), (index[10], 200.0, 1, index[12])], columns=X.SWING_COLUMNS)
    return frame, swings, pd.Series(5.0, index=index)


def test_retracement_tests_by_hand():
    # 23.6% 的价位 176.4 在第 12 根（确认那一根）之前就碰过了，不算；38.2% 的价位 161.8 在第 14 根第一次被碰到
    frame, swings, atr = retracement_bars([(172, 168, 170), (165, 161, 163), (167, 162, 166), (170, 166, 169)])
    found = Pt.retracement_tests(frame["high"], frame["low"], frame["close"], swings, atr, [0.236, 0.382, 0.5])
    assert found["ratio"].tolist() == [0.382]
    assert found["touched_at"].iloc[0] == frame.index[14] and found["level"].iloc[0] == pytest.approx(161.8)
    assert found["held"].iloc[0] == 1.0                                       # 第 15 根最高 167 ≥ 161.8 + 5
    frame, swings, atr = retracement_bars([(172, 168, 170), (165, 161, 163), (164, 156, 158)])
    found = Pt.retracement_tests(frame["high"], frame["low"], frame["close"], swings, atr, [0.382])
    assert found["held"].iloc[0] == 0.0                                       # 第 15 根最低 156 ≤ 161.8 - 5


def test_retracement_tests_down_leg_mirror():
    frame, swings, atr = retracement_bars([(172, 168, 170), (165, 161, 163), (167, 162, 166), (170, 166, 169)])
    flipped = pd.DataFrame({"high": -frame["low"], "low": -frame["high"], "close": -frame["close"]})
    flipped_swings = swings.assign(price=-swings["price"], kind=-swings["kind"])
    up = Pt.retracement_tests(frame["high"], frame["low"], frame["close"], swings, atr, [0.236, 0.382, 0.5])
    down = Pt.retracement_tests(flipped["high"], flipped["low"], flipped["close"], flipped_swings, atr, [0.236, 0.382, 0.5])
    assert down["touched_at"].tolist() == up["touched_at"].tolist() and down["held"].tolist() == up["held"].tolist()
    assert down["level"].tolist() == pytest.approx((-up["level"]).tolist())


def test_retracement_tests_never_use_the_future():
    rng = np.random.default_rng(180)
    close = pd.Series(100 * np.exp(np.cumsum(rng.normal(0, 0.02, 1500))), index=pd.date_range("2020-01-01", periods=1500, freq="D"))
    high, low = close * (1 + rng.uniform(0, 0.01, 1500)), close * (1 - rng.uniform(0, 0.01, 1500))
    atr = pd.Series(0.02 * close.to_numpy(), index=close.index)
    ratios = [0.382, 0.5, 0.618]
    full = Pt.retracement_tests(high, low, close, X.zigzag(close, 0.05), atr, ratios)
    assert len(full) > 20
    for k in [700, 1200]:
        part = Pt.retracement_tests(high.iloc[:k], low.iloc[:k], close.iloc[:k], X.zigzag(close.iloc[:k], 0.05), atr.iloc[:k], ratios)
        settled = lambda frame: frame[frame["touched_at"] <= close.index[k - 1 - 20]].reset_index(drop=True)
        pd.testing.assert_frame_equal(settled(part), settled(full))

pytest -q

........................................................................ [ 91%]
.............                                                            [100%]
157 passed in 0.61s