technical-patterns-lab/docs/技术形态识别-对称三角形（含三角形家族）-Python实现调研.md

ad3590a2-2133-43ee-8ecd-d64a32760224

技术形态识别调研：对称三角形（含三角形家族）如何用 Python 实现

目标：把“图上能看出来的三角形”落地为可计算、可回测、可解释的规则与代码框架。本文以对称三角形为主，同时覆盖上升三角形/下降三角形的差异化判定。

---

1. 输入数据与基本约定

1.1 数据字段（最低要求）

• OHLC：open/high/low/close
• 成交量：volume（可选但强烈建议，用于形态可靠性/突破确认）
• 时间索引：datetime 或顺序索引（实现里用 bar_index = 0..n-1）

1.2 周期选择（建议）

• 日线：最常用，适合识别 1~6 个月级别三角形。
• 周线：更稳定，噪声更低，触碰次数更少但信号更“硬”。

1.3 你在图里画的三角形是什么（术语对齐）

• 对称三角形（Symmetrical Triangle）：上边线下倾、下边线上升，区间收敛；本身不天然看涨/看跌，方向以突破为准。
• 上升三角形（Ascending Triangle，看涨更典型）：上边线近似水平、下边线上升。
• 下降三角形（Descending Triangle，看跌更典型）：上边线下降、下边线近似水平。

---

2. 从“肉眼形态”到“可编码规则”的拆解

形态识别一般分 4 层：

1. 先找到“有意义的拐点”（枢轴点 / swing points）
2. 用拐点拟合两条趋势线（上沿/下沿）
3. 校验是否满足某类三角形的几何约束（斜率、收敛、触碰次数、包含性）
4. 识别突破与确认（价格 + 成交量 + 失败回撤）

下面给出可落地的判定口径。

---

3. 枢轴点（Pivot / Fractal）的提取方法

3.1 为什么要用枢轴点

三角形的上沿/下沿本质是“多次触碰的压力/支撑”，用所有 K 线拟合会被噪声拖偏；用“局部极值点”更接近画线逻辑。

3.2 经典实现：左右窗口分形（推荐作为 baseline）

定义窗口 k：

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

参数建议：

• 日线：k=3~5
• 周线：k=2~3

注意：该方法会引入 k 根的“确认延迟”（右侧需要 k 根数据才能确认 pivot）。

---

4. 趋势线拟合（上沿/下沿）

4.1 最简可用：线性回归/最小二乘

对窗口内的枢轴高点集合 (x_h, y_h) 拟合上沿：

y = a_u x + b_u

对枢轴低点集合 (x_l, y_l) 拟合下沿：

y = a_l x + b_l

实现可用：

• numpy.polyfit(x, y, deg=1)
• 或 scipy.stats.linregress（可选）

4.2 鲁棒拟合（可选增强）

真实行情会有“假突破/尖刺”，可以用：

• RANSAC 回归（sklearn.linear_model.RANSACRegressor） 来降低离群点影响。

---

5. 三角形家族的“几何判定条件”

下面所有条件都应在一个滑动窗口内判断（例如最近 W=60~180 根）。

5.1 基础条件（所有三角形通用）

• 拐点数量：至少 >=2 个枢轴高点 + >=2 个枢轴低点
• 触碰有效性：枢轴点到趋势线的垂直距离（或相对误差）足够小
  • 例如：abs(pivot_y - line_y) / line_y <= tol，tol 可取 0.3%~1.0%
• 包含性：窗口内大多数 K 线价格在两条线之间（允许少量穿越）
• 收敛性：两线间距随时间缩小（后段宽度 < 前段宽度）

5.2 对称三角形（Symmetrical）

• 斜率：a_u < 0 且 a_l > 0
• 收敛：两线交点（apex）在当前窗口右侧不远处（避免“几乎平行”）
  • 计算 apex：解 a_u x + b_u = a_l x + b_l
  • x_{apex} = (b_l - b_u)/(a_u - a_l)
  • 约束：x_apex 在窗口右侧的合理范围内，例如 end < x_apex < end + 2W
• 宽度缩小：(upper(end) - lower(end)) / (upper(start) - lower(start)) <= shrink_ratio
  • shrink_ratio 可取 0.3~0.7

5.3 上升三角形（Ascending，典型看涨）

• 上沿近似水平：abs(a_u) <= slope_eps
• 下沿上升：a_l > 0

5.4 下降三角形（Descending，典型看跌）

• 上沿下降：a_u < 0
• 下沿近似水平：abs(a_l) <= slope_eps

