technical-patterns-lab/docs/对称三角形识别-Python实现.md

对称三角形识别（Symmetrical Triangle）——规则口径 + Python 实现

目标：只实现对称三角形的自动识别，并输出可解释结果：上沿/下沿、收敛度、触碰次数、是否突破、是否假突破（失败回撤）。

---

1. 概念对齐：上沿、下沿、突破、失败回撤

• 上沿（Upper line / Resistance）
  用“枢轴高点”（一串冲高回落的峰）拟合出的压力线；对称三角形要求它向右下倾。

• 下沿（Lower line / Support）
  用“枢轴低点”（一串探底回升的谷）拟合出的支撑线；对称三角形要求它向右上升。

• 向上突破（Upward breakout）
  价格从两线之间运行，某一根开始收盘价明显站上上沿（不是影线擦边）。

• 向下跌破（Downward breakdown）
  收盘价明显跌到下沿之下。

• 失败回撤 / 假突破（False breakout / Failed retest）
  “看起来突破了”，但在接下来 m 根内又回到三角形内部：

  • 向上突破后：close 又回到 upper_line 下方
  • 向下跌破后：close 又回到 lower_line 上方
    含义：突破没站稳，信号可靠性下降（回测/复盘里非常重要）。

---

2. 输入数据要求

df: pandas.DataFrame 至少包含：

• open, high, low, close
• volume（可选；用于“放量确认突破”，没有则跳过确认）

实现里用 bar_index = 0..n-1 作为横轴。

---

3. 可编码的识别口径（只针对对称三角形）

3.1 枢轴点（Pivot / Fractal）

用左右窗口 k 识别局部极值：

• 枢轴高点：high[i] 是 [i-k, i+k] 最大值
• 枢轴低点：low[i] 是 [i-k, i+k] 最小值

直觉：枢轴点就是你手工画线时会“圈出来”的那些关键峰/谷。

3.2 趋势线拟合

在一个检测窗口 window 内：

• 取该窗口里的枢轴高点集合拟合上沿：y = a_u x + b_u
• 取该窗口里的枢轴低点集合拟合下沿：y = a_l x + b_l

3.3 对称三角形硬条件（建议）

在窗口内必须满足：

• 斜率方向：a_u < 0 且 a_l > 0
• 收敛：末端宽度显著小于起始宽度
  width_end / width_start <= shrink_ratio
• 触碰次数：上沿触碰 ≥ 2 且下沿触碰 ≥ 2
  触碰判定：abs(pivot_y - line_y) / line_y <= touch_tol
• Apex 合理性（可选）：两线交点在窗口右侧不远处，避免两线近似平行导致的伪形态

3.4 突破/确认/假突破（信号层）

突破（价格）：

• 向上：close[t] > upper[t] * (1 + break_tol)
• 向下：close[t] < lower[t] * (1 - break_tol)

成交量确认（可选）：

• volume[t] > MA(volume, vol_window)[t] * vol_k

假突破（回测/复盘可用）：

• 若在 m 根内再次回到两线之间，则标记为 false_breakout=True

---

4. 参数建议（先跑通，再调参）

日线常用默认值（你可以直接用）：

• window=120（约 6 个月）
• pivot_k=3~5
• touch_tol=0.006（0.6% 贴线容差）
• shrink_ratio=0.6（末端宽度 ≤ 起始宽度的 60%）
• break_tol=0.003（突破需离线 0.3%）
• vol_window=20, vol_k=1.5
• false_break_m=5

---

5. Python 实现（可直接复制使用）

from __future__ import annotations

from dataclasses import dataclass
from typing import Dict, List, Literal, Optional, Tuple

import numpy as np
import pandas as pd


@dataclass
class SymTriangleResult:
    start: int
    end: int
    upper_coef: Tuple[float, float]  # (a_u, b_u)
    lower_coef: Tuple[float, float]  # (a_l, b_l)
    width_start: float
    width_end: float
    width_ratio: float
    touches_upper: int
    touches_lower: int
    apex_x: float
    breakout: Literal["up", "down", "none"]
    breakout_idx: Optional[int]
    volume_confirmed: Optional[bool]
    false_breakout: Optional[bool]


