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)