参数建议：

• slope_eps 需要与价格尺度无关，建议在归一化 x \in [0,1] 之后拟合（或把斜率换算成“每根 K 的百分比变化”）。

---

6. 成交量与突破确认（把“形态”变成“信号”）

6.1 成交量收缩（形态期）

常见经验：收敛阶段成交量逐步下降。可做一个弱约束：

• mean(volume[first_third]) > mean(volume[last_third]) 或
• volume 的线性回归斜率为负（噪声大，建议弱约束）

6.2 向上突破（看涨确认）

对称三角形/上升三角形的看涨信号可定义为：

• close[t] > upper_line[t] * (1 + break_tol)
• 且成交量：volume[t] > SMA(volume, n)[t] * vol_k
• 可选：close 连续 1~2 根站在线上方（避免一根假突破）

建议：

• break_tol：0.2%~0.8% 或用 ATR 的一部分（更稳）
• vol_k：1.2~2.0

6.3 失败回撤（False Breakout）处理

突破后若在 m 根内收回形态内部（close < upper_line），可标记为失败突破或降低评分。

---

7. 量度目标（可解释输出）

经典量度：

• 形态初期高度：H = upper(start) - lower(start)
• 向上突破目标：target = breakout_price + H
• 向下跌破目标：target = breakout_price - H

注意：量度目标是“参考”，实际还需结合前高/筹码/均线等约束。

---

8. Python 实现骨架（pandas/numpy baseline，可直接跑）

说明：下面代码优先做到“可读、可调参、可画图验算”。性能优化（向量化/numba）可在验证有效后再做。

from __future__ import annotations

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

import numpy as np
import pandas as pd


@dataclass
class TriangleCandidate:
    start: int
    end: int
    kind: Literal["sym", "asc", "desc"]
    upper_coef: Tuple[float, float]  # (a_u, b_u)
    lower_coef: Tuple[float, float]  # (a_l, b_l)
    apex_x: float
    width_start: float
    width_end: float


def _pivots_fractal(high: np.ndarray, low: np.ndarray, k: int = 3) -> Tuple[np.ndarray, np.ndarray]:
    """
    返回 pivot_high_idx, pivot_low_idx
    需要右侧 k 根确认，所以最后 k 根无法成为 pivot。
    """
    n = len(high)
    ph = []
    pl = []
    for i in range(k, n - k):
        win_h = high[i - k : i + k + 1]
        win_l = low[i - k : i + k + 1]
        if high[i] == np.max(win_h):
            ph.append(i)
        if low[i] == np.min(win_l):
            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_triangles(
    df: pd.DataFrame,
    window: int = 120,
    pivot_k: int = 3,
    tol: float = 0.006,
    slope_eps: float = 0.0005,
    shrink_ratio: float = 0.6,
) -> List[TriangleCandidate]:
    """
    在最近窗口内寻找三角形候选（返回候选列表，通常取最近一个做信号）。

    参数口径：
    - tol: 枢轴点到趋势线的相对误差容忍（0.6%）
    - slope_eps: 近似水平线斜率阈值（需结合 x 的尺度；此处 x 用 bar_index）
    - shrink_ratio: 末端宽度/初端宽度阈值，越小要求越严格
    """
    high = df["high"].to_numpy(dtype=float)
    low = df["low"].to_numpy(dtype=float)
    close = df["close"].to_numpy(dtype=float)
    n = len(df)

    ph_idx, pl_idx = _pivots_fractal(high, low, k=pivot_k)
    out: List[TriangleCandidate] = []

    # 只做最近窗口的检测（也可扩展为滑动扫描全历史）
    end = n - 1
    start = max(0, end - window + 1)
    x = 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 out

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

    # apex（两线交点）
    denom = (a_u - a_l)
    if abs(denom) < 1e-12:
        return out
    apex_x = (b_l - b_u) / denom

    upper_start = _line_y(a_u, b_u, np.array([start]))[0]
    lower_start = _line_y(a_l, b_l, np.array([start]))[0]
    upper_end = _line_y(a_u, b_u, np.array([end]))[0]
    lower_end = _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 out
    if width_end / width_start > shrink_ratio:
        return out

    # 触碰判定：pivot 点距离线足够近
    ph_dist = np.abs(high[ph_in] - _line_y(a_u, b_u, x[ph_in])) / np.maximum(_line_y(a_u, b_u, x[ph_in]), 1e-9)
    pl_dist = np.abs(low[pl_in] - _line_y(a_l, b_l, x[pl_in])) / np.maximum(_line_y(a_l, b_l, x[pl_in]), 1e-9)
    if (ph_dist <= tol).sum() < 2 or (pl_dist <= tol).sum() < 2:
        return out

    # apex 合理性：在窗口右侧一定范围内（可调）
    if not (end < apex_x < end + 2 * window):
        return out

    # 分类
    kind: Optional[str] = None
    if a_u < 0 and a_l > 0:
        kind = "sym"
    elif abs(a_u) <= slope_eps and a_l > 0:
        kind = "asc"
    elif a_u < 0 and abs(a_l) <= slope_eps:
        kind = "desc"

    if kind is None:
        return out

    out.append(
        TriangleCandidate(
            start=start,
            end=end,
            kind=kind,  # type: ignore[arg-type]
            upper_coef=(a_u, b_u),
            lower_coef=(a_l, b_l),
            apex_x=float(apex_x),
            width_start=float(width_start),
            width_end=float(width_end),
        )
    )
    return out


def breakout_signal(
    df: pd.DataFrame,
    tri: TriangleCandidate,
    break_tol: float = 0.003,
    vol_window: int = 20,
    vol_k: float = 1.5,
) -> Literal["up", "down", "none"]:
    """
    简化版突破信号：
    - up: close > upper*(1+break_tol) 且 volume 放大
    - down: close < lower*(1-break_tol) 且 volume 放大
    """
    a_u, b_u = tri.upper_coef
    a_l, b_l = tri.lower_coef
    end = tri.end
    x = float(end)
    upper = a_u * x + b_u
    lower = a_l * x + b_l

    close = float(df["close"].iloc[end])
    vol = float(df.get("volume", pd.Series([np.nan] * len(df))).iloc[end])
    vol_ma = float(df.get("volume", pd.Series([np.nan] * len(df))).rolling(vol_window).mean().iloc[end])

    vol_ok = np.isfinite(vol) and np.isfinite(vol_ma) and (vol > vol_ma * vol_k)

    if close > upper * (1 + break_tol) and vol_ok:
        return "up"
    if close < lower * (1 - break_tol) and vol_ok:
        return "down"
    return "none"

---

9. 可视化验算（强烈建议做，否则很容易“识别对了但画线不对”）

import matplotlib.pyplot as plt


def plot_triangle(df, tri):
    a_u, b_u = tri.upper_coef
    a_l, b_l = tri.lower_coef
    start, end = tri.start, tri.end
    x = np.arange(start, end + 1)
    upper = a_u * x + b_u
    lower = a_l * x + b_l

    close = df["close"].to_numpy()
    plt.figure(figsize=(12, 5))
    plt.plot(np.arange(len(close)), close, linewidth=1, label="close")
    plt.plot(x, upper, linewidth=2, label="upper")
    plt.plot(x, lower, linewidth=2, label="lower")
    plt.axvline(end, color="gray", linestyle="--", linewidth=1)
    plt.title(f"triangle={tri.kind}, width_end/width_start={tri.width_end/tri.width_start:.2f}")
    plt.legend()
    plt.show()

---

10. 工程化落地建议（给“评级表/打分维度”用）

10.1 输出结构（可解释）

建议把形态识别输出做成结构化字段，方便评分与回测：

• kind: sym/asc/desc
• touches_upper/touches_lower
• shrink_ratio = width_end/width_start
• apex_distance = apex_x - end
• breakout: none/up/down
• break_strength: (close - upper)/ATR 或百分比
• volume_ratio = volume / MA(volume, n)
• measured_target: breakout_price ± H

10.2 常见坑

• 没有 pivot：k 取太大/窗口太短，会导致拐点不足。
• 斜率尺度问题：不同价格/周期导致斜率阈值不可比；建议用归一化横轴或用“每根百分比斜率”。
• 尖刺离群点：极端上影线会影响拟合；可用 RANSAC 或在 pivot 上做 winsorize。
• 形态过度拟合：条件太多会导致回测看起来很强但实盘稀疏；建议分层评分（硬条件 + 软条件加分）。

---

11. 下一步（如果要继续深化）

• 增加全历史滑动扫描：输出每个窗口的候选并去重（同一形态合并）。
• 引入 ATR 作为容差：tol = c * ATR/price，比固定百分比更稳。
• 加入“前置趋势”判定：三角形属于延续还是反转（给评级体系加解释）。
• 将“画线/形态”统一到可视化回放工具（方便人工抽检）。