第 17 篇：经典图表形态

t = pd.Timestamp("2025-01-08", tz="UTC")
known = day.loc[:t]                                           # 只用决策当天收盘为止的数据
close = known["close"]
swings = X.zigzag(close, 0.05)
print(swings[swings["time"] >= "2024-11-01"].to_string(index=False))
found = Pt.head_and_shoulders(close, swings)
row = found.iloc[-1]
print(row[["left", "neck_1", "head", "neck_2", "right", "confirmed_at", "left_price", "head_price", "right_price",
           "height"]].to_string())
pos = {x: k for k, x in enumerate(close.index)}
slope = (swings.set_index("time")["price"][row["neck_2"]] - swings.set_index("time")["price"][row["neck_1"]]) / \
        (pos[row["neck_2"]] - pos[row["neck_1"]])
neck_today = swings.set_index("time")["price"][row["neck_1"]] + slope * (pos[t] - pos[row["neck_1"]])
print(f"颈线每天抬高 {slope:,.2f}；1 月 8 日的颈线 {neck_today:,.2f}，收盘 {close[t]:,.2f}，高出 {close[t] / neck_today - 1:.2%}")
print(f"如果跌破，量度目标约为 {neck_today - row['height']:,.0f}（颈线减去高度 {row['height']:,.2f}）")

                     time     price  kind              confirmed_at
2024-11-04 00:00:00+00:00  67850.01    -1 2024-11-06 00:00:00+00:00
2024-11-22 00:00:00+00:00  98892.00     1 2024-11-25 00:00:00+00:00
2024-11-26 00:00:00+00:00  91965.16    -1 2024-11-29 00:00:00+00:00
2024-12-17 00:00:00+00:00 106133.74     1 2024-12-18 00:00:00+00:00
2024-12-30 00:00:00+00:00  92792.05    -1 2025-01-03 00:00:00+00:00
2025-01-06 00:00:00+00:00 102235.60     1 2025-01-07 00:00:00+00:00
left            2024-11-22 00:00:00+00:00
neck_1          2024-11-26 00:00:00+00:00
head            2024-12-17 00:00:00+00:00
neck_2          2024-12-30 00:00:00+00:00
right           2025-01-06 00:00:00+00:00
confirmed_at    2025-01-07 00:00:00+00:00
left_price                        98892.0
head_price                      106133.74
right_price                      102235.6
height                       13657.853824
颈线每天抬高 24.32；1 月 8 日的颈线 93,010.93，收盘 95,060.61，高出 2.20%
如果跌破，量度目标约为 79,353（颈线减去高度 13,657.85）

points = [(0, 100), (10, 110), (16, 102), (28, 120), (36, 104), (44, 112), (60, 96)]
path = np.interp(np.arange(61), [p[0] for p in points], [p[1] for p in points])
toy = pd.Series(path, index=pd.date_range("2024-01-01", periods=61, freq="D"))
toy_swings = X.zigzag(toy, 0.05)
print(toy_swings.to_string(index=False))
toy_found = Pt.head_and_shoulders(toy, toy_swings)
print(toy_found.drop(columns=["left", "neck_1", "neck_2"]).round(2).T.to_string())

      time  price  kind confirmed_at
2024-01-01  100.0    -1   2024-01-06
2024-01-11  110.0     1   2024-01-16
2024-01-17  102.0    -1   2024-01-21
2024-01-29  120.0     1   2024-02-01
2024-02-06  104.0    -1   2024-02-12
2024-02-14  112.0     1   2024-02-20
                                     0
head               2024-01-29 00:00:00
right              2024-02-14 00:00:00
confirmed_at       2024-02-20 00:00:00
left_price                       110.0
head_price                       120.0
right_price                      112.0
neckline                         105.4
height                            16.8
broken_at          2024-02-21 00:00:00
neckline_at_break                105.5
target                            88.7

rows = []
for threshold in [0.03, 0.05, 0.08, 0.10, 0.15]:
    sw = X.zigzag(day["close"], threshold)
    hs_top, hs_bottom = Pt.head_and_shoulders(day["close"], sw), Pt.head_and_shoulders(day["close"], sw, "bottom")
    rows.append({"ZigZag 阈值": f"{threshold:.0%}", "摆动点": len(sw), "头肩顶": len(hs_top), "头肩底": len(hs_bottom),
                 "双顶": len(Pt.double_tops(day["close"], sw)), "双底": len(Pt.double_tops(day["close"], sw, "bottom")),
                 "三角形和楔形": int(Pt.converging(day["close"], sw)["name"].str.contains("三角|楔").sum()),
                 "决策点的头肩顶在不在": bool(((hs_top["head"] == pd.Timestamp("2024-12-17", tz="UTC"))
                                              & (hs_top["right"] == pd.Timestamp("2025-01-06", tz="UTC"))).any())})
print("BTC 日线，换 ZigZag 阈值：")
print(pd.DataFrame(rows).to_string(index=False))
rows = []
sw = X.zigzag(day["close"], 0.05)
for tol in [0.1, 0.2, 0.3, 0.5, 1.0]:
    rows.append({"两肩高度差上限（占高度）": tol, "头肩顶": len(Pt.head_and_shoulders(day["close"], sw, shoulder_tol=tol)),
                 "双顶": len(Pt.double_tops(day["close"], sw, tol=tol))})