def pivots_fractal(high: np.ndarray, low: np.ndarray, k: int = 3) -> Tuple[np.ndarray, np.ndarray]:
    """左右窗口分形：返回 pivot_high_idx, pivot_low_idx。"""
    n = len(high)
    ph: List[int] = []
    pl: List[int] = []
    for i in range(k, n - k):
        if high[i] == np.max(high[i - k : i + k + 1]):
            ph.append(i)
        if low[i] == np.min(low[i - k : i + k + 1]):
            pl.append(i)
    return np.array(ph, dtype=int), np.array(pl, dtype=int)


def fit_line(x: np.ndarray, y: np.ndarray) -> Tuple[float, float]:
    """拟合 y = a*x + b"""
    a, b = np.polyfit(x, y, deg=1)
    return float(a), float(b)


def line_y(a: float, b: float, x: np.ndarray) -> np.ndarray:
    return a * x + b


def detect_sym_triangle(
    df: pd.DataFrame,
    window: int = 120,
    pivot_k: int = 3,
    touch_tol: float = 0.006,
    shrink_ratio: float = 0.6,
    break_tol: float = 0.003,
    vol_window: int = 20,
    vol_k: float = 1.5,
    false_break_m: int = 5,
) -> Optional[SymTriangleResult]:
    """
    只检测“最近一个窗口”是否存在对称三角形，并给出突破/确认/假突破信息。
    - 实时：false_breakout 只能返回 None（因为需要未来 m 根确认）
    - 回测：如果 df 包含突破后的 m 根数据，才会输出 false_breakout True/False
    """
    required = {"open", "high", "low", "close"}
    if not required.issubset(df.columns):
        raise ValueError(f"df must contain columns: {sorted(required)}")

    n = len(df)
    if n < max(window, 2 * pivot_k + 5):
        return None

    high = df["high"].to_numpy(dtype=float)
    low = df["low"].to_numpy(dtype=float)
    close = df["close"].to_numpy(dtype=float)
    has_volume = "volume" in df.columns and df["volume"].notna().any()
    volume = df["volume"].to_numpy(dtype=float) if has_volume else None

    ph_idx, pl_idx = pivots_fractal(high, low, k=pivot_k)

    end = n - 1
    start = max(0, end - window + 1)
    x_all = np.arange(n, dtype=float)

    ph_in = ph_idx[(ph_idx >= start) & (ph_idx <= end)]
    pl_in = pl_idx[(pl_idx >= start) & (pl_idx <= end)]
    if len(ph_in) < 2 or len(pl_in) < 2:
        return None

    # 拟合两条线
    a_u, b_u = fit_line(x_all[ph_in], high[ph_in])
    a_l, b_l = fit_line(x_all[pl_in], low[pl_in])

    # 斜率：对称三角形硬条件
    if not (a_u < 0 and a_l > 0):
        return None

    # 宽度收敛
    upper_start = float(line_y(a_u, b_u, np.array([start]))[0])
    lower_start = float(line_y(a_l, b_l, np.array([start]))[0])
    upper_end = float(line_y(a_u, b_u, np.array([end]))[0])
    lower_end = float(line_y(a_l, b_l, np.array([end]))[0])
    width_start = upper_start - lower_start
    width_end = upper_end - lower_end
    if width_start <= 0 or width_end <= 0:
        return None
    width_ratio = width_end / width_start
    if width_ratio > shrink_ratio:
        return None

    # 触碰次数（用 pivot 点贴线程度判定）
    ph_dist = np.abs(high[ph_in] - line_y(a_u, b_u, x_all[ph_in])) / np.maximum(
        line_y(a_u, b_u, x_all[ph_in]), 1e-9
    )
    pl_dist = np.abs(low[pl_in] - line_y(a_l, b_l, x_all[pl_in])) / np.maximum(
        line_y(a_l, b_l, x_all[pl_in]), 1e-9
    )
    touches_upper = int((ph_dist <= touch_tol).sum())
    touches_lower = int((pl_dist <= touch_tol).sum())
    if touches_upper < 2 or touches_lower < 2:
        return None

    # apex（两线交点）仅做合理性输出（可扩展为硬条件）
    denom = (a_u - a_l)
    apex_x = float((b_l - b_u) / denom) if abs(denom) > 1e-12 else float("inf")

    # === 突破判定（用最后一根 close 做实时判断）===
    upper_last = upper_end
    lower_last = lower_end
    breakout: Literal["up", "down", "none"] = "none"
    breakout_idx: Optional[int] = None
    if close[end] > upper_last * (1 + break_tol):
        breakout = "up"
        breakout_idx = end
    elif close[end] < lower_last * (1 - break_tol):
        breakout = "down"
        breakout_idx = end

    # 成交量确认（可选）
    volume_confirmed: Optional[bool] = None
    if breakout != "none" and has_volume and volume is not None:
        vol_ma = pd.Series(volume).rolling(vol_window).mean().to_numpy()
        if np.isfinite(vol_ma[end]) and vol_ma[end] > 0:
            volume_confirmed = bool(volume[end] > vol_ma[end] * vol_k)
        else:
            volume_confirmed = None

    # 假突破（需要未来 m 根数据，实时无法得知）
    false_breakout: Optional[bool] = None
    if breakout != "none" and breakout_idx is not None:
        # 如果数据不足以观察未来 m 根，则返回 None
        if breakout_idx + false_break_m < n:
            false_breakout = False
            for t in range(breakout_idx + 1, breakout_idx + false_break_m + 1):
                upper_t = float(line_y(a_u, b_u, np.array([t]))[0])
                lower_t = float(line_y(a_l, b_l, np.array([t]))[0])
                if breakout == "up" and close[t] < upper_t:
                    false_breakout = True
                    break
                if breakout == "down" and close[t] > lower_t:
                    false_breakout = True
                    break
        else:
            false_breakout = None

    return SymTriangleResult(
        start=start,
        end=end,
        upper_coef=(a_u, b_u),
        lower_coef=(a_l, b_l),
        width_start=float(width_start),
        width_end=float(width_end),
        width_ratio=float(width_ratio),
        touches_upper=touches_upper,
        touches_lower=touches_lower,
        apex_x=apex_x,
        breakout=breakout,
        breakout_idx=breakout_idx,
        volume_confirmed=volume_confirmed,
        false_breakout=false_breakout,
    )


