technical-patterns-lab/docs/2026-01-16_converging_triangle_algorithm.md

收敛三角形算法实现日志

日期: 2026-01-16
任务: 基于对称三角形算法，开发支持批量滚动计算的收敛三角形检测算法

---

1. 需求概述

• 函数名: detect_converging_triangle (收敛三角形)
• 参数和返回值使用对象结构
• 支持二维矩阵数据处理 (n_stocks × n_days)
• 历史滚动计算: 每个交易日往过去看 window 天，不使用未来数据
• 返回突破强度指标 (0~1 连续分数)

---

2. 新建文件

文件	说明
src/converging_triangle.py	核心算法模块
scripts/run_converging_triangle.py	批量运行脚本

---

3. 参数对象设计

@dataclass
class ConvergingTriangleParams:
    # 窗口设置
    window: int = 400
    
    # 枢轴点检测
    pivot_k: int = 20
    
    # 边界线拟合
    boundary_n_segments: int = 2
    boundary_source: str = "full"  # "full" | "pivots"
    
    # 斜率约束
    upper_slope_max: float = 0.10
    lower_slope_min: float = -0.10
    
    # 触碰判定
    touch_tol: float = 0.10
    touch_loss_max: float = 0.10
    
    # 收敛要求
    shrink_ratio: float = 0.8
    
    # 突破判定
    break_tol: float = 0.001
    vol_window: int = 20
    vol_k: float = 1.3
    false_break_m: int = 5

---

4. 返回对象设计

@dataclass
class ConvergingTriangleResult:
    # 基础标识
    stock_idx: int
    date_idx: int
    is_valid: bool  # 是否识别到有效三角形
    
    # 突破强度 (0~1 连续分数)
    breakout_strength_up: float    # 向上突破强度
    breakout_strength_down: float  # 向下突破强度
    
    # 几何属性
    upper_slope: float
    lower_slope: float
    width_ratio: float
    touches_upper: int
    touches_lower: int
    apex_x: float
    
    # 突破状态
    breakout_dir: str  # "up" | "down" | "none"
    volume_confirmed: Optional[bool]
    false_breakout: Optional[bool]
    
    # 窗口范围
    window_start: int
    window_end: int

---

5. 突破强度计算公式

def calc_breakout_strength(
    close: float,
    upper_line: float,
    lower_line: float,
    volume_ratio: float,
    width_ratio: float,
) -> Tuple[float, float]:
    """
    计算向上/向下突破强度 (0~1)
    
    综合考虑:
    - 价格突破幅度 (close 相对于上/下沿的距离)
    - 成交量放大倍数
    - 收敛程度 (width_ratio 越小越强)
    """
    # 价格突破分数
    price_up = max(0, (close - upper_line) / upper_line)
    price_down = max(0, (lower_line - close) / lower_line)
    
    # 收敛加成 (越收敛, 突破越有效)
    convergence_bonus = max(0, 1 - width_ratio)
    
    # 成交量加成 (放量2倍=满分)
    vol_bonus = min(1, max(0, volume_ratio - 1))
    
    # 加权合成
    strength_up = min(1.0, price_up * 5 * (1 + convergence_bonus * 0.5) * (1 + vol_bonus * 0.5))
    strength_down = min(1.0, price_down * 5 * (1 + convergence_bonus * 0.5) * (1 + vol_bonus * 0.5))
    
    return strength_up, strength_down

---

6. 核心函数签名

单点检测

def detect_converging_triangle(
    high: np.ndarray,
    low: np.ndarray,
    close: np.ndarray,
    volume: Optional[np.ndarray],
    params: ConvergingTriangleParams,
    stock_idx: int = 0,
    date_idx: int = 0,
) -> ConvergingTriangleResult:

批量滚动检测

def detect_converging_triangle_batch(
    open_mtx: np.ndarray,   # (n_stocks, n_days)
    high_mtx: np.ndarray,
    low_mtx: np.ndarray,
    close_mtx: np.ndarray,
    volume_mtx: np.ndarray,
    params: ConvergingTriangleParams,
    start_day: Optional[int] = None,
    end_day: Optional[int] = None,
    only_valid: bool = False,
    verbose: bool = False,
) -> pd.DataFrame:

---

7. 滚动计算逻辑

遍历 stock_idx: 0..107
  遍历 date_idx: start_day..end_day
    提取窗口数据 ohlcv[:, date_idx-window+1 : date_idx+1]
    如果数据足够 (date_idx >= window):
      调用 detect_converging_triangle()
      计算突破强度
      添加到结果列表
转换为 DataFrame 返回

---

8. 测试运行结果

运行命令:

cd D:\project\technical-patterns-lab
.\.venv\Scripts\Activate.ps1
python scripts/run_converging_triangle.py

检测范围:

• 股票数量: 108
• 计算天数: 最近 500 天
• 总检测点: 54,000

检测结果:

• 有效三角形: 2,912 个 (5.4%)
• 向上突破: 545 次
• 向下突破: 273 次
• 未突破: 2,094 次

突破强度统计 (有效三角形):

• 向上强度: mean=0.1315, max=1.0000
• 向下强度: mean=0.0473, max=1.0000

高强度突破 (strength > 0.3):

• 向上: 444 个
• 向下: 174 个

---

9. 输出文件

outputs/converging_triangles/
├── all_results.csv          # 全部有效记录 (2,912 行)
├── strong_breakout_up.csv   # 高强度向上突破 (444 行)
└── strong_breakout_down.csv # 高强度向下突破 (174 行)

CSV 字段说明

字段	说明
stock_idx	股票索引 (0~107)
date_idx	日期索引
is_valid	是否识别到有效三角形
breakout_strength_up	向上突破强度 (0~1)
breakout_strength_down	向下突破强度 (0~1)
upper_slope	上沿斜率
lower_slope	下沿斜率
width_ratio	收敛比 (末端宽度/起始宽度)
touches_upper	上沿触碰次数
touches_lower	下沿触碰次数
apex_x	顶点位置
breakout_dir	突破方向 (up/down/none)
volume_confirmed	成交量确认
false_breakout	假突破标记
window_start	窗口起始位置
window_end	窗口结束位置

---

10. 使用说明

修改计算范围

编辑 scripts/run_converging_triangle.py:

# 只算最近 500 天 (默认)
RECENT_DAYS = 500

# 计算全部历史 (耗时较长)
RECENT_DAYS = None

# 只算最近 100 天 (快速测试)
RECENT_DAYS = 100

修改检测参数

PARAMS = ConvergingTriangleParams(
    window=400,           # 调整窗口大小
    shrink_ratio=0.8,     # 调整收敛要求
    upper_slope_max=0.10, # 调整斜率容差
    # ...
)

只输出有效记录

ONLY_VALID = True   # 只输出识别到三角形的记录
ONLY_VALID = False  # 输出所有检测点

---

11. 相关文件

文件	说明
src/converging_triangle.py	收敛三角形核心算法
src/sym_triangle.py	对称三角形算法 (原有)
scripts/run_converging_triangle.py	批量运行脚本
scripts/run_sym_triangle_pkl.py	对称三角形脚本 (原有)

---

12. 待优化项

• 性能优化: 向量化 pivot 检测，减少循环
• 并行计算: 多进程处理多只股票
• 关联股票代码: 结合 data/map/stock_name.json 显示真实代码
• 可视化: 为高强度突破生成图表
• 回测验证: 检验突破后 N 天的收益率