print("BTC 日线，ZigZag 5%，换两肩（两顶）的高度差上限：")
print(pd.DataFrame(rows).to_string(index=False))
base = Pt.head_and_shoulders(h4["close"], X.zigzag(h4["close"], 0.02))
for threshold in [0.015, 0.025, 0.03]:
    other = Pt.head_and_shoulders(h4["close"], X.zigzag(h4["close"], threshold))
    same = base["head"].isin(other["head"]).mean()
    print(f"BTC 4 小时线：阈值 2% 找到的 {len(base)} 个头肩顶里，阈值换成 {threshold:.1%} 还能找到（头在同一根 K 线）的占 {same:.1%}；"
          f"那边一共 {len(other)} 个")

BTC 日线，换 ZigZag 阈值：
ZigZag 阈值  摆动点  头肩顶  头肩底  双顶  双底  三角形和楔形  决策点的头肩顶在不在
       3%  585   21   20  25  28     295       False
       5%  344   15   11  15  21     180        True
       8%  219    7    3   5  17     103       False
      10%  159    6    2   6  10      71       False
      15%   85    2    1   1   5      43       False
BTC 日线，ZigZag 5%，换两肩（两顶）的高度差上限：
 两肩高度差上限（占高度）  头肩顶  双顶
          0.1    4  15
          0.2   11  30
          0.3   15  49
          0.5   18  90
          1.0   23 171
BTC 4 小时线：阈值 2% 找到的 76 个头肩顶里，阈值换成 1.5% 还能找到（头在同一根 K 线）的占 73.7%；那边一共 91 个
BTC 4 小时线：阈值 2% 找到的 76 个头肩顶里，阈值换成 2.5% 还能找到（头在同一根 K 线）的占 68.4%；那边一共 63 个
BTC 4 小时线：阈值 2% 找到的 76 个头肩顶里，阈值换成 3.0% 还能找到（头在同一根 K 线）的占 47.4%；那边一共 57 个

SPY 日线：NATR 中位数 1.09%
AAPL 日线：NATR 中位数 2.08%
BTC 日线：NATR 中位数 4.40%
BTC 4 小时线：NATR 中位数 1.64%
BTC 1 小时线：NATR 中位数 0.79%

c = day["close"]
i = c.index.get_loc(t)
for n in [1, 2, 5, 12, 30, 61]:
    print(f"{n} 天后（{c.index[i + n].date()}）：收盘 {c.iloc[i + n]:,.2f}（{c.iloc[i + n] / c[t] - 1:+.2%}）")
full = Pt.head_and_shoulders(c, X.zigzag(c, 0.05))
pattern = full[full["head"] == pd.Timestamp("2024-12-17", tz="UTC")].iloc[0]
print(f"跌破颈线：{pattern['broken_at'].date()}，那天的颈线 {pattern['neckline_at_break']:,.2f}，收盘 {c[pattern['broken_at']]:,.2f}；"
      f"量度目标 {pattern['target']:,.2f}")
after = day.loc[pattern["broken_at"]:].iloc[1:]
over = after[after["close"] > pattern["right_price"]].index[0]
print(f"跌破之后，收盘价第一次回到右肩 {pattern['right_price']:,.2f} 上方：{over.date()}（收盘 {c[over]:,.2f}）；"
      f"之后到 1 月 31 日的最高价 {day['high'].loc[pattern['broken_at']:pd.Timestamp('2025-01-31', tz='UTC')].max():,.2f}")
reach = after[after["close"] <= pattern["target"]].index[0]
print(f"收盘价第一次到达量度目标：{reach.date()}（收盘 {c[reach]:,.2f}），跌破颈线之后第 {c.index.get_loc(reach) - c.index.get_loc(pattern['broken_at'])} 天")

1 天后（2025-01-09）：收盘 92,552.49（-2.64%）
2 天后（2025-01-10）：收盘 94,726.11（-0.35%）
5 天后（2025-01-13）：收盘 94,536.10（-0.55%）
12 天后（2025-01-20）：收盘 102,260.01（+7.57%）
30 天后（2025-02-07）：收盘 96,506.80（+1.52%）
61 天后（2025-03-10）：收盘 78,595.86（-17.32%）
跌破颈线：2025-01-09，那天的颈线 93,035.25，收盘 92,552.49；量度目标 79,377.40
跌破之后，收盘价第一次回到右肩 102,235.60 上方：2025-01-17（收盘 104,077.48）；之后到 1 月 31 日的最高价 109,588.00
收盘价第一次到达量度目标：2025-03-10（收盘 78,595.86），跌破颈线之后第 60 天

def trade(close, high, low, start, stop, target, sign, horizon=60):
    """在第 start 根收盘价做空（sign = 1）或做多（sign = -1），之后 horizon 根里先碰到目标记 1，先碰到止损记 0。

    同一根 K 线两个都碰到，分不清先后，按先碰到止损算；horizon 根里都没碰到记 NaN。
    """
    for j in range(start + 1, min(start + 1 + horizon, len(close))):
        hit_target = low[j] <= target if sign == 1 else high[j] >= target
        hit_stop = high[j] >= stop if sign == 1 else low[j] <= stop
        if hit_stop:
            return 0.0
        if hit_target:
            return 1.0
    return np.nan


