diff --git a/discuss/20260129-讨论.md b/discuss/20260129-讨论.md new file mode 100644 index 0000000..e69de29 diff --git a/docs/2026-01-29_迁移到data_server.md b/docs/2026-01-29_迁移到data_server.md new file mode 100644 index 0000000..4e04440 --- /dev/null +++ b/docs/2026-01-29_迁移到data_server.md @@ -0,0 +1,111 @@ +## 1. `三角形强度()` - 全市场批量筛选 + +**作用**：对全市场所有股票、所有日期进行批量检测，返回强度矩阵，用于筛选和排序。 + +**入参**： +| 参数 | 类型 | 默认值 | 说明 | +|-----|------|--------|------| +| `window` | int | 240 | 检测窗口（交易日） | +| `min_convergence` | float | 0.45 | 最小收敛比例（末端/起始宽度） | +| `breakout_threshold` | float | 0.005 | 突破阈值（0.5%） | +| `volume_multiplier` | float | 1.5 | 放量确认倍数 | +| `start_day` | int | -1 | 起始日索引（-1=自动） | +| `end_day` | int | -1 | 结束日索引（-1=自动） | + +**返回值**： +```python +np.ndarray # shape: (n_stocks, n_days) +``` +- 正值 `+0 ~ +1`：向上突破/潜力，值越大越强 +- 负值 `-1 ~ -0`：向下突破/潜力，绝对值越大越强 +- `NaN`：无有效形态 + +**使用场景**： +```python +强度 = 三角形强度() + +# 筛选最新一天强向上的股票 +强向上 = 强度[:, -1] > 0.5 + +# 按强度排序 +排名 = np.argsort(强度[:, -1])[::-1] +``` + +--- + +## 2. `三角形详情(ticker)` - 单股票详情 + +**作用**：对单只股票进行检测，返回完整详情（强度分量、几何属性、图表数据），用于展示和分析。 + +**入参**： +| 参数 | 类型 | 默认值 | 说明 | +|-----|------|--------|------| +| `ticker` | str | 必填 | 股票代码，如 `"SH600519"` | +| `window` | int | 240 | 检测窗口 | +| `min_convergence` | float | 0.45 | 最小收敛比例 | +| `breakout_threshold` | float | 0.005 | 突破阈值 | +| `volume_multiplier` | float | 1.5 | 放量倍数 | +| `display_days` | int | 300 | 图表显示天数 | + +**返回值**：`TriangleDetail` 对象 + +| 字段 | 类型 | 说明 | +|-----|------|------| +| `strength` | float | 综合强度 (-1 ~ +1) | +| `is_valid` | bool | 是否有效形态 | +| `direction` | str | `"up"` / `"down"` / `"none"` | +| `strength_up` | float | 向上强度原始值 (0~1) | +| `strength_down` | float | 向下强度原始值 (0~1) | +| `convergence_score` | float | 收敛分 (0~1) | +| `volume_score` | float | 成交量分 (0~1) | +| `fitting_score` | float | 拟合分 (0~1) | +| `width_ratio` | float | 收敛比例 | +| `upper_slope` | float | 上沿斜率 | +| `lower_slope` | float | 下沿斜率 | +| `touches_upper` | int | 触碰上沿次数 | +| `touches_lower` | int | 触碰下沿次数 | +| `volume_confirmed` | bool | 成交量是否确认 | +| `chart_data` | dict | 图表数据（ECharts 格式） | + +**`chart_data` 结构**： +```python +{ + 'dates': [20260101, 20260102, ...], # 日期 + 'candlestick': [[o,c,l,h], ...], # K线数据 + 'upper_line': [[x1,y1], [x2,y2]], # 上沿线 + 'lower_line': [[x1,y1], [x2,y2]], # 下沿线 + 'detection_window': [start, end], # 检测窗口 + 'ticker': "SH600519", + 'strength': 0.65, + 'direction': "up", +} +``` + +**使用场景**： +```python +详情 = 三角形详情("SH600519") + +# 查看结果 +print(f"强度: {详情.strength}") +print(f"方向: {详情.direction}") +print(f"收敛度: {详情.width_ratio}") + +# 前端绑定 +图表配置 = 详情.chart_data +``` + +--- + +## 典型工作流 + +```python +# 1. 批量筛选 +强度 = 三角形强度() +目标股票索引 = np.where(强度[:, -1] > 0.5)[0] + +# 2. 获取详情 +tkrs = g.load_pkl()['tkrs'] +for idx in 目标股票索引[:10]: + 详情 = 三角形详情(tkrs[idx]) + print(f"{tkrs[idx]}: {详情.strength:.2f}, {详情.direction}") +``` \ No newline at end of file diff --git a/docs/triangle_api_reference.md b/docs/triangle_api_reference.md new file mode 100644 index 0000000..6e438aa --- /dev/null +++ b/docs/triangle_api_reference.md @@ -0,0 +1,475 @@ +# 收敛三角形检测 API 参考文档 + +## 概览 + +`triangle_detector_api.py` 提供收敛三角形形态检测的完整封装，核心设计： + +- **主函数 `detect_matrix()`**：全市场矩阵批量检测（如万得全A所有A股） +- **一个 `strength` 搞定筛选**：正值=向上，负值=向下，绝对值=强度 +- **4 个核心参数**：都有合理默认值，大多数场景无需调整 + +--- + +## 快速开始 + +```python +from triangle_detector_api import detect_matrix + +# 输入：OHLCV 矩阵 (n_stocks, n_days) +df = detect_matrix( + high_mtx=high_mtx, # shape: (n_stocks, n_days) + low_mtx=low_mtx, + close_mtx=close_mtx, + volume_mtx=volume_mtx, + dates=dates, # shape: (n_days,) + tkrs=tkrs, # shape: (n_stocks,) 股票代码 + tkrs_name=tkrs_name, # shape: (n_stocks,) 股票名称 +) + +# 筛选 +df[df['strength'] > 0.5] # 强向上突破 +df[df['strength'] < -0.5] # 强向下突破 +df.sort_values('strength', ascending=False) # 排序 +``` + +--- + +## 核心函数 + +### `detect_matrix()` ⭐ 主函数 + +全市场矩阵批量检测：每只股票每天都进行检测。适用于万得全A等全市场扫描场景。 + +```python +def detect_matrix( + high_mtx: np.ndarray, # (n_stocks, n_days) + low_mtx: np.ndarray, + close_mtx: np.ndarray, + volume_mtx: np.ndarray, + dates: np.ndarray, # (n_days,) + tkrs: np.ndarray = None, # (n_stocks,) 股票代码 + tkrs_name: np.ndarray = None, # (n_stocks,) 股票名称 + params: DetectionParams = None, + start_day: int = None, + end_day: int = None, + only_valid: bool = True, + verbose: bool = False, +) -> pd.DataFrame +``` + +**参数说明：** + +| 参数 | 类型 | 必填 | 说明 | +|------|------|------|------| +| `high_mtx` | ndarray | ✅ | 最高价矩阵 (n_stocks, n_days) | +| `low_mtx` | ndarray | ✅ | 最低价矩阵 | +| `close_mtx` | ndarray | ✅ | 收盘价矩阵 | +| `volume_mtx` | ndarray | ✅ | 成交量矩阵 | +| `dates` | ndarray | ✅ | 日期数组 (n_days,) | +| `tkrs` | ndarray | ❌ | 股票代码数组 | +| `tkrs_name` | ndarray | ❌ | 股票名称数组 | +| `params` | DetectionParams | ❌ | 检测参数 | +| `start_day` | int | ❌ | 起始日索引 | +| `end_day` | int | ❌ | 结束日索引 | +| `only_valid` | bool | ❌ | 只返回有效形态，默认 True | +| `verbose` | bool | ❌ | 打印进度 | + +**返回值：** DataFrame + +| 列名 | 说明 | +|------|------| +| `stock_idx` | 股票索引 | +| `stock_code` | 股票代码 | +| `stock_name` | 股票名称 | +| `date_idx` | 日期索引 | +| `date` | 日期 | +| `strength` | **强度分（核心输出）** | +| `is_valid` | 是否有效形态 | +| `direction` | 突破方向 | +| `width_ratio` | 收敛比例 | + +**示例：** + +```python +from triangle_detector_api import detect_matrix + +# 批量检测 +df = detect_matrix(high, low, close, volume, dates, tkrs, tkrs_name, verbose=True) + +# 筛选 +strong_up = df[df['strength'] > 0.5] +strong_down = df[df['strength'] < -0.5] + +# 某日最强股票 +latest_date = df['date'].max() +df[df['date'] == latest_date].sort_values('strength', ascending=False).head(10) +``` + +--- + +### `detect_triangle()` + +单股票检测（用于生成图表数据）。 + +```python +def detect_triangle( + ohlc_data: Dict, + params: DetectionParams = None, + include_pivots: bool = True, + display_window: int = None, +) -> DetectionResult +``` + +**ohlc_data 格式：** + +```python +{'dates': [...], 'open': [...], 'high': [...], 'low': [...], 'close': [...], 'volume': [...]} +``` + +**返回值：** `DetectionResult` 对象，包含 `chart_data` 可生成 ECharts 配置 + +--- + +## 入参配置 + +### `DetectionParams` + +检测参数类，所有参数都有合理默认值。 + +```python +from triangle_detector_api import DetectionParams + +# 使用默认参数（推荐） +params = DetectionParams() + +# 自定义参数 +params = DetectionParams( + window=120, + min_convergence=0.6, +) +``` + +**可调参数：** + +| 参数 | 类型 | 默认值 | 说明 | +|------|------|--------|------| +| `window` | int | 240 | 检测窗口大小（交易日） | +| `min_convergence` | float | 0.45 | 收敛比例阈值（末端宽度/起始宽度） | +| `breakout_threshold` | float | 0.005 | 突破阈值（0.5%） | +| `volume_multiplier` | float | 1.5 | 放量确认倍数 | + +**参数调整建议：** + +| 参数 | 较小值效果 | 较大值效果 | +|------|-----------|-----------| +| `window` | 检测短期形态，信号及时 | 检测中长期形态，信号可靠 | +| `min_convergence` | 要求更强收敛，形态标准 | 放宽收敛，更多候选 | +| `breakout_threshold` | 更敏感，可能假突破 | 减少假突破，可能漏信号 | +| `volume_multiplier` | 允许轻度放量 | 要求更强放量确认 | + +--- + +## 输出结构 + +### `DetectionResult` + +检测结果类。 + +```python +@dataclass +class DetectionResult: + # 核心输出 + strength: float # 综合强度分 (-1 ~ +1) + is_valid: bool # 是否检测到有效形态 + direction: BreakoutDirection # 突破方向 + + # 辅助输出 + strength_up: float # 向上强度原始值 (0~1) + strength_down: float # 向下强度原始值 (0~1) + strength_components: StrengthComponents # 强度分分量 + + # 形态几何属性 + width_ratio: float # 收敛比例 + touches_upper: int # 触碰上沿次数 + touches_lower: int # 触碰下沿次数 + volume_confirmed: bool # 成交量确认 + + # 前端数据 + chart_data: ChartData # 图表数据 +``` + +### 核心字段：`strength` + +**范围：-1 ~ +1** + +| 范围 | 含义 | 建议操作 | +|------|------|----------| +| +0.7 ~ +1.0 | 强向上突破 ⬆️⬆️⬆️ | 重点关注 | +| +0.4 ~ +0.7 | 中等向上 ⬆️⬆️ | 关注 | +| +0.2 ~ +0.4 | 向上潜力 ⬆️ | 观察 | +| 0 | 无效形态 | 忽略 | +| -0.2 ~ -0.4 | 向下潜力 ⬇️ | 观察 | +| -0.4 ~ -0.7 | 中等向下 ⬇️⬇️ | 关注 | +| -0.7 ~ -1.0 | 强向下突破 ⬇️⬇️⬇️ | 重点关注 | + +### `StrengthComponents` + +强度分各分量，便于理解分数构成。 + +| 分量 | 权重 | 说明 | +|------|------|------| +| `price_score` | 50% | 价格突破幅度 | +| `convergence_score` | 15% | 收敛程度 | +| `volume_score` | 10% | 放量程度 | +| `fitting_score` | 10% | 拟合贴合度 | +| `utilization_score` | 15% | 边界利用率 | + +--- + +## 筛选示例 + +### 基础筛选 + +```python +from triangle_detector_api import detect_matrix + +# 批量检测 +df = detect_matrix(high, low, close, volume, dates, tkrs, tkrs_name) + +# 强向上突破 +df[df['strength'] > 0.5] + +# 强向下突破 +df[df['strength'] < -0.5] + +# 任意强信号 +df[abs(df['strength']) > 0.3] + +# 按强度排序 +df.sort_values('strength', ascending=False) +``` + +### 按日期筛选 + +```python +# 某日的检测结果 +target_date = 20260120 +df_day = df[df['date'] == target_date] + +# 某日最强的10只股票 +df_day.sort_values('strength', ascending=False).head(10) +``` + +### 按股票筛选 + +```python +# 某只股票的历史检测结果 +df_stock = df[df['stock_code'] == '000001.SZ'] + +# 查看该股票的强度分变化 +df_stock[['date', 'strength', 'direction']].tail(20) +``` + +### 统计分析 + +```python +# 每日有效形态数量 +df.groupby('date')['is_valid'].sum() + +# 每日强向上突破数量 +df[df['strength'] > 0.5].groupby('date').size() + +# 强度分分布 +df['strength'].describe() +``` + +--- + +## 前端集成 + +### Vue3 + ECharts + +**1. 安装依赖** + +```bash +npm install echarts vue-echarts +``` + +**2. 获取 ECharts 配置** + +```python +result = detect_triangle(ohlc_data) + +if result.chart_data: + echarts_option = result.chart_data.to_echarts_option("股票名称") + # echarts_option 可直接用于前端 +``` + +**3. Vue3 组件使用** + +```vue + + + +``` + +### ChartData 结构 + +```python +@dataclass +class ChartData: + candlestick: List[List[float]] # K线数据 [[open, close, low, high], ...] + dates: List[str] # 日期数组 + volumes: List[float] # 成交量 + upper_line: Optional[TrendLine] # 上沿趋势线 + lower_line: Optional[TrendLine] # 下沿趋势线 + pivots: List[PivotPoint] # 枢轴点列表 + detection_window: Tuple[int, int] # 检测窗口范围 +``` + +**注意：** ECharts K线格式为 `[open, close, low, high]`，与标准 OHLC 顺序不同。 + +--- + +## 完整示例 + +```python +from triangle_detector_api import detect_triangle, DetectionParams, quick_detect + +# ============ 示例1：基础使用 ============ +ohlc_data = { + 'dates': dates_list, + 'open': open_list, + 'high': high_list, + 'low': low_list, + 'close': close_list, + 'volume': volume_list, +} + +result = detect_triangle(ohlc_data) + +print(f"是否有效形态: {result.is_valid}") +print(f"强度分: {result.strength}") +print(f"突破方向: {result.direction.value}") + +if result.is_valid: + print(f"收敛比例: {result.width_ratio:.2f}") + print(f"触碰次数: 上沿 {result.touches_upper}, 下沿 {result.touches_lower}") + +# ============ 示例2：自定义参数 ============ +params = DetectionParams( + window=120, # 短窗口，检测短期形态 + min_convergence=0.6, # 放宽收敛要求 + breakout_threshold=0.01, # 更严格的突破阈值 +) +result = detect_triangle(ohlc_data, params) + +# ============ 示例3：快速筛选 ============ +strength = quick_detect( + dates=dates_list, + open_=open_list, + high=high_list, + low=low_list, + close=close_list, + volume=volume_list, +) + +if strength > 0.5: + print("发现强向上突破信号！") +elif strength < -0.5: + print("发现强向下突破信号！") + +# ============ 示例4：获取前端数据 ============ +if result.chart_data: + # 获取 ECharts 配置 + echarts_option = result.chart_data.to_echarts_option("000001.SZ 平安银行") + + # 转为 JSON（用于 API 响应） + import json + json_data = json.dumps(echarts_option, ensure_ascii=False) +``` + +--- + +## 常见问题 + +### Q: 为什么强度分为 0？ + +**可能原因：** +1. 数据长度不足（需要至少 `window` 天的数据） +2. 未形成有效的收敛三角形形态 +3. 斜率不满足收敛要求（上沿应向下，下沿应向上） + +### Q: 如何调整检测敏感度？ + +```python +# 更严格（高质量形态） +params = DetectionParams( + min_convergence=0.4, # 要求更强收敛 + breakout_threshold=0.01, # 要求更明显突破 + volume_multiplier=1.8, # 要求更强放量 +) + +# 更宽松（更多候选） +params = DetectionParams( + min_convergence=0.7, # 放宽收敛要求 + breakout_threshold=0.002, # 允许轻微突破 + volume_multiplier=1.2, # 允许轻度放量 +) +``` + +### Q: 图表数据格式与其他库不兼容？ + +ECharts K线格式为 `[open, close, low, high]`。 + +如需标准 OHLC 格式 `[open, high, low, close]`，可自行转换： + +```python +# ECharts -> 标准 OHLC +def to_standard_ohlc(echarts_data): + return [[row[0], row[3], row[2], row[1]] for row in echarts_data] +``` + +--- + +## 版本信息 + +- **模块**: `triangle_detector_api.py` +- **核心依赖**: `converging_triangle.py`（自动使用 Numba 优化版本） +- **Python**: 3.8+ +- **NumPy**: 1.20+ +- **Numba**: 0.55+（可选，安装后自动启用加速） + +> **性能优化说明**：`converging_triangle.py` 会自动检测并加载 `converging_triangle_optimized.py`（Numba 加速版）。启动时会打印： +> - `[性能优化] 已启用Numba加速` - 优化版本生效 +> - `[性能优化] 未启用Numba加速` - 回退到原版（numba 未安装） diff --git a/src/triangle_detector_api.py b/src/triangle_detector_api.py new file mode 100644 index 0000000..edd3dc0 --- /dev/null +++ b/src/triangle_detector_api.py @@ -0,0 +1,1001 @@ +""" +收敛三角形检测 API + +主函数: detect_matrix() - 全市场矩阵批量检测 +核心输出: strength (-1 ~ +1)，正=向上，负=向下，绝对值=强度 + +Example: + df = detect_matrix(high, low, close, volume, dates, tkrs, tkrs_name) + df[df['strength'] > 0.5] # 强向上突破 + +详细文档: docs/triangle_api_reference.md +""" + +from __future__ import annotations + +import sys +import os +from dataclasses import dataclass, field, asdict +from typing import Dict, List, Literal, Optional, Tuple, Any, Union +from enum import Enum + +import numpy as np + +# 添加路径以导入核心模块 +sys.path.append(os.path.dirname(__file__)) + +from converging_triangle import ( + ConvergingTriangleParams, + detect_converging_triangle, + pivots_fractal, + pivots_fractal_hybrid, + fit_pivot_line_dispatch, + line_y, +) + + +# ============================================================================ +# 枚举类型定义 +# ============================================================================ + +class BreakoutDirection(str, Enum): + """突破方向""" + UP = "up" # 向上突破 + DOWN = "down" # 向下突破 + NONE = "none" # 未突破 + + +# ============================================================================ +# 参数对象（用户可配置） +# ============================================================================ + +@dataclass +class DetectionParams: + """检测参数，都有合理默认值，大多数场景无需调整""" + # ===== 用户可调参数 ===== + window: int = 240 + min_convergence: float = 0.45 + breakout_threshold: float = 0.005 + volume_multiplier: float = 1.5 + + # ===== 内部参数（已优化，无需调整） ===== + _realtime_mode: bool = field(default=True, repr=False) + _pivot_lookback: int = field(default=15, repr=False) + + def to_internal_params(self) -> ConvergingTriangleParams: + """转换为内部检测参数""" + return ConvergingTriangleParams( + window=self.window, + pivot_k=self._pivot_lookback, + boundary_n_segments=2, + boundary_source="full", + fitting_method="anchor", + upper_slope_max=0, # 上沿必须向下或水平 + lower_slope_min=0, # 下沿必须向上或水平 + touch_tol=0.10, + touch_loss_max=0.10, + shrink_ratio=self.min_convergence, + break_tol=self.breakout_threshold, + vol_window=20, + vol_k=self.volume_multiplier, + false_break_m=5, + ) + + +# ============================================================================ +# 图表数据结构（前端绑定用） +# ============================================================================ + +@dataclass +class TrendLine: + """ + 趋势线数据（用于 ECharts markLine） + + 前端使用示例（ECharts）： + markLine: { + data: [ + [{coord: [startIndex, startPrice]}, {coord: [endIndex, endPrice]}] + ] + } + """ + start_index: int # 起点X坐标（K线索引） + start_price: float # 起点Y坐标（价格） + end_index: int # 终点X坐标 + end_price: float # 终点Y坐标 + slope: float # 斜率 + intercept: float # 截距 + + def to_echarts_markline(self) -> List[Dict]: + """转换为 ECharts markLine 格式""" + return [ + {"coord": [self.start_index, self.start_price]}, + {"coord": [self.end_index, self.end_price]} + ] + + +@dataclass +class PivotPoint: + """枢轴点数据""" + index: int + price: float + is_high: bool + is_confirmed: bool = True + + +@dataclass +class ChartData: + """前端图表数据，to_echarts_option() 直接生成 ECharts 配置""" + # K线数据（ECharts candlestick 格式） + candlestick: List[List[float]] # [[open, close, low, high], ...] + dates: List[str] # 日期数组 + volumes: List[float] # 成交量数组 + + # 趋势线 + upper_line: Optional[TrendLine] = None + lower_line: Optional[TrendLine] = None + + # 枢轴点 + pivots: List[PivotPoint] = field(default_factory=list) + + # 检测窗口范围 + detection_window: Optional[Tuple[int, int]] = None + + # 交汇点（三角形顶点） + apex_index: Optional[float] = None # 可能在未来 + + def to_echarts_option(self, title: str = "") -> Dict[str, Any]: + """ + 生成完整的 ECharts 配置项 + + 可直接用于 Vue3 + ECharts: + + + Returns: + ECharts option 配置对象 + """ + option = { + "title": {"text": title, "left": "center"}, + "tooltip": { + "trigger": "axis", + "axisPointer": {"type": "cross"} + }, + "legend": { + "data": ["K线", "成交量"], + "top": "30" + }, + "grid": [ + {"left": "10%", "right": "10%", "top": "15%", "height": "55%"}, + {"left": "10%", "right": "10%", "top": "75%", "height": "15%"} + ], + "xAxis": [ + { + "type": "category", + "data": self.dates, + "gridIndex": 0, + "axisLabel": {"rotate": 45} + }, + { + "type": "category", + "data": self.dates, + "gridIndex": 1, + "axisLabel": {"show": False} + } + ], + "yAxis": [ + {"type": "value", "gridIndex": 0, "scale": True}, + {"type": "value", "gridIndex": 1, "scale": True} + ], + "dataZoom": [ + {"type": "inside", "xAxisIndex": [0, 1], "start": 50, "end": 100}, + {"type": "slider", "xAxisIndex": [0, 1], "start": 50, "end": 100} + ], + "series": [] + } + + # K线系列 + candlestick_series = { + "name": "K线", + "type": "candlestick", + "data": self.candlestick, + "xAxisIndex": 0, + "yAxisIndex": 0, + "itemStyle": { + "color": "#ef232a", # 涨（阳线） + "color0": "#14b143", # 跌（阴线） + "borderColor": "#ef232a", + "borderColor0": "#14b143" + } + } + + # 添加趋势线（markLine） + if self.upper_line or self.lower_line: + mark_line_data = [] + + if self.upper_line: + mark_line_data.append(self.upper_line.to_echarts_markline() + [ + {"lineStyle": {"color": "#ef232a", "type": "dashed", "width": 2}} + ]) + + if self.lower_line: + mark_line_data.append(self.lower_line.to_echarts_markline() + [ + {"lineStyle": {"color": "#14b143", "type": "dashed", "width": 2}} + ]) + + candlestick_series["markLine"] = { + "symbol": "none", + "data": mark_line_data + } + + option["series"].append(candlestick_series) + + # 成交量系列 + option["series"].append({ + "name": "成交量", + "type": "bar", + "data": self.volumes, + "xAxisIndex": 1, + "yAxisIndex": 1, + "itemStyle": {"color": "#7fbbe9"} + }) + + # 枢轴点系列（如果有） + if self.pivots: + high_pivots = [[p.index, p.price] for p in self.pivots if p.is_high] + low_pivots = [[p.index, p.price] for p in self.pivots if not p.is_high] + + if high_pivots: + option["series"].append({ + "name": "高点", + "type": "scatter", + "data": high_pivots, + "symbol": "triangle", + "symbolSize": 10, + "itemStyle": {"color": "#ef232a"} + }) + + if low_pivots: + option["series"].append({ + "name": "低点", + "type": "scatter", + "data": low_pivots, + "symbol": "triangle", + "symbolRotate": 180, + "symbolSize": 10, + "itemStyle": {"color": "#14b143"} + }) + + return option + + def to_dict(self) -> Dict[str, Any]: + """转换为字典（JSON序列化友好）""" + return { + "candlestick": self.candlestick, + "dates": self.dates, + "volumes": self.volumes, + "upper_line": asdict(self.upper_line) if self.upper_line else None, + "lower_line": asdict(self.lower_line) if self.lower_line else None, + "pivots": [asdict(p) for p in self.pivots], + "detection_window": self.detection_window, + "apex_index": self.apex_index, + } + + +# ============================================================================ +# 检测结果 +# ============================================================================ + +@dataclass +class StrengthComponents: + """ + 强度分分量（用于分析和可视化） + + 总强度 = 价格分×50% + 收敛分×15% + 成交量分×10% + 拟合分×10% + 边界利用率×15% + """ + price_score: float # 价格突破分数 (0~1) + convergence_score: float # 收敛分数 (0~1) + volume_score: float # 成交量分数 (0~1) + fitting_score: float # 拟合贴合度分数 (0~1) + utilization_score: float # 边界利用率分数 (0~1) + + def to_dict(self) -> Dict[str, float]: + return asdict(self) + + +@dataclass +class DetectionResult: + """ +检测结果，strength 为核心输出（正=向上，负=向下）""" + # 核心输出 + strength: float # 正=向上突破, 负=向下突破, 0=未突破/无效 + is_valid: bool + direction: BreakoutDirection + + # 辅助输出 + strength_up: float = 0.0 # 向上强度原始值 (0~1) + strength_down: float = 0.0 # 向下强度原始值 (0~1) + strength_components: Optional[StrengthComponents] = None + + # 形态几何属性 + width_ratio: float = 0.0 # 收敛比例（末端/起始宽度） + touches_upper: int = 0 # 触碰上沿次数 + touches_lower: int = 0 # 触碰下沿次数 + volume_confirmed: Optional[bool] = None # 成交量确认 + + # 前端数据 + chart_data: Optional[ChartData] = None + + # 检测模式信息 + detection_mode: str = "standard" + + def to_dict(self) -> Dict[str, Any]: + """转换为字典（JSON序列化友好）""" + return { + "strength": self.strength, + "is_valid": self.is_valid, + "direction": self.direction.value, + "strength_up": self.strength_up, + "strength_down": self.strength_down, + "strength_components": self.strength_components.to_dict() if self.strength_components else None, + "width_ratio": self.width_ratio, + "touches_upper": self.touches_upper, + "touches_lower": self.touches_lower, + "volume_confirmed": self.volume_confirmed, + "chart_data": self.chart_data.to_dict() if self.chart_data else None, + "detection_mode": self.detection_mode, + } + + +# ============================================================================ +# 核心检测函数 +# ============================================================================ + +def detect_triangle( + ohlc_data: Dict[str, Union[List, np.ndarray]], + params: Optional[DetectionParams] = None, + include_pivots: bool = True, + display_window: Optional[int] = None, +) -> DetectionResult: + """ + 主检测函数 + + Args: + ohlc_data: OHLCV 数据字典 {dates, open, high, low, close, volume?} + params: 检测参数，默认使用推荐值 + include_pivots: 是否包含枢轴点数据 + display_window: 图表显示窗口大小 + + Returns: + DetectionResult: strength 为核心输出 + + Example: + >>> data = { + ... 'dates': ['2026-01-20', '2026-01-21', ...], + ... 'open': [10.1, 10.2, ...], + ... 'high': [10.5, 10.6, ...], + ... 'low': [9.9, 10.0, ...], + ... 'close': [10.3, 10.4, ...], + ... 'volume': [1000000, 1200000, ...] + ... } + >>> result = detect_triangle(data) + >>> print(f"强度: {result.strength}, 方向: {result.direction}") + + >>> # 自定义参数 + >>> params = DetectionParams( + ... window=180, + ... min_convergence=0.6, + ... ) + >>> result = detect_triangle(data, params) + + >>> # 获取 ECharts 配置 + >>> if result.chart_data: + ... echarts_option = result.chart_data.to_echarts_option("股票名称") + """ + # 使用默认参数 + if params is None: + params = DetectionParams() + + # 转换为内部参数 + internal_params = params.to_internal_params() + + # 提取数据 + dates = ohlc_data.get('dates', []) + open_arr = np.array(ohlc_data.get('open', []), dtype=float) + high_arr = np.array(ohlc_data.get('high', []), dtype=float) + low_arr = np.array(ohlc_data.get('low', []), dtype=float) + close_arr = np.array(ohlc_data.get('close', []), dtype=float) + volume_arr = np.array(ohlc_data.get('volume', []), dtype=float) if 'volume' in ohlc_data else None + + n = len(close_arr) + + # 数据验证 + if n < internal_params.window: + return _create_invalid_result( + "数据长度不足", + ohlc_data, dates, open_arr, high_arr, low_arr, close_arr, volume_arr, + display_window + ) + + # 过滤NaN + valid_mask = ~np.isnan(close_arr) + valid_indices = np.where(valid_mask)[0] + + if len(valid_indices) < internal_params.window: + return _create_invalid_result( + "有效数据长度不足", + ohlc_data, dates, open_arr, high_arr, low_arr, close_arr, volume_arr, + display_window + ) + + # 提取检测窗口数据 + valid_end = len(valid_indices) - 1 + detect_start = valid_end - internal_params.window + 1 + + high_win = high_arr[valid_mask][detect_start:valid_end + 1] + low_win = low_arr[valid_mask][detect_start:valid_end + 1] + close_win = close_arr[valid_mask][detect_start:valid_end + 1] + volume_win = volume_arr[valid_mask][detect_start:valid_end + 1] if volume_arr is not None else None + + # 灵活区域大小（与 pivot_lookback 一致） + flexible_zone = params.pivot_lookback + + # 调用核心检测函数 + result = detect_converging_triangle( + high=high_win, + low=low_win, + close=close_win, + volume=volume_win, + params=internal_params, + stock_idx=0, + date_idx=n - 1, + real_time_mode=params._realtime_mode, + flexible_zone=flexible_zone, + ) + + # 计算综合强度分（正=向上，负=向下，0=无效/未突破） + if result.is_valid: + if result.breakout_dir == "up": + strength = result.breakout_strength_up + direction = BreakoutDirection.UP + elif result.breakout_dir == "down": + strength = -result.breakout_strength_down # 向下为负值 + direction = BreakoutDirection.DOWN + else: + # 形态有效但未突破，返回较大方向的强度的较小版本 + if result.breakout_strength_up >= result.breakout_strength_down: + strength = result.breakout_strength_up * 0.5 # 潜在向上 + else: + strength = -result.breakout_strength_down * 0.5 # 潜在向下 + direction = BreakoutDirection.NONE + else: + strength = 0.0 + direction = BreakoutDirection.NONE + + # 构建强度分量 + strength_components = StrengthComponents( + price_score=result.price_score_up if result.breakout_dir != "down" else result.price_score_down, + convergence_score=result.convergence_score, + volume_score=result.volume_score, + fitting_score=result.fitting_score, + utilization_score=result.boundary_utilization, + ) + + # 构建前端图表数据 + chart_data = _build_chart_data( + dates=dates, + open_arr=open_arr, + high_arr=high_arr, + low_arr=low_arr, + close_arr=close_arr, + volume_arr=volume_arr, + valid_mask=valid_mask, + valid_indices=valid_indices, + detect_start=detect_start, + internal_params=internal_params, + result=result, + include_pivots=include_pivots, + display_window=display_window, + realtime_mode=params._realtime_mode, + flexible_zone=flexible_zone, + ) + + return DetectionResult( + strength=round(strength, 4), + is_valid=result.is_valid, + direction=direction, + strength_up=round(result.breakout_strength_up, 4), + strength_down=round(result.breakout_strength_down, 4), + strength_components=strength_components, + width_ratio=round(result.width_ratio, 4), + touches_upper=result.touches_upper, + touches_lower=result.touches_lower, + volume_confirmed=result.volume_confirmed, + chart_data=chart_data, + detection_mode="realtime" if params._realtime_mode else "standard", + ) + + +def _create_invalid_result( + reason: str, + ohlc_data: Dict, + dates: List, + open_arr: np.ndarray, + high_arr: np.ndarray, + low_arr: np.ndarray, + close_arr: np.ndarray, + volume_arr: Optional[np.ndarray], + display_window: Optional[int], +) -> DetectionResult: + """创建无效结果（数据不足等情况）""" + # 构建基础图表数据（仅K线，无趋势线） + chart_data = _build_basic_chart_data( + dates=dates, + open_arr=open_arr, + high_arr=high_arr, + low_arr=low_arr, + close_arr=close_arr, + volume_arr=volume_arr, + display_window=display_window, + ) + + return DetectionResult( + strength=0.0, + is_valid=False, + direction=BreakoutDirection.NONE, + chart_data=chart_data, + detection_mode="invalid", + ) + + +def _build_basic_chart_data( + dates: List, + open_arr: np.ndarray, + high_arr: np.ndarray, + low_arr: np.ndarray, + close_arr: np.ndarray, + volume_arr: Optional[np.ndarray], + display_window: Optional[int], +) -> ChartData: + """构建基础图表数据（仅K线，无趋势线）""" + n = len(close_arr) + + # 确定显示范围 + if display_window and display_window < n: + start = n - display_window + else: + start = 0 + + # 日期格式转换 + display_dates = _format_dates(dates[start:]) + + # 构建 K 线数据（ECharts 格式：[open, close, low, high]） + candlestick = [] + for i in range(start, n): + candlestick.append([ + float(open_arr[i]) if not np.isnan(open_arr[i]) else 0, + float(close_arr[i]) if not np.isnan(close_arr[i]) else 0, + float(low_arr[i]) if not np.isnan(low_arr[i]) else 0, + float(high_arr[i]) if not np.isnan(high_arr[i]) else 0, + ]) + + # 成交量 + volumes = [] + if volume_arr is not None: + for i in range(start, n): + volumes.append(float(volume_arr[i]) if not np.isnan(volume_arr[i]) else 0) + + return ChartData( + candlestick=candlestick, + dates=display_dates, + volumes=volumes, + ) + + +def _build_chart_data( + dates: List, + open_arr: np.ndarray, + high_arr: np.ndarray, + low_arr: np.ndarray, + close_arr: np.ndarray, + volume_arr: Optional[np.ndarray], + valid_mask: np.ndarray, + valid_indices: np.ndarray, + detect_start: int, + internal_params: ConvergingTriangleParams, + result, + include_pivots: bool, + display_window: Optional[int], + realtime_mode: bool, + flexible_zone: int, +) -> ChartData: + """构建完整图表数据（包含趋势线和枢轴点）""" + valid_end = len(valid_indices) - 1 + n_valid = len(valid_indices) + + # 确定显示范围 + if display_window and display_window < n_valid: + display_start = n_valid - display_window + else: + display_start = 0 + + # 提取显示数据 + display_indices = valid_indices[display_start:valid_end + 1] + display_dates_raw = [dates[i] for i in display_indices] + display_dates = _format_dates(display_dates_raw) + + display_open = open_arr[valid_mask][display_start:valid_end + 1] + display_high = high_arr[valid_mask][display_start:valid_end + 1] + display_low = low_arr[valid_mask][display_start:valid_end + 1] + display_close = close_arr[valid_mask][display_start:valid_end + 1] + display_volume = volume_arr[valid_mask][display_start:valid_end + 1] if volume_arr is not None else [] + + # 构建 K 线数据 + candlestick = [] + for i in range(len(display_close)): + candlestick.append([ + float(display_open[i]) if not np.isnan(display_open[i]) else 0, + float(display_close[i]) if not np.isnan(display_close[i]) else 0, + float(display_low[i]) if not np.isnan(display_low[i]) else 0, + float(display_high[i]) if not np.isnan(display_high[i]) else 0, + ]) + + volumes = [float(v) if not np.isnan(v) else 0 for v in display_volume] if len(display_volume) > 0 else [] + + # 如果没有有效形态，返回基础数据 + if not result.is_valid: + return ChartData( + candlestick=candlestick, + dates=display_dates, + volumes=volumes, + ) + + # 计算趋势线（在检测窗口内） + high_win = high_arr[valid_mask][detect_start:valid_end + 1] + low_win = low_arr[valid_mask][detect_start:valid_end + 1] + close_win = close_arr[valid_mask][detect_start:valid_end + 1] + n_win = len(close_win) + + # 计算枢轴点 + if realtime_mode: + confirmed_ph, confirmed_pl, candidate_ph, candidate_pl = pivots_fractal_hybrid( + high_win, low_win, k=internal_params.pivot_k, flexible_zone=flexible_zone + ) + ph_idx = np.concatenate([confirmed_ph, candidate_ph]) if len(candidate_ph) > 0 else confirmed_ph + pl_idx = np.concatenate([confirmed_pl, candidate_pl]) if len(candidate_pl) > 0 else confirmed_pl + ph_idx = np.sort(ph_idx) + pl_idx = np.sort(pl_idx) + else: + ph_idx, pl_idx = pivots_fractal(high_win, low_win, k=internal_params.pivot_k) + + # 拟合边界线 + a_u, b_u, selected_ph = fit_pivot_line_dispatch( + pivot_indices=ph_idx, + pivot_values=high_win[ph_idx], + mode="upper", + method=internal_params.fitting_method, + all_prices=high_win, + window_start=0, + window_end=n_win - 1, + ) + a_l, b_l, selected_pl = fit_pivot_line_dispatch( + pivot_indices=pl_idx, + pivot_values=low_win[pl_idx], + mode="lower", + method=internal_params.fitting_method, + all_prices=low_win, + window_start=0, + window_end=n_win - 1, + ) + + # 计算显示窗口中的偏移 + triangle_offset_in_display = len(candlestick) - n_win + + # 构建趋势线 + upper_line = TrendLine( + start_index=triangle_offset_in_display, + start_price=float(a_u * 0 + b_u), + end_index=triangle_offset_in_display + n_win - 1, + end_price=float(a_u * (n_win - 1) + b_u), + slope=float(a_u), + intercept=float(b_u), + ) + + lower_line = TrendLine( + start_index=triangle_offset_in_display, + start_price=float(a_l * 0 + b_l), + end_index=triangle_offset_in_display + n_win - 1, + end_price=float(a_l * (n_win - 1) + b_l), + slope=float(a_l), + intercept=float(b_l), + ) + + # 构建枢轴点 + pivots = [] + if include_pivots: + # 高点枢轴点 + for idx in ph_idx: + pivots.append(PivotPoint( + index=int(triangle_offset_in_display + idx), + price=float(high_win[idx]), + is_high=True, + is_confirmed=idx in confirmed_ph if realtime_mode else True, + )) + # 低点枢轴点 + for idx in pl_idx: + pivots.append(PivotPoint( + index=int(triangle_offset_in_display + idx), + price=float(low_win[idx]), + is_high=False, + is_confirmed=idx in confirmed_pl if realtime_mode else True, + )) + + # 计算交汇点 + denom = a_u - a_l + apex_x = float((b_l - b_u) / denom) if abs(denom) > 1e-12 else None + if apex_x is not None: + apex_x = triangle_offset_in_display + apex_x + + return ChartData( + candlestick=candlestick, + dates=display_dates, + volumes=volumes, + upper_line=upper_line, + lower_line=lower_line, + pivots=pivots, + detection_window=(triangle_offset_in_display, triangle_offset_in_display + n_win - 1), + apex_index=apex_x, + ) + + +def _format_dates(dates: List) -> List[str]: + """格式化日期列表为字符串""" + result = [] + for d in dates: + if isinstance(d, (int, np.integer)): + # 整数格式如 20260120 -> "2026-01-20" + s = str(d) + if len(s) == 8: + result.append(f"{s[:4]}-{s[4:6]}-{s[6:]}") + else: + result.append(str(d)) + else: + result.append(str(d)) + return result + + +# ============================================================================ +# 核心函数：矩阵批量检测 +# ============================================================================ + +def detect_matrix( + high_mtx: np.ndarray, + low_mtx: np.ndarray, + close_mtx: np.ndarray, + volume_mtx: np.ndarray, + dates: np.ndarray, + tkrs: Optional[np.ndarray] = None, + tkrs_name: Optional[np.ndarray] = None, + params: Optional[DetectionParams] = None, + start_day: Optional[int] = None, + end_day: Optional[int] = None, + only_valid: bool = True, + verbose: bool = False, +): + """ + 全市场矩阵批量检测：每只股票每天都进行检测 + + 适用于万得全A等全市场扫描场景 + + Args: + high_mtx: 最高价矩阵 (n_stocks, n_days) + low_mtx: 最低价矩阵 + close_mtx: 收盘价矩阵 + volume_mtx: 成交量矩阵 + dates: 日期数组 (n_days,) + tkrs: 股票代码数组 (n_stocks,) + tkrs_name: 股票名称数组 (n_stocks,) + params: 检测参数 + start_day: 起始日索引，默认 window-1 + end_day: 结束日索引，默认最后一天 + only_valid: 只返回有效形态，默认 True + verbose: 打印进度 + + Returns: + DataFrame: stock_idx, stock_code, stock_name, date_idx, date, + strength, is_valid, direction, width_ratio + """ + import pandas as pd + from converging_triangle import detect_converging_triangle_batch + + if params is None: + params = DetectionParams() + + internal_params = params.to_internal_params() + + n_stocks, n_days = close_mtx.shape + + # 计算范围 + if start_day is None: + start_day = internal_params.window - 1 + if end_day is None: + # 找最后有效日 + any_valid = np.any(~np.isnan(close_mtx), axis=0) + valid_day_idx = np.where(any_valid)[0] + end_day = valid_day_idx[-1] if len(valid_day_idx) > 0 else n_days - 1 + + start_day = max(internal_params.window - 1, start_day) + end_day = min(n_days - 1, end_day) + + if verbose: + total_points = n_stocks * (end_day - start_day + 1) + print(f"检测范围: {n_stocks} 只股票 × {end_day - start_day + 1} 天 = {total_points} 个检测点") + + # 调用底层批量检测 + df = detect_converging_triangle_batch( + open_mtx=close_mtx, # 开盘价用收盘价代替（不影响三角形检测） + high_mtx=high_mtx, + low_mtx=low_mtx, + close_mtx=close_mtx, + volume_mtx=volume_mtx, + params=internal_params, + start_day=start_day, + end_day=end_day, + only_valid=only_valid, + verbose=verbose, + real_time_mode=params._realtime_mode, + flexible_zone=params._pivot_lookback, + ) + + if len(df) == 0: + return pd.DataFrame() + + # 计算带符号的 strength + df['strength'] = df.apply( + lambda row: row['breakout_strength_up'] if row['breakout_dir'] == 'up' + else -row['breakout_strength_down'] if row['breakout_dir'] == 'down' + else max(row['breakout_strength_up'], row['breakout_strength_down']) + if row['breakout_strength_up'] >= row['breakout_strength_down'] + else -row['breakout_strength_down'], + axis=1 + ) + + # 添加股票代码和名称 + if tkrs is not None: + df['stock_code'] = df['stock_idx'].map(lambda x: tkrs[x] if x < len(tkrs) else '') + if tkrs_name is not None: + df['stock_name'] = df['stock_idx'].map(lambda x: tkrs_name[x] if x < len(tkrs_name) else '') + + # 添加日期 + df['date'] = df['date_idx'].map(lambda x: dates[x] if x < len(dates) else 0) + + # 重命名列 + df = df.rename(columns={ + 'breakout_dir': 'direction', + }) + + # 选择输出列 + output_cols = ['stock_idx', 'date_idx', 'date', 'strength', 'is_valid', 'direction', 'width_ratio'] + if tkrs is not None: + output_cols.insert(1, 'stock_code') + if tkrs_name is not None: + output_cols.insert(2 if tkrs is not None else 1, 'stock_name') + + # 保留存在的列 + output_cols = [c for c in output_cols if c in df.columns] + + return df[output_cols] + + +# ============================================================================ +# 便捷函数 +# ============================================================================ + +def quick_detect( + dates: List, + open_: List[float], + high: List[float], + low: List[float], + close: List[float], + volume: Optional[List[float]] = None, +) -> float: + """快速检测，直接返回强度分""" + ohlc_data = { + 'dates': dates, + 'open': open_, + 'high': high, + 'low': low, + 'close': close, + } + if volume is not None: + ohlc_data['volume'] = volume + + params = DetectionParams() + result = detect_triangle(ohlc_data, params, include_pivots=False) + return result.strength + + +# ============================================================================ +# 测试/演示代码 +# ============================================================================ + +if __name__ == "__main__": + print("=" * 70) + print("收敛三角形检测 API 示例") + print("=" * 70) + + # 生成模拟数据（收敛形态） + np.random.seed(42) + n_days = 300 + + # 基础价格序列（带收敛趋势） + base_price = 10.0 + prices = [base_price] + + for i in range(1, n_days): + # 添加随机波动 + change = np.random.normal(0, 0.02) * prices[-1] + # 添加收敛效果（振幅逐渐缩小） + decay = 1 - (i / n_days) * 0.5 + change *= decay + prices.append(prices[-1] + change) + + prices = np.array(prices) + + # 生成 OHLC + dates = [f"2025-{(i//30)+1:02d}-{(i%30)+1:02d}" for i in range(n_days)] + opens = prices * (1 + np.random.normal(0, 0.005, n_days)) + highs = np.maximum(prices, opens) * (1 + np.abs(np.random.normal(0, 0.01, n_days))) + lows = np.minimum(prices, opens) * (1 - np.abs(np.random.normal(0, 0.01, n_days))) + closes = prices + volumes = np.random.randint(1000000, 5000000, n_days).astype(float) + + # 构建数据 + ohlc_data = { + 'dates': dates, + 'open': opens.tolist(), + 'high': highs.tolist(), + 'low': lows.tolist(), + 'close': closes.tolist(), + 'volume': volumes.tolist(), + } + + print("\n[1] 使用默认参数检测...") + result = detect_triangle(ohlc_data) + print(f" 有效形态: {result.is_valid}") + print(f" 强度分: {result.strength}") + print(f" 突破方向: {result.direction.value}") + + if result.strength_components: + print(f" 分量: 价格={result.strength_components.price_score:.3f}, " + f"收敛={result.strength_components.convergence_score:.3f}, " + f"成交量={result.strength_components.volume_score:.3f}") + + print("\n[2] 使用宽松参数检测...") + params_loose = DetectionParams(min_convergence=0.8, breakout_threshold=0.001) + result_loose = detect_triangle(ohlc_data, params_loose) + print(f" 有效形态: {result_loose.is_valid}") + print(f" 强度分: {result_loose.strength}") + + print("\n[3] 快速检测...") + strength = quick_detect( + dates=dates, + open_=opens.tolist(), + high=highs.tolist(), + low=lows.tolist(), + close=closes.tolist(), + volume=volumes.tolist(), + ) + print(f" 强度分: {strength}") + + print("\n[4] 图表数据...") + if result.chart_data: + print(f" K线数量: {len(result.chart_data.candlestick)}") + print(f" 有上沿线: {result.chart_data.upper_line is not None}") + print(f" 有下沿线: {result.chart_data.lower_line is not None}") + print(f" 枢轴点数: {len(result.chart_data.pivots)}") + + # 生成 ECharts 配置示例 + echarts_option = result.chart_data.to_echarts_option("示例股票") + print(f" ECharts配置键: {list(echarts_option.keys())}") + + print("\n详细文档: docs/triangle_api_reference.md") + print("=" * 70)