ChanPivotMonitor: 实时中枢特征跟踪 + Telegram推送

- ChanPivotClassifier: 提取 calc_duration/contraction/shift 为 @staticmethod，新增 compute_features()
- ChanPivotMonitor: 实时追踪当前中枢，bi_count 增长时重新计算 shift/contraction/duration
- bsp_monitor/fetcher: 改用 data_provider HTTP API 替代直连 CCXT
- bsp_monitor/notify: 新增 send_telegram_message() 通用推送
- bsp_monitor/main: 集成 ChanPivotMonitor，有新笔或 BSP 时推送到 Telegram

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

ChanPivotClassifier.py

@@ -31,14 +31,16 @@ class ChanPivotClassifier:
     # Feature extraction
     # ------------------------------------------------------------------
 
-    def _calc_duration(self, zs) -> int:
+    @staticmethod
+    def calc_duration(zs) -> int:
         """持续时间: 第一笔首K → 最后一笔末K 的 index 差"""
         bi_list = zs.bi_list
         start_idx = bi_list[0].start_klc.index
         end_idx = bi_list[-1].end_klc.index
         return end_idx - start_idx
 
-    def _calc_contraction(self, zs) -> float:
+    @staticmethod
+    def calc_contraction(zs) -> float:
         """收敛率: 后窗口振幅均值 / 前窗口振幅均值"""
         bi_list = zs.bi_list
         if len(bi_list) < 4:
@@ -55,7 +57,8 @@ class ChanPivotClassifier:
             return 1.0
         return last_mean / first_mean
 
-    def _calc_shift(self, zs) -> tuple[float, float]:
+    @staticmethod
+    def calc_shift(zs) -> tuple[float, float]:
         """重心漂移: 前后半段重心均值差 (原始值, 归一化值)"""
         bi_list = zs.bi_list
         mid = len(bi_list) // 2
@@ -76,6 +79,39 @@ class ChanPivotClassifier:
 
         return shift_raw, shift_norm
 
+    @staticmethod
+    def compute_duration_norm(duration_raw: int, historical_durations: list) -> float:
+        """用历史窗口均值归一化 duration"""
+        if not historical_durations:
+            return 1.0
+        avg = sum(historical_durations) / len(historical_durations)
+        if avg == 0:
+            return 1.0
+        return duration_raw / avg
+
+    @staticmethod
+    def compute_features(zs, historical_durations: list | None = None):
+        """计算单个中枢的全部结构特征（实时友好）"""
+        duration_raw = ChanPivotClassifier.calc_duration(zs)
+        contraction = ChanPivotClassifier.calc_contraction(zs)
+        shift_raw, shift_norm = ChanPivotClassifier.calc_shift(zs)
+
+        if historical_durations is not None and len(historical_durations) > 0:
+            duration_norm = ChanPivotClassifier.compute_duration_norm(
+                duration_raw, historical_durations
+            )
+        else:
+            duration_norm = 1.0
+
+        return {
+            "duration_raw": duration_raw,
+            "duration_norm": round(duration_norm, 4),
+            "contraction": round(contraction, 4),
+            "shift_raw": round(shift_raw, 6),
+            "shift_norm": round(shift_norm, 4),
+            "zs_height": round(zs.zg - zs.zd, 6),
+        }
+
     # ------------------------------------------------------------------
     # Label computation
     # ------------------------------------------------------------------
@@ -189,9 +225,9 @@ class ChanPivotClassifier:
             if not zs.is_sure or len(zs.bi_list) < 3:
                 continue
 
-            duration_raw = self._calc_duration(zs)
+            duration_raw = ChanPivotClassifier.calc_duration(zs)
-            contraction = self._calc_contraction(zs)
+            contraction = ChanPivotClassifier.calc_contraction(zs)
-            shift_raw, shift_norm = self._calc_shift(zs)
+            shift_raw, shift_norm = ChanPivotClassifier.calc_shift(zs)
 
             raw.append({
                 "zs": zs,
@@ -203,17 +239,14 @@ class ChanPivotClassifier:
                 "zs_height": zs.zg - zs.zd,
             })
 
-        # 归一化 duration: 除以均值
-        if raw:
-            avg_duration = sum(r["duration_raw"] for r in raw) / len(raw)
-        else:
-            avg_duration = 1
-
         # 第二遍：组装输出 + 计算 label
         result = []
         for r in raw:
             zs = r["zs"]
-            duration_norm = r["duration_raw"] / avg_duration if avg_duration > 0 else 1.0
+            historical = [x["duration_raw"] for x in raw]
+            duration_norm = ChanPivotClassifier.compute_duration_norm(
+                r["duration_raw"], historical
+            )
             label_info = self._compute_label(zs, r["contraction"], r["shift_norm"])
 
             # 时间处理