def binomial_p(k, n, p0):
    """二项检验的双侧 p 值：成功概率为 p0 时，出现概率不高于「恰好 k 次」的所有结果的概率之和。"""
    probs = [math.comb(n, i) * p0 ** i * (1 - p0) ** (n - i) for i in range(n + 1)]
    return sum(p for p in probs if p <= probs[k] * (1 + 1e-9))


def plain_breaks(close, threshold, sign):
    """「普通的跌破」：收盘价第一次跌破最近一个已确认的摆动低点的那一根（sign = -1 时是升破高点）。"""
    levels = X.trend_state(X.zigzag(close, threshold), close, close)
    line = levels["last_low"] if sign == 1 else levels["last_high"]
    crossed = (sign * (close - line) < 0) & (sign * (close.shift(1) - line.shift(1)) >= 0)
    return np.flatnonzero(crossed.to_numpy())


kinds = [("头肩顶", Pt.head_and_shoulders, "top"), ("头肩底", Pt.head_and_shoulders, "bottom"),
         ("双顶", Pt.double_tops, "top"), ("双底", Pt.double_tops, "bottom")]
rows_a, rows_b = [], []
for name, df in datasets:
    c = df["close"]
    high, low, close_ = df["high"].to_numpy(float), df["low"].to_numpy(float), c.to_numpy(float)
    sw = X.zigzag(c, thresholds[name])
    atr = I.atr(df["high"], df["low"], c).to_numpy()
    where = {x: k for k, x in enumerate(c.index)}
    for label, func, kind in kinds:
        sign = 1 if kind == "top" else -1
        stop_column = "right_price" if func is Pt.head_and_shoulders else "second_price"
        found = func(c, sw, kind)
        if found.empty:
            continue
        target = found["neckline"] - sign * found["height"]                   # 决策时刻就能算出的目标
        at_seen = np.array([trade(close_, high, low, where[x], s, g, sign)
                            for x, s, g in zip(found["confirmed_at"], found[stop_column], target)])
        broken = found["broken_at"].notna().to_numpy()
        rows_a.append({"数据": name, "形态": label, "候选": len(found), "后来突破": int(broken.sum()),
                       "决策时刻入场 全部候选": np.nanmean(at_seen), "只算后来突破的": np.nanmean(at_seen[broken])})
        plain = plain_breaks(c, thresholds[name], sign)
        plain = plain[~np.isnan(atr[plain])]
        wins, need, plain_rate, r_multiple, when = [], [], [], [], []
        for x in found[broken].itertuples():
            b = where[x.broken_at]
            stop = getattr(x, stop_column)
            goal = x.target
            risk, reward = abs(stop - close_[b]), abs(close_[b] - goal)
            outcome = trade(close_, high, low, b, stop, goal, sign)
            if np.isnan(outcome) or risk == 0:
                continue
            wins.append(outcome)
            when.append(x.broken_at)
            need.append(risk / (risk + reward))
            r_multiple.append(reward / risk if outcome == 1 else -1.0)
            risk_atr, reward_atr = risk / atr[b], reward / atr[b]                 # 对照用 ATR 为单位的同样距离
            starts = rng.choice(plain, 30)
            plain_rate.append(np.nanmean([trade(close_, high, low, s, close_[s] + sign * risk_atr * atr[s],
                                                close_[s] - sign * reward_atr * atr[s], sign) for s in starts]))
        if wins:
            k, n = int(sum(wins)), len(wins)
            early = np.array(when) < pd.Timestamp("2022-01-01", tz=c.index.tz)
            rows_b.append({"数据": name, "形态": label, "交易": n, "胜率": k / n, "盈亏平衡胜率": np.mean(need),
                           "平均盈亏（R）": np.mean(r_multiple), "普通跌破 同样距离": np.mean(plain_rate),
                           "p": binomial_p(k, n, float(np.mean(plain_rate))),
                           "2022 年前 胜率": np.mean(np.array(wins)[early]) if early.any() else np.nan,
                           "2022 年起 胜率": np.mean(np.array(wins)[~early]) if (~early).any() else np.nan})
table_a = pd.DataFrame(rows_a)
print("按教科书的止损（右肩或第二个顶外侧）和量度目标，先碰到目标的比例：")
print(table_a.to_string(index=False, float_format=lambda v: f"{v:.3f}"))
table_b = pd.DataFrame(rows_b)
print("突破颈线的那根收盘入场，止损和目标同上：")
print(table_b.to_string(index=False, float_format=lambda v: f"{v:.3f}"))

