diff --git a/.cursor/plans/4-a1a89b85.plan.md b/.cursor/plans/4-a1a89b85.plan.md new file mode 100644 index 0000000..10f2513 --- /dev/null +++ b/.cursor/plans/4-a1a89b85.plan.md @@ -0,0 +1,132 @@ + +# 拟合线迭代离群点移除优化 + +## 问题描述 + +当前的分段选择算法会选中一些"弱"枢轴点用于拟合： + +- 上沿线：某个高点虽然是时间段内最高，但明显低于其他高点（如图中第二个点 5.8 元） +- 这些点会拉低/拉高拟合线，导致与主观判断不符 + +## 解决方案：迭代离群点移除 + +### 核心逻辑 + +``` +1. 初始拟合：用所有枢轴点做线性回归 +2. 计算残差：每个点到拟合线的偏差 +3. 识别离群点： + - 上沿线：价格明显低于拟合线的点 = 弱高点 + - 下沿线：价格明显高于拟合线的点 = 弱低点 +4. 移除最差的离群点 +5. 重新拟合 +6. 重复直到收敛 +``` + +### 算法流程图 + +```mermaid +flowchart TD + A[输入枢轴点] --> B[初始线性回归] + B --> C[计算残差] + C --> D{存在离群点?} + D -->|是| E[移除最大离群点] + E --> F{剩余点 >= 3?} + F -->|是| G{迭代次数 < 3?} + G -->|是| B + G -->|否| H[返回当前拟合] + F -->|否| H + D -->|否| H +``` + +## 代码修改 + +### 文件: [src/converging_triangle.py](src/converging_triangle.py) + +重写 `fit_pivot_line` 函数（第 230-350 行）: + +```python +def fit_pivot_line( + pivot_indices: np.ndarray, + pivot_values: np.ndarray, + mode: str = "upper", + min_points: int = 2, + outlier_threshold: float = 1.5, # 新增：离群点阈值（标准差倍数） + max_iterations: int = 3, # 新增：最大迭代次数 +) -> Tuple[float, float, np.ndarray]: + """ + 迭代离群点移除的枢轴点拟合算法 + + 策略： + 1. 先用所有点做初始拟合 + 2. 识别并移除偏离拟合线的"弱"点 + 3. 迭代直到收敛 + + 对于上沿线：移除价格明显低于拟合线的点 + 对于下沿线：移除价格明显高于拟合线的点 + """ +``` + +### 关键参数 + +| 参数 | 默认值 | 说明 | + +|------|--------|------| + +| `outlier_threshold` | 1.5 | 残差超过 1.5 倍标准差视为离群点 | + +| `max_iterations` | 3 | 最多迭代 3 次，避免过度过滤 | + +| `min_points` | 3 | 至少保留 3 个点用于拟合 | + +### 离群点判定逻辑 + +**上沿线（upper）**: + +```python +# 残差 = 拟合值 - 实际值 +# 正残差表示点在拟合线下方（弱高点） +residuals = fitted_values - actual_values +outliers = residuals > threshold # 弱高点 +``` + +**下沿线（lower）**: + +```python +# 残差 = 实际值 - 拟合值 +# 正残差表示点在拟合线上方（弱低点） +residuals = actual_values - fitted_values +outliers = residuals > threshold # 弱低点 +``` + +### 预期效果 + +以图中 SZ300278 为例： + +- 第二个点（5.8元）明显低于拟合线 +- 在第一次迭代后会被识别为离群点并移除 +- 最终拟合线只使用剩余 3 个更有代表性的高点 + +## 测试计划 + +1. 使用 SZ300278 验证修复效果 +2. 对比修改前后的图表 +3. 确保不会过度过滤正常的枢轴点 + +## 文档更新 + +更新 [docs/枢轴点分段选择算法详解.md](docs/枢轴点分段选择算法详解.md)，添加迭代离群点移除的说明。 + +### To-dos + +- [x] 在ConvergingTriangleResult中增加detection_mode等3个新字段 +- [x] 实现pivots_fractal_hybrid()函数 +- [x] 修改detect_converging_triangle()支持real_time_mode参数 +- [x] 修改detect_converging_triangle_batch()传递实时模式参数 +- [x] 在triangle_config.py中添加REALTIME_MODE和FLEXIBLE_ZONE配置 +- [x] 更新run_converging_triangle.py导入和使用实时配置 +- [x] 创建test_realtime_mode.py对比测试脚本 +- [x] 更新README.md和创建实时模式使用指南 +- [ ] 重写 fit_pivot_line 函数，实现迭代离群点移除算法 +- [ ] 测试 SZ300278，验证修复效果 +- [ ] 更新算法文档，添加迭代离群点移除说明 \ No newline at end of file diff --git a/USAGE.md b/USAGE.md index 7f8e3d7..0831bd2 100644 --- a/USAGE.md +++ b/USAGE.md @@ -37,6 +37,12 @@ python scripts/pipeline_converging_triangle.py # 指定日期 python scripts/pipeline_converging_triangle.py --date 20260120 +# 生成详情模式图片（显示所有枢轴点和拟合点） +python scripts/pipeline_converging_triangle.py --show-details + +# 组合使用 +python scripts/pipeline_converging_triangle.py --date 20260120 --show-details + # 跳过检测（仅生成报告与图表） python scripts/pipeline_converging_triangle.py --skip-detection ``` diff --git a/discuss/20260126-讨论.md b/discuss/20260126-讨论.md index e65fd39..f0e4c58 100644 --- a/discuss/20260126-讨论.md +++ b/discuss/20260126-讨论.md @@ -1,16 +1,61 @@ +## 问题1: 末端枢轴点未识别 + ![](images/2026-01-26-15-44-13.png) -下沿线，明显不对。 + +**现象**: 图表最右边明显的低点/高点没有被标记为枢轴点 + +**根因**: +1. 数据中有空值(NaN)导致比较失效 +2. 只检测窗口前235天的枢轴点，最后5天被忽略 +3. 绘图时丢弃了末端的"候选枢轴点" + +**解决**: +1. 用 `nanmin/nanmax` 替代 `min/max`，自动跳过空值 +2. 把"灵活区域"从5天扩大到15天，覆盖更多末端数据 +3. 对称短窗口：右边只有N天时，左边也只看N天（而不是固定15天） +4. 绘图时把"确认枢轴点"和"候选枢轴点"合并显示 + +详见 [枢轴点检测与可视化修复](../docs/2026-01-26_枢轴点检测与可视化修复.md) + +--- + +## 问题2: 拟合点选择不合理 ![](images/2026-01-26-15-48-56.png) -拟合线的时候，有些点应该去掉，跟主观判断不对齐，比如图中第二个点 + +**现象**: 某些"弱"枢轴点（如5.8元的低高点）被用于拟合，拉偏趋势线 + ![](images/2026-01-26-15-50-58.png) +**解决**: 迭代离群点移除算法 +1. 先用所有点画一条拟合线 +2. 找出离拟合线太远的"异常点" +3. 去掉最差的那个点，重新画线 +4. 重复2-3次，直到没有异常点 -### 强度分：是否符合三角形的形态 + 突破强度 +详见 [枢轴点拟合算法详解](../docs/枢轴点分段选择算法详解.md) -108个个股按照分数排序 +--- -可调参数、速度快 -后续回测后继续调优 +## 问题3: 非收敛形态误判 -历史曲线的每个点，都需要计算强度分。 \ No newline at end of file +**现象**: "上升三角形"（上沿水平）被误判为"收敛三角形" +![](images/2026-01-26-18-36-50.png) +**解决**: 收紧斜率限制 +- 上沿必须向下（或至少水平），不能向上 +- 下沿必须向上（或至少水平），不能向下 +- 这样就能过滤掉"上升三角形"和"下降通道"等非收敛形态 + +--- + +## 待办: 突破强度评分 + +| 分量 | 权重 | 说明 | +|------|------|------| +| 价格突破 | 60% | 突破幅度 | +| 收敛程度 | 25% | 蓄势充分度 | +| 成交量 | 15% | 放量确认 | + +- 枢轴点和拟合线距离。 + +后续回测调优。 diff --git a/discuss/images/2026-01-26-18-36-50.png b/discuss/images/2026-01-26-18-36-50.png new file mode 100644 index 0000000..266e90a Binary files /dev/null and b/discuss/images/2026-01-26-18-36-50.png differ diff --git a/docs/2026-01-26_图表详细模式功能.md b/docs/2026-01-26_图表详细模式功能.md index c227162..517cd16 100644 --- a/docs/2026-01-26_图表详细模式功能.md +++ b/docs/2026-01-26_图表详细模式功能.md @@ -1,16 +1,293 @@ # 图表详细模式功能说明 **日期**: 2026-01-26 -**功能**: 图表可视化改进 - 简洁模式与详细模式 +**功能**: 图表可视化改进 - 简洁模式与详细模式 +**版本**: v2.0 (移除分段竖线 + 流水线支持) --- ## 📋 功能概述 -为了满足不同使用场景的需求，我们为收敛三角形图表增加了**简洁模式**和**详细模式**两种显示方式： +为了满足不同使用场景的需求,我们为收敛三角形图表增加了**简洁模式**和**详细模式**两种显示方式： - **简洁模式（默认）**: 仅显示收盘价、上沿线、下沿线，图表清爽易读 -- **详细模式**: 显示所有枢轴点、拟合点、分段线等调试信息，便于理解算法 +- **详细模式**: 显示所有枢轴点、拟合点，便于理解算法 + +**v2.0 更新**: 移除分段竖线（算法已改为迭代离群点移除） + +--- + +## 🎯 使用场景 + +### 简洁模式 +- ✅ 日常使用和实盘选股 +- ✅ 快速查看三角形形态 +- ✅ 对外展示和报告 +- ✅ 减少视觉干扰 + +### 详细模式 +- ✅ 算法调试和验证 +- ✅ 理解枢轴点识别逻辑 +- ✅ 验证迭代拟合算法 +- ✅ 学习和教学用途 + +--- + +## 📊 两种模式对比 + +| 显示元素 | 简洁模式 | 详细模式 | 说明 | +|---------|---------|---------|------| +| **收盘价线** | ✅ | ✅ | 黑色实线 | +| **上沿线** | ✅ | ✅ | 红色虚线 | +| **下沿线** | ✅ | ✅ | 绿色虚线 | +| **所有高点枢轴点** | ❌ | ✅ | 浅红色小实心圆 | +| **所有低点枢轴点** | ❌ | ✅ | 浅绿色小实心圆 | +| **上沿拟合点** | ❌ | ✅ | 深红色大空心圆（迭代算法选出） | +| **下沿拟合点** | ❌ | ✅ | 深绿色大空心圆（迭代算法选出） | +| **分段竖线** | ❌ | ❌ | v2.0已移除 | +| **输出文件名** | `YYYYMMDD_代码_名称.png` | `YYYYMMDD_代码_名称_detail.png` | + +**文件名说明**: +- 简洁模式文件不带后缀 +- 详细模式文件带 `_detail` 后缀 +- 两种模式可以同时保留，方便对比查看 + +--- + +## 🛠️ 如何启用 + +### 方法1: 流水线脚本（推荐） + +```bash +# 简洁模式（默认） +python scripts/pipeline_converging_triangle.py + +# 详细模式 +python scripts/pipeline_converging_triangle.py --show-details + +# 组合使用 +python scripts/pipeline_converging_triangle.py --date 20260120 --show-details +``` + +### 方法2: 单独绘图脚本 + +```bash +# 简洁模式（默认） +python scripts/plot_converging_triangles.py + +# 详细模式 +python scripts/plot_converging_triangles.py --show-details +``` + +### 方法3: 配置文件（默认设置） + +编辑 `scripts/triangle_config.py`: + +```python +# 图表详细模式（显示枢轴点等调试信息） +SHOW_CHART_DETAILS = False # False=简洁模式（默认），True=详细模式 +``` + +**优先级**: 命令行参数 > 配置文件 + +**智能清理**: +- 简洁模式运行时，只清理简洁模式的旧图片 +- 详细模式运行时，只清理详细模式的旧图片 +- 两种模式互不影响，可以共存 + +--- + +## 📈 详细模式显示元素说明 + +### 1. 所有枢轴点（小实心圆） + +**作用**: 显示算法识别的所有局部高点和低点 + +- **高点枢轴点**: 浅红色，size=50，alpha=0.4 +- **低点枢轴点**: 浅绿色，size=50，alpha=0.4 + +**标签**: `所有高点枢轴点(N)` / `所有低点枢轴点(N)` + +### 2. 拟合点（大空心圆） + +**作用**: 显示迭代离群点移除算法最终选出的关键点 + +- **上沿拟合点**: 深红色空心圆，size=120，linewidth=2.5 +- **下沿拟合点**: 深绿色空心圆，size=120，linewidth=2.5 + +**标签**: `上沿拟合点(N)` / `下沿拟合点(N)` + +**说明**: +- 使用迭代离群点移除算法选出代表性强的枢轴点 +- 自动过滤"弱"枢轴点（偏离趋势线过大的点） +- 详见 [枢轴点拟合算法详解](./枢轴点分段选择算法详解.md) + +### 3. ~~分段竖线~~（已移除） + +**v2.0 更新**: 由于算法改为迭代离群点移除，不再使用分段策略，因此移除了分段竖线。 + +--- + +## 💡 图表解读示例 + +### 简洁模式（默认） + +``` +图表内容： +├─ 黑色实线：收盘价 +├─ 红色虚线：上沿线（向下） +└─ 绿色虚线：下沿线（向上） + +图例： +- 收盘价 +- 上沿 +- 下沿 +``` + +**优点**: +- 清晰直观 +- 无视觉干扰 +- 适合日常使用 + +--- + +### 详细模式（`--show-details`） + +``` +图表内容： +├─ 黑色实线：收盘价 +├─ 红色虚线：上沿线 +├─ 绿色虚线：下沿线 +├─ 浅红小圆：所有高点枢轴点 +├─ 浅绿小圆：所有低点枢轴点 +├─ 深红大圆：上沿拟合点（迭代算法选出） +└─ 深绿大圆：下沿拟合点（迭代算法选出） + +图例： +- 收盘价 +- 上沿 +- 下沿 +- 所有高点枢轴点(N) +- 所有低点枢轴点(N) +- 上沿拟合点(N) +- 下沿拟合点(N) +``` + +**优点**: +- 完整展示算法逻辑 +- 便于验证和调试 +- 有助于理解原理 + +--- + +## 🎨 视觉层次设计 + +图表元素的 Z-order（从后到前）： + +``` +1. 网格线（alpha=0.3） +2. 价格曲线（黑色，zorder=默认） +3. 所有枢轴点（浅色小圆，alpha=0.4，zorder=4） +4. 拟合点（深色大圆，zorder=5） +5. 趋势线（红/绿虚线，zorder=默认） +``` + +**设计原则**: +- 详细信息放在后层（不遮挡主要信息） +- 关键信息（拟合点）突出显示（zorder高） +- 使用透明度（alpha）区分重要性 + +--- + +## 📝 代码实现要点 + +### 1. 流水线参数传递 + +```python +# pipeline_converging_triangle.py +parser.add_argument("--show-details", action="store_true") + +# 传递给绘图脚本 +if args.show_details: + cmd_args.append("--show-details") +``` + +### 2. 绘图脚本参数 + +```python +# plot_converging_triangles.py +parser.add_argument( + "--show-details", + action="store_true", + help="显示详细调试信息（枢轴点、拟合点等）", +) +``` + +### 3. 条件渲染 + +```python +# 详细模式：显示所有枢轴点和拟合点 +if show_details: + # 绘制所有枢轴点 + ax1.scatter(ph_display_idx, high_win[ph_idx], ...) + + # 绘制拟合点 + ax1.scatter(selected_ph_display, high_win[selected_ph_pos], ...) +``` + +--- + +## ✅ 测试验证 + +### 测试用例1: 流水线简洁模式（默认） + +```bash +python scripts/pipeline_converging_triangle.py +``` + +**预期结果**: +- ✅ 控制台显示："图表模式: 简洁模式（仅显示价格和趋势线）" +- ✅ 图表只显示收盘价、上沿、下沿 +- ✅ 输出文件: `20260120_SZ300278_华昌达.png` + +### 测试用例2: 流水线详细模式 + +```bash +python scripts/pipeline_converging_triangle.py --show-details +``` + +**预期结果**: +- ✅ 控制台显示："图表模式: 详情模式（显示所有枢轴点）" +- ✅ 图表显示所有枢轴点和拟合点 +- ✅ 输出文件: `20260120_SZ300278_华昌达_detail.png` + +--- + +## 📚 相关文档 + +- [README.md](../README.md) - 项目概述 +- [USAGE.md](../USAGE.md) - 使用指南 +- [枢轴点拟合算法详解.md](./枢轴点分段选择算法详解.md) - 迭代离群点移除算法 ⭐ +- [枢轴点检测原理.md](./枢轴点检测原理.md) - 枢轴点算法说明 + +--- + +## 🎉 总结 + +通过添加**简洁模式**和**详细模式**，图表功能更加灵活： + +- **对于日常用户**: 简洁清爽的图表，快速查看形态 +- **对于研究者**: 完整的算法细节，深入理解逻辑 +- **对于开发者**: 便于调试和验证算法正确性 + +**v2.0 改进**: +- ✅ 移除了不再适用的分段竖线 +- ✅ 流水线脚本支持详情模式 +- ✅ 更符合新的迭代拟合算法 + +这个功能提升了项目的易用性和专业性！🎯 + + --- diff --git a/docs/2026-01-26_枢轴点检测与可视化修复.md b/docs/2026-01-26_枢轴点检测与可视化修复.md new file mode 100644 index 0000000..94f0f7b --- /dev/null +++ b/docs/2026-01-26_枢轴点检测与可视化修复.md @@ -0,0 +1,157 @@ +# 枢轴点检测与可视化修复 + +**日期**: 2026-01-26 +**版本**: v1.0 + +## 问题概述 + +收敛三角形检测系统存在多个问题，导致： +1. 非收敛形态（如上升三角形）被误判 +2. 绘图与检测结果不一致 +3. 窗口末端枢轴点无法识别 + +--- + +## 问题1: 斜率约束过于宽松 + +### 现象 +"上升三角形"（上沿水平+下沿向上）被误判为"收敛三角形"。 + +### 原因 +```python +upper_slope_max=0.10, # 允许上沿向上 +lower_slope_min=-0.10, # 允许下沿向下 +``` + +### 解决方案 +**文件**: `scripts/triangle_config.py` +```python +upper_slope_max=0, # 上沿必须向下或水平（≤0） +lower_slope_min=0, # 下沿必须向上或水平（≥0） +``` + +--- + +## 问题2: 绘图与检测枢轴点不一致 + +### 现象 +图表显示的枢轴点和趋势线与实际检测结果不同。 + +### 原因 +- 检测代码使用 `pivots_fractal_hybrid`（实时模式） +- 绘图代码使用 `pivots_fractal`（标准模式） + +### 解决方案 +**文件**: `scripts/plot_converging_triangles.py` +```python +from triangle_config import REALTIME_MODE, FLEXIBLE_ZONE + +if REALTIME_MODE: + confirmed_ph, confirmed_pl, candidate_ph, candidate_pl = pivots_fractal_hybrid( + high_win, low_win, k=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) +``` + +--- + +## 问题3: 窗口末端枢轴点识别失败 + +### 现象 +图表最右边明显的低点/高点没有被识别为枢轴点。 + +### 原因 +`FLEXIBLE_ZONE=5` 太小，末端枢轴点超出灵活区域范围。 + +### 解决方案 +**文件**: `scripts/triangle_config.py` +```python +FLEXIBLE_ZONE = 15 # 与 pivot_k 一致，确保末端枢轴点能被识别 +``` + +--- + +## 问题4: NaN值导致比较失败 + +### 现象 +包含 NaN 值的数据窗口中，枢轴点检测完全失效。 + +### 原因 +```python +np.min([30.0, np.nan, 29.0]) # 返回 nan +30.0 == nan # 永远为 False +``` + +### 解决方案 +**文件**: `src/converging_triangle.py` +```python +# 使用 nanmin/nanmax 忽略 NaN 值 +if not np.isnan(high[i]) and high[i] == np.nanmax(high[i - k : i + k + 1]): + confirmed_ph.append(i) +if not np.isnan(low[i]) and low[i] == np.nanmin(low[i - k : i + k + 1]): + confirmed_pl.append(i) +``` + +--- + +## 问题5: 候选枢轴点未合并 + +### 现象 +`pivots_fractal_hybrid` 返回的候选枢轴点在绘图时被丢弃。 + +### 原因 +```python +ph_idx, pl_idx, _, _ = pivots_fractal_hybrid(...) # 候选点被忽略 +``` + +### 解决方案 +见问题2的解决方案，合并 confirmed 和 candidate。 + +--- + +## 问题6: 对称窗口逻辑优化 + +### 现象 +窗口末端的局部低点/高点，因右侧数据不足无法识别。 + +### 解决方案 +**文件**: `src/converging_triangle.py` +```python +# 候选枢轴点使用对称的短窗口 +for i in range(max(k, n - flexible_zone), n): + right_avail = n - 1 - i + left_look = min(k, max(right_avail + 1, 3)) # 至少看3天 + left_start = max(0, i - left_look) + right_end = min(n, i + right_avail + 1) + + if high[i] == np.nanmax(high[left_start : right_end]): + candidate_ph.append(i) +``` + +--- + +## 修改文件清单 + +| 文件 | 修改内容 | +|------|----------| +| `scripts/triangle_config.py` | 斜率约束（`upper_slope_max=0`, `lower_slope_min=0`）、`FLEXIBLE_ZONE=15` | +| `scripts/plot_converging_triangles.py` | 实时模式枢轴点检测、候选点合并、排序 | +| `src/converging_triangle.py` | NaN处理（`nanmin/nanmax`）、对称窗口逻辑 | + +--- + +## 效果验证 + +修复前后对比： + +| 指标 | 修复前 | 修复后 | +|------|--------|--------| +| SZ300892 低点枢轴点 | 4个 | 7个 | +| 末端低点识别 | ❌ | ✅ | +| 图表/检测一致性 | ❌ | ✅ | +| 非收敛形态过滤 | ❌ | ✅ | + diff --git a/docs/2026-01-26_流水线详情模式支持.md b/docs/2026-01-26_流水线详情模式支持.md new file mode 100644 index 0000000..26df639 --- /dev/null +++ b/docs/2026-01-26_流水线详情模式支持.md @@ -0,0 +1,214 @@ +# 流水线脚本详情模式支持 + +**日期**: 2026-01-26 +**功能**: 为 `pipeline_converging_triangle.py` 添加 `--show-details` 参数 + +--- + +## 📋 需求背景 + +用户希望在使用流水线脚本 `pipeline_converging_triangle.py` 时，能够通过参数控制是否生成详情模式的图片，而不是默认生成。 + +**之前**：流水线脚本没有详情模式参数，只能生成简洁模式图片。 + +**现在**：流水线脚本支持 `--show-details` 参数，可以按需生成详情模式图片。 + +--- + +## ✅ 实现内容 + +### 1. 添加命令行参数 + +在 `pipeline_converging_triangle.py` 中添加： + +```python +parser.add_argument( + "--show-details", + action="store_true", + help="生成详情模式图片（显示所有枢轴点和拟合点）", +) +``` + +### 2. 参数传递逻辑 + +将参数传递给绘图脚本： + +```python +# 步骤 3: 绘制图表 +cmd_args = [sys.argv[0]] +if args.date: + cmd_args.extend(["--date", str(args.date)]) +if args.show_details: + cmd_args.append("--show-details") + +sys.argv = cmd_args +run_plot() +``` + +### 3. 控制台提示信息 + +在流水线开始时显示当前模式： + +```python +if args.show_details: + print(f"图表模式: 详情模式（显示所有枢轴点）") +else: + print(f"图表模式: 简洁模式（仅显示价格和趋势线）") +``` + +--- + +## 🛠️ 使用方法 + +### 基本用法 + +```bash +# 简洁模式（默认）- 不生成详情图片 +python scripts/pipeline_converging_triangle.py + +# 详情模式 - 生成详情图片 +python scripts/pipeline_converging_triangle.py --show-details +``` + +### 组合参数 + +```bash +# 指定日期 + 详情模式 +python scripts/pipeline_converging_triangle.py --date 20260120 --show-details + +# 跳过检测 + 详情模式 +python scripts/pipeline_converging_triangle.py --skip-detection --show-details + +# 跳过检测和报告，只生成详情图表 +python scripts/pipeline_converging_triangle.py --skip-detection --skip-report --show-details +``` + +--- + +## 📊 输出对比 + +### 简洁模式（默认） + +``` +输出文件: outputs/converging_triangles/charts/20260120_SZ300278_华昌达.png +文件内容: 收盘价 + 上沿线 + 下沿线 +``` + +### 详情模式（--show-details） + +``` +输出文件: outputs/converging_triangles/charts/20260120_SZ300278_华昌达_detail.png +文件内容: 收盘价 + 上沿线 + 下沿线 + 所有枢轴点 + 拟合点 +``` + +**注意**：两种模式的文件可以共存，互不覆盖。 + +--- + +## 🎯 参数优先级 + +流水线脚本的参数传递链： + +``` +用户命令行 + └─ pipeline_converging_triangle.py + ├─ --show-details (可选) + └─ 传递给 plot_converging_triangles.py + ├─ --show-details (如果用户指定) + └─ 调用 plot_triangle(show_details=True/False) +``` + +**优先级规则**： +1. 流水线脚本的 `--show-details` 参数 +2. 传递到绘图脚本 +3. 绘图脚本的配置文件 `SHOW_CHART_DETAILS`（命令行参数优先） + +--- + +## 🔍 测试验证 + +### 测试1：帮助信息 + +```bash +python scripts/pipeline_converging_triangle.py --help +``` + +**预期**：显示 `--show-details` 参数说明 + +### 测试2：简洁模式（默认） + +```bash +python scripts/pipeline_converging_triangle.py --skip-detection --skip-report +``` + +**预期**： +- ✅ 控制台显示："图表模式: 简洁模式（仅显示价格和趋势线）" +- ✅ 生成不带 `_detail` 后缀的图片 + +### 测试3：详情模式 + +```bash +python scripts/pipeline_converging_triangle.py --skip-detection --skip-report --show-details +``` + +**预期**： +- ✅ 控制台显示："图表模式: 详情模式（显示所有枢轴点）" +- ✅ 控制台显示："详细模式: 开启 (--show-details)" +- ✅ 生成带 `_detail` 后缀的图片 + +--- + +## 📝 修改文件列表 + +1. **scripts/pipeline_converging_triangle.py** + - 添加 `--show-details` 参数 + - 添加模式提示信息 + - 参数传递逻辑 + +2. **USAGE.md** + - 更新流水线脚本的使用说明 + - 添加 `--show-details` 参数示例 + +3. **docs/2026-01-26_图表详细模式功能.md** + - 更新文档，添加流水线使用方法 + - 标注为 v2.0 版本 + +--- + +## 🎉 效果总结 + +### 用户体验改进 + +**之前**： +```bash +# 想要详情图片 +python scripts/pipeline_converging_triangle.py # ❌ 只能生成简洁模式 +# 需要单独运行绘图脚本 +python scripts/plot_converging_triangles.py --show-details +``` + +**现在**： +```bash +# 一键生成详情图片 +python scripts/pipeline_converging_triangle.py --show-details # ✅ 直接搞定 +``` + +### 优势 + +1. **一致性**：流水线和绘图脚本参数统一 +2. **便捷性**：无需分步操作 +3. **灵活性**：默认简洁，按需详细 +4. **清晰性**：控制台明确提示当前模式 + +--- + +## 📚 相关文档 + +- [USAGE.md](../USAGE.md) - 使用指南 +- [2026-01-26_图表详细模式功能.md](./2026-01-26_图表详细模式功能.md) - 详情模式完整说明 +- [枢轴点拟合算法详解.md](./枢轴点分段选择算法详解.md) - 拟合算法原理 + +--- + +**总结**：通过添加 `--show-details` 参数，流水线脚本现在完全支持详情模式，用户可以更方便地按需生成详细的调试图表。✅ + diff --git a/docs/拟合线迭代离群点移除优化.md b/docs/拟合线迭代离群点移除优化.md new file mode 100644 index 0000000..4fe823c --- /dev/null +++ b/docs/拟合线迭代离群点移除优化.md @@ -0,0 +1,117 @@ +# 拟合线迭代离群点移除优化 + +## 问题描述 + +当前的分段选择算法会选中一些"弱"枢轴点用于拟合： + +- 上沿线：某个高点虽然是时间段内最高，但明显低于其他高点（如图中第二个点 5.8 元） +- 这些点会拉低/拉高拟合线，导致与主观判断不符 + +## 解决方案：迭代离群点移除 + +### 核心逻辑 + +``` +1. 初始拟合：用所有枢轴点做线性回归 +2. 计算残差：每个点到拟合线的偏差 +3. 识别离群点： + - 上沿线：价格明显低于拟合线的点 = 弱高点 + - 下沿线：价格明显高于拟合线的点 = 弱低点 +4. 移除最差的离群点 +5. 重新拟合 +6. 重复直到收敛 +``` + +### 算法流程图 + +```mermaid +flowchart TD + A[输入枢轴点] --> B[初始线性回归] + B --> C[计算残差] + C --> D{存在离群点?} + D -->|是| E[移除最大离群点] + E --> F{剩余点 >= 3?} + F -->|是| G{迭代次数 < 3?} + G -->|是| B + G -->|否| H[返回当前拟合] + F -->|否| H + D -->|否| H +``` + +## 代码修改 + +### 文件: [src/converging_triangle.py](src/converging_triangle.py) + +重写 `fit_pivot_line` 函数（第 230-350 行）: + +```python +def fit_pivot_line( + pivot_indices: np.ndarray, + pivot_values: np.ndarray, + mode: str = "upper", + min_points: int = 2, + outlier_threshold: float = 1.5, # 新增：离群点阈值（标准差倍数） + max_iterations: int = 3, # 新增：最大迭代次数 +) -> Tuple[float, float, np.ndarray]: + """ + 迭代离群点移除的枢轴点拟合算法 + + 策略： + 1. 先用所有点做初始拟合 + 2. 识别并移除偏离拟合线的"弱"点 + 3. 迭代直到收敛 + + 对于上沿线：移除价格明显低于拟合线的点 + 对于下沿线：移除价格明显高于拟合线的点 + """ +``` + +### 关键参数 + +| 参数 | 默认值 | 说明 | + +|------|--------|------| + +| `outlier_threshold` | 1.5 | 残差超过 1.5 倍标准差视为离群点 | + +| `max_iterations` | 3 | 最多迭代 3 次，避免过度过滤 | + +| `min_points` | 3 | 至少保留 3 个点用于拟合 | + +### 离群点判定逻辑 + +**上沿线（upper）**: + +```python +# 残差 = 拟合值 - 实际值 +# 正残差表示点在拟合线下方（弱高点） +residuals = fitted_values - actual_values +outliers = residuals > threshold # 弱高点 +``` + +**下沿线（lower）**: + +```python +# 残差 = 实际值 - 拟合值 +# 正残差表示点在拟合线上方（弱低点） +residuals = actual_values - fitted_values +outliers = residuals > threshold # 弱低点 +``` + +### 预期效果 + +以图中 SZ300278 为例： + +- 第二个点（5.8元）明显低于拟合线 +- 在第一次迭代后会被识别为离群点并移除 +- 最终拟合线只使用剩余 3 个更有代表性的高点 + +## 测试计划 + +1. 使用 SZ300278 验证修复效果 +2. 对比修改前后的图表 +3. 确保不会过度过滤正常的枢轴点 + +## 文档更新 + +更新 [docs/枢轴点分段选择算法详解.md](docs/枢轴点分段选择算法详解.md)，添加迭代离群点移除的说明。 \ No newline at end of file diff --git a/docs/文档索引.md b/docs/文档索引.md index d1ae9c2..083a5ec 100644 --- a/docs/文档索引.md +++ b/docs/文档索引.md @@ -140,6 +140,12 @@ - 文档更新说明 - 项目结构整理 +7. **[2026-01-26_枢轴点检测与可视化修复.md](./2026-01-26_枢轴点检测与可视化修复.md)** 🔧 + - 斜率约束收紧 + - NaN值处理修复 + - 候选枢轴点合并 + - 对称窗口逻辑优化 + ### 实时模式实施 7. **[方案4实施完成报告.md](./方案4实施完成报告.md)** diff --git a/docs/枢轴点分段选择算法详解.md b/docs/枢轴点分段选择算法详解.md index 433644c..095ad9b 100644 --- a/docs/枢轴点分段选择算法详解.md +++ b/docs/枢轴点分段选择算法详解.md @@ -1,6 +1,6 @@ -# 枢轴点分段选择算法详解 +# 枢轴点拟合算法详解 -**日期**: 2026-01-26 +**日期**: 2026-01-26 (v2 更新) **文件**: `src/converging_triangle.py` - `fit_pivot_line()` 函数 --- @@ -8,9 +8,9 @@ ## 📋 目录 1. [算法概述](#算法概述) -2. [为什么需要分段](#为什么需要分段) -3. [分段算法详解](#分段算法详解) -4. [独立分段机制](#独立分段机制) +2. [迭代离群点移除（当前算法）](#迭代离群点移除当前算法) +3. [历史版本：分段选择算法](#历史版本分段选择算法) +4. [独立处理机制](#独立处理机制) 5. [代码实现](#代码实现) 6. [实际案例分析](#实际案例分析) 7. [边界情况处理](#边界情况处理) @@ -22,13 +22,131 @@ ### 核心思想 -对于识别出的所有枢轴点，我们不是简单地用所有点或仅用首尾两点来拟合趋势线，而是： +对于识别出的所有枢轴点，我们采用**迭代离群点移除**算法来拟合趋势线： + +1. **初始拟合**：先用所有枢轴点做线性回归 +2. **识别离群点**：计算残差，找出偏离拟合线的"弱"点 +3. **移除离群点**：移除最差的离群点 +4. **迭代优化**：重复步骤1-3直到收敛 + +### 设计目标 + +```python +# 上沿线：移除价格明显低于拟合线的点（弱高点） +# 下沿线：移除价格明显高于拟合线的点（弱低点） +``` + +--- + +## 迭代离群点移除（当前算法） + +### 为什么需要这个算法？ + +**问题场景**： + +``` +价格 + ^ + | * ← 强高点 (7.9元) + | \ + | * ← 弱高点 (5.8元) ← 问题！ + | \ + | * ← 强高点 (6.8元) + | \ + | * ← 强高点 (6.0元) + └──────────────> 时间 +``` + +如果使用所有4个点拟合，5.8元的"弱高点"会把上沿线拉低，导致： +- ❌ 拟合线不是真正的上边界 +- ❌ 与主观判断不符 +- ❌ 可能错过真实的突破信号 + +### 解决方案：迭代移除 + +``` +第1次迭代: + 拟合线: y = ax + b (用全部4点) + 残差: [0.1, 1.8, 0.2, 0.3] ← 5.8元点的残差最大(1.8) + 判定: 1.8 > 1.5*std → 离群点 + 操作: 移除5.8元点 + +第2次迭代: + 拟合线: y = a'x + b' (用剩余3点) + 残差: [0.05, 0.08, 0.12] + 判定: 无离群点 → 收敛 + +最终结果: 只用3个强高点拟合 +``` + +### 算法流程图 + +```mermaid +flowchart TD + A[输入全部枢轴点] --> B[线性回归拟合] + B --> C[计算方向性残差] + C --> D{残差 > 阈值?} + D -->|是| E[移除最大残差点] + E --> F{剩余点 >= 3?} + F -->|是| G{迭代次数 < 3?} + G -->|是| B + G -->|否| H[返回当前拟合] + F -->|否| H + D -->|否| H +``` + +### 关键参数 + +| 参数 | 默认值 | 说明 | +|------|--------|------| +| `outlier_threshold` | 1.5 | 残差超过 1.5 倍标准差视为离群点 | +| `max_iterations` | 3 | 最多迭代 3 次，避免过度过滤 | +| `min_keep` | 3 | 至少保留 3 个点用于拟合 | + +### 残差计算逻辑 + +**上沿线（upper）**： + +```python +# 残差 = 拟合值 - 实际值 +# 正残差 = 点在拟合线下方 = 弱高点（应移除） +residuals = fitted_values - actual_values +outliers = residuals > threshold +``` + +**下沿线（lower）**： + +```python +# 残差 = 实际值 - 拟合值 +# 正残差 = 点在拟合线上方 = 弱低点（应移除） +residuals = actual_values - fitted_values +outliers = residuals > threshold +``` + +### 算法优势 + +| 优势 | 说明 | +|------|------| +| **自适应** | 自动识别偏离点，无需人工干预 | +| **稳健性** | 不会被单个异常点影响整条线 | +| **可控性** | 通过阈值和迭代次数控制过滤强度 | +| **保守性** | 至少保留3个点，避免过度过滤 | + +--- + +## 历史版本：分段选择算法 + +> **注意**：以下是 v1 版本的分段算法，已被迭代离群点移除算法替代。保留此节供参考。 + +### 原核心思想 + +对于识别出的所有枢轴点： 1. **分段策略**：将枢轴点按时间顺序分成 **3 个时间段** 2. **代表点选择**：从每段中选出 **最极端的点**（上沿选最高，下沿选最低） 3. **线性回归**：用这 3 个代表点进行线性回归，拟合趋势线 -### 触发条件 +### 原触发条件 ```python if 枢轴点数量 > 4: @@ -37,6 +155,12 @@ else: 使用全部枢轴点 ``` +### 原算法的局限性 + +分段算法会选中一些"弱"枢轴点： +- 某个高点虽然是时间段内最高，但可能明显低于其他时间段的高点 +- 这些点会拉低/拉高拟合线，导致与主观判断不符 + --- ## 为什么需要分段 @@ -254,7 +378,7 @@ n = 4 ≤ 4，不分段，全部使用 ✓ --- -## 独立分段机制 +## 独立处理机制 ### 关键点：高点和低点分别独立处理 @@ -323,29 +447,28 @@ else: ## 代码实现 -### 完整代码（带详细注释） +### 当前算法：迭代离群点移除（v2） ```python def fit_pivot_line( - pivot_indices: np.ndarray, # 枢轴点的时间索引 - pivot_values: np.ndarray, # 枢轴点的价格 - mode: str = "upper", # "upper" 或 "lower" + pivot_indices: np.ndarray, + pivot_values: np.ndarray, + mode: str = "upper", min_points: int = 2, + outlier_threshold: float = 1.5, # 离群点阈值（标准差倍数） + max_iterations: int = 3, # 最大迭代次数 ) -> Tuple[float, float, np.ndarray]: """ - 拟合枢轴点趋势线（使用分段选择策略） + 迭代离群点移除的枢轴点拟合算法 策略： - - 如果枢轴点 > 4：分3段，每段选1个极值点（共3个拟合点） - - 如果枢轴点 ≤ 4：全部使用 + 1. 先用所有点做初始拟合 + 2. 识别并移除偏离拟合线的"弱"点 + 3. 迭代直到收敛 - Returns: - (斜率a, 截距b, 选中的枢轴点索引) + 对于上沿线：移除价格明显低于拟合线的点（弱高点） + 对于下沿线：移除价格明显高于拟合线的点（弱低点） """ - - # ───────────────────────────────────────────────────────── - # 第1步：基本检查和排序 - # ───────────────────────────────────────────────────────── if len(pivot_indices) < min_points: return 0.0, 0.0, np.array([]) @@ -353,115 +476,113 @@ def fit_pivot_line( sort_idx = np.argsort(pivot_indices) x_sorted = pivot_indices[sort_idx].astype(float) y_sorted = pivot_values[sort_idx] - n = len(x_sorted) if n < 2: return 0.0, 0.0, np.array([]) - # ───────────────────────────────────────────────────────── - # 第2步：决定是否分段 - # ───────────────────────────────────────────────────────── - if n <= 4: - # 点数少，全部使用 - selected_mask = np.ones(n, dtype=bool) - else: - # 点数多，使用分段策略 - selected_mask = np.zeros(n, dtype=bool) - - # 计算每段大小 - segment_size = n // 3 - if segment_size < 1: - segment_size = 1 - - # 定义三个时间段 - segments = [ - range(0, min(segment_size, n)), # 第1段 - range(segment_size, min(2 * segment_size, n)), # 第2段 - range(2 * segment_size, n), # 第3段 - ] - - # ───────────────────────────────────────────────────── - # 第3步：从每段选择极值点 - # ───────────────────────────────────────────────────── - for seg in segments: - if len(seg) == 0: - continue - - seg_list = list(seg) - seg_y = y_sorted[seg_list] - - if mode == "upper": - # 上沿：选该段最高点 - best_idx_in_seg = np.argmax(seg_y) - else: # mode == "lower" - # 下沿：选该段最低点 - best_idx_in_seg = np.argmin(seg_y) - - # 标记选中 - selected_mask[seg_list[best_idx_in_seg]] = True + # 初始化：所有点都参与 + active_mask = np.ones(n, dtype=bool) + min_keep = max(3, min_points) # 至少保留3个点 # ───────────────────────────────────────────────────────── - # 第4步：提取选中的点 + # 迭代离群点移除 # ───────────────────────────────────────────────────────── - selected_x = x_sorted[selected_mask] - selected_y = y_sorted[selected_mask] - selected_indices_sorted = np.where(selected_mask)[0] + for iteration in range(max_iterations): + active_indices = np.where(active_mask)[0] + if len(active_indices) <= min_keep: + break + + # 当前活跃点 + x_active = x_sorted[active_mask] + y_active = y_sorted[active_mask] + + # 线性回归 + a, b = fit_line(x_active, y_active) + + # 计算拟合值 + fitted_values = a * x_active + b + + # ───────────────────────────────────────────────────── + # 计算方向性残差 + # ───────────────────────────────────────────────────── + if mode == "upper": + # 上沿：残差 = 拟合值 - 实际值 + # 正残差表示点在拟合线下方（弱高点） + residuals = fitted_values - y_active + else: + # 下沿：残差 = 实际值 - 拟合值 + # 正残差表示点在拟合线上方（弱低点） + residuals = y_active - fitted_values + + # 计算标准差 + std_residual = np.std(residuals) + if std_residual < 1e-10: + # 所有点几乎在一条线上，无需移除 + break + + # ───────────────────────────────────────────────────── + # 识别并移除最大离群点 + # ───────────────────────────────────────────────────── + threshold = outlier_threshold * std_residual + max_outlier_idx = np.argmax(residuals) + + if residuals[max_outlier_idx] <= threshold: + # 最大残差不超过阈值，收敛 + break + + if np.sum(active_mask) <= min_keep: + break + + # 移除离群点 + original_idx = active_indices[max_outlier_idx] + active_mask[original_idx] = False + + # ───────────────────────────────────────────────────────── + # 最终拟合 + # ───────────────────────────────────────────────────────── + selected_indices_sorted = np.where(active_mask)[0] + selected_x = x_sorted[active_mask] + selected_y = y_sorted[active_mask] - # 保底：至少选首尾两点 if len(selected_x) < 2: - selected_mask = np.zeros(n, dtype=bool) - selected_mask[0] = True - selected_mask[-1] = True - selected_x = x_sorted[selected_mask] - selected_y = y_sorted[selected_mask] - selected_indices_sorted = np.where(selected_mask)[0] + # 兜底：使用首尾两点 + active_mask = np.zeros(n, dtype=bool) + active_mask[0] = True + active_mask[-1] = True + selected_x = x_sorted[active_mask] + selected_y = y_sorted[active_mask] + selected_indices_sorted = np.where(active_mask)[0] - # ───────────────────────────────────────────────────────── - # 第5步：线性回归 - # ───────────────────────────────────────────────────────── a, b = fit_line(selected_x, selected_y) - # ───────────────────────────────────────────────────────── - # 第6步：覆盖性验证（确保线不穿透任何枢轴点） - # ───────────────────────────────────────────────────────── - fitted_all = a * x_sorted + b - tolerance = 0.03 # 3%容差 - - if mode == "upper": - # 上沿线不应低于任何高点 - violations = y_sorted > fitted_all + tolerance * np.mean(y_sorted) - if np.any(violations): - # 强制包含全局最高点 - global_max_idx = np.argmax(y_sorted) - if not selected_mask[global_max_idx]: - selected_mask[global_max_idx] = True - selected_x = x_sorted[selected_mask] - selected_y = y_sorted[selected_mask] - selected_indices_sorted = np.where(selected_mask)[0] - a, b = fit_line(selected_x, selected_y) - else: # mode == "lower" - # 下沿线不应高于任何低点 - violations = y_sorted < fitted_all - tolerance * np.mean(y_sorted) - if np.any(violations): - # 强制包含全局最低点 - global_min_idx = np.argmin(y_sorted) - if not selected_mask[global_min_idx]: - selected_mask[global_min_idx] = True - selected_x = x_sorted[selected_mask] - selected_y = y_sorted[selected_mask] - selected_indices_sorted = np.where(selected_mask)[0] - a, b = fit_line(selected_x, selected_y) - - # ───────────────────────────────────────────────────────── - # 第7步：返回结果 - # ───────────────────────────────────────────────────────── - # 将排序后的索引映射回原始索引 + # 返回原始索引 selected_original = sort_idx[selected_indices_sorted] return float(a), float(b), selected_original ``` +### 历史算法：分段选择（v1） + +> 以下代码已被替代，保留供参考。 + +```python +def fit_pivot_line_v1( + pivot_indices: np.ndarray, + pivot_values: np.ndarray, + mode: str = "upper", + min_points: int = 2, +) -> Tuple[float, float, np.ndarray]: + """ + 分段选择策略（已废弃） + + 策略： + - 如果枢轴点 > 4：分3段，每段选1个极值点（共3个拟合点） + - 如果枢轴点 ≤ 4：全部使用 + """ + # ... 详见历史版本 +``` + --- ## 实际案例分析 @@ -616,90 +737,45 @@ if len(selected_x) < 2: ``` 图表元素 对应内容 ───────────────────────────────────────────── -浅红色小实心圆（6个） 所有高点枢轴点 -深红色大空心圆（3个） 上沿拟合点（从6个中选出） -红色点划竖线 高点分段边界 - - "高1|2" 标签 第1段和第2段分界 - - "高2|3" 标签 第2段和第3段分界 - -浅绿色小实心圆（4个） 所有低点枢轴点 -深绿色大空心圆（4个） 下沿拟合点（全部使用，因≤4） -（无绿色竖线） 低点不分段（≤4） +浅红色小实心圆 所有高点枢轴点 +深红色大空心圆 上沿拟合点（迭代算法选出） +浅绿色小实心圆 所有低点枢轴点 +深绿色大空心圆 下沿拟合点（迭代算法选出） ``` -### 分段线的位置 - -**分段边界的时间索引**： - -```python -# 以高点为例，n=6, segment_size=2 -排序后的高点索引: [10, 30, 50, 60, 80, 90] - 位置0 1 2 3 4 5 - -第1段结束 = 第2段开始: - boundary_1 = 索引[segment_size] = 索引[2] = 50 - → 在时间50处画竖线 - -第2段结束 = 第3段开始: - boundary_2 = 索引[2*segment_size] = 索引[4] = 80 - → 在时间80处画竖线 - -分段结果: - 第1段: [10, 30] | 第2段: [50, 60] | 第3段: [80, 90] - ↑ boundary_1 ↑ boundary_2 -``` - -**代码实现**： - -```python -if len(ph_idx) > 4: - n_high = len(ph_idx) - segment_size_high = n_high // 3 - - # 第1条竖线 - if segment_size_high < n_high: - boundary_1 = ph_idx[segment_size_high] - ax.axvline(boundary_1, color='red', linestyle='-.', ...) - ax.text(boundary_1, y_max*0.96, '高1|2', ...) - - # 第2条竖线 - if 2 * segment_size_high < n_high: - boundary_2 = ph_idx[2 * segment_size_high] - ax.axvline(boundary_2, color='red', linestyle='-.', ...) - ax.text(boundary_2, y_max*0.96, '高2|3', ...) -``` +**注意**：自 v2.0 起，分段竖线已被移除（因算法不再使用分段策略）。 --- ## 总结 -### 分段策略的核心要点 +### 迭代离群点移除算法的核心要点 -1. **触发条件**: 枢轴点 > 4 -2. **分段数量**: 固定3段 -3. **分段方式**: 时间均分（每段 `n//3` 个点） -4. **选择策略**: 每段选1个极值点 -5. **独立性**: 高点和低点各自独立分段 -6. **保底机制**: 覆盖性验证 + 全局极值保护 +1. **初始状态**: 所有枢轴点参与拟合 +2. **迭代移除**: 每次移除最大残差的离群点 +3. **方向性残差**: 上沿移除低于线的点，下沿移除高于线的点 +4. **收敛条件**: 无离群点 / 达到最大迭代次数 / 点数达到下限 +5. **独立性**: 高点和低点各自独立处理 +6. **保底机制**: 至少保留3个点用于拟合 ### 算法优势 | 优势 | 说明 | |------|------| -| **时间均衡** | 前、中、后三个时期都有代表点 | -| **代表性强** | 每段选最极端的点，确保边界性 | -| **抗噪性好** | 不易被局部密集点影响 | -| **稳健性高** | 多点回归比两点连线更稳定 | -| **可扩展性** | 点数增加时仍保持3个拟合点 | +| **自适应** | 自动识别并移除偏离点 | +| **稳健性** | 不受单个异常点影响 | +| **可控性** | 通过阈值和迭代次数控制 | +| **保守性** | 至少保留3个点，避免过度过滤 | +| **符合直觉** | 拟合结果与主观判断一致 | ### 与其他方法的对比 | 方法 | 优点 | 缺点 | |------|------|------| | **两点连线** | 简单快速 | 忽略中间点，易受噪声影响 | -| **全部点回归** | 利用所有信息 | 权重不均，点密集区域主导 | -| **分段选择（当前）** | 时间均衡，代表性强 | 略复杂，需要分段逻辑 | -| **滑动窗口** | 平滑效果好 | 计算复杂，参数敏感 | +| **全部点回归** | 利用所有信息 | 权重不均，异常点影响大 | +| **分段选择（v1）** | 时间均衡 | 可能选中弱枢轴点 | +| **迭代离群点移除（当前）** | 自适应，稳健 | 需要迭代计算 | --- @@ -711,6 +787,9 @@ if len(ph_idx) > 4: --- -**文档版本**: v1.0 -**最后更新**: 2026-01-26 +**文档版本**: v2.0 +**最后更新**: 2026-01-26 +**变更记录**: +- v2.0: 采用迭代离群点移除算法，替代原分段选择策略 +- v1.0: 初始版本，分段选择算法 diff --git a/scripts/__pycache__/plot_converging_triangles.cpython-313.pyc b/scripts/__pycache__/plot_converging_triangles.cpython-313.pyc index cf6f904..2f82301 100644 Binary files a/scripts/__pycache__/plot_converging_triangles.cpython-313.pyc and b/scripts/__pycache__/plot_converging_triangles.cpython-313.pyc differ diff --git a/scripts/__pycache__/triangle_config.cpython-313.pyc b/scripts/__pycache__/triangle_config.cpython-313.pyc index 7dad114..7434990 100644 Binary files a/scripts/__pycache__/triangle_config.cpython-313.pyc and b/scripts/__pycache__/triangle_config.cpython-313.pyc differ diff --git a/scripts/pipeline_converging_triangle.py b/scripts/pipeline_converging_triangle.py index e0feaf3..197933f 100644 --- a/scripts/pipeline_converging_triangle.py +++ b/scripts/pipeline_converging_triangle.py @@ -9,6 +9,7 @@ 用法: python scripts/pipeline_converging_triangle.py python scripts/pipeline_converging_triangle.py --date 20260120 + python scripts/pipeline_converging_triangle.py --show-details # 生成详情模式图片 """ from __future__ import annotations @@ -54,6 +55,11 @@ def main() -> None: default=None, help="指定日期（YYYYMMDD），用于报告和图表生成（默认为数据最新日）", ) + parser.add_argument( + "--show-details", + action="store_true", + help="生成详情模式图片（显示所有枢轴点和拟合点）", + ) parser.add_argument( "--skip-detection", action="store_true", @@ -79,6 +85,10 @@ def main() -> None: print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") if args.date: print(f"指定日期: {args.date}") + if args.show_details: + print(f"图表模式: 详情模式（显示所有枢轴点）") + else: + print(f"图表模式: 简洁模式（仅显示价格和趋势线）") print("=" * 80) results = [] @@ -149,10 +159,13 @@ def main() -> None: try: # 设置命令行参数 + cmd_args = [sys.argv[0]] if args.date: - sys.argv = [sys.argv[0], "--date", str(args.date)] - else: - sys.argv = [sys.argv[0]] + cmd_args.extend(["--date", str(args.date)]) + if args.show_details: + cmd_args.append("--show-details") + + sys.argv = cmd_args run_plot() success = True diff --git a/scripts/plot_converging_triangles.py b/scripts/plot_converging_triangles.py index 61d174b..0c24de0 100644 --- a/scripts/plot_converging_triangles.py +++ b/scripts/plot_converging_triangles.py @@ -37,10 +37,11 @@ from converging_triangle import ( fit_pivot_line, line_y, pivots_fractal, + pivots_fractal_hybrid, ) # 导入统一的参数配置 -from triangle_config import DETECTION_PARAMS, DISPLAY_WINDOW, SHOW_CHART_DETAILS +from triangle_config import DETECTION_PARAMS, DISPLAY_WINDOW, SHOW_CHART_DETAILS, REALTIME_MODE, FLEXIBLE_ZONE class FakeModule: @@ -159,7 +160,11 @@ def plot_triangle( display_volume = volume_stock[valid_mask][display_start:valid_end + 1] display_dates = dates[valid_indices[display_start:valid_end + 1]] - # 检测三角形（使用检测窗口数据） + # ======================================================================== + # 计算三角形参数（用于绘图） + # 注意：不验证 is_valid，因为CSV中已经验证通过了 + # 这里只是重新计算参数用于可视化 + # ======================================================================== result = detect_converging_triangle( high=high_win, low=low_win, @@ -170,9 +175,8 @@ def plot_triangle( date_idx=date_idx, ) - if not result.is_valid: - print(f" [跳过] {stock_code} {stock_name}: 未识别到有效三角形") - return + # 不再检查 is_valid，直接绘图 + # 原因：CSV中已经包含了通过验证的股票，这里只需要可视化 # 绘图准备 x_display = np.arange(len(display_close), dtype=float) @@ -184,8 +188,19 @@ def plot_triangle( n = len(close_win) x_win = np.arange(n, dtype=float) - # 计算枢轴点（与检测算法一致） - ph_idx, pl_idx = pivots_fractal(high_win, low_win, k=params.pivot_k) + # 计算枢轴点（与检测算法一致，考虑实时模式） + if REALTIME_MODE: + confirmed_ph, confirmed_pl, candidate_ph, candidate_pl = pivots_fractal_hybrid( + high_win, low_win, k=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=params.pivot_k) # 使用枢轴点连线法拟合边界线（与检测算法一致） a_u, b_u, selected_ph = fit_pivot_line( @@ -281,108 +296,6 @@ def plot_triangle( zorder=5, label=f'下沿拟合点({len(selected_pl_pos)})', ) - - # 绘制分段竖线（显示算法如何分段选择枢轴点） - # 高点和低点分别独立分段，用不同颜色显示 - y_min, y_max = ax1.get_ylim() - - # 绘制高点枢轴点的分段线（红色） - if len(ph_idx) > 4: - n_high = len(ph_idx) - segment_size_high = n_high // 3 - - # 第1段结束 = 第2段开始 - if segment_size_high < n_high: - boundary_1 = ph_idx[segment_size_high] + triangle_offset - ax1.axvline( - boundary_1, - color='red', - linestyle='-.', - linewidth=1.2, - alpha=0.4, - zorder=3, - ) - ax1.text( - boundary_1, - y_max * 0.96, - '高1|2', - ha='center', - va='top', - fontsize=7, - color='red', - bbox=dict(boxstyle='round,pad=0.2', facecolor='white', edgecolor='red', alpha=0.7), - ) - - # 第2段结束 = 第3段开始 - if 2 * segment_size_high < n_high: - boundary_2 = ph_idx[2 * segment_size_high] + triangle_offset - ax1.axvline( - boundary_2, - color='red', - linestyle='-.', - linewidth=1.2, - alpha=0.4, - zorder=3, - ) - ax1.text( - boundary_2, - y_max * 0.96, - '高2|3', - ha='center', - va='top', - fontsize=7, - color='red', - bbox=dict(boxstyle='round,pad=0.2', facecolor='white', edgecolor='red', alpha=0.7), - ) - - # 绘制低点枢轴点的分段线（绿色） - if len(pl_idx) > 4: - n_low = len(pl_idx) - segment_size_low = n_low // 3 - - # 第1段结束 = 第2段开始 - if segment_size_low < n_low: - boundary_1 = pl_idx[segment_size_low] + triangle_offset - ax1.axvline( - boundary_1, - color='green', - linestyle='-.', - linewidth=1.2, - alpha=0.4, - zorder=3, - ) - ax1.text( - boundary_1, - y_min + (y_max - y_min) * 0.04, - '低1|2', - ha='center', - va='bottom', - fontsize=7, - color='green', - bbox=dict(boxstyle='round,pad=0.2', facecolor='white', edgecolor='green', alpha=0.7), - ) - - # 第2段结束 = 第3段开始 - if 2 * segment_size_low < n_low: - boundary_2 = pl_idx[2 * segment_size_low] + triangle_offset - ax1.axvline( - boundary_2, - color='green', - linestyle='-.', - linewidth=1.2, - alpha=0.4, - zorder=3, - ) - ax1.text( - boundary_2, - y_min + (y_max - y_min) * 0.04, - '低2|3', - ha='center', - va='bottom', - fontsize=7, - color='green', - bbox=dict(boxstyle='round,pad=0.2', facecolor='white', edgecolor='green', alpha=0.7), - ) ax1.set_title( f"{stock_code} {stock_name} - 收敛三角形 (检测窗口: {detect_dates[0]} ~ {detect_dates[-1]})\n" diff --git a/scripts/triangle_config.py b/scripts/triangle_config.py index 1908bde..6ea155c 100644 --- a/scripts/triangle_config.py +++ b/scripts/triangle_config.py @@ -31,9 +31,9 @@ DETECTION_PARAMS = ConvergingTriangleParams( boundary_n_segments=2, # 边界线分段数 boundary_source="full", # 边界拟合数据源: "full"(全部) 或 "pivot"(仅枢轴点) - # 斜率约束 - upper_slope_max=0.10, # 上沿最大斜率（正值，向上倾斜） - lower_slope_min=-0.10, # 下沿最小斜率（负值，向下倾斜） + # 斜率约束（严格收敛三角形） + upper_slope_max=0, # 上沿必须向下或水平（≤0） + lower_slope_min=0, # 下沿必须向上或水平（≥0） # 注意：算法会自动过滤"同向通道"（上下沿都向上或都向下） # 只保留真正的收敛形态（上下沿相向运动） @@ -92,8 +92,8 @@ DISPLAY_WINDOW = 500 REALTIME_MODE = True # True=实时模式（默认）, False=标准模式 # 灵活区域大小（仅在实时模式下生效） -FLEXIBLE_ZONE = 5 # 最近5天使用降低标准 -# 建议: 3-7天，太大会引入噪音 +FLEXIBLE_ZONE = 15 # 最近15天使用降低标准（与pivot_k一致） +# 说明: 设为与pivot_k相同，确保窗口末端的所有潜在枢轴点都能被识别 # 实时模式说明： # - 实时模式（推荐）：使用确认+候选枢轴点（允许右边数据不完整），适合实时选股 diff --git a/src/__pycache__/converging_triangle.cpython-313.pyc b/src/__pycache__/converging_triangle.cpython-313.pyc index 75ea239..67896b3 100644 Binary files a/src/__pycache__/converging_triangle.cpython-313.pyc and b/src/__pycache__/converging_triangle.cpython-313.pyc differ diff --git a/src/converging_triangle.py b/src/converging_triangle.py index 0c08b37..65ede3b 100644 --- a/src/converging_triangle.py +++ b/src/converging_triangle.py @@ -103,14 +103,17 @@ class ConvergingTriangleResult: def pivots_fractal( high: np.ndarray, low: np.ndarray, k: int = 3 ) -> Tuple[np.ndarray, np.ndarray]: - """左右窗口分形：返回 pivot_high_idx, pivot_low_idx""" + """左右窗口分形：返回 pivot_high_idx, pivot_low_idx + + 使用 nanmax/nanmin 来忽略 NaN 值 + """ n = len(high) ph: List[int] = [] pl: List[int] = [] for i in range(k, n - k): - if high[i] == np.max(high[i - k : i + k + 1]): + if not np.isnan(high[i]) and high[i] == np.nanmax(high[i - k : i + k + 1]): ph.append(i) - if low[i] == np.min(low[i - k : i + k + 1]): + if not np.isnan(low[i]) and low[i] == np.nanmin(low[i - k : i + k + 1]): pl.append(i) return np.array(ph, dtype=int), np.array(pl, dtype=int) @@ -138,22 +141,36 @@ def pivots_fractal_hybrid( n = len(high) # 确认枢轴点（完整窗口） + # 使用 nanmax/nanmin 来忽略 NaN 值 confirmed_ph: List[int] = [] confirmed_pl: List[int] = [] for i in range(k, n - k): - if high[i] == np.max(high[i - k : i + k + 1]): + if not np.isnan(high[i]) and high[i] == np.nanmax(high[i - k : i + k + 1]): confirmed_ph.append(i) - if low[i] == np.min(low[i - k : i + k + 1]): + if not np.isnan(low[i]) and low[i] == np.nanmin(low[i - k : i + k + 1]): confirmed_pl.append(i) # 候选枢轴点（灵活窗口，最近flexible_zone天） + # 优化逻辑：使用对称窗口，左右各看 min(k, right_avail+1) 天 + # 这样可以识别出窗口末端的局部枢轴点 + # 使用 nanmax/nanmin 来忽略 NaN 值 candidate_ph: List[int] = [] candidate_pl: List[int] = [] for i in range(max(k, n - flexible_zone), n): + if np.isnan(high[i]) or np.isnan(low[i]): + continue # 跳过 NaN 点 + right_avail = n - 1 - i - if high[i] == np.max(high[i - k : i + right_avail + 1]): + # 使用对称的短窗口：左右各看 right_avail+1 天（至少1天） + # 但左边最多看 k 天（保证有足够数据） + left_look = min(k, max(right_avail + 1, 3)) # 至少看3天 + left_start = max(0, i - left_look) + right_end = min(n, i + right_avail + 1) + + # 在对称窗口内是最大/最小即为候选枢轴点 + if high[i] == np.nanmax(high[left_start : right_end]): candidate_ph.append(i) - if low[i] == np.min(low[i - k : i + right_avail + 1]): + if low[i] == np.nanmin(low[left_start : right_end]): candidate_pl.append(i) return ( @@ -232,21 +249,27 @@ def fit_pivot_line( pivot_values: np.ndarray, mode: str = "upper", min_points: int = 2, + outlier_threshold: float = 2.5, + max_iterations: int = 2, ) -> Tuple[float, float, np.ndarray]: """ - 枢轴点连线法：选择合适的枢轴点连成边界线 + 迭代离群点移除的枢轴点拟合算法 - 改进策略（2026-01-26）： - - 使用多个枢轴点进行线性回归，而不是只选2个点 - - 将时间轴分为3段，每段选择最具代表性的点 - - 上沿选最高点，下沿选最低点 - - 使用选中的点进行线性回归，充分利用所有信息 + 改进策略（2026-01-26 v2）： + 1. 先用所有点做初始拟合 + 2. 识别并移除偏离拟合线的"弱"点 + 3. 迭代直到收敛 + + 对于上沿线：移除价格明显低于拟合线的点（弱高点） + 对于下沿线：移除价格明显高于拟合线的点（弱低点） Args: pivot_indices: 枢轴点的X坐标（索引） pivot_values: 枢轴点的Y值（价格） mode: "upper"(上沿) 或 "lower"(下沿) - min_points: 最少需要的枢轴点数 + min_points: 最少需要的枢轴点数（默认2，实际保留至少3个） + outlier_threshold: 离群点阈值（标准差倍数，默认2.5，更宽松） + max_iterations: 最大迭代次数（默认2，避免过度过滤） Returns: (slope, intercept, selected_indices): 斜率、截距、选中的枢轴点索引 @@ -263,87 +286,82 @@ def fit_pivot_line( if n < 2: return 0.0, 0.0, np.array([]) - # 新策略：分段选择代表性点进行回归 - if n <= 4: - # 点数少，直接用所有点 - selected_mask = np.ones(n, dtype=bool) - else: - # 点数多，分3段选择 - selected_mask = np.zeros(n, dtype=bool) - - # 分3段 - segment_size = n // 3 - if segment_size < 1: - segment_size = 1 - - segments = [ - range(0, min(segment_size, n)), - range(segment_size, min(2 * segment_size, n)), - range(2 * segment_size, n), - ] - - for seg in segments: - if len(seg) == 0: - continue - - seg_list = list(seg) - seg_y = y_sorted[seg_list] - - if mode == "upper": - # 上沿：选该段最高点 - best_idx_in_seg = np.argmax(seg_y) - else: - # 下沿：选该段最低点 - best_idx_in_seg = np.argmin(seg_y) - - selected_mask[seg_list[best_idx_in_seg]] = True + # 初始化：所有点都参与 + active_mask = np.ones(n, dtype=bool) + min_keep = max(3, min_points) # 至少保留3个点 - # 获取选中的点 - selected_x = x_sorted[selected_mask] - selected_y = y_sorted[selected_mask] - selected_indices_sorted = np.where(selected_mask)[0] + # 迭代离群点移除 + for iteration in range(max_iterations): + active_indices = np.where(active_mask)[0] + if len(active_indices) <= min_keep: + break + + # 当前活跃点 + x_active = x_sorted[active_mask] + y_active = y_sorted[active_mask] + + # 线性回归 + a, b = fit_line(x_active, y_active) + + # 计算拟合值 + fitted_values = a * x_active + b + + # 计算残差（方向性残差） + if mode == "upper": + # 上沿：残差 = 拟合值 - 实际值 + # 正残差表示点在拟合线下方（弱高点） + residuals = fitted_values - y_active + else: + # 下沿：残差 = 实际值 - 拟合值 + # 正残差表示点在拟合线上方（弱低点） + residuals = y_active - fitted_values + + # 计算标准差 + std_residual = np.std(residuals) + if std_residual < 1e-10: + # 所有点几乎在一条线上，无需移除 + break + + # 识别离群点（残差 > threshold * std） + threshold = outlier_threshold * std_residual + outlier_mask_active = residuals > threshold + + if not np.any(outlier_mask_active): + # 没有离群点，收敛 + break + + # 找到最大离群点 + max_outlier_idx_in_active = np.argmax(residuals) + + # 检查是否确实是离群点 + if residuals[max_outlier_idx_in_active] <= threshold: + break + + # 检查移除后是否仍有足够的点 + if np.sum(active_mask) <= min_keep: + break + + # 在原始索引中标记移除 + original_idx = active_indices[max_outlier_idx_in_active] + active_mask[original_idx] = False + + # 最终拟合 + selected_indices_sorted = np.where(active_mask)[0] + selected_x = x_sorted[active_mask] + selected_y = y_sorted[active_mask] if len(selected_x) < 2: # 兜底：使用首尾两点 - selected_mask = np.zeros(n, dtype=bool) - selected_mask[0] = True - selected_mask[-1] = True - selected_x = x_sorted[selected_mask] - selected_y = y_sorted[selected_mask] - selected_indices_sorted = np.where(selected_mask)[0] + active_mask = np.zeros(n, dtype=bool) + active_mask[0] = True + active_mask[-1] = True + selected_x = x_sorted[active_mask] + selected_y = y_sorted[active_mask] + selected_indices_sorted = np.where(active_mask)[0] # 线性回归 a, b = fit_line(selected_x, selected_y) - # 验证覆盖情况 - fitted_all = a * x_sorted + b - tolerance = 0.03 # 3% 容差 - - if mode == "upper": - # 上沿：确保所有点在线下方或附近 - violations = y_sorted > fitted_all + tolerance * np.mean(y_sorted) - if np.any(violations): - # 有点严重超出，添加全局最高点重新拟合 - global_max_idx = np.argmax(y_sorted) - if not selected_mask[global_max_idx]: - selected_mask[global_max_idx] = True - selected_x = x_sorted[selected_mask] - selected_y = y_sorted[selected_mask] - selected_indices_sorted = np.where(selected_mask)[0] - a, b = fit_line(selected_x, selected_y) - else: - # 下沿：确保所有点在线上方或附近 - violations = y_sorted < fitted_all - tolerance * np.mean(y_sorted) - if np.any(violations): - # 有点严重低于，添加全局最低点重新拟合 - global_min_idx = np.argmin(y_sorted) - if not selected_mask[global_min_idx]: - selected_mask[global_min_idx] = True - selected_x = x_sorted[selected_mask] - selected_y = y_sorted[selected_mask] - selected_indices_sorted = np.where(selected_mask)[0] - a, b = fit_line(selected_x, selected_y) - # 返回原始索引 selected_original = sort_idx[selected_indices_sorted]