bsp_monitor: 支持全部20个币对 + 数据源切换至data_provider + Python 3.9兼容

- fetcher.py: 数据源从CCXT改为data_provider HTTP API，新增get_symbols()自动获取所有币对
- main.py: 重构为多币对架构，每个币对独立SymbolState（pivot_monitor/BSP去重/首轮抑制）
- engine.py: format_bsp_detail()支持动态币对名
- ChanPivotMonitor/Classifier: 修复Python 3.9类型注解兼容（X|None → Optional[X]）
- 首轮初始化时不推送中枢和BSP，避免启动时20条消息轰炸

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

ChanPivotClassifier.py

@@ -8,6 +8,7 @@ Feature 描述中枢内部结构，Label 记录中枢后实际演化。
 
 import math
 import json
+from typing import Optional
 from ChanEnum import Chan_BI_DIR
 
 
@@ -90,7 +91,7 @@ class ChanPivotClassifier:
         return duration_raw / avg
 
     @staticmethod
-    def compute_features(zs, historical_durations: list | None = None):
+    def compute_features(zs, historical_durations: Optional[list] = None):
         """计算单个中枢的全部结构特征（实时友好）"""
         duration_raw = ChanPivotClassifier.calc_duration(zs)
         contraction = ChanPivotClassifier.calc_contraction(zs)

ChanPivotMonitor.py

@@ -8,6 +8,7 @@ shift / contraction / duration。
 """
 
 from collections import deque
+from typing import Optional
 from ChanPivotClassifier import ChanPivotClassifier
 
 
@@ -22,17 +23,17 @@ class ChanPivotMonitor:
     def __init__(self, window_size: int = 10):
         self._window_size = window_size
         self._duration_history: deque[int] = deque(maxlen=window_size)
-        self._current_zs_id: tuple | None = None
+        self._current_zs_id: Optional[tuple] = None
         self._current_bi_count: int = 0
         self._current_is_sure: bool = False
-        self._current_state: dict | None = None
+        self._current_state: Optional[dict] = None
         self._duration_added_for_zs: set = set()  # 已加入窗口的中枢 ID
 
     # ------------------------------------------------------------------
     # Public API
     # ------------------------------------------------------------------
 
-    def update(self, bi_zs_list: list) -> dict | None:
+    def update(self, bi_zs_list: list) -> Optional[dict]:
         """
         主入口：检测当前中枢特征变化。
 
@@ -96,7 +97,7 @@ class ChanPivotMonitor:
 
         return self._current_state
 
-    def get_current(self) -> dict | None:
+    def get_current(self) -> Optional[dict]:
         """返回当前中枢的最新特征"""
         return self._current_state
 

bsp_monitor/engine.py

@@ -127,7 +127,7 @@ class ChanEngine:
         cst = dt.astimezone(timezone(timedelta(hours=8)))
         return cst.strftime("%Y-%m-%d %H:%M:%S CST")
 
-    def format_bsp_detail(self, bsp: ChanBSP) -> str:
+    def format_bsp_detail(self, bsp: ChanBSP, symbol: str = "BTC/USDT:USDT") -> str:
         bi = bsp.bi
         klc = bsp.klc
         bsp_type = bsp.type
@@ -136,8 +136,9 @@ class ChanEngine:
         emoji = "🟢" if bsp_dir == Chan_BSP_DIR.BUY else "🔴"
         dir_label = "买点" if bsp_dir == Chan_BSP_DIR.BUY else "卖点"
 