按教科书的止损（右肩或第二个顶外侧）和量度目标，先碰到目标的比例：
       数据  形态  候选  后来突破  决策时刻入场 全部候选  只算后来突破的
   SPY 日线 头肩顶   5     5        0.400    0.400
   SPY 日线 头肩底   2     2        0.000    0.000
   SPY 日线  双顶  14     7        0.500    0.857
   SPY 日线  双底  13     8        0.545    0.833
  AAPL 日线 头肩顶   5     5        0.200    0.200
  AAPL 日线 头肩底   6     5        0.600    0.750
  AAPL 日线  双顶   4     3        0.250    0.333
  AAPL 日线  双底  17    11        0.438    0.545
   BTC 日线 头肩顶  15    12        0.333    0.417
   BTC 日线 头肩底  11    10        0.636    0.700
   BTC 日线  双顶  15     6        0.333    0.667
   BTC 日线  双底  21    13        0.476    0.692
BTC 4 小时线 头肩顶  76    60        0.294    0.370
BTC 4 小时线 头肩底  56    44        0.320    0.359
BTC 4 小时线  双顶  83    33        0.388    0.871
BTC 4 小时线  双底 101    46        0.258    0.452
BTC 1 小时线 头肩顶 256   189        0.393    0.517
BTC 1 小时线 头肩底 235   174        0.362    0.476
BTC 1 小时线  双顶 345   171        0.422    0.683
BTC 1 小时线  双底 339   172        0.424    0.687

突破颈线的那根收盘入场，止损和目标同上：
       数据  形态  交易    胜率  盈亏平衡胜率  平均盈亏（R）  普通跌破 同样距离     p  2022 年前 胜率  2022 年起 胜率
   SPY 日线 头肩顶   5 0.400   0.451   -0.323      0.393 1.000       1.000       0.250
   SPY 日线 头肩底   2 0.000   0.448   -1.000      0.626 0.140       0.000       0.000
   SPY 日线  双顶   7 0.857   0.648    0.352      0.558 0.142       0.800       1.000
   SPY 日线  双底   6 0.833   0.624    0.343      0.741 1.000       1.000       0.750
  AAPL 日线 头肩顶   5 0.600   0.346    0.883      0.304 0.168       0.333       1.000
  AAPL 日线 头肩底   4 0.750   0.469    0.831      0.471 0.349       0.750         NaN
  AAPL 日线  双顶   3 0.333   0.573   -0.537      0.567 0.583       0.500       0.000
  AAPL 日线  双底  11 0.636   0.601    0.056      0.668 0.760       0.333       1.000
   BTC 日线 头肩顶  12 0.500   0.442    0.187      0.342 0.361       0.600       0.429
   BTC 日线 头肩底  10 0.700   0.519    0.342      0.514 0.346       0.750       0.667
   BTC 日线  双顶   6 0.667   0.619    0.059      0.586 1.000       0.333       1.000
   BTC 日线  双底  13 0.692   0.606    0.125      0.570 0.417       0.667       0.714
BTC 4 小时线 头肩顶  55 0.400   0.428   -0.153      0.440 0.589       0.346       0.448
BTC 4 小时线 头肩底  38 0.421   0.465   -0.067      0.515 0.260       0.318       0.562
BTC 4 小时线  双顶  32 0.844   0.705    0.204      0.710 0.119       0.917       0.800
BTC 4 小时线  双底  43 0.512   0.622   -0.162      0.656 0.054       0.586       0.357
BTC 1 小时线 头肩顶 182 0.549   0.471    0.214      0.456 0.014       0.636       0.391
BTC 1 小时线 头肩底 163 0.540   0.466    0.237      0.478 0.117       0.531       0.554
BTC 1 小时线  双顶 162 0.741   0.664    0.136      0.646 0.011       0.787       0.648
BTC 1 小时线  双底 162 0.722   0.653    0.101      0.658 0.097       0.709       0.746

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]
    if f.sum() == 0 or (~f).sum() == 0:
        return np.nan, np.nan
    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 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


names = ["上升三角形", "下降三角形", "对称三角形", "上升楔形", "下降楔形", "矩形", "通道", "扩散"]


def up_share(df, threshold):
    found = Pt.converging(df["close"], X.zigzag(df["close"], threshold))
    found = found[found["broken_at"].notna()]
    return found.groupby("name")["direction"].apply(lambda d: (d == 1).mean()).reindex(names)


rows = []
for name, df in datasets:
    c = df["close"]
    a = I.atr(df["high"], df["low"], c)
    found = Pt.converging(c, X.zigzag(c, thresholds[name]))
    found = found[found["broken_at"].notna() & a.reindex(found["broken_at"]).notna().to_numpy()]
    found["顺向"] = X.first_passage(c, df["high"], df["low"], a, found["broken_at"], found["direction"])
    shuffled = pd.concat([up_share(shuffle_bars(df, rng), thresholds[name]) for _ in range(10)], axis=1).mean(axis=1)
    for shape in names:
        m = (found["name"] == shape).to_numpy()
        _, p = label_test(found["顺向"], m)
        rows.append({"数据": name, "形态": shape, "突破": int(m.sum()), "向上突破": (found["direction"][m] == 1).mean(),
                     "打乱后 向上突破": shuffled[shape], "顺着突破方向": found["顺向"][m].mean(),
                     "其他形态的突破": found["顺向"][~m].mean(), "p": p})
