From 93949d5f8ac64f2a0adb7bfa5ff39a67440cdbe7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=A4=9A=E5=AE=8F=E5=85=89?= <542672041@qq.com> Date: Wed, 4 Mar 2026 14:46:19 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=94=B6=E6=95=9B=E4=B8=89=E8=A7=92?= =?UTF-8?q?=E5=BD=A2=E5=BD=A2=E6=80=81=E5=8F=AF=E8=A7=86=E5=8C=96=E9=AA=8C?= =?UTF-8?q?=E8=AF=81=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 功能： - 支持中文名称搜索股票（如"中控技术"、"英伟达"） - 支持 A 股和美股 - 趋势线从窗口起点延伸到终点（与前端一致） - Y 轴自动调整以包含趋势线 - 支持 --window 参数（1Y/3Y/5Y/ALL） 用法： python validate.py 中控技术 日 --window 3Y --save python validate.py 英伟达 日 --window 3Y --save Co-Authored-By: Claude Opus 4.6 --- triangle-validator/.gitignore | 26 ++ triangle-validator/CLAUDE.md | 75 ++++++ triangle-validator/README.md | 152 +++++++++++ triangle-validator/batch_validate.py | 116 +++++++++ triangle-validator/chart.py | 361 ++++++++++++++++++++++++++ triangle-validator/config.py | 28 ++ triangle-validator/requirements.txt | 5 + triangle-validator/run.bat | 5 + triangle-validator/run.sh | 9 + triangle-validator/validate.py | 365 +++++++++++++++++++++++++++ 10 files changed, 1142 insertions(+) create mode 100644 triangle-validator/.gitignore create mode 100644 triangle-validator/CLAUDE.md create mode 100644 triangle-validator/README.md create mode 100644 triangle-validator/batch_validate.py create mode 100644 triangle-validator/chart.py create mode 100644 triangle-validator/config.py create mode 100644 triangle-validator/requirements.txt create mode 100644 triangle-validator/run.bat create mode 100644 triangle-validator/run.sh create mode 100644 triangle-validator/validate.py diff --git a/triangle-validator/.gitignore b/triangle-validator/.gitignore new file mode 100644 index 0000000..6619d0f --- /dev/null +++ b/triangle-validator/.gitignore @@ -0,0 +1,26 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +env/ +venv/ +.venv/ + +# Output +output/*.png +output/*.jpg + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# OS +.DS_Store +Thumbs.db + +# Local config (optional, if you want to keep token private) +# config.py diff --git a/triangle-validator/CLAUDE.md b/triangle-validator/CLAUDE.md new file mode 100644 index 0000000..b3eddfd --- /dev/null +++ b/triangle-validator/CLAUDE.md @@ -0,0 +1,75 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## 项目概述 + +收敛三角形（Converging Triangle）形态检测可视化验证工具。通过 HTTP API 调用远程公式执行，绘制 K 线图 + 三角形趋势线，帮助快速目测验证形态检测结果。**支持 A 股和美股**。 + +## 常用命令 + +```bash +# 安装依赖 +pip install requests mplfinance matplotlib numpy pandas + +# 单股验证 +python validate.py <股票名或代码> [频率] [--save] [--no-show] [--date YYYYMMDD] [--mode 0-3] [--window 1Y/3Y/5Y/ALL] + +# 示例 +python validate.py 中远海控 # 日K，弹窗显示 +python validate.py 中控技术 日 --window 3Y --save # 3年K线，保存图片 +python validate.py 英伟达 日 --window 3Y --save # 美股也支持 +python validate.py SH600519 月 --no-show --save # 月K，只保存不弹窗 + +# 批量验证 +python batch_validate.py --no-show --save +``` + +**参数说明**： +- `freq`: 日 / 周 / 月，默认日 +- `--save`: 保存 PNG 到 output/ 目录 +- `--no-show`: 不弹出交互窗口 +- `--mode`: 强度模式 0=等权 1=激进 2=保守 3=放量 +- `--window`: K 线时间窗口 1Y/3Y/5Y/ALL，默认 3Y + +## 架构 + +``` +数据流： +validate.py + ① GET {SEARCH_API_URL}/smartstock/assets/search?q=<关键词> + → 获取正确的 ticker（中控技术 → SH688777，英伟达 → NVDA.O） + + ② POST {GPT_SERVER_URL}/innerServer/runOneFormula + body: { formula_str: "结果=收敛三角形详情(ticker,freq,True,mode)" } + → response.output = MongoDB _id + + ③ GET {DATA_SERVER_URL}/api/timeSeries/requestIndexDetail?id=<_id> + header: Authorization: Bearer + → data.data_info.detail (含 strength/direction/chart_data) + + ④ GET {KLINE_API_URL}/dataSupport/getKLineDataByTicker?ticker=&begin_date= + → 获取 K 线 OHLC 数据 + + ⑤ chart.draw_triangle_chart(detail) + → mplfinance 绘制 K 线 + 趋势线 + → Y 轴自动调整以包含趋势线 + → output/{ticker}_{freq}K_{date}.png +``` + +**模块职责**： +- `validate.py`: CLI 入口、资产搜索、公式执行、详情获取、K 线数据获取 +- `batch_validate.py`: 批量验证，修改 `BATCH_CASES` 列表自定义验证标的 +- `chart.py`: mplfinance 绘图，趋势线延伸逻辑（与前端 TechPattern.vue 一致） +- `config.py`: API 端点配置，Token 过期时在此更新 + +## 与前端一致性 + +`chart.py` 中的趋势线绘制逻辑与 `D:\project\frontend_mobile\src\views\workflow\components\assetRatingV2\components\TechPattern.vue` 完全一致： +- 趋势线从 `window_start_date` 延伸到 `window_end_date` +- 使用公式数据中的 `index` 计算斜率 +- Y 轴范围包含趋势线价格 + 5% 边距 + +## 配置 + +Token 过期时，更新 `config.py` 中的 `AUTH_TOKEN`。 diff --git a/triangle-validator/README.md b/triangle-validator/README.md new file mode 100644 index 0000000..f8345ff --- /dev/null +++ b/triangle-validator/README.md @@ -0,0 +1,152 @@ +# triangle-validator + +收敛三角形（Converging Triangle）形态检测可视化验证工具。 + +通过调用远程 HTTP API 执行公式并拉取结果，绘制 K 线图 + 三角形趋势线， +帮助快速目测验证形态检测结果是否准确。**无需本地数据库，无需 dunhe 虚拟环境。** + +**支持 A 股和美股**（如英伟达、苹果等）。 + +--- + +## 目录结构 + +``` +triangle-validator/ +├── validate.py # 主入口：单股验证 +├── batch_validate.py # 批量验证多只股票 +├── chart.py # K 线 + 三角形绘图模块 +├── config.py # API 端点 & 认证配置 +├── requirements.txt # Python 依赖 +├── output/ # 生成的图片（自动创建） +└── README.md +``` + +--- + +## 前提条件 + +只需要系统 Python（3.8+）及两个依赖包： + +```bash +pip install requests mplfinance +``` + +--- + +## 单股验证 + +```bash +# 基本用法（日K，弹出窗口） +python validate.py 中远海控 + +# 使用中文名称（自动搜索转换为股票代码） +python validate.py 中控技术 日 --window 3Y --save + +# 美股也支持 +python validate.py 英伟达 日 --window 3Y --save + +# 周K，保存图片 +python validate.py 中远海控 周 --save + +# 月K，不弹窗，只保存 +python validate.py SH600519 月 --no-show --save + +# 指定截止日期 +python validate.py 贵州茅台 日 --date 20260120 --save + +# 使用激进模式（mode=1） +python validate.py 宁德时代 周 --mode 1 --save +``` + +### 参数说明 + +| 参数 | 说明 | +|------|------| +| `ticker` | 股票名称（如"中远海控"、"英伟达"）或代码（如"SH601919"）| +| `freq` | 频率：`日` / `周` / `月`，默认 `日` | +| `--save` | 保存 PNG 到 `output/` 目录 | +| `--no-show` | 不弹出交互窗口（适合服务器/批量场景）| +| `--date` | 截止日期，格式 `YYYYMMDD`，默认最新数据日 | +| `--mode` | 强度模式：0=等权 1=激进 2=保守 3=放量，默认 0 | +| `--window` | K 线时间窗口：`1Y` / `3Y` / `5Y` / `ALL`，默认 `3Y` | + +--- + +## 批量验证 + +```bash +# 默认批量（内置测试列表，日/周/月各一批） +python batch_validate.py + +# 只保存，不弹窗 +python batch_validate.py --no-show --save +``` + +修改 `batch_validate.py` 中的 `BATCH_CASES` 列表可自定义验证标的。 + +--- + +## 图表说明 + +生成的图表包含以下元素： + +| 元素 | 颜色 | 含义 | +|------|------|------| +| 红色趋势线 | `#ef4444` | 上沿（压力线），从窗口起点延伸到终点 | +| 绿色趋势线 | `#10b981` | 下沿（支撑线），从窗口起点延伸到终点 | +| 红色虚线 | `#ef4444` | 上突破价位 | +| 绿色虚线 | `#10b981` | 下突破价位 | + +标题格式：`{ticker} {频率}K 强度={score:.3f} {方向} 窗口 yyyyMMdd ~ yyyyMMdd` + +--- + +## 数据流 + +``` +validate.py + ① GET {SEARCH_API_URL}/smartstock/assets/search?q=<关键词> + → 获取正确的 ticker（如 "中控技术" → "SH688777"） + + ② POST {GPT_SERVER_URL}/innerServer/runOneFormula + body: { formula_str: "结果=收敛三角形详情(ticker,freq,True,mode)" } + → response.output = MongoDB _id + + ③ GET {DATA_SERVER_URL}/api/timeSeries/requestIndexDetail?id=<_id> + header: Authorization: Bearer + → data.data_info.detail + ├─ strength / direction / is_valid + ├─ breakout_price_up / breakout_price_down + └─ chart_data + ├─ upper_line / lower_line (趋势线端点) + ├─ upper_pivots / lower_pivots (枢轴点) + └─ window_start_date / window_end_date + + ④ GET {KLINE_API_URL}/dataSupport/getKLineDataByTicker?ticker=&begin_date= + → 获取 K 线 OHLC 数据（3年历史） + + ⑤ chart.draw_triangle_chart(detail) + → mplfinance 绘制 K 线 + 趋势线 + → Y 轴自动调整以包含趋势线 + → output/{ticker}_{freq}K_{date}.png +``` + +--- + +## 常见问题 + +**Q: 报错 `ModuleNotFoundError: No module named 'requests'`** +A: 执行 `pip install requests mplfinance`。 + +**Q: 报错 `ConnectionError` 或 `404`** +A: 检查 `config.py` 中的 `GPT_SERVER_URL` / `DATA_SERVER_URL` 是否正确，以及 token 是否过期。 + +**Q: 公式执行失败，未返回 _id** +A: 确认 ticker 名称正确，或改用标准代码（如 `SH600519`）。 + +**Q: 图表中文乱码（显示方框）** +A: 安装中文字体，推荐 `Microsoft YaHei`（Windows 自带）或 `SimHei`。chart.py 会自动检测并应用。 + +**Q: 美股 K 线数据获取失败** +A: 使用中文名称（如"英伟达"）而非股票代码，程序会自动通过搜索接口获取正确的 ticker。 diff --git a/triangle-validator/batch_validate.py b/triangle-validator/batch_validate.py new file mode 100644 index 0000000..4add6a1 --- /dev/null +++ b/triangle-validator/batch_validate.py @@ -0,0 +1,116 @@ +""" +batch_validate.py — 批量验证多只股票的收敛三角形形态（HTTP 版） + +用法： + python batch_validate.py [--no-show] [--save] + +--no-show 不弹窗，直接生成所有图（适合无显示器/批量场景） +--save 保存图片到 output/ 目录 + +修改下方 BATCH_CASES 可自定义验证列表。 +""" + +import sys +import argparse +from pathlib import Path + +from validate import run_formula, fetch_detail + +_THIS_DIR = Path(__file__).parent.resolve() + + +# ── 批量验证列表（可自由修改）─────────────────────────────────────────────────── +# 格式：(ticker, freq, mode=0, date=None) +BATCH_CASES = [ + # 日 K —— 知名个股 + ('中远海控', '日', 0, None), + ('贵州茅台', '日', 0, None), + ('宁德时代', '日', 0, None), + ('中芯国际', '日', 0, None), + # 周 K —— 宽基指数 + ('沪深300', '周', 0, None), + ('中证500', '周', 0, None), + # 月 K + ('贵州茅台', '月', 0, None), + ('中远海控', '月', 0, None), +] + + +# ── 单条处理 ────────────────────────────────────────────────────────────────── +def process_one(ticker: str, freq: str, mode: int, date, save: bool, show: bool) -> bool: + print(f'[batch] → {ticker!r} {freq}K ...') + + try: + index_id = run_formula(ticker, freq, mode, date) + except Exception as e: + print(f'[batch] ✗ 公式执行失败: {e}') + return False + + try: + detail = fetch_detail(index_id) + except Exception as e: + print(f'[batch] ✗ 读取详情失败: {e}') + return False + + cd = detail.get('chart_data', {}) + strength = detail.get('strength', 0) + direction = detail.get('direction', 'none') + print( + f'[batch] ✓ 强度={strength:.4f} 方向={direction} ' + f'窗口={cd.get("window_start_date", "?")}~{cd.get("window_end_date", "?")}' + ) + + if not cd.get('klines'): + print('[batch] ✗ 没有 klines，跳过绘图') + return False + + from chart import draw_triangle_chart + + safe_ticker = ticker.replace('/', '_').replace('\\', '_') + output_path = None + if save: + output_dir = _THIS_DIR / 'output' + fname = f'{safe_ticker}_{freq}K_{cd.get("target_date", "")}.png' + output_path = str(output_dir / fname) + + try: + draw_triangle_chart(detail=detail, output_path=output_path, show=show) + except Exception as e: + print(f'[batch] ✗ 绘图失败: {e}') + return False + + return True + + +# ── 主函数 ──────────────────────────────────────────────────────────────────── +def main(): + parser = argparse.ArgumentParser(description='收敛三角形批量验证（HTTP 版）') + parser.add_argument('--no-show', action='store_true', help='不弹出图形窗口') + parser.add_argument('--save', action='store_true', help='保存图片到 output/') + args = parser.parse_args() + + show = not args.no_show + save = args.save + + if not show and not save: + print('[batch] 提示：既未开启 --save 也未弹窗，将只打印摘要。') + + ok_count = 0 + fail_count = 0 + + for case in BATCH_CASES: + ticker, freq = case[0], case[1] + mode = case[2] if len(case) > 2 else 0 + date = case[3] if len(case) > 3 else None + + success = process_one(ticker, freq, mode, date, save, show) + if success: + ok_count += 1 + else: + fail_count += 1 + + print(f'\n[batch] 完成 成功={ok_count} 失败={fail_count} 总计={ok_count + fail_count}') + + +if __name__ == '__main__': + main() diff --git a/triangle-validator/chart.py b/triangle-validator/chart.py new file mode 100644 index 0000000..71b8740 --- /dev/null +++ b/triangle-validator/chart.py @@ -0,0 +1,361 @@ +""" +收敛三角形可视化绘制模块 + +根据 chart_data (包含 klines) 绘制 K 线图 + 三角形趋势线 +""" +import os +import numpy as np +import pandas as pd +import mplfinance as mpf +import matplotlib.pyplot as plt +import matplotlib.font_manager as fm + + +# ── 中文字体支持 ────────────────────────────────────────────────────────────── +def _setup_chinese_font(): + """尝试配置中文字体，避免乱码""" + candidates = [ + 'Microsoft YaHei', 'SimHei', 'PingFang SC', 'STSong', + 'Noto Sans CJK SC', 'WenQuanYi Zen Hei', + ] + available = {f.name for f in fm.fontManager.ttflist} + for name in candidates: + if name in available: + plt.rcParams['font.family'] = name + plt.rcParams['axes.unicode_minus'] = False + return name + # 找不到就用默认，可能出现方框 + return None + + +_CHINESE_FONT = _setup_chinese_font() + + +# ── 工具函数 ────────────────────────────────────────────────────────────────── +def _date_int_to_str(d: int) -> str: + """20250327 → '2025-03-27'""" + s = str(int(d)) + return f'{s[:4]}-{s[4:6]}-{s[6:]}' + + +def _build_dataframe(klines: dict) -> pd.DataFrame: + """从 klines 字典构建 mplfinance 需要的 DataFrame""" + dates_str = [_date_int_to_str(d) for d in klines['dates']] + df = pd.DataFrame( + { + 'Open': [v if v is not None else np.nan for v in klines['open']], + 'High': [v if v is not None else np.nan for v in klines['high']], + 'Low': [v if v is not None else np.nan for v in klines['low']], + 'Close': [v if v is not None else np.nan for v in klines['close']], + 'Volume': [v if v is not None else np.nan for v in klines['volume']], + }, + index=pd.DatetimeIndex(dates_str), + ) + return df, dates_str + + +def _make_aline(line_pts: list): + """ + 将 [{"date": int, "price": float}, ...] 转换为 mplfinance aline 格式的 2-端点元组 + + Returns: + (("2025-03-27", price1), ("2026-01-20", price2)) or None + """ + if not line_pts or len(line_pts) < 2: + return None + p1 = (_date_int_to_str(line_pts[0]['date']), line_pts[0]['price']) + p2 = (_date_int_to_str(line_pts[-1]['date']), line_pts[-1]['price']) + return (p1, p2) + + +def _build_pivot_arrays(pivots: list, dates_str: list, n: int) -> np.ndarray: + """ + 将枢轴点列表映射为长度 n 的 float 数组（非枢轴位置为 NaN） + """ + date_to_idx = {d: i for i, d in enumerate(dates_str)} + arr = np.full(n, np.nan) + for pt in pivots: + ds = _date_int_to_str(pt['date']) + idx = date_to_idx.get(ds) + if idx is not None: + arr[idx] = pt['price'] + return arr + + +# ── 主绘图函数 ──────────────────────────────────────────────────────────────── +def draw_triangle_chart( + detail: dict, + output_path: str = None, + show: bool = True, + show_pivots: bool = False, # 是否显示枢轴点（默认不显示，与前端一致） +): + """ + 绘制收敛三角形 K 线图 + + Args: + detail: 收敛三角形详情字典，即 ``result.info['detail']`` + output_path: 保存路径（如 ``'output/CSCO_日.png'``），None 则不保存 + show: 是否弹出交互窗口（在无 GUI 环境下设 False） + show_pivots: 是否显示枢轴点标记（默认 False，与前端一致） + + Returns: + output_path (便于链式调用) + """ + chart = detail.get('chart_data', {}) + klines = chart.get('klines') + if not klines: + raise ValueError( + "chart_data 中没有 klines，请用 include_klines=True 调用收敛三角形详情" + ) + + # ── 基本属性 ───────────────────────────────── + ticker = chart.get('ticker', '') + freq = chart.get('freq', 'D') + freq_label = {'D': '日K', 'W': '周K', 'M': '月K'}.get(freq, freq) + strength = detail.get('strength', 0.0) + direction = detail.get('direction', 'none') + dir_label = {'up': '↑上涨突破', 'down': '↓下跌突破', 'none': '整理中'}.get(direction, direction) + window_start = chart.get('window_start_date', '') + window_end = chart.get('window_end_date', '') + touches_upper = detail.get('touches_upper', '?') + touches_lower = detail.get('touches_lower', '?') + + # ── 构建 DataFrame ───────────────────────────── + df, dates_str = _build_dataframe(klines) + n = len(df) + + # ── 趋势线：从 window_start_date 延伸到 window_end_date（与前端一致）────────── + alines_list = [] + alines_colors = [] + + upper_line = chart.get('upper_line', []) + lower_line = chart.get('lower_line', []) + + # 将日期字符串转为集合，用于查找 + date_to_idx = {d: i for i, d in enumerate(dates_str)} + window_start_str = _date_int_to_str(window_start) if window_start else None + window_end_str = _date_int_to_str(window_end) if window_end else None + + # 辅助函数：查找最近的日期（与前端 getClosestDateCategory 一致） + def get_closest_date_idx(target_date: int, dates_str: list) -> int: + """查找目标日期在 K 线数据中最接近的位置""" + target_str = _date_int_to_str(target_date) + if target_str in date_to_idx: + return date_to_idx[target_str] + + # 找不到精确匹配时，找最近的日期 + from datetime import datetime + target_dt = datetime.strptime(target_str, '%Y-%m-%d') + min_diff = float('inf') + closest_idx = 0 + + for i, d in enumerate(dates_str): + dt = datetime.strptime(d, '%Y-%m-%d') + diff = abs((dt - target_dt).days) + if diff < min_diff: + min_diff = diff + closest_idx = i + + return closest_idx + + # 辅助函数：计算延伸后的趋势线（与前端 getExtendedLinePoints 完全一致） + def make_extended_aline(line_pts, pivot_point, color): + """ + 与前端 TechPattern.vue 的 getExtendedLinePoints 函数逻辑完全一致 + 趋势线从 window_start_date 延伸到 window_end_date + """ + if not line_pts or len(line_pts) < 2: + return None + + p1 = line_pts[0] + p2 = line_pts[-1] + + # 检查是否有 index 字段（公式返回的数据） + has_index = 'index' in p1 and 'index' in p2 and pivot_point and 'index' in pivot_point + + if has_index and window_start and window_end: + # 使用原始 index 计算斜率（与前端一致） + slope = (p2['price'] - p1['price']) / (p2['index'] - p1['index']) if p2['index'] != p1['index'] else 0 + + # 前端: startIdx = 0 (window_start_date 对应 index 0) + start_idx = 0 + + # 查找 window_end_date 对应的 index（与前端一致） + end_idx = p2['index'] + all_points = (chart.get('upper_line', []) + chart.get('lower_line', []) + + chart.get('upper_pivots', []) + chart.get('lower_pivots', [])) + for pt in all_points: + if pt.get('date') == window_end: + end_idx = pt.get('index', end_idx) + break + + # 计算延伸后的价格（与前端一致） + pivot_index = pivot_point['index'] + pivot_price = pivot_point['price'] + start_price = pivot_price + slope * (start_idx - pivot_index) + end_price = pivot_price + slope * (end_idx - pivot_index) + + # 将 window_start_date 和 window_end_date 映射到 K 线数据的索引 + kline_start_idx = get_closest_date_idx(window_start, dates_str) + kline_end_idx = get_closest_date_idx(window_end, dates_str) + + print(f'[chart] 趋势线延伸: {window_start_str}({kline_start_idx}) -> {window_end_str}({kline_end_idx})') + print(f'[chart] 价格: {start_price:.2f} -> {end_price:.2f}') + + return ( + (kline_start_idx, start_price), + (kline_end_idx, end_price) + ) + + # 降级处理：仅连接两点（与前端一致） + d1 = _date_int_to_str(p1['date']) + d2 = _date_int_to_str(p2['date']) + + idx1 = date_to_idx.get(d1) + idx2 = date_to_idx.get(d2) + + if idx1 is None or idx2 is None: + print(f'[chart] 警告: 趋势线日期不在 K 线数据中: {d1} 或 {d2}') + return None + + return ( + (idx1, p1['price']), + (idx2, p2['price']) + ) + + # 绘制上轨线（红色）- 与前端一致 + if upper_line and window_start and window_end: + upper_aline = make_extended_aline(upper_line, upper_line[0] if upper_line else None, '#ef4444') + if upper_aline: + alines_list.append(upper_aline) + alines_colors.append('#ef4444') # 红色 + + # 绘制下轨线（绿色）- 与前端一致 + if lower_line and window_start and window_end: + lower_aline = make_extended_aline(lower_line, lower_line[0] if lower_line else None, '#10b981') + if lower_aline: + alines_list.append(lower_aline) + alines_colors.append('#10b981') # 绿色 + + # ── 枢轴点散点图（可选）───────────────────────── + addplots = [] + if show_pivots: + upper_pivots_arr = _build_pivot_arrays(chart.get('upper_pivots', []), dates_str, n) + lower_pivots_arr = _build_pivot_arrays(chart.get('lower_pivots', []), dates_str, n) + + if np.any(~np.isnan(upper_pivots_arr)): + addplots.append(mpf.make_addplot( + upper_pivots_arr, type='scatter', + markersize=100, marker='^', color='#ef4444', alpha=0.85, + )) + if np.any(~np.isnan(lower_pivots_arr)): + addplots.append(mpf.make_addplot( + lower_pivots_arr, type='scatter', + markersize=100, marker='v', color='#10b981', alpha=0.85, + )) + + # ── 突破价位水平线 ───────────────────────────── + bp_up = detail.get('breakout_price_up') + bp_down = detail.get('breakout_price_down') + hlines_vals = [] + hlines_colors = [] + if bp_up and float(bp_up) > 0: + hlines_vals.append(float(bp_up)) + hlines_colors.append('#ef4444') + if bp_down and float(bp_down) > 0: + hlines_vals.append(float(bp_down)) + hlines_colors.append('#10b981') + + # ── 标题 ──────────────────────────────────────── + title = ( + f'{ticker} {freq_label} ' + f'强度={strength:.3f} {dir_label}\n' + f'窗口 {window_start}~{window_end} ' + f'(上沿触碰={touches_upper} 下沿触碰={touches_lower})' + ) + + # ── 组装 plot 参数 ──────────────────────────────── + _rc = {'axes.unicode_minus': False} + if _CHINESE_FONT: + _rc['font.family'] = _CHINESE_FONT + _rc['font.sans-serif'] = [_CHINESE_FONT, 'DejaVu Sans'] + + _style = mpf.make_mpf_style(base_mpf_style='yahoo', rc=_rc) + + kwargs = dict( + type='candle', + volume=False, # 不显示成交量 + title=title, + style=_style, + figsize=(18, 10), + tight_layout=True, + ) + + if alines_list: + kwargs['alines'] = dict( + alines=alines_list, + colors=alines_colors, + linewidths=2.0, + alpha=0.90, + ) + + if addplots: + kwargs['addplot'] = addplots + + if hlines_vals: + kwargs['hlines'] = dict( + hlines=hlines_vals, + colors=hlines_colors, + linestyle='--', + linewidths=1.2, + alpha=0.7, + ) + + if output_path: + os.makedirs(os.path.dirname(os.path.abspath(output_path)), exist_ok=True) + + # ── 绘图 ───────────────────────────────────────── + # 先不绘制趋势线，获取 figure 和 axes + _kwargs = {k: v for k, v in kwargs.items() if k not in ['alines', 'savefig']} + fig, axes = mpf.plot(df, **_kwargs, returnfig=True, warn_too_much_data=1000) + + # 使用 matplotlib 直接在主图上绘制趋势线 + # alines_list 格式: [((idx1, price1), (idx2, price2)), ...] + if alines_list and len(axes) > 0: + main_ax = axes[0] # 主图（K线图） + + # 收集趋势线价格，用于调整 Y 轴范围（与前端一致） + trend_line_prices = [] + for aline, color in zip(alines_list, alines_colors): + (idx1, p1), (idx2, p2) = aline + trend_line_prices.extend([p1, p2]) + # 直接使用整数索引作为 x 坐标（与 mplfinance 内部一致） + main_ax.plot([idx1, idx2], [p1, p2], color=color, linewidth=2, alpha=0.9) + + # 调整 Y 轴范围以包含趋势线（与前端一致，增加 5% 边距） + if trend_line_prices: + current_ylim = main_ax.get_ylim() + kline_min, kline_max = current_ylim + + # 合并 K 线和趋势线的价格范围 + all_min = min(kline_min, min(trend_line_prices)) + all_max = max(kline_max, max(trend_line_prices)) + price_range = all_max - all_min + + # 增加 5% 边距（与前端一致） + new_min = all_min - price_range * 0.05 + new_max = all_max + price_range * 0.05 + + main_ax.set_ylim(new_min, new_max) + print(f'[chart] Y 轴范围调整: {kline_min:.2f}~{kline_max:.2f} -> {new_min:.2f}~{new_max:.2f}') + + if show: + plt.show() + + if output_path: + fig.savefig(output_path, dpi=150, bbox_inches='tight') + print(f'[chart] 已保存: {output_path}') + + plt.close(fig) + + return output_path diff --git a/triangle-validator/config.py b/triangle-validator/config.py new file mode 100644 index 0000000..2b2d9d4 --- /dev/null +++ b/triangle-validator/config.py @@ -0,0 +1,28 @@ +""" +triangle-validator 配置文件 + +当 Token 过期时，在此处更新即可。 +""" + +# ── gpt_server (公式执行) +# 运行公式用的 API，与 MCP 工具走同一路径 +GPT_SERVER_URL = "http://test.guanzhao12.com:3006" + +# ── timeSeries (读取详情结果) +# 读取公式执行结果 (含 chart_data / klines) +DATA_SERVER_URL = "http://test.guanzhao12.com:8081" + +# ── K线数据 API (与前端一致) +# 单独获取 K 线 OHLC 数据 +KLINE_API_URL = "http://test.guanzhao12.com:3000" + +# ── 资产搜索 API (与前端一致) +# 搜索股票/指数等资产 +SEARCH_API_URL = "http://test.guanzhao12.com:3009" + +# JWT Token —— 过期后在此处替换新 Token +AUTH_TOKEN = ( + "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9" + ".eyJ1c2VySWQiOjczLCJ1c2VyTmFtZSI6Im1jcCIsImlhdCI6MTc3MjQzODg1MCwiZXhwIjoxNzczMDQzNjUwfQ" + ".NGSoODRLjB1lHYL7bFGRGCRzIkzO_ebUEbqcakAhGvE" +) diff --git a/triangle-validator/requirements.txt b/triangle-validator/requirements.txt new file mode 100644 index 0000000..0cd0911 --- /dev/null +++ b/triangle-validator/requirements.txt @@ -0,0 +1,5 @@ +requests>=2.28 +mplfinance>=0.12.10 +matplotlib>=3.7 +numpy>=1.24 +pandas>=2.0 diff --git a/triangle-validator/run.bat b/triangle-validator/run.bat new file mode 100644 index 0000000..25716bb --- /dev/null +++ b/triangle-validator/run.bat @@ -0,0 +1,5 @@ +@echo off +REM 使用 dunhe_dataServer 的 venv 运行 validate.py +REM 用法: run.bat 中远海控 [日|周|月] [--save] [--no-show] [--date 20260120] +set PYTHON=..\dunhe_dataServer\.venv\Scripts\python.exe +%PYTHON% validate.py %* diff --git a/triangle-validator/run.sh b/triangle-validator/run.sh new file mode 100644 index 0000000..ddb0f82 --- /dev/null +++ b/triangle-validator/run.sh @@ -0,0 +1,9 @@ +#!/usr/bin/env bash +# 使用 dunhe_dataServer 的 venv 运行 validate.py +# 用法: ./run.sh 中远海控 [日|周|月] [--save] [--no-show] [--date 20260120] +PYTHON="../dunhe_dataServer/.venv/Scripts/python.exe" +# Linux/macOS fallback +if [ ! -f "$PYTHON" ]; then + PYTHON="../dunhe_dataServer/.venv/bin/python" +fi +"$PYTHON" validate.py "$@" diff --git a/triangle-validator/validate.py b/triangle-validator/validate.py new file mode 100644 index 0000000..6cd5c6b --- /dev/null +++ b/triangle-validator/validate.py @@ -0,0 +1,365 @@ +""" +triangle-validator / validate.py + +收敛三角形形态可视化验证脚本（HTTP 版） + +流程： + 1. 调 gpt_server POST /api/gptAgent/runFormula 执行公式 → 拿到 _id + 2. 调 timeSeries GET /api/timeSeries/requestIndexDetail?id=_id → 拿到完整 detail（含 klines） + 3. 调 chart.draw_triangle_chart() 绘图 + +用法： + python validate.py <股票名或代码> [频率] [--save] [--no-show] [--date 日期] [--window 窗口] + +示例： + python validate.py 中远海控 + python validate.py 中远海控 周 --save + python validate.py SH600519 月 --no-show --save + python validate.py 贵州茅台 日 --date 20260120 --save + python validate.py SH688777 日 --window 1Y --save + python validate.py SH688777 日 --window ALL --save +""" + +import sys +import json +import argparse +from pathlib import Path + +import requests + +from config import GPT_SERVER_URL, DATA_SERVER_URL, KLINE_API_URL, SEARCH_API_URL, AUTH_TOKEN +from datetime import datetime, timedelta + +_THIS_DIR = Path(__file__).parent.resolve() + + +# ── 时间窗口计算（与前端一致）─────────────────────────────────────────────── +def get_window_start_date(window: str) -> str: + """ + 根据时间窗口字符串计算相对于今天的开始日期（与前端 getWindowStartDate 一致） + + Args: + window: 时间窗口，如 '1Y', '3Y', '5Y', 'ALL' + + Returns: + 开始日期，格式为 YYYYMMDD，如 '20240130' + """ + if window == 'ALL': + return '' + + today = datetime.now() + + # 提取数字和单位 + import re + match = re.match(r'^(\d+)([YMD])$', window, re.IGNORECASE) + if not match: + print(f'[WARN] 无效的时间窗口格式: {window}，使用默认3年') + window = '3Y' + match = re.match(r'^(\d+)([YMD])$', window, re.IGNORECASE) + + amount = int(match.group(1)) + unit = match.group(2).upper() + + if unit == 'Y': # 年 + start_date = today - timedelta(days=amount * 365) + elif unit == 'M': # 月 + start_date = today - timedelta(days=amount * 30) + elif unit == 'D': # 日 + start_date = today - timedelta(days=amount) + else: + start_date = today + + return start_date.strftime('%Y%m%d') + + +# ── CLI 解析 ────────────────────────────────────────────────────────────────── +def parse_args(): + parser = argparse.ArgumentParser( + description='收敛三角形形态可视化验证', + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__, + ) + parser.add_argument('ticker', help='股票名称或代码，如：中远海控 / SH601919') + parser.add_argument( + 'freq', nargs='?', default='日', + choices=['日', '周', '月'], + help='K 线频率（默认：日）', + ) + parser.add_argument('--save', action='store_true', help='保存图片到 output/ 目录') + parser.add_argument('--no-show', action='store_true', help='不弹出图形窗口') + parser.add_argument('--date', type=int, default=None, metavar='YYYYMMDD', + help='截止日期，如 20260120') + parser.add_argument('--mode', type=int, default=0, choices=[0, 1, 2, 3], + help='强度模式：0=等权 1=激进 2=保守 3=放量（默认：0）') + parser.add_argument('--window', type=str, default='3Y', + choices=['1Y', '3Y', '5Y', 'ALL'], + help='K线时间窗口：1Y=1年 3Y=3年 5Y=5年 ALL=全部（默认：3Y，与前端一致）') + return parser.parse_args() + + +# ── Step 1：执行公式，拿 _id ────────────────────────────────────────────────── +def run_formula(ticker: str, freq: str, mode: int, date) -> str: + """ + 调 innerServer/runOneFormula 接口执行收敛三角形详情公式。 + Returns: + index_info._id (str) — 直接从 response.output 取 + """ + import uuid + + if date: + formula = f'结果=收敛三角形详情({ticker},{freq},True,{mode},{date})' + else: + formula = f'结果=收敛三角形详情({ticker},{freq},True,{mode})' + + url = f'{GPT_SERVER_URL}/innerServer/runOneFormula' + payload = { + 'formula_str': formula, + 'begin_date': 0, + 'user_name': 'mcp', + 'tool_name': 'triangle_validate', + 'session_id': f'tv-{uuid.uuid4().hex[:8]}', + } + + print(f'[validate] 执行公式: {formula}') + resp = requests.post(url, json=payload, timeout=120) + resp.raise_for_status() + + body = resp.json() + _id = (body.get('response', {}) or {}).get('output') + if not _id: + _id = ( + (body.get('response', {}) or {}) + .get('data', {}) + .get('index_info', {}) + .get('_id') + ) + if not _id: + raise RuntimeError( + f'公式执行失败，未返回 _id\n响应: {json.dumps(body, ensure_ascii=False)[:500]}' + ) + + print(f'[validate] 公式执行完成，_id = {_id}') + return str(_id) + + +# ── Step 2：读取完整 detail（含 klines）─────────────────────────────────────── +def fetch_detail(index_id: str) -> dict: + """ + 调 requestIndexDetail 接口读取 detail（含 chart_data.klines）。 + Returns: + detail dict：包含 strength / chart_data / klines 等所有字段 + """ + url = f'{DATA_SERVER_URL}/api/timeSeries/requestIndexDetail' + headers = {'Authorization': f'Bearer {AUTH_TOKEN}'} + params = {'id': index_id} + + print(f'[validate] 读取详情: {url}?id={index_id}') + resp = requests.get(url, headers=headers, params=params, timeout=30) + resp.raise_for_status() + + body = resp.json() + if body.get('code') != 0: + raise RuntimeError(f'requestIndexDetail 返回错误: {body}') + + return body['data']['data_info']['detail'] + + +# ── Step 2.4：搜索资产获取 ticker（与前端一致）─────────────────────────────────── +def search_asset(keyword: str) -> dict: + """ + 调用与前端相同的资产搜索 API。 + + Args: + keyword: 搜索关键词，如 "中控技术"、"SH688777"、"英伟达" + + Returns: + dict: { code, name, ticker, type } 或 None + """ + url = f'{SEARCH_API_URL}/smartstock/assets/search' + params = {'q': keyword} + + print(f'[validate] 搜索资产: {url}?q={keyword}') + resp = requests.get(url, params=params, timeout=10) + resp.raise_for_status() + + body = resp.json() + if body.get('code') != 0: + raise RuntimeError(f'搜索返回错误: {body}') + + items = body.get('data', {}).get('items', []) + if items: + result = items[0] + print(f'[validate] 搜索结果: {result}') + return result + else: + print(f'[validate] 搜索无结果') + return None + + +# ── Step 2.5：单独获取 K 线数据（与前端一致）─────────────────────────────────── +def fetch_kline_data(ticker: str, begin_date: int | str = None) -> dict: + """ + 调用与前端相同的 K 线数据 API。 + + Args: + ticker: 股票代码，如 "SH601919" + begin_date: 开始日期 YYYYMMDD 格式，如 20250101 + + Returns: + dict: { labels: [...], open: [...], high: [...], low: [...], close: [...] } + """ + url = f'{KLINE_API_URL}/dataSupport/getKLineDataByTicker' + headers = {'Authorization': f'Bearer {AUTH_TOKEN}'} + params = {'ticker': ticker} + if begin_date: + params['begin_date'] = begin_date + + print(f'[validate] 获取K线数据: {url}?ticker={ticker}&begin_date={begin_date}') + resp = requests.get(url, headers=headers, params=params, timeout=30) + resp.raise_for_status() + + body = resp.json() + if body.get('code') != 0: + raise RuntimeError(f'getKLineDataByTicker 返回错误: {body}') + + # 数据嵌套在 body['data'] 中 + data = body.get('data', {}) + print(f'[validate] K线数据获取成功，共 {len(data.get("labels", []))} 条') + return data + + +# ── Step 3：打印摘要 ────────────────────────────────────────────────────────── +def print_summary(detail: dict, ticker: str, freq: str): + cd = detail.get('chart_data', {}) + strength = detail.get('strength', 0) + direction = detail.get('direction', 'none') + is_valid = detail.get('is_valid', False) + bp_up = detail.get('breakout_price_up', '-') + bp_down = detail.get('breakout_price_down', '-') + klines = cd.get('klines', {}) + + print(f'\n{"=" * 52}') + print(f' {ticker} {freq}K 收敛三角形') + print(f'{"=" * 52}') + print(f' 有效形态 : {is_valid}') + print(f' 综合强度 : {strength:.4f}') + print(f' 突破方向 : {direction}') + print(f' 上突破价 : {bp_up}') + print(f' 下突破价 : {bp_down}') + print(f' 上沿触碰 : {detail.get("touches_upper", "?")}') + print(f' 下沿触碰 : {detail.get("touches_lower", "?")}') + print(f' 收敛比例 : {detail.get("width_ratio", 0):.4f}') + print(f' 窗口日期 : {cd.get("window_start_date", "?")} ~ {cd.get("window_end_date", "?")}') + print(f' K 线根数 : {len(klines.get("dates", []))}') + print(f'{"=" * 52}\n') + + +# ── 主函数 ──────────────────────────────────────────────────────────────────── +def run(args): + # 0. 搜索资产获取正确的 ticker（与前端一致） + search_result = search_asset(args.ticker) + if search_result: + kline_ticker = search_result.get('ticker') # 用于 K 线 API + asset_name = search_result.get('name') # 资产名称 + print(f'[validate] 资产: {asset_name}, Ticker: {kline_ticker}') + else: + # 搜索失败，使用原始输入 + kline_ticker = args.ticker + print(f'[WARN] 搜索失败，使用原始输入: {kline_ticker}') + + # 1. 执行公式 + try: + index_id = run_formula(args.ticker, args.freq, args.mode, args.date) + except Exception as e: + print(f'[ERROR] 公式执行失败: {e}') + sys.exit(1) + + # 2. 读取结果 + try: + detail = fetch_detail(index_id) + except Exception as e: + print(f'[ERROR] 读取详情失败: {e}') + sys.exit(1) + + # 3. 摘要 + print_summary(detail, args.ticker, args.freq) + + cd = detail.get('chart_data', {}) + + # 4. 获取 K 线数据（使用搜索到的 ticker） + # 根据 window 参数计算开始日期（与前端一致） + begin_date = get_window_start_date(args.window) + print(f'[validate] 时间窗口: {args.window}, 开始日期: {begin_date or "(全部)"}') + + # 获取 K 线数据 + klines = None + try: + kline_data = fetch_kline_data(kline_ticker, begin_date=begin_date) + + # 🔧 过滤掉 None 值（非交易日），与前端 filterAndProcessKLineData 一致 + raw_labels = kline_data.get('labels', []) + raw_open = kline_data.get('open', []) + raw_high = kline_data.get('high', []) + raw_low = kline_data.get('low', []) + raw_close = kline_data.get('close', []) + + valid_indices = [ + i for i in range(len(raw_labels)) + if raw_open[i] is not None + and raw_high[i] is not None + and raw_low[i] is not None + and raw_close[i] is not None + ] + + if valid_indices: + # 转换为过滤后的 klines 格式 + klines = { + 'dates': [raw_labels[i] for i in valid_indices], + 'open': [raw_open[i] for i in valid_indices], + 'high': [raw_high[i] for i in valid_indices], + 'low': [raw_low[i] for i in valid_indices], + 'close': [raw_close[i] for i in valid_indices], + 'volume': [0] * len(valid_indices), # 暂时用0填充 + } + print(f'[INFO] K 线数据已获取，原始 {len(raw_labels)} 条，有效 {len(klines["dates"])} 条') + else: + print(f'[WARN] K 线 API 返回空数据') + except Exception as e: + print(f'[WARN] 获取 K 线数据失败: {e}') + + # 如果 K 线 API 失败或返回空数据，使用公式返回的 klines + if not klines or not klines.get('dates'): + print('[INFO] 使用公式返回的 K 线数据') + klines = cd.get('klines') + if not klines or not klines.get('dates'): + print('[ERROR] 没有 K 线数据，跳过绘图。') + return + + cd['klines'] = klines + + # 5. 绘图 + from chart import draw_triangle_chart + + safe_ticker = args.ticker.replace('/', '_').replace('\\', '_') + output_path = None + if args.save: + output_dir = _THIS_DIR / 'output' + fname = f'{safe_ticker}_{args.freq}K_{cd.get("target_date", "")}.png' + output_path = str(output_dir / fname) + + try: + draw_triangle_chart( + detail=detail, + output_path=output_path, + show=not args.no_show, + ) + except Exception as e: + print(f'[ERROR] 绘图失败: {e}') + import traceback + traceback.print_exc() + sys.exit(1) + + +# ── 入口 ────────────────────────────────────────────────────────────────────── +if __name__ == '__main__': + run(parse_args())