From 8eb50e3eae1a97ea5475c7174b78d12594d3ee36 Mon Sep 17 00:00:00 2001 From: jackyu66git Date: Tue, 26 May 2026 14:42:53 +0800 Subject: [PATCH] =?UTF-8?q?bsp=5Fmonitor:=20=E5=A4=9A=E5=91=A8=E6=9C=9F=20?= =?UTF-8?q?BSP=20=E6=8E=A8=E9=80=81=20(1m/5m/15m/1h)=EF=BC=8C=E4=B8=AD?= =?UTF-8?q?=E6=9E=A2=E7=9B=91=E6=8E=A7=E4=BB=A3=E7=A0=81=E4=BF=9D=E7=95=99?= =?UTF-8?q?=E4=BD=86=E6=9A=82=E5=81=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 --- bsp_monitor/fetcher.py | 7 +- bsp_monitor/main.py | 145 +++++++++++++++++------------------------ 2 files changed, 61 insertions(+), 91 deletions(-) diff --git a/bsp_monitor/fetcher.py b/bsp_monitor/fetcher.py index 9f3bd2f..a1d066f 100644 --- a/bsp_monitor/fetcher.py +++ b/bsp_monitor/fetcher.py @@ -10,7 +10,6 @@ import logging logger = logging.getLogger(__name__) -TIMEFRAME = "1m" PROVIDER_URL = "http://103.179.242.166" FETCH_LIMIT = 1000 @@ -33,12 +32,12 @@ def get_symbols() -> list[str]: return _symbols_cache -def fetch_ohlcv(symbol: str) -> pd.DataFrame: - """从 data_provider API 拉取某个币对最近 FETCH_LIMIT 根 1m K 线。""" +def fetch_ohlcv(symbol: str, tf: str = "1m") -> pd.DataFrame: + """从 data_provider API 拉取某个币对最近 FETCH_LIMIT 根 K 线。""" url = f"{PROVIDER_URL}/api/candles" params = { "symbol": symbol, - "tf": TIMEFRAME, + "tf": tf, "limit": FETCH_LIMIT, } resp = requests.get(url, params=params, timeout=30) diff --git a/bsp_monitor/main.py b/bsp_monitor/main.py index c2fa77f..6857b30 100644 --- a/bsp_monitor/main.py +++ b/bsp_monitor/main.py @@ -1,12 +1,11 @@ #!/usr/bin/env python3 """ -main.py - 缠论买卖点监控主程序。 +main.py - 缠论多周期买卖点监控。 每整分钟： - 1. 从 data_provider 拉取所有币对最新 1m K 线 - 2. 每个币对独立跑缠论管线 - 3. 检测新笔 → 检查上一笔终点是否 BSP → 推送 - 4. 检测中枢特征变化 → 推送 + 1. 从 data_provider 拉取所有币对多周期 K 线 + 2. 每个币对 × 每个周期独立跑缠论管线 + 3. 检测新笔确认 → BSP 推送 """ import asyncio import logging @@ -21,13 +20,13 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from fetcher import fetch_ohlcv, get_symbols from engine import ChanEngine -from notify import send_bsp_alert, send_telegram_message, BOT_TOKEN, CHAT_ID +from notify import send_bsp_alert, 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 from ChanEnum import Chan_BI_DIR +# from ChanPivotMonitor import ChanPivotMonitor # 暂停中枢监控 logging.basicConfig( level=logging.INFO, @@ -35,6 +34,8 @@ logging.basicConfig( ) logger = logging.getLogger("bsp_monitor") +TIMEFRAMES = ["1m", "5m", "15m", "1h"] + def _short(symbol: str) -> str: """BTC/USDT:USDT → BTCUSDT""" @@ -48,26 +49,39 @@ def _bi_id(bi) -> Optional[tuple]: return (bi.start_klc.start_time,) -def _push_bsp(engine: ChanEngine, bsp, symbol: str, name: str) -> bool: +def _push_bsp(engine: ChanEngine, bsp, symbol: str, tf: str) -> bool: """推送 BSP 到 Telegram（带去重）。""" if bsp.klc is None: return False - key = f"{symbol}_{bsp.type}_{bsp.klc.end_time}" + key = f"{symbol}_{bsp.type}_{bsp.klc.end_time}_{tf}" 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}") + logger.info(f"[{_short(symbol)} {tf}] ✅ BSP: {key}") return True return False +@dataclass +class TfState: + """单个周期的状态。""" + last_bi_id: Optional[tuple] = None + last_df_ts: object = None + first_run: bool = True + # pivot_monitor: ChanPivotMonitor = None # 暂停中枢监控 + + # def __post_init__(self): + # if self.pivot_monitor is None: + # self.pivot_monitor = ChanPivotMonitor() + + @dataclass class SymbolState: symbol: str - pivot_monitor: ChanPivotMonitor = field(default_factory=ChanPivotMonitor) - last_bi_id: Optional[tuple] = None # 上次检查过的最后一笔确认 ID - last_df_ts: object = None # 最新已处理的 K 线时间戳 - first_run: bool = True + tfs: dict = field(default_factory=dict) + + def __post_init__(self): + self.tfs = {tf: TfState() for tf in TIMEFRAMES} class BSPMonitor: @@ -76,7 +90,8 @@ class BSPMonitor: 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)}") + logger.info(f"监控 {len(symbols)}×{len(TIMEFRAMES)} 币对×周期: " + f"{', '.join(_short(s) for s in symbols)}") async def tick(self): tick_start = time.monotonic() @@ -91,31 +106,40 @@ class BSPMonitor: async def _tick_symbol(self, symbol: str, st: SymbolState): name = _short(symbol) + for tf in TIMEFRAMES: + await self._check_tf(symbol, tf, st.tfs[tf], name) + + async def _check_tf(self, symbol: str, tf: str, ts: TfState, name: str): # 1. 拉取 K 线 try: - df = fetch_ohlcv(symbol) + df = fetch_ohlcv(symbol, tf) except Exception as e: - logger.error(f"[{name}] 拉取失败: {e}") + logger.error(f"[{name} {tf}] 拉取失败: {e}") return if df.empty: - logger.warning(f"[{name}] DataFrame 为空") return # 2. 检查是否有新 K 线 latest_ts = df.iloc[-1]["timestamp"] - if st.last_df_ts and latest_ts <= st.last_df_ts: + if ts.last_df_ts and latest_ts <= ts.last_df_ts: return - st.last_df_ts = latest_ts + ts.last_df_ts = latest_ts # 3. 运行缠论管线 try: engine = ChanEngine(df) except Exception as e: - logger.error(f"[{name}] 缠论计算失败: {e}", exc_info=True) + logger.error(f"[{name} {tf}] 缠论计算失败: {e}", exc_info=True) return - # 4. 获取已确认的笔 + # 4. 中枢特征更新（暂停） + # try: + # ts.pivot_monitor.update(engine.bi_zs_list) + # except Exception as e: + # logger.debug(f"[{name} {tf}] 中枢特征更新失败: {e}") + + # 5. BSP 检测 confirmed = [b for b in engine.bi_list if b.is_sure] if len(confirmed) < 2: return @@ -125,90 +149,37 @@ class BSPMonitor: if current_bi_id is None: return - # 5. 首轮：记录状态，检查最后一笔的 BSP（可能刚被分型确认） - if st.first_run: - st.first_run = False - st.last_bi_id = current_bi_id - st.pivot_monitor.update(engine.bi_zs_list) + if ts.first_run: + ts.first_run = False + ts.last_bi_id = current_bi_id bsp = engine.get_bsp_for_bi(last_confirmed) if bsp: - _push_bsp(engine, bsp, symbol, name) + _push_bsp(engine, bsp, symbol, tf) logger.info( - f"[{name}] 首次完成 — {len(confirmed)} 笔, " - f"{len(engine.bsp_list)} BSP" + f"[{name} {tf}] 首次完成 — " + f"{len(confirmed)} 笔, {len(engine.bsp_list)} BSP" ) return - # 6. 检测新笔确认（反向二类分型触发 is_sure=True） - if current_bi_id == st.last_bi_id: + if current_bi_id == ts.last_bi_id: return - st.last_bi_id = current_bi_id + ts.last_bi_id = current_bi_id bi_dir = "⬆️" if last_confirmed.dir == Chan_BI_DIR.UP else "⬇️" - logger.info(f"[{name}] 新笔确认 — #{len(confirmed)} " + logger.info(f"[{name} {tf}] 新笔确认 — #{len(confirmed)} " f"{bi_dir} 高度: ${last_confirmed.height:.2f}") - # 7. 检查刚被分型确认这笔的终点是否为 BSP bsp = engine.get_bsp_for_bi(last_confirmed) if bsp: - _push_bsp(engine, bsp, symbol, name) - - # 8. 中枢特征更新 - pivot_state = st.pivot_monitor.update(engine.bi_zs_list) - 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) - - 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) + _push_bsp(engine, bsp, symbol, tf) async def run(self): logger.info("=" * 50) - logger.info(f"bsp_monitor 启动 — {len(self._states)} 币对 1m 缠论监控") + logger.info(f"bsp_monitor 启动 — {len(self._states)} 币对 " + f"× {len(TIMEFRAMES)} 周期 ({', '.join(TIMEFRAMES)})") logger.info(f"Telegram: {'已配置' if BOT_TOKEN and CHAT_ID else '⚠️ 未配置'}") logger.info("=" * 50)