table_c = pd.DataFrame(rows)
print(table_c.to_string(index=False, float_format=lambda v: f"{v:.3f}"))
print(f"「顺着突破方向」一共 {table_c['p'].notna().sum()} 次检验，p < 0.05 的有 {(table_c['p'] < 0.05).sum()} 次")

       数据    形态   突破  向上突破  打乱后 向上突破  顺着突破方向  其他形态的突破     p
   SPY 日线 上升三角形   15 0.267     0.348   0.214    0.514 0.048
   SPY 日线 下降三角形   19 0.526     0.727   0.500    0.486 1.000
   SPY 日线 对称三角形   13 0.538     0.404   0.333    0.500 0.385
   SPY 日线  上升楔形   34 0.176     0.229   0.394    0.512 0.242
   SPY 日线  下降楔形   19 0.895     0.809   0.722    0.457 0.043
   SPY 日线    矩形    9 0.778     0.774   0.875    0.467 0.034
   SPY 日线    通道   49 0.327     0.440   0.458    0.500 0.729
   SPY 日线    扩散    9 0.333     0.674   0.857    0.470 0.055
  AAPL 日线 上升三角形   12 0.417     0.320   0.667    0.494 0.499
  AAPL 日线 下降三角形   24 0.792     0.715   0.591    0.491 0.511
  AAPL 日线 对称三角形   16 0.625     0.497   0.600    0.494 0.602
  AAPL 日线  上升楔形   44 0.227     0.278   0.310    0.558 0.005
  AAPL 日线  下降楔形   29 0.724     0.852   0.429    0.516 0.412
  AAPL 日线    矩形    1 1.000     0.701   0.000    0.505 0.503
  AAPL 日线    通道   67 0.463     0.426   0.621    0.450 0.046
  AAPL 日线    扩散   15 0.533     0.687   0.429    0.509 0.582
   BTC 日线 上升三角形   20 0.450     0.355   0.556    0.580 1.000
   BTC 日线 下降三角形   40 0.725     0.699   0.733    0.554 0.082
   BTC 日线 对称三角形   18 0.500     0.484   0.538    0.580 0.778
   BTC 日线  上升楔形   48 0.188     0.235   0.537    0.587 0.593
   BTC 日线  下降楔形   53 0.792     0.803   0.575    0.578 1.000
   BTC 日线    矩形    6 0.333     0.444   0.400    0.582 0.649
   BTC 日线    通道   66 0.515     0.466   0.545    0.588 0.633
   BTC 日线    扩散   28 0.500     0.625   0.609    0.574 0.837
BTC 4 小时线 上升三角形  144 0.347     0.341   0.514    0.530 0.763
BTC 4 小时线 下降三角形  180 0.689     0.681   0.593    0.520 0.110
BTC 4 小时线 对称三角形  137 0.438     0.479   0.471    0.534 0.260
BTC 4 小时线  上升楔形  252 0.214     0.233   0.467    0.540 0.064
BTC 4 小时线  下降楔形  233 0.781     0.816   0.547    0.525 0.619
BTC 4 小时线    矩形   24 0.458     0.466   0.400    0.531 0.257
BTC 4 小时线    通道  491 0.487     0.489   0.553    0.517 0.257
BTC 4 小时线    扩散  105 0.505     0.532   0.525    0.529 1.000
BTC 1 小时线 上升三角形  622 0.310     0.328   0.487    0.532 0.063
BTC 1 小时线 下降三角形  634 0.662     0.688   0.531    0.527 0.885
BTC 1 小时线 对称三角形  575 0.489     0.509   0.537    0.526 0.688
BTC 1 小时线  上升楔形  953 0.168     0.208   0.544    0.524 0.326
BTC 1 小时线  下降楔形  916 0.784     0.808   0.498    0.533 0.090
BTC 1 小时线    矩形   86 0.477     0.502   0.515    0.528 0.896
BTC 1 小时线    通道 1852 0.487     0.493   0.537    0.523 0.357
BTC 1 小时线    扩散  463 0.464     0.510   0.551    0.525 0.336
「顺着突破方向」一共 40 次检验，p < 0.05 的有 5 次

# ---------------------------------------------------------------------------
# 四、图表形态（第 17 篇）
# ---------------------------------------------------------------------------
#
# 图表形态由摆动点（structure.zigzag 或 structure.fractals）连成。每个函数返回全部「候选形态」：
# 最后一个摆动点被确认的那一刻（confirmed_at），形状条件已经满足，就记一行，不管之后有没有突破。
# broken_at 是之后第一次收盘越过颈线（或趋势线）的时刻，等不到就是 NaT。
# 这样既能统计「成立的形态」，也能统计「当时看得到、后来没有成立的形态」。

def _runs(swings: pd.DataFrame, kinds: list[int]) -> list[pd.DataFrame]:
    """按时间排序，找出种类依次等于 kinds 的每一组相邻摆动点。"""
    ordered = swings.sort_values("time").reset_index(drop=True)
    k = ordered["kind"].to_numpy()
    n = len(kinds)
    return [ordered.iloc[i:i + n] for i in range(len(ordered) - n + 1) if list(k[i:i + n]) == kinds]


def _first_close_beyond(close: np.ndarray, start: int, stop: int, line, sign: int):
    """从第 start 根到第 stop 根（不含），第一次 sign × (收盘价 - line(根号)) < 0 的根号；没有返回 None。"""
    for j in range(start, min(stop, len(close))):
        if sign * (close[j] - line(j)) < 0:
            return j
    return None


