technical-patterns-lab/docs/plans/converging-triangle-algorithm_e49f62bc.plan.md

name	overview	todos
converging-triangle-algorithm	基于现有对称三角形算法，创建一个新的"收敛三角形"算法，支持二维矩阵批量输入，历史滚动计算，返回包含突破强度分数的完整 DataFrame。	id content status create-params 创建 ConvergingTriangleParams 参数对象 completed id content status create-result 创建 ConvergingTriangleResult 返回对象 completed id content status impl-strength 实现突破强度计算函数 calc_breakout_strength completed id content status impl-single 实现单点检测函数 detect_converging_triangle completed id content status impl-batch 实现批量滚动检测函数 detect_converging_triangle_batch completed id content status create-script 创建运行脚本 run_converging_triangle.py completed id content status test-run 测试运行并验证输出 in_progress id content status todo-1768985498252-e4ly4s7r9 只计算最近2年，可以变成一个默认的参数，方便后续修改 pending

收敛三角形算法设计

1. 核心设计

函数名

• detect_converging_triangle (单点检测)
• detect_converging_triangle_batch (批量滚动计算)

数据流

flowchart LR
    subgraph input [Input]
        OHLCV["OHLCV Matrices<br/>(n_stocks, n_days)"]
        Params[ConvergingTriangleParams]
    end
    
    subgraph process [Processing]
        Rolling["Rolling Window<br/>每天往过去看 window 天"]
        Detect["收敛三角形检测"]
        Score["突破强度计算"]
    end
    
    subgraph output [Output]
        DF["DataFrame<br/>(stock_idx, date_idx, 各指标...)"]
    end
    
    OHLCV --> Rolling
    Params --> Rolling
    Rolling --> Detect
    Detect --> Score
    Score --> DF

---

2. 参数对象设计

新建文件: src/converging_triangle.py

@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

---

3. 返回对象设计

@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

---

4. 突破强度计算公式

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 = 1 - width_ratio
    
    # 成交量加成
    vol_bonus = min(1, volume_ratio / 2)  # 放量2倍=满分
    
    # 加权合成
    strength_up = min(1, price_up * 5 * (1 + convergence_bonus) * (1 + vol_bonus))
    strength_down = min(1, price_down * 5 * (1 + convergence_bonus) * (1 + vol_bonus))
    
    return strength_up, strength_down

---

5. 核心函数签名

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: int = None,  # 从哪一天开始计算 (默认: window)
    end_day: int = None,    # 到哪一天结束 (默认: 最后一天)
) -> pd.DataFrame:
    """
    批量滚动检测收敛三角形
    
    Returns:
        DataFrame with columns:
    - stock_idx, date_idx
    - is_valid
    - breakout_strength_up, breakout_strength_down
    - upper_slope, lower_slope, width_ratio
    - touches_upper, touches_lower, apex_x
    - breakout_dir, volume_confirmed, false_breakout
    - window_start, window_end
    """

---

6. 滚动计算逻辑

flowchart TD
    Start[开始] --> Loop1["遍历 stock_idx: 0..107"]
    Loop1 --> Loop2["遍历 date_idx: start_day..end_day"]
    Loop2 --> Extract["提取窗口数据<br/>ohlcv[:, date_idx-window+1 : date_idx+1]"]
    Extract --> Check{"数据足够?<br/>date_idx >= window"}
    Check -->|No| Skip[跳过, is_valid=False]
    Check -->|Yes| Detect["调用 detect_converging_triangle()"]
    Detect --> CalcStrength["计算突破强度"]
    CalcStrength --> Append["添加到结果列表"]
    Skip --> Append
    Append --> Loop2
    Loop2 -->|done| Loop1
    Loop1 -->|done| ToDF["转换为 DataFrame"]
    ToDF --> End[返回]

---

7. 文件结构

technical-patterns-lab/
├── src/
│   ├── sym_triangle.py          # 现有算法 (保留)
│   └── converging_triangle.py   # 新算法 (新建)
├── scripts/
│   ├── run_sym_triangle_pkl.py  # 现有脚本 (保留)
│   └── run_converging_triangle.py # 新脚本 (新建)

---

8. 性能考虑

• 108 只股票 × 5113 天 = 55 万次检测
• 预估耗时: 约 2-5 分钟 (可优化)
• 可选优化: - 只计算最近 N 天 - 并行处理多只股票 - 向量化 pivot 检测