def plot_sym_triangle(df: pd.DataFrame, res: SymTriangleResult) -> None:
    """简单可视化验算（需要 matplotlib）。"""
    import matplotlib.pyplot as plt

    close = df["close"].to_numpy(dtype=float)
    x = np.arange(len(df), dtype=float)
    a_u, b_u = res.upper_coef
    a_l, b_l = res.lower_coef

    start, end = res.start, res.end
    xw = np.arange(start, end + 1, dtype=float)
    upper = line_y(a_u, b_u, xw)
    lower = line_y(a_l, b_l, xw)

    plt.figure(figsize=(12, 5))
    plt.plot(x, close, linewidth=1.2, label="close")
    plt.plot(xw, upper, linewidth=2, label="upper")
    plt.plot(xw, lower, linewidth=2, label="lower")
    plt.axvline(end, color="gray", linestyle="--", linewidth=1)
    plt.title(
        f"sym_triangle: width_ratio={res.width_ratio:.2f}, touches=({res.touches_upper},{res.touches_lower}), "
        f"breakout={res.breakout}, vol_ok={res.volume_confirmed}, false={res.false_breakout}"
    )
    plt.legend()
    plt.show()

---

6. 输出如何用于“评级/打分”（建议字段）

你可以直接把 SymTriangleResult 的字段映射为评分输入：

• 形态质量：width_ratio（越小越收敛）、touches_upper/lower（越多越可信）
• 突破质量：breakout + volume_confirmed（放量加分）
• 风险提示：false_breakout（为 True 则大幅降分；为 None 表示实时待确认）

---

7. 你该如何验证“程序画的线”是否接近你手工画线

1. 先挑一个你手工画过的例子（如你图中的那段）
2. 跑 detect_sym_triangle 得到 res
3. 用 plot_sym_triangle(df, res) 画出 close + upper/lower
4. 调整 pivot_k / touch_tol / window / shrink_ratio，直到“触点选择”和你肉眼一致