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 实现（可直接复制使用）

```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`，直到“触点选择”和你肉眼一致