ChanPivotMonitor.py

@@ -0,0 +1,144 @@
+"""
+实时中枢特征跟踪器
+Real-time Pivot Feature Tracker
+
+定位: 观察者 — 不修改管线，只观察 bi_zs_list 中当前中枢的特征变化。
+每次管线重算后调用 update()，检测 bi_count 是否增长，若增长则重新计算
+shift / contraction / duration。
+"""
+
+from collections import deque
+from ChanPivotClassifier import ChanPivotClassifier
+
+
+class ChanPivotMonitor:
+    """
+    实时追踪当前中枢的结构特征。
+
+    update() 每次管线重算后调用，对比 bi_count 判断是否有新笔加入中枢。
+    若 bi_count 增长则重新计算 3 个结构特征并返回最新值。
+    """
+
+    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_bi_count: int = 0
+        self._current_is_sure: bool = False
+        self._current_state: dict | None = None
+        self._duration_added_for_zs: set = set()  # 已加入窗口的中枢 ID
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def update(self, bi_zs_list: list) -> dict | None:
+        """
+        主入口：检测当前中枢特征变化。
+
+        参数:
+            bi_zs_list: 当前管线产出的笔中枢列表
+
+        返回:
+            特征 dict（有变化时），无变化返回 None
+        """
+        if not bi_zs_list:
+            self._current_zs_id = None
+            self._current_bi_count = 0
+            self._current_is_sure = False
+            self._current_state = None
+            return None
+
+        zs = self._find_current_zs(bi_zs_list)
+        if zs is None:
+            return None
+
+        zs_id = self._make_zs_id(zs)
+        bi_count = len(zs.bi_list)
+        is_sure = zs.is_sure
+
+        # 无变化 → 跳过
+        if (zs_id == self._current_zs_id
+                and bi_count == self._current_bi_count
+                and is_sure == self._current_is_sure):
+            return None
+
+        # 中枢切换 → 将旧中枢 duration 加入窗口
+        if zs_id != self._current_zs_id:
+            self._maybe_add_to_history()
+
+        self._current_zs_id = zs_id
+        self._current_bi_count = bi_count
+        self._current_is_sure = is_sure
+
+        features = ChanPivotClassifier.compute_features(
+            zs, list(self._duration_history)
+        )
+
+        self._current_state = {
+            "zs_id": zs_id,
+            "zs_index": zs.index,
+            "zs_dir": str(zs.dir),
+            "bi_count": bi_count,
+            "is_sure": zs.is_sure,
+            "zg": round(zs.zg, 6),
+            "zd": round(zs.zd, 6),
+            "gg": round(zs.gg, 6),
+            "dd": round(zs.dd, 6),
+            **features,
+            "start_time": str(zs.start_time) if hasattr(zs, "start_time") and zs.start_time else None,
+        }
+
+        # 中枢刚变为已确认时，将其 duration 加入滚动窗口
+        if is_sure and zs_id not in self._duration_added_for_zs:
+            self._add_duration(features["duration_raw"])
+            self._duration_added_for_zs.add(zs_id)
+
+        return self._current_state
+
+    def get_current(self) -> dict | None:
+        """返回当前中枢的最新特征"""
+        return self._current_state
+
+    def get_duration_history(self) -> list[int]:
+        """返回用于归一化的 duration 滚动窗口"""
+        return list(self._duration_history)
+
+    # ------------------------------------------------------------------
+    # Internal
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _make_zs_id(zs) -> tuple:
+        """生成中枢的稳定标识（基于首笔首K线索引，不依赖 zs.index）"""
+        bi0 = zs.bi_list[0]
+        return (bi0.start_klc.index,)
+
+    @staticmethod
+    def _find_current_zs(bi_zs_list: list):
+        """
+        找到当前活跃中枢：
+        优先取最后一个 is_sure=False（形成中）的中枢，
+        没有则取最后一个 is_sure=True 的中枢。
+        """
+        forming = None
+        last_sure = None
+        for zs in bi_zs_list:
+            if len(zs.bi_list) < 3:
+                continue
+            if not zs.is_sure:
+                forming = zs
+            else:
+                last_sure = zs
+        return forming if forming is not None else last_sure
+
+    def _add_duration(self, duration_raw: int):
+        """将已确认中枢的 duration 加入滚动窗口"""
+        self._duration_history.append(duration_raw)
+
+    def _maybe_add_to_history(self):
+        """旧中枢切换前，若已确认且未记录过，则将其 duration 加入窗口"""
+        if (self._current_state and self._current_state["is_sure"]
+                and self._current_zs_id not in self._duration_added_for_zs):
+            self._add_duration(self._current_state["duration_raw"])
+            self._duration_added_for_zs.add(self._current_zs_id)