+        symbol_short = symbol.split(":")[0].replace("/", "")
         lines = [
-            f"{emoji} [{dir_label}] {self._bsp_type_name(bsp_type)} — <b>BTC/USDT 1m</b>",
+            f"{emoji} [{dir_label}] {self._bsp_type_name(bsp_type)} — <b>{symbol_short} 1m</b>",
             "",
             f"⏰ 确认: <code>{self._utc_to_cst(klc.end_time)}</code>",
             f"💰 价格: <b>{klc.close:.2f}</b>",

bsp_monitor/fetcher.py

@@ -2,23 +2,42 @@
 fetcher.py - 从 data_provider HTTP API 拉取 K 线数据。
 """
 
+from typing import List, Optional
+
 import requests
 import pandas as pd
 import logging
 
 logger = logging.getLogger(__name__)
 
-SYMBOL = "BTC/USDT:USDT"
 TIMEFRAME = "1m"
 PROVIDER_URL = "http://103.179.242.166"
 FETCH_LIMIT = 1000
 
+_symbols_cache: Optional[List[str]] = None
 
-def fetch_ohlcv() -> pd.DataFrame:
+
-    """从 data_provider API 拉取最近 FETCH_LIMIT 根 1m K 线。"""
+def get_symbols() -> list[str]:
+    """从 data_provider /health 获取所有可用币对（缓存）。"""
+    global _symbols_cache
+    if _symbols_cache is not None:
+        return _symbols_cache
+    try:
+        resp = requests.get(f"{PROVIDER_URL}/health", timeout=10)
+        resp.raise_for_status()
+        _symbols_cache = resp.json().get("symbols", [])
+        logger.info(f"获取到 {len(_symbols_cache)} 个币对")
+    except Exception as e:
+        logger.error(f"获取币对列表失败: {e}")
+        _symbols_cache = ["BTC/USDT:USDT"]
+    return _symbols_cache
+
+
+def fetch_ohlcv(symbol: str) -> pd.DataFrame:
+    """从 data_provider API 拉取某个币对最近 FETCH_LIMIT 根 1m K 线。"""
     url = f"{PROVIDER_URL}/api/candles"
     params = {
-        "symbol": SYMBOL,
+        "symbol": symbol,
         "tf": TIMEFRAME,
         "limit": FETCH_LIMIT,
     }
@@ -27,7 +46,7 @@ def fetch_ohlcv() -> pd.DataFrame:
     data = resp.json()
 
     if not data:
-        logger.warning("API 返回空数据")
+        logger.warning(f"{symbol}: API 返回空数据")
         return pd.DataFrame()
 
     df = pd.DataFrame(data)
@@ -35,5 +54,4 @@ def fetch_ohlcv() -> pd.DataFrame:
     df["date"] = df["timestamp"]
 
     df = df.drop_duplicates(subset="timestamp").sort_values("timestamp").reset_index(drop=True)
-    logger.info(f"拉取 {len(df)} 根 {TIMEFRAME} K 线 from {PROVIDER_URL}")
     return df

bsp_monitor/main.py

@@ -3,21 +3,22 @@
 main.py - 缠论买卖点监控主程序。
 
 每整分钟：
-  1. 从 Binance 拉取最新 1m K 线
+  1. 从 data_provider 拉取所有币对最新 1m K 线
-  2. 跑完整缠论管线
+  2. 每个币对独立跑缠论管线
-  3. 检测新出现的买卖点（BSP）
+  3. 检测新出现的买卖点（BSP）+ 中枢特征变化
-  4. 推送到 Telegram（同一 BSP 只推一次）
+  4. 推送到 Telegram
 """
 import asyncio
 import logging
 import sys
 import os
 import time
+from dataclasses import dataclass, field
 from datetime import datetime, timezone, timedelta
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 
-from fetcher import fetch_ohlcv
+from fetcher import fetch_ohlcv, get_symbols
 from engine import ChanEngine
 from notify import send_bsp_alert, send_telegram_message, BOT_TOKEN, CHAT_ID
 
@@ -33,149 +34,148 @@ logging.basicConfig(
 logger = logging.getLogger("bsp_monitor")
 
 
-def _bsp_stable_key(bsp) -> str:
+def _short(symbol: str) -> str:
-    """生成 BSP 的稳定唯一键（不依赖笔边界的微小变化）。"""
+    """BTC/USDT:USDT → BTCUSDT"""
-    return f"{bsp.type}_{bsp.klc.end_time}"
+    return symbol.split(":")[0].replace("/", "")
+
+
+@dataclass
+class SymbolState:
+    symbol: str
+    pivot_monitor: ChanPivotMonitor = field(default_factory=ChanPivotMonitor)
+    known_bsp_keys: set = field(default_factory=set)
+    last_df_ts: object = None
+    first_run: bool = True
 
 
 class BSPMonitor:
     def __init__(self):
-        self._first_run = True
+        symbols = get_symbols()
-        self._last_df_ts = None
+        self._states: dict[str, SymbolState] = {
-        self._known_bsp_keys: set = set()  # 已见过的 BSP 键（含已推送和历史的）
+            s: SymbolState(symbol=s) for s in symbols
-        self.pivot_monitor = ChanPivotMonitor(window_size=10)
+        }
+        logger.info(f"监控 {len(symbols)} 个币对: {', '.join(_short(s) for s in symbols)}")
 
     async def tick(self):
-        """单次 tick。"""
         tick_start = time.monotonic()
         logger.info("── tick 开始 ──")
 
+        for symbol, st in self._states.items():
+            await self._tick_symbol(symbol, st)
+
+        elapsed = (time.monotonic() - tick_start) * 1000
+        logger.info(f"── tick 结束 ({elapsed:.0f}ms) ──")
+
+    async def _tick_symbol(self, symbol: str, st: SymbolState):
+        name = _short(symbol)
+
         # 1. 拉取 K 线
         try:
-            df = fetch_ohlcv()
+            df = fetch_ohlcv(symbol)
         except Exception as e:
-            logger.error(f"拉取 K 线失败: {e}")
+            logger.error(f"[{name}] 拉取失败: {e}")
             return
 
         if df.empty:
-            logger.warning("DataFrame 为空，跳过")
+            logger.warning(f"[{name}] DataFrame 为空")
             return
 
         latest_ts = df.iloc[-1]["timestamp"]
-        if self._last_df_ts and latest_ts <= self._last_df_ts:
+        if st.last_df_ts and latest_ts <= st.last_df_ts:
-            logger.info("无新 K 线，跳过")
+            return  # 无新K线
-            return
+        st.last_df_ts = latest_ts
-        self._last_df_ts = latest_ts
 
         # 2. 运行缠论管线
         try:
             engine = ChanEngine(df)
         except Exception as e:
-            logger.error(f"缠论计算失败: {e}", exc_info=True)
+            logger.error(f"[{name}] 缠论计算失败: {e}", exc_info=True)
             return
 
-        # 2.5 更新中枢特征监控 + 推送
+        # 3. 中枢特征监控 + 推送
-        pivot_state = self.pivot_monitor.update(engine.bi_zs_list)
+        pivot_state = st.pivot_monitor.update(engine.bi_zs_list)
-        if pivot_state:
-            logger.info(
-                f"中枢特征更新: bi_count={pivot_state['bi_count']} "
-                f"is_sure={pivot_state['is_sure']} "
-                f"contraction={pivot_state['contraction']:.4f} "
-                f"shift_norm={pivot_state['shift_norm']:+.4f} "
-                f"duration_norm={pivot_state['duration_norm']:.4f}"
-            )
-            # Telegram 推送
-            if pivot_state["is_sure"]:
-                phase = "✅ 已确认"
-            elif pivot_state["bi_count"] > 3:
-                phase = "🔄 延伸中"
-            else:
-                phase = "🆕 刚形成"
-            zs_dir = pivot_state["zs_dir"]
-            dir_label = "⬆️ 向上" if "UP" in zs_dir else "⬇️ 向下"
 
-            # 白话解释
+        # 4. BSP 检测 + 推送
-            c = pivot_state["contraction"]
-            if c < 0.85:
-                contraction_note = "收敛（振幅缩小，可能快出方向）"
-            elif c > 1.15:
-                contraction_note = "扩张（振幅放大，波动加剧）"
-            else:
-                contraction_note = "稳定"
-
-            s = pivot_state["shift_norm"]
-            if s > 0.3:
-                shift_note = "重心上移（偏多）"
-            elif s < -0.3:
-                shift_note = "重心下移（偏空）"
-            else:
-                shift_note = "重心居中"
-
-            msg = (
-                f"🏠 <b>中枢更新</b> — BTC/USDT 1m\n"
-                f"\n"
-                f"📐 笔数: <b>{pivot_state['bi_count']}</b>  {dir_label}  {phase}\n"
-                f"📏 收敛率: <b>{pivot_state['contraction']:.4f}</b> → {contraction_note}\n"
-                f"⚖️ 重心漂移: <b>{pivot_state['shift_norm']:+.4f}</b> → {shift_note}\n"
-                f"⏱️ 持续: {pivot_state['duration_raw']}K  "
-                f"(norm: {pivot_state['duration_norm']:.2f})\n"
-                f"📦 区间: {pivot_state['zd']:.2f} – {pivot_state['zg']:.2f}  "
-                f"(gg/dd: {pivot_state['gg']:.2f}/{pivot_state['dd']:.2f})"
-            )
-            send_telegram_message(msg)
-
-        # 3. 检测新 BSP（用 stable key 去重）
         current_bsps = engine.bsp_list
-        current_keys = {_bsp_stable_key(b) for b in current_bsps}
+        current_keys = {_bsp_stable_key(b, symbol) for b in current_bsps}
 
-        if self._first_run:
+        if st.first_run:
-            self._first_run = False
+            st.first_run = False
-            self._known_bsp_keys = current_keys
+            st.known_bsp_keys = current_keys
             confirmed = [b for b in engine.bi_list if b.is_sure]
             logger.info(
-                f"首次运行完成 — {len(confirmed)} 笔已确认, "
+                f"[{name}] 首次完成 — {len(confirmed)} 笔, "
-                f"{len(current_bsps)} 个买卖点 (不推送历史)"
+                f"{len(current_bsps)} BSP (不推送)"
             )
-            elapsed = (time.monotonic() - tick_start) * 1000
-            logger.info(f"── tick 结束 ({elapsed:.0f}ms) ──")
             return
 
-        new_keys = current_keys - self._known_bsp_keys
+        if pivot_state:
-        self._known_bsp_keys |= current_keys  # 永增，BSP 一旦见过就不会再"新"
+            logger.info(
+                f"[{name}] 中枢更新: bi_count={pivot_state['bi_count']} "
+                f"contraction={pivot_state['contraction']:.4f} "
+                f"shift={pivot_state['shift_norm']:+.4f}"
+            )
+            self._push_pivot(pivot_state, name)
+
+        new_keys = current_keys - st.known_bsp_keys
+        st.known_bsp_keys |= current_keys
 
         if not new_keys:
-            logger.debug("无新买卖点")
-            elapsed = (time.monotonic() - tick_start) * 1000
-            logger.info(f"── tick 结束 ({elapsed:.0f}ms) ──")
             return
 
-        logger.info(f"检测到 {len(new_keys)} 个新买卖点")
+        logger.info(f"[{name}] {len(new_keys)} 个新 BSP")
-
-        # 4. 推送新 BSP
         for bsp in current_bsps:
-            key = _bsp_stable_key(bsp)
+            key = _bsp_stable_key(bsp, symbol)
             if key not in new_keys:
                 continue
-            msg = engine.format_bsp_detail(bsp)
+            msg = engine.format_bsp_detail(bsp, symbol)
-            # HTML 转义（Telegram parse_mode=HTML 要求）
+            msg = _escape_html(msg)
-            msg = msg.replace("&", "&amp;")
+            if send_bsp_alert(msg, bsp_key=key):
-            msg = msg.replace("<b>", "\x00B\x00").replace("</b>", "\x00/B\x00")
+                logger.info(f"[{name}] ✅ BSP: {key}")
-            msg = msg.replace("<code>", "\x00C\x00").replace("</code>", "\x00/C\x00")
-            msg = msg.replace("<", "&lt;").replace(">", "&gt;")
-            msg = msg.replace("\x00B\x00", "<b>").replace("\x00/B\x00", "</b>")
-            msg = msg.replace("\x00C\x00", "<code>").replace("\x00/C\x00", "</code>")
-            ok = send_bsp_alert(msg, bsp_key=key)
-            if ok:
-                logger.info(f"✅ 推送: {key}")
-            else:
-                logger.warning(f"❌ 推送失败: {key}")
 
-        elapsed = (time.monotonic() - tick_start) * 1000
+    def _push_pivot(self, state: dict, name: str):
-        logger.info(f"── tick 结束 ({elapsed:.0f}ms) ──")
+        if state["is_sure"]:
+            phase = "✅ 已确认"
+        elif state["bi_count"] > 3:
+            phase = "🔄 延伸中"
+        else:
+            phase = "🆕 刚形成"
+
+        zs_dir = state["zs_dir"]
+        dir_label = "⬆️ 向上" if "UP" in zs_dir else "⬇️ 向下"
+
+        c = state["contraction"]
+        if c < 0.85:
+            contraction_note = "收敛（振幅缩小，可能快出方向）"
+        elif c > 1.15:
+            contraction_note = "扩张（振幅放大，波动加剧）"
+        else:
+            contraction_note = "稳定"
+
+        s = state["shift_norm"]
+        if s > 0.3:
+            shift_note = "重心上移（偏多）"
+        elif s < -0.3:
+            shift_note = "重心下移（偏空）"
+        else:
+            shift_note = "重心居中"
+
+        msg = (
+            f"🏠 <b>中枢更新</b> — {name} 1m\n"
+            f"\n"
+            f"📐 笔数: <b>{state['bi_count']}</b>  {dir_label}  {phase}\n"
+            f"📏 收敛率: <b>{state['contraction']:.4f}</b> → {contraction_note}\n"
+            f"⚖️ 重心漂移: <b>{state['shift_norm']:+.4f}</b> → {shift_note}\n"
+            f"⏱️ 持续: {state['duration_raw']}K  "
+            f"(norm: {state['duration_norm']:.2f})\n"
+            f"📦 区间: {state['zd']:.2f} – {state['zg']:.2f}  "
+            f"(gg/dd: {state['gg']:.2f}/{state['dd']:.2f})"
+        )
+        send_telegram_message(msg)
 
     async def run(self):
         logger.info("=" * 50)
-        logger.info("bsp_monitor 启动 — BTC/USDT 1m 缠论买卖点监控")
+        logger.info(f"bsp_monitor 启动 — {len(self._states)} 币对 1m 缠论监控")
         logger.info(f"Telegram: {'已配置' if BOT_TOKEN and CHAT_ID else '⚠️ 未配置'}")
         logger.info("=" * 50)
 
@@ -197,6 +197,21 @@ class BSPMonitor:
                 await asyncio.sleep(5)
 
 
+def _bsp_stable_key(bsp, symbol: str) -> str:
+    return f"{symbol}_{bsp.type}_{bsp.klc.end_time}"
+
+
+def _escape_html(msg: str) -> str:
+    """HTML 转义，保留已有的 <b>/<code> 标签。"""
+    msg = msg.replace("&", "&amp;")
+    msg = msg.replace("<b>", "\x00B\x00").replace("</b>", "\x00/B\x00")
+    msg = msg.replace("<code>", "\x00C\x00").replace("</code>", "\x00/C\x00")
+    msg = msg.replace("<", "&lt;").replace(">", "&gt;")
+    msg = msg.replace("\x00B\x00", "<b>").replace("\x00/B\x00", "</b>")
+    msg = msg.replace("\x00C\x00", "<code>").replace("\x00/C\x00", "</code>")
+    return msg
+
+
 if __name__ == "__main__":
     monitor = BSPMonitor()
     try: