褚宏光
543572667b
- Created README.md and USAGE.md for project overview and usage instructions. - Added core algorithm in src/converging_triangle.py for batch processing of stock data. - Introduced data files (open.pkl, high.pkl, low.pkl, close.pkl, volume.pkl) for OHLCV data. - Developed output documentation for results and breakout strength calculations. - Implemented scripts for running the detection and generating reports. - Added SVG visualizations and markdown documentation for algorithm details and usage examples.
385 lines
13 KiB
Markdown
385 lines
13 KiB
Markdown
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）可在验证有效后再做。
|
||
|
||
```python
|
||
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. 可视化验算（强烈建议做，否则很容易“识别对了但画线不对”）
|
||
|
||
```python
|
||
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`，比固定百分比更稳。
|
||
- 加入“前置趋势”判定：三角形属于延续还是反转（给评级体系加解释）。
|
||
- 将“画线/形态”统一到可视化回放工具（方便人工抽检）。
|
||
|