def head_and_shoulders(close: pd.Series, swings: pd.DataFrame, kind: str = "top", shoulder_tol: float = 0.3,
                       neck_tol: float = 0.5, time_ratio: float = 3.0, wait: int | None = None) -> pd.DataFrame:
    """头肩顶（kind="top"）或头肩底（kind="bottom"）的全部候选。

    头肩顶是五个相邻的摆动点：左肩（高）、颈线点 1（低）、头（高）、颈线点 2（低）、右肩（高）。
    颈线：连接两个颈线点的直线，向右延长。高度：头的价格减去头所在位置的颈线值。
    形状条件：
    1. 头比两个肩都高
    2. 两个肩的高度差不超过高度的 shoulder_tol
    3. 两个颈线点的高度差不超过高度的 neck_tol（颈线不能太斜）
    4. 头到右肩的根数 ÷ 左肩到头的根数，在 1 / time_ratio 到 time_ratio 之间
    confirmed_at：右肩被确认的时刻。在这之前，右肩还不存在，形态也就还不存在。
    broken_at：从 confirmed_at 这一根起，wait 根以内（默认等于左肩到右肩的根数），第一次收盘价低于颈线。
    target：跌破时的颈线值减去高度，也就是经典的「量度目标」；没有跌破为 NaN。
    头肩底把高低全部颠倒：颈线在上方，向上突破，目标在上方。返回的价格都是原始价格，height 总是正数。
    """
    if kind not in ("top", "bottom"):
        raise ValueError('kind 只能是 "top" 或 "bottom"')
    sign = 1 if kind == "top" else -1
    c = close.to_numpy(float)
    pos = {t: i for i, t in enumerate(close.index)}
    rows = []
    for run in _runs(swings, [sign, -sign, sign, -sign, sign]):
        left, neck_1, head, neck_2, right = (sign * run["price"].to_numpy())
        t = [pos[x] for x in run["time"]]
        slope = (neck_2 - neck_1) / (t[3] - t[1])
        line = lambda j, a=neck_1, b=t[1], m=slope: sign * (a + m * (j - b))           # 原始价格上的颈线
        height = head - (neck_1 + slope * (t[2] - t[1]))
        if not (head > left and head > right and height > 0):
            continue
        if abs(left - right) > shoulder_tol * height or abs(neck_2 - neck_1) > neck_tol * height:
            continue
        if not 1 / time_ratio <= (t[4] - t[2]) / (t[2] - t[0]) <= time_ratio:
            continue
        seen = pos[run["confirmed_at"].iloc[4]]
        limit = t[4] + (wait if wait is not None else t[4] - t[0]) + 1
        j = _first_close_beyond(c, seen, limit, line, sign)
        rows.append((*run["time"], run["confirmed_at"].iloc[4], sign * left, sign * head, sign * right,
                     line(seen), height, close.index[j] if j is not None else pd.NaT,
                     line(j) if j is not None else np.nan, line(j) - sign * height if j is not None else np.nan))
    return pd.DataFrame(rows, columns=["left", "neck_1", "head", "neck_2", "right", "confirmed_at", "left_price",
                                       "head_price", "right_price", "neckline", "height", "broken_at",
                                       "neckline_at_break", "target"])


def double_tops(close: pd.Series, swings: pd.DataFrame, kind: str = "top", tol: float = 0.1,
                wait: int | None = None) -> pd.DataFrame:
    """双顶（kind="top"）或双底（kind="bottom"）的全部候选。

    双顶是三个相邻的摆动点：第一个顶（高）、中间的低点、第二个顶（高）。
    颈线：中间低点的价格（水平线）。高度：两个顶中较高的那个减去颈线。
    形状条件：两个顶的高度差不超过高度的 tol。
    confirmed_at：第二个顶被确认的时刻。
    broken_at：从 confirmed_at 起，wait 根以内（默认等于两个顶之间的根数），第一次收盘价低于颈线。
    target：颈线减去高度（双底是加上高度），不管有没有跌破都给出。双底上下颠倒，height 总是正数。
    """
    if kind not in ("top", "bottom"):
        raise ValueError('kind 只能是 "top" 或 "bottom"')
    sign = 1 if kind == "top" else -1
    c = close.to_numpy(float)
    pos = {t: i for i, t in enumerate(close.index)}
    rows = []
    for run in _runs(swings, [sign, -sign, sign]):
        first, middle, second = sign * run["price"].to_numpy()
        t = [pos[x] for x in run["time"]]
        height = max(first, second) - middle
        if height <= 0 or abs(first - second) > tol * height:
            continue
        seen = pos[run["confirmed_at"].iloc[2]]
        limit = t[2] + (wait if wait is not None else t[2] - t[0]) + 1
        j = _first_close_beyond(c, seen, limit, lambda _, m=sign * middle: m, sign)
        rows.append((*run["time"], run["confirmed_at"].iloc[2], sign * first, sign * second, sign * middle,
                     height, close.index[j] if j is not None else pd.NaT, sign * (middle - height)))
    return pd.DataFrame(rows, columns=["first", "middle", "second", "confirmed_at", "first_price", "second_price",
                                       "neckline", "height", "broken_at", "target"])


