diff --git a/ChanPivotClassifier.py b/ChanPivotClassifier.py index 8718d4c..52b1af0 100644 --- a/ChanPivotClassifier.py +++ b/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) diff --git a/ChanPivotMonitor.py b/ChanPivotMonitor.py index 69b8e70..d6e29f7 100644 --- a/ChanPivotMonitor.py +++ b/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 diff --git a/bsp_monitor/engine.py b/bsp_monitor/engine.py index be01e6a..f3ed73d 100644 --- a/bsp_monitor/engine.py +++ b/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)} — BTC/USDT 1m", + f"{emoji} [{dir_label}] {self._bsp_type_name(bsp_type)} — {symbol_short} 1m", "", f"⏰ 确认: {self._utc_to_cst(klc.end_time)}", f"💰 价格: {klc.close:.2f}", diff --git a/bsp_monitor/fetcher.py b/bsp_monitor/fetcher.py index 76ea0e9..9f3bd2f 100644 --- a/bsp_monitor/fetcher.py +++ 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 diff --git a/bsp_monitor/main.py b/bsp_monitor/main.py index 35e4dbc..a3ae592 100644 --- a/bsp_monitor/main.py +++ b/bsp_monitor/main.py @@ -3,21 +3,22 @@ main.py - 缠论买卖点监控主程序。 每整分钟： - 1. 从 Binance 拉取最新 1m K 线 - 2. 跑完整缠论管线 - 3. 检测新出现的买卖点（BSP） - 4. 推送到 Telegram（同一 BSP 只推一次） + 1. 从 data_provider 拉取所有币对最新 1m K 线 + 2. 每个币对独立跑缠论管线 + 3. 检测新出现的买卖点（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: - """生成 BSP 的稳定唯一键（不依赖笔边界的微小变化）。""" - return f"{bsp.type}_{bsp.klc.end_time}" +def _short(symbol: str) -> str: + """BTC/USDT:USDT → BTCUSDT""" + 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 - self._last_df_ts = None - self._known_bsp_keys: set = set() # 已见过的 BSP 键（含已推送和历史的） - self.pivot_monitor = ChanPivotMonitor(window_size=10) + symbols = get_symbols() + self._states: dict[str, SymbolState] = { + s: SymbolState(symbol=s) for s in symbols + } + 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: - logger.info("无新 K 线，跳过") - return - self._last_df_ts = latest_ts + if st.last_df_ts and latest_ts <= st.last_df_ts: + return # 无新K线 + st.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 更新中枢特征监控 + 推送 - 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 "⬇️ 向下" + # 3. 中枢特征监控 + 推送 + pivot_state = st.pivot_monitor.update(engine.bi_zs_list) - # 白话解释 - 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"🏠 中枢更新 — BTC/USDT 1m\n" - f"\n" - f"📐 笔数: {pivot_state['bi_count']} {dir_label} {phase}\n" - f"📏 收敛率: {pivot_state['contraction']:.4f} → {contraction_note}\n" - f"⚖️ 重心漂移: {pivot_state['shift_norm']:+.4f} → {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 去重） + # 4. BSP 检测 + 推送 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: - self._first_run = False - self._known_bsp_keys = current_keys + if st.first_run: + st.first_run = False + st.known_bsp_keys = current_keys confirmed = [b for b in engine.bi_list if b.is_sure] logger.info( - f"首次运行完成 — {len(confirmed)} 笔已确认, " - f"{len(current_bsps)} 个买卖点 (不推送历史)" + f"[{name}] 首次完成 — {len(confirmed)} 笔, " + 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 - self._known_bsp_keys |= current_keys # 永增，BSP 一旦见过就不会再"新" + if pivot_state: + 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)} 个新买卖点") - - # 4. 推送新 BSP + logger.info(f"[{name}] {len(new_keys)} 个新 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) - # HTML 转义（Telegram parse_mode=HTML 要求） - msg = msg.replace("&", "&") - msg = msg.replace("", "\x00B\x00").replace("", "\x00/B\x00") - msg = msg.replace("", "\x00C\x00").replace("", "\x00/C\x00") - msg = msg.replace("<", "<").replace(">", ">") - msg = msg.replace("\x00B\x00", "").replace("\x00/B\x00", "") - msg = msg.replace("\x00C\x00", "").replace("\x00/C\x00", "") - ok = send_bsp_alert(msg, bsp_key=key) - if ok: - logger.info(f"✅ 推送: {key}") - else: - logger.warning(f"❌ 推送失败: {key}") + msg = engine.format_bsp_detail(bsp, symbol) + msg = _escape_html(msg) + if send_bsp_alert(msg, bsp_key=key): + logger.info(f"[{name}] ✅ BSP: {key}") - elapsed = (time.monotonic() - tick_start) * 1000 - logger.info(f"── tick 结束 ({elapsed:.0f}ms) ──") + def _push_pivot(self, state: dict, name: str): + 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"🏠 中枢更新 — {name} 1m\n" + f"\n" + f"📐 笔数: {state['bi_count']} {dir_label} {phase}\n" + f"📏 收敛率: {state['contraction']:.4f} → {contraction_note}\n" + f"⚖️ 重心漂移: {state['shift_norm']:+.4f} → {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 转义，保留已有的 / 标签。""" + msg = msg.replace("&", "&") + msg = msg.replace("", "\x00B\x00").replace("", "\x00/B\x00") + msg = msg.replace("", "\x00C\x00").replace("", "\x00/C\x00") + msg = msg.replace("<", "<").replace(">", ">") + msg = msg.replace("\x00B\x00", "").replace("\x00/B\x00", "") + msg = msg.replace("\x00C\x00", "").replace("\x00/C\x00", "") + return msg + + if __name__ == "__main__": monitor = BSPMonitor() try: