diff --git a/ChanPivotClassifier.py b/ChanPivotClassifier.py
index 4b319d5..8718d4c 100644
--- a/ChanPivotClassifier.py
+++ b/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)
- contraction = self._calc_contraction(zs)
- shift_raw, shift_norm = self._calc_shift(zs)
+ duration_raw = ChanPivotClassifier.calc_duration(zs)
+ contraction = ChanPivotClassifier.calc_contraction(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"])
# 时间处理
diff --git a/ChanPivotMonitor.py b/ChanPivotMonitor.py
new file mode 100644
index 0000000..69b8e70
--- /dev/null
+++ b/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)
diff --git a/bsp_monitor/fetcher.py b/bsp_monitor/fetcher.py
index 054e571..76ea0e9 100644
--- a/bsp_monitor/fetcher.py
+++ b/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 数据）
-_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
+PROVIDER_URL = "http://103.179.242.166"
+FETCH_LIMIT = 1000
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 线窗口。
- 用 limit 而非 since 避免 API 500 根限制截断新数据。
- """
- exchange = _get_exchange()
- raw = exchange.fetch_ohlcv(SYMBOL, TIMEFRAME, limit=_FETCH_LIMIT)
+ if not data:
+ logger.warning("API 返回空数据")
+ return pd.DataFrame()
- 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
diff --git a/bsp_monitor/main.py b/bsp_monitor/main.py
index 2d1c950..35e4dbc 100644
--- a/bsp_monitor/main.py
+++ b/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"🏠 中枢更新 — 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 去重）
current_bsps = engine.bsp_list
current_keys = {_bsp_stable_key(b) for b in current_bsps}
diff --git a/bsp_monitor/notify.py b/bsp_monitor/notify.py
index d7a75e7..ae3d85e 100644
--- a/bsp_monitor/notify.py
+++ b/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 推送买卖点消息（自动去重）。