def converging(close: pd.Series, swings: pd.DataFrame, flat: float = 0.2, wait: int | None = None) -> pd.DataFrame:
    """用最近两个摆动高点连上轨、两个摆动低点连下轨，按两条线的方向给形态命名。

    四个相邻的摆动点（两高两低，高低交替）决定两条线。设第一个点的位置两线之间的宽度为 W，
    四个点从头到尾的根数为 T，每条线在 T 根里的变化 ÷ W 叫这条线的「斜度」，绝对值不超过 flat 算水平。
    上轨水平、下轨向上：上升三角形；上轨向下、下轨水平：下降三角形；上轨向下、下轨向上：对称三角形
    两条线都向上且下轨更陡（在收窄）：上升楔形；都向下且上轨更陡：下降楔形
    两条线都水平：矩形；上轨向上、下轨向下：扩散；其余同方向而不收窄的：通道
    confirmed_at：第四个摆动点被确认的时刻。
    broken_at / direction：从 confirmed_at 起，wait 根以内（默认 T），第一次收盘价在上轨上方（1）或下轨下方（-1）。
    """
    c = close.to_numpy(float)
    pos = {t: i for i, t in enumerate(close.index)}
    rows = []
    ordered = swings.sort_values("time").reset_index(drop=True)
    for i in range(len(ordered) - 3):
        run = ordered.iloc[i:i + 4]
        kinds = run["kind"].to_numpy()
        if not all(kinds[1:] == -kinds[:-1]):
            continue
        t = np.array([pos[x] for x in run["time"]])
        price = run["price"].to_numpy(float)
        highs, lows = np.flatnonzero(kinds == 1), np.flatnonzero(kinds == -1)
        slope_up = (price[highs[1]] - price[highs[0]]) / (t[highs[1]] - t[highs[0]])
        slope_down = (price[lows[1]] - price[lows[0]]) / (t[lows[1]] - t[lows[0]])
        upper = lambda j, a=price[highs[0]], b=t[highs[0]], m=slope_up: a + m * (j - b)
        lower = lambda j, a=price[lows[0]], b=t[lows[0]], m=slope_down: a + m * (j - b)
        width, span = upper(t[0]) - lower(t[0]), t[3] - t[0]
        if width <= 0:
            continue
        du, dl = slope_up * span / width, slope_down * span / width
        up_flat, down_flat = abs(du) <= flat, abs(dl) <= flat
        if up_flat and down_flat:
            name = "矩形"
        elif up_flat and dl > 0:
            name = "上升三角形"
        elif down_flat and du < 0:
            name = "下降三角形"
        elif du < 0 < dl:
            name = "对称三角形"
        elif du > 0 and dl > du:
            name = "上升楔形"
        elif dl < 0 and du < dl:
            name = "下降楔形"
        elif du > 0 > dl:
            name = "扩散"
        else:
            name = "通道"
        seen = pos[run["confirmed_at"].iloc[3]]
        stop = t[3] + (wait if wait is not None else span) + 1
        j, direction = None, 0
        for k in range(seen, min(stop, len(c))):
            if c[k] > upper(k) or c[k] < lower(k):
                j, direction = k, 1 if c[k] > upper(k) else -1
                break
        rows.append((run["time"].iloc[0], run["time"].iloc[3], run["confirmed_at"].iloc[3], name, du, dl,
                     upper(seen), lower(seen), close.index[j] if j is not None else pd.NaT, direction))
    return pd.DataFrame(rows, columns=["start", "end", "confirmed_at", "name", "upper_slope", "lower_slope",
                                       "upper", "lower", "broken_at", "direction"])

# ---------------------------------------------------------------------------
# 图表形态（第 17 篇）
# ---------------------------------------------------------------------------

def toy_path(points, n):
    """按 (根号, 价格) 的折点线性插值出 n 根收盘价。"""
    values = np.interp(np.arange(n), [p[0] for p in points], [p[1] for p in points])
    return pd.Series(values, index=pd.date_range("2024-01-01", periods=n, freq="D"))


def manual_swings(close, rows):
    """rows 是 (根号, 种类) 的列表，价格取收盘价，确认时刻放在下一根。"""
    return pd.DataFrame([(close.index[k], close.iloc[k], kind, close.index[k + 1]) for k, kind in rows],
                        columns=["time", "price", "kind", "confirmed_at"])


def test_head_and_shoulders_by_hand():
    close = toy_path([(0, 100), (10, 110), (16, 102), (28, 120), (36, 104), (44, 112), (60, 96)], 61)
    found = Pt.head_and_shoulders(close, X.zigzag(close, 0.05))
    assert len(found) == 1
    row = found.iloc[0]
    # 颈线过 (16, 102) 和 (36, 104)，每根抬高 0.1；头在第 28 根，颈线 103.2，高度 120 - 103.2 = 16.8
    # 右肩 112 在第 44 根，价格每根跌 1，第 50 根跌到 106，跌够 5%，右肩被确认；那时颈线 105.4，还没跌破
    # 第 51 根收盘 105 低于颈线 105.5：跌破；目标 105.5 - 16.8 = 88.7
    assert row["head"] == close.index[28] and row["confirmed_at"] == close.index[50]
    assert row["neckline"] == pytest.approx(105.4) and row["height"] == pytest.approx(16.8)
    assert row["broken_at"] == close.index[51] and row["target"] == pytest.approx(88.7)