bsp_monitor/fetcher.py

@@ -1,45 +1,39 @@
 """
-fetcher.py - CCXT REST 拉取 Binance 永续合约 1m K 线，从固定起点累积。
+fetcher.py - 从 data_provider HTTP API 拉取 K 线数据。
 """
-import ccxt
+
+import requests
 import pandas as pd
 import logging
-from datetime import datetime, timezone
 
 logger = logging.getLogger(__name__)
 
 SYMBOL = "BTC/USDT:USDT"
 TIMEFRAME = "1m"
-# 每次拉取最近 LIMIT 根 K 线（Binance 上限 1500，足够缠论管线用 ~25h 数据）
+PROVIDER_URL = "http://103.179.242.166"
-_FETCH_LIMIT = 1000
+FETCH_LIMIT = 1000
-
-_exchange = None
-
-
-def _get_exchange():
-    global _exchange
-    if _exchange is None:
-        _exchange = ccxt.binance({
-            "enableRateLimit": True,
-            "options": {"defaultType": "future"},
-        })
-        _exchange.load_markets()
-        logger.info("ccxt binance 已初始化")
-    return _exchange
 
 
 def fetch_ohlcv() -> pd.DataFrame:
-    """拉取最近 _FETCH_LIMIT 根 1m K 线。
+    """从 data_provider API 拉取最近 FETCH_LIMIT 根 1m K 线。"""
+    url = f"{PROVIDER_URL}/api/candles"
+    params = {
+        "symbol": SYMBOL,
+        "tf": TIMEFRAME,
+        "limit": FETCH_LIMIT,
+    }
+    resp = requests.get(url, params=params, timeout=30)
+    resp.raise_for_status()
+    data = resp.json()
 
-    管线每次重跑最新的 K 线窗口。
+    if not data:
-    用 limit 而非 since 避免 API 500 根限制截断新数据。
+        logger.warning("API 返回空数据")
-    """
+        return pd.DataFrame()
-    exchange = _get_exchange()
-    raw = exchange.fetch_ohlcv(SYMBOL, TIMEFRAME, limit=_FETCH_LIMIT)
 
-    df = pd.DataFrame(raw, columns=["timestamp", "open", "high", "low", "close", "volume"])
+    df = pd.DataFrame(data)
     df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
     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

@@ -19,7 +19,12 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 
 from fetcher import fetch_ohlcv
 from engine import ChanEngine
-from notify import send_bsp_alert, BOT_TOKEN, CHAT_ID
+from notify import send_bsp_alert, send_telegram_message, BOT_TOKEN, CHAT_ID
+
+_PARENT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if _PARENT not in sys.path:
+    sys.path.insert(0, _PARENT)
+from ChanPivotMonitor import ChanPivotMonitor
 
 logging.basicConfig(
     level=logging.INFO,
@@ -38,6 +43,7 @@ class BSPMonitor:
         self._first_run = True
         self._last_df_ts = None
         self._known_bsp_keys: set = set()  # 已见过的 BSP 键（含已推送和历史的）
+        self.pivot_monitor = ChanPivotMonitor(window_size=10)
 
     async def tick(self):
         """单次 tick。"""
@@ -68,6 +74,56 @@ class BSPMonitor:
             logger.error(f"缠论计算失败: {e}", exc_info=True)
             return
 
+        # 2.5 更新中枢特征监控 + 推送
+        pivot_state = self.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 "⬇️ 向下"
+
+            # 白话解释
+            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}

bsp_monitor/notify.py

@@ -54,6 +54,39 @@ def _save_pushed():
 _load_pushed()
 
 
+def send_telegram_message(text: str) -> bool:
+    """发送 Telegram 消息（不去重，每次调用都发）。
+
+    Args:
+        text: HTML 格式的消息文本
+
+    Returns:
+        True 如果发送成功
+    """
+    if not BOT_TOKEN or not CHAT_ID:
+        logger.warning("Telegram 未配置，跳过推送")
+        return False
+
+    url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
+    try:
+        resp = requests.post(
+            url,
+            json={
+                "chat_id": CHAT_ID,
+                "text": text,
+                "parse_mode": "HTML",
+                "disable_web_page_preview": True,
+            },
+            timeout=10,
+        )
+        resp.raise_for_status()
+        logger.info(f"Telegram 推送成功")
+        return True
+    except Exception as e:
+        logger.error(f"Telegram 推送失败: {e}")
+        return False
+
+
 def send_bsp_alert(text: str, bsp_key: str = "") -> bool:
     """通过 Telegram Bot API 推送买卖点消息（自动去重）。