From f0ea6a6065b0a6ac17d34320b7add551a12ee363 Mon Sep 17 00:00:00 2001 From: jackyu66git Date: Mon, 18 May 2026 08:42:43 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=20bsp=5Fmonitor:=20BTC/USDT?= =?UTF-8?q?=201m=20=E7=BC=A0=E8=AE=BA=E4=B9=B0=E5=8D=96=E7=82=B9=E5=AE=9E?= =?UTF-8?q?=E6=97=B6=E7=9B=91=E6=8E=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 每整分钟拉取 Binance 永续合约 1m K 线 - 运行完整缠论管线检测买卖点 (BSP) - 新 BSP 推送到 Telegram - fix: fetcher 用 limit=1000 替代固定 since，避免 API 500 根限制截断新数据 --- bsp_monitor/.env | 3 + bsp_monitor/__init__.py | 1 + bsp_monitor/engine.py | 181 ++++++++++++++++++++++++++++++++++++++++ bsp_monitor/fetcher.py | 45 ++++++++++ bsp_monitor/main.py | 149 +++++++++++++++++++++++++++++++++ bsp_monitor/notify.py | 96 +++++++++++++++++++++ bsp_monitor/run.sh | 11 +++ 7 files changed, 486 insertions(+) create mode 100644 bsp_monitor/.env create mode 100644 bsp_monitor/__init__.py create mode 100644 bsp_monitor/engine.py create mode 100644 bsp_monitor/fetcher.py create mode 100644 bsp_monitor/main.py create mode 100644 bsp_monitor/notify.py create mode 100755 bsp_monitor/run.sh diff --git a/bsp_monitor/.env b/bsp_monitor/.env new file mode 100644 index 0000000..df69084 --- /dev/null +++ b/bsp_monitor/.env @@ -0,0 +1,3 @@ +# bsp_monitor 复用 Hermes Agent 的 Telegram bot +# notify.py 从 ~/.hermes/.env 直接读取 TELEGRAM_BOT_TOKEN +# 此处无需重复配置 diff --git a/bsp_monitor/__init__.py b/bsp_monitor/__init__.py new file mode 100644 index 0000000..e9d7b81 --- /dev/null +++ b/bsp_monitor/__init__.py @@ -0,0 +1 @@ +# bsp_monitor - 缠论买卖点监控 (BTC/USDT 1m) diff --git a/bsp_monitor/engine.py b/bsp_monitor/engine.py new file mode 100644 index 0000000..aae1f74 --- /dev/null +++ b/bsp_monitor/engine.py @@ -0,0 +1,181 @@ +""" +engine.py - 缠论管线封装：DataFrame → KLC → BI → ZS → BSP。 + +复用 ~/Project/Chan/ 下的 TF_DF 模块。 +注意：TF_DF.__init__ 有 bug（get_zs_list 不存在），这里手动调用各步骤。 +""" +import sys +import os +from typing import List, Optional + +_PARENT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if _PARENT not in sys.path: + sys.path.insert(0, _PARENT) + +import pandas as pd +import talib.abstract as ta +from ChanEnum import ( + Chan_BSP_DIR, Chan_BSP_TYPE, Chan_KLC_FX, Chan_BI_DIR, + Chan_FX_TYPE, Chan_KLINE_DIR, Chan_SEG_DIR, Chan_ZS_DIR, +) +from ChanBSP import ChanBSP +from ChanBI import ChanBI +from ChanZS import ChanZS + +# 仅导入类，不触发 TF_DF.__init__ +from TF_DF import TF_DF as _TF_DF_Class + + +class ChanEngine: + """手动执行缠论管线，绕过 TF_DF.__init__ 的 bug。""" + + def __init__(self, df: pd.DataFrame): + if df.empty or len(df) < 50: + raise ValueError("DataFrame 至少需要 50 根 K 线") + + if "date" not in df.columns and "timestamp" in df.columns: + df["date"] = df["timestamp"] + + self.df = df + self._tf = _TF_DF_Class.__new__(_TF_DF_Class) # 不调用 __init__ + + # Step 0: 添加 TA 指标 + self._df_with_indicators = self._tf.add_indicators(df.copy()) + + # Step 1: KLU (K-line unit) + self.klu_list = self._tf.cal_kl_data(self._df_with_indicators) + + # Step 1.5: MACD state + from ChanMACD import ChanMACD + chanmacd = ChanMACD(self.klu_list) + self.klu_list = chanmacd.cal_macd_state() + + # Step 2: KLC (combined K-line) + self.klc_list = self._tf.get_klc_list(self.klu_list) + + # Step 3: BI (stroke) + self.bi_list = self._tf.cal_bi_list(self.klc_list) + + # Step 4: SEG (segment) + self.seg_list = self._tf.get_seg_list(self.bi_list) + + # Step 5: ZS (bi-level center) — 供 find_all_bsp 使用 + self.bi_zs_list: List = self._tf.cal_bi_zs_list(self.bi_list) + + # Step 6: BSP (buy/sell points) + self.bsp_list: List[ChanBSP] = self._tf.find_all_bsp( + self.bi_list, self.bi_zs_list + ) + + def get_second_last_bi(self) -> Optional[ChanBI]: + """获取倒数第二笔（最新确认的笔）。""" + confirmed = [b for b in self.bi_list if b.is_sure] + if len(confirmed) >= 2: + return confirmed[-2] + elif len(confirmed) == 1: + return confirmed[-1] + return None + + def get_bsp_for_bi(self, bi: ChanBI) -> Optional[ChanBSP]: + """检查某个 Bi 的 end_klc 是否是买卖点。""" + if bi is None or not bi.is_sure: + return None + klc = bi.end_klc + if klc is None: + return None + if klc.bsp and klc.bsp_type != Chan_BSP_TYPE.NONE: + for bsp in self.bsp_list: + if bsp.klc is klc: + return bsp + return None + + # ── 格式化 ── + + @staticmethod + def _bsp_type_name(t: Chan_BSP_TYPE) -> str: + import ChanEnum + names = { + Chan_BSP_TYPE.B1: "一类买点(B1)", + Chan_BSP_TYPE.B2: "二类买点(B2)", + Chan_BSP_TYPE.B3: "三类买点(B3)", + Chan_BSP_TYPE.S1: "一类卖点(S1)", + Chan_BSP_TYPE.S2: "二类卖点(S2)", + Chan_BSP_TYPE.S3: "三类卖点(S3)", + } + return names.get(t, str(t)) + + @staticmethod + def _bi_dir_name(d) -> str: + return "⬆️ 向上" if d == Chan_BI_DIR.UP else "⬇️ 向下" + + @staticmethod + def _fx_strength_name(klc_fx_type) -> str: + import ChanEnum + names = { + Chan_KLC_FX.TOP0: "TOP0(弱)", Chan_KLC_FX.TOP1: "TOP1(标准)", + Chan_KLC_FX.TOP2: "TOP2(强)", Chan_KLC_FX.TOP3: "TOP3(二类)", + Chan_KLC_FX.TOP4: "TOP4(BB上轨)", Chan_KLC_FX.TOP5: "TOP5", + Chan_KLC_FX.TOP6: "TOP6(高位空)", Chan_KLC_FX.TOP7: "TOP7(背驰)", + Chan_KLC_FX.TOP8: "TOP8(信号线)", + Chan_KLC_FX.BOTTOM0: "BOTTOM0(弱)", Chan_KLC_FX.BOTTOM1: "BOTTOM1(标准)", + Chan_KLC_FX.BOTTOM2: "BOTTOM2(强)", Chan_KLC_FX.BOTTOM3: "BOTTOM3(二类)", + Chan_KLC_FX.BOTTOM4: "BOTTOM4(BB下轨)", Chan_KLC_FX.BOTTOM5: "BOTTOM5(零轴下)", + Chan_KLC_FX.BOTTOM6: "BOTTOM6(高位空)", Chan_KLC_FX.BOTTOM7: "BOTTOM7(背驰)", + Chan_KLC_FX.BOTTOM8: "BOTTOM8(信号线)", + } + return names.get(klc_fx_type, f"UNKNOWN({klc_fx_type})") + + @staticmethod + def _utc_to_cst(time_str: str) -> str: + """UTC 时间字符串 → 东八区 (UTC+8)。""" + from datetime import datetime, timedelta, timezone + dt = datetime.fromisoformat(str(time_str)) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + 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: + bi = bsp.bi + klc = bsp.klc + bsp_type = bsp.type + bsp_dir = bsp.dir + + emoji = "🟢" if bsp_dir == Chan_BSP_DIR.BUY else "🔴" + dir_label = "买点" if bsp_dir == Chan_BSP_DIR.BUY else "卖点" + + lines = [ + f"{emoji} [{dir_label}] {self._bsp_type_name(bsp_type)} — BTC/USDT 1m", + "", + f"⏰ 确认: {self._utc_to_cst(klc.end_time)}", + f"💰 价格: {klc.close:.2f}", + f"📐 笔方向: {self._bi_dir_name(bi.dir)}", + f"📏 笔高度: ${bi.height:.2f} 宽度: {bi.width}K 斜率: {bi.slop:.2f}", + f"🔩 分型强度: {self._fx_strength_name(klc.klc_fx_type)}", + ] + + if bsp.zs: + zs = bsp.zs + zs_dir = "UP" if hasattr(zs, 'dir') and hasattr(Chan_ZS_DIR, 'UP') and zs.dir == Chan_ZS_DIR.UP else "DOWN" + lines.append(f"🏠 中枢: {zs.zd:.2f} – {zs.zg:.2f} ({zs_dir}, #{getattr(zs, 'index', 0) + 1})") + + if bsp.type in (Chan_BSP_TYPE.B1, Chan_BSP_TYPE.S1): + lines.append("📊 MACD背驰: 有 (离开段能量 < 进入段)") + + if hasattr(klc, 'ema_status') and klc.ema_status: + ema52 = klc.ema_status.get('ema52', {}) + if ema52: + pos = str(ema52.get('pos', '?')) + lines.append(f"📈 EMA52: {pos} (值: {klc.ema52:.2f})") + + lines.append(f"📋 KLC状态: {klc.klc_state}") + + if bi.pre: + prev = bi.pre + lines.extend([ + "────", + f"⬅️ 前一笔: {self._bi_dir_name(prev.dir)} " + f"高度: ${prev.height:.2f} 宽度: {prev.width}K", + ]) + + return "\n".join(lines) diff --git a/bsp_monitor/fetcher.py b/bsp_monitor/fetcher.py new file mode 100644 index 0000000..054e571 --- /dev/null +++ b/bsp_monitor/fetcher.py @@ -0,0 +1,45 @@ +""" +fetcher.py - CCXT REST 拉取 Binance 永续合约 1m K 线，从固定起点累积。 +""" +import ccxt +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 数据） +_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 线。 + + 管线每次重跑最新的 K 线窗口。 + 用 limit 而非 since 避免 API 500 根限制截断新数据。 + """ + exchange = _get_exchange() + raw = exchange.fetch_ohlcv(SYMBOL, TIMEFRAME, limit=_FETCH_LIMIT) + + df = pd.DataFrame(raw, columns=["timestamp", "open", "high", "low", "close", "volume"]) + 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) + return df diff --git a/bsp_monitor/main.py b/bsp_monitor/main.py new file mode 100644 index 0000000..2d1c950 --- /dev/null +++ b/bsp_monitor/main.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python3 +""" +main.py - 缠论买卖点监控主程序。 + +每整分钟： + 1. 从 Binance 拉取最新 1m K 线 + 2. 跑完整缠论管线 + 3. 检测新出现的买卖点（BSP） + 4. 推送到 Telegram（同一 BSP 只推一次） +""" +import asyncio +import logging +import sys +import os +import time +from datetime import datetime, timezone, timedelta + +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 + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", +) +logger = logging.getLogger("bsp_monitor") + + +def _bsp_stable_key(bsp) -> str: + """生成 BSP 的稳定唯一键（不依赖笔边界的微小变化）。""" + return f"{bsp.type}_{bsp.klc.end_time}" + + +class BSPMonitor: + def __init__(self): + self._first_run = True + self._last_df_ts = None + self._known_bsp_keys: set = set() # 已见过的 BSP 键（含已推送和历史的） + + async def tick(self): + """单次 tick。""" + tick_start = time.monotonic() + logger.info("── tick 开始 ──") + + # 1. 拉取 K 线 + try: + df = fetch_ohlcv() + except Exception as e: + logger.error(f"拉取 K 线失败: {e}") + return + + if df.empty: + logger.warning("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 + + # 2. 运行缠论管线 + try: + engine = ChanEngine(df) + except Exception as e: + logger.error(f"缠论计算失败: {e}", exc_info=True) + return + + # 3. 检测新 BSP（用 stable key 去重） + current_bsps = engine.bsp_list + current_keys = {_bsp_stable_key(b) for b in current_bsps} + + if self._first_run: + self._first_run = False + self._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)} 个买卖点 (不推送历史)" + ) + 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 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 + for bsp in current_bsps: + key = _bsp_stable_key(bsp) + 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}") + + elapsed = (time.monotonic() - tick_start) * 1000 + logger.info(f"── tick 结束 ({elapsed:.0f}ms) ──") + + async def run(self): + logger.info("=" * 50) + logger.info("bsp_monitor 启动 — BTC/USDT 1m 缠论买卖点监控") + logger.info(f"Telegram: {'已配置' if BOT_TOKEN and CHAT_ID else '⚠️ 未配置'}") + logger.info("=" * 50) + + logger.info("首次运行（初始化，不推送）...") + await self.tick() + + while True: + now = datetime.now(timezone.utc) + next_minute = now.replace(second=0, microsecond=0) + timedelta(minutes=1) + wait_seconds = max(0.1, (next_minute - now).total_seconds()) + + logger.info(f"等待 {wait_seconds:.0f}s 到 {next_minute.strftime('%H:%M:%S')}UTC") + await asyncio.sleep(wait_seconds) + + try: + await self.tick() + except Exception as e: + logger.error(f"tick 异常: {e}", exc_info=True) + await asyncio.sleep(5) + + +if __name__ == "__main__": + monitor = BSPMonitor() + try: + asyncio.run(monitor.run()) + except KeyboardInterrupt: + logger.info("收到中断信号，退出") diff --git a/bsp_monitor/notify.py b/bsp_monitor/notify.py new file mode 100644 index 0000000..d7a75e7 --- /dev/null +++ b/bsp_monitor/notify.py @@ -0,0 +1,96 @@ +""" +notify.py - Telegram 推送，复用 Hermes Agent 的 bot token。 +直接从 ~/.hermes/.env 读取 token（独立进程不受 Hermes 遮罩影响）。 +支持持久化去重：已推送的 bsp_key 保存到文件，重启不重复推送。 +""" +import os +import json +import logging +import requests + +logger = logging.getLogger(__name__) + +_HERMES_ENV = os.path.expanduser("~/.hermes/.env") +BOT_TOKEN = "" +CHAT_ID = "" + +if os.path.exists(_HERMES_ENV): + with open(_HERMES_ENV) as f: + for line in f: + line = line.strip() + if line.startswith("TELEGRAM_BOT_TOKEN="): + BOT_TOKEN = line.split("=", 1)[1] + elif line.startswith("TELEGRAM_ALLOWED_USERS="): + CHAT_ID = line.split("=", 1)[1] + +# 持久化去重文件 +_PUSHED_FILE = os.path.expanduser("~/.hermes/bsp_pushed.json") +_pushed_bsp_keys: set = set() + + +def _load_pushed(): + """从文件加载已推送的 bsp_key 集合。""" + global _pushed_bsp_keys + if os.path.exists(_PUSHED_FILE): + try: + with open(_PUSHED_FILE) as f: + data = json.load(f) + _pushed_bsp_keys = set(data.get("keys", [])) + logger.info(f"加载已推送记录: {len(_pushed_bsp_keys)} 条") + except Exception: + _pushed_bsp_keys = set() + + +def _save_pushed(): + """持久化已推送的 bsp_key。""" + try: + with open(_PUSHED_FILE, "w") as f: + json.dump({"keys": list(_pushed_bsp_keys)}, f) + except Exception as e: + logger.error(f"保存去重记录失败: {e}") + + +# 启动时加载 +_load_pushed() + + +def send_bsp_alert(text: str, bsp_key: str = "") -> bool: + """通过 Telegram Bot API 推送买卖点消息（自动去重）。 + + Args: + text: HTML 格式的消息文本 + bsp_key: 唯一标识，用于持久化去重 + + Returns: + True 如果发送成功 + """ + if not BOT_TOKEN or not CHAT_ID: + logger.warning("Telegram 未配置，跳过推送") + return False + + if bsp_key: + if bsp_key in _pushed_bsp_keys: + logger.info(f"已推送过，跳过: {bsp_key}") + 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() + if bsp_key: + _pushed_bsp_keys.add(bsp_key) + _save_pushed() # 立即持久化 + logger.info(f"Telegram 推送成功: {bsp_key or 'no-key'}") + return True + except Exception as e: + logger.error(f"Telegram 推送失败: {e}") + return False diff --git a/bsp_monitor/run.sh b/bsp_monitor/run.sh new file mode 100755 index 0000000..eb1588b --- /dev/null +++ b/bsp_monitor/run.sh @@ -0,0 +1,11 @@ +#!/bin/bash +# bsp_monitor 启动脚本 +# 用法: bash run.sh + +cd "$(dirname "$0")" +echo "=== bsp_monitor ===" +echo "启动时间: $(date -u '+%Y-%m-%d %H:%M:%S UTC')" +echo "监控: BTC/USDT:USDT 1m 缠论买卖点" +echo "推送: Telegram (复用 Hermes bot)" +echo "===================" +exec /usr/bin/python3 -u main.py