def test_head_and_shoulders_rules_and_bottom_mirror():
    close = toy_path([(0, 100), (10, 110), (16, 102), (28, 120), (36, 104), (44, 119), (60, 96)], 61)
    swings = X.zigzag(close, 0.05)
    assert Pt.head_and_shoulders(close, swings).empty                         # 两肩差 9，超过高度 16.8 的 30%
    assert len(Pt.head_and_shoulders(close, swings, shoulder_tol=0.6)) == 1
    top_close = toy_path([(0, 100), (10, 110), (16, 102), (28, 120), (36, 104), (44, 112), (60, 96)], 61)
    top_swings = X.zigzag(top_close, 0.05)
    flipped = 300 - top_close                                                  # 上下颠倒：头肩顶变成头肩底
    flipped_swings = top_swings.assign(price=300 - top_swings["price"], kind=-top_swings["kind"])
    top, bottom = Pt.head_and_shoulders(top_close, top_swings), Pt.head_and_shoulders(flipped, flipped_swings, "bottom")
    assert list(bottom["head"]) == list(top["head"]) and list(bottom["broken_at"]) == list(top["broken_at"])
    assert bottom["height"].iloc[0] == pytest.approx(top["height"].iloc[0])
    assert bottom["target"].iloc[0] == pytest.approx(300 - top["target"].iloc[0])


def test_double_tops_by_hand():
    close = toy_path([(0, 100), (10, 120), (20, 108), (30, 119), (45, 100)], 46)
    found = Pt.double_tops(close, X.zigzag(close, 0.05))
    assert len(found) == 1
    row = found.iloc[0]
    # 两个顶 120 和 119，中间低点 108：高度 12，两顶差 1 ≤ 1.2；第二个顶之后每根跌 19/15
    # 第 35 根跌到 112.67（跌够 5%）确认；第 39 根收盘 107.6 跌破 108；目标 108 - 12 = 96
    assert row["confirmed_at"] == close.index[35] and row["broken_at"] == close.index[39]
    assert row["neckline"] == 108 and row["height"] == 12 and row["target"] == 96
    assert Pt.double_tops(close, X.zigzag(close, 0.05), tol=0.05).empty      # 两顶差 1 > 12 × 5%


@pytest.mark.parametrize("highs,lows,name", [
    ((110, 108), (100, 102), "对称三角形"), ((110, 110), (100, 104), "上升三角形"), ((110, 106), (100, 100), "下降三角形"),
    ((110, 112), (100, 106), "上升楔形"), ((110, 104), (100, 98), "下降楔形"), ((110, 110), (100, 100), "矩形"),
    ((110, 114), (100, 96), "扩散"), ((110, 114), (100, 104), "通道")])
def test_converging_names_by_hand(highs, lows, name):
    close = pd.Series(105.0, index=pd.date_range("2024-01-01", periods=41, freq="D"))
    for k, price in zip([0, 20, 10, 30], [*highs, *lows]):
        close.iloc[k] = price
    swings = manual_swings(close, [(0, 1), (10, -1), (20, 1), (30, -1)])
    assert Pt.converging(close, swings)["name"].tolist() == [name]


def test_converging_breakout_by_hand():
    close = pd.Series(104.5, index=pd.date_range("2024-01-01", periods=61, freq="D"))
    for k, price in [(0, 110), (10, 100), (20, 108), (30, 102)]:
        close.iloc[k] = price
    close.iloc[36] = 120                                                      # 上轨在第 36 根是 106.4，收盘 120 升破
    found = Pt.converging(close, manual_swings(close, [(0, 1), (10, -1), (20, 1), (30, -1)]))
    assert found["name"].iloc[0] == "对称三角形"
    assert found["broken_at"].iloc[0] == close.index[36] and found["direction"].iloc[0] == 1


def test_chart_patterns_never_use_the_future():
    rng = np.random.default_rng(170)
    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"))
    full_swings = X.zigzag(close, 0.03)
    for func, kinds in [(Pt.head_and_shoulders, ["top", "bottom"]), (Pt.double_tops, ["top", "bottom"])]:
        for kind in kinds:
            full = func(close, full_swings, kind)
            assert len(full) > 3
            for k in [600, 1100]:
                cut = close.iloc[:k]
                part = func(cut, X.zigzag(cut, 0.03), kind)
                expected = full[full["confirmed_at"] <= cut.index[-1]].reset_index(drop=True)
                expected["broken_at"] = expected["broken_at"].where(expected["broken_at"] <= cut.index[-1])
                shape = [col for col in part.columns if col not in ("broken_at", "neckline_at_break", "target")]
                pd.testing.assert_frame_equal(part[shape], expected[shape])
                assert part["broken_at"].equals(expected["broken_at"])

pytest -q

........................................................................ [ 94%]
........                                                                 [100%]
152 passed in 0.56s