{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
deleted file mode 100644
index ac4dc90..0000000
--- a/bsp_monitor/fetcher.py
+++ /dev/null
@@ -1,62 +0,0 @@
-"""
-fetcher.py - 从 data_provider HTTP API 拉取 K 线数据。
-"""
-
-from typing import List, Optional
-
-import requests
-import pandas as pd
-import logging
-
-logger = logging.getLogger(__name__)
-
-PROVIDER_URL = "http://103.179.242.166"
-PROVIDER_URL = "http://127.0.0.1"
-FETCH_LIMIT = 1000
-
-_symbols_cache: Optional[List[str]] = None
-
-
-# 只推送 BTC，其他币对暂不监控
-_SYMBOL_WHITELIST = {"BTC/USDT:USDT", "ETH/USDT:USDT", "SOL/USDT:USDT"}
-
-
-def get_symbols() -> list[str]:
- """获取要监控的币对列表（目前只监控 BTC）。"""
- 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()
- all_symbols = resp.json().get("symbols", [])
- _symbols_cache = [s for s in all_symbols if s in _SYMBOL_WHITELIST]
- logger.info(f"获取到 {len(all_symbols)} 个币对，过滤后监控 {len(_symbols_cache)} 个: {_symbols_cache}")
- except Exception as e:
- logger.error(f"获取币对列表失败: {e}")
- _symbols_cache = ["BTC/USDT:USDT"]
- return _symbols_cache
-
-
-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": tf,
- "limit": FETCH_LIMIT,
- }
- resp = requests.get(url, params=params, timeout=30)
- resp.raise_for_status()
- data = resp.json()
-
- if not data:
- logger.warning(f"{symbol}: API 返回空数据")
- return pd.DataFrame()
-
- 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)
- return df
diff --git a/bsp_monitor/main.py b/bsp_monitor/main.py
deleted file mode 100644
index aa3a304..0000000
--- a/bsp_monitor/main.py
+++ /dev/null
@@ -1,220 +0,0 @@
-#!/usr/bin/env python3
-"""
-main.py - 缠论多周期买卖点监控。
-
-每整分钟：
- 1. 从 data_provider 拉取所有币对多周期 K 线
- 2. 每个币对 × 每个周期独立跑缠论管线
- 3. 检测新笔确认 → BSP 推送
-"""
-import asyncio
-import logging
-import sys
-import os
-import time
-from dataclasses import dataclass, field
-from datetime import datetime, timezone, timedelta
-from typing import Optional
-
-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, 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 ChanEnum import Chan_BI_DIR
-# from ChanPivotMonitor import ChanPivotMonitor # 暂停中枢监控
-
-logging.basicConfig(
- level=logging.INFO,
- format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-logger = logging.getLogger("bsp_monitor")
-
-TIMEFRAMES = ["1m", "5m", "15m", "1h"]
-
-
-def _short(symbol: str) -> str:
- """BTC/USDT:USDT → BTCUSDT"""
- return symbol.split(":")[0].replace("/", "")
-
-
-def _bi_id(bi) -> Optional[tuple]:
- """笔的稳定标识，基于首K线时间戳。"""
- if bi.start_klc is None:
- return None
- return (bi.start_klc.start_time,)
-
-
-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}_{tf}"
- msg = engine.format_bsp_detail(bsp, symbol, tf)
- msg = _escape_html(msg)
- if send_bsp_alert(msg, bsp_key=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
- tfs: dict = field(default_factory=dict)
-
- def __post_init__(self):
- self.tfs = {tf: TfState() for tf in TIMEFRAMES}
-
-
-class BSPMonitor:
- def __init__(self):
- symbols = get_symbols()
- self._states: dict[str, SymbolState] = {
- s: SymbolState(symbol=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()
- 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)
-
- 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, tf)
- except Exception as e:
- logger.error(f"[{name} {tf}] 拉取失败: {e}")
- return
-
- if df.empty:
- return
-
- # 2. 检查是否有新 K 线
- latest_ts = df.iloc[-1]["timestamp"]
- if ts.last_df_ts and latest_ts <= ts.last_df_ts:
- return
- ts.last_df_ts = latest_ts
-
- # 3. 运行缠论管线
- try:
- engine = ChanEngine(df)
- except Exception as e:
- logger.error(f"[{name} {tf}] 缠论计算失败: {e}", exc_info=True)
- return
-
- # 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
-
- last_confirmed = confirmed[-1]
- current_bi_id = _bi_id(last_confirmed)
- if current_bi_id is None:
- return
-
- 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, tf)
-
- logger.info(
- f"[{name} {tf}] 首次完成 — "
- f"{len(confirmed)} 笔, {len(engine.bsp_list)} BSP"
- )
- return
-
- if current_bi_id == ts.last_bi_id:
- return
-
- ts.last_bi_id = current_bi_id
-
- bi_dir = "⬆️" if last_confirmed.dir == Chan_BI_DIR.UP else "⬇️"
- logger.info(f"[{name} {tf}] 新笔确认 — #{len(confirmed)} "
- f"{bi_dir} 高度: ${last_confirmed.height:.2f}")
-
- bsp = engine.get_bsp_for_bi(last_confirmed)
- if bsp:
- _push_bsp(engine, bsp, symbol, tf)
-
- async def run(self):
- logger.info("=" * 50)
- 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)
-
- 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)
-
-
-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:
- asyncio.run(monitor.run())
- except KeyboardInterrupt:
- logger.info("收到中断信号，退出")
diff --git a/bsp_monitor/notify.py b/bsp_monitor/notify.py
deleted file mode 100644
index ee114b3..0000000
--- a/bsp_monitor/notify.py
+++ /dev/null
@@ -1,61 +0,0 @@
-"""
-notify.py - Telegram 推送。
-"""
-import logging
-import requests
-
-logger = logging.getLogger(__name__)
-
-BOT_TOKEN = "8742822093:AAGzD1vS7ru7ROhgcOjA-UyHb4R8Cfcqv3Q"
-CHAT_ID = "580807463"
-
-
-def send_telegram_message(text: str) -> bool:
- """发送 Telegram 消息（不去重，每次调用都发）。"""
- 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()
- return True
- except Exception as e:
- logger.error(f"Telegram 推送失败: {e}")
- return False
-
-
-def send_bsp_alert(text: str, bsp_key: str = "") -> bool:
- """推送 BSP 消息。"""
- 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 推送成功: {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
deleted file mode 100755
index eb1588b..0000000
--- a/bsp_monitor/run.sh
+++ /dev/null
@@ -1,11 +0,0 @@
-#!/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
diff --git a/bsp_monitor/twitter_web.py b/bsp_monitor/twitter_web.py
deleted file mode 100644
index cbdec57..0000000
--- a/bsp_monitor/twitter_web.py
+++ /dev/null
@@ -1,380 +0,0 @@
-#!/usr/bin/env python3
-"""
-twitter_web.py — Twitter 监控账号管理 Web 界面。
-单文件，零依赖，只用到 Python 标准库。
-"""
-
-import json
-import os
-import sys
-import re
-from datetime import datetime, timezone
-from http.server import HTTPServer, BaseHTTPRequestHandler
-from urllib.parse import urlparse, parse_qs
-
-SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
-WATCHLIST_PATH = os.path.join(SCRIPT_DIR, "twitter_watchlist.json")
-STATE_PATH = os.path.join(SCRIPT_DIR, "twitter_state.json")
-
-PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8010
-
-
-def extract_username(value: str) -> str:
- value = value.strip().rstrip("/")
- if value.startswith("@"):
- return value[1:]
- for pattern in [r"(?:twitter\.com|x\.com)/(\w+)(?:/|$)", r"/(\w+)$"]:
- m = re.search(pattern, value)
- if m:
- return m.group(1)
- if re.match(r"^\w+$", value):
- return value
- raise ValueError(f"无法提取用户名: {value}")
-
-
-def load_json(path):
- if os.path.exists(path):
- with open(path) as f:
- return json.load(f)
- return {}
-
-
-def save_json(path, data):
- with open(path, "w") as f:
- json.dump(data, f, indent=2, ensure_ascii=False)
-
-
-def get_watchlist():
- return load_json(WATCHLIST_PATH).get("users", [])
-
-
-def save_watchlist(users):
- save_json(WATCHLIST_PATH, {"users": users})
-
-
-def get_state():
- return load_json(STATE_PATH)
-
-
-HTML = """
-
-
-
-
-
-
-
-Twitter 监控管理
-
-
-
-🐦 Twitter 账号监控
-管理 twitterapi.io 监控账号 · 增删改查
-
-
-
-
-
-
-
-
-
-"""
-
-
-class Handler(BaseHTTPRequestHandler):
- def log_message(self, format, *args):
- pass # silent
-
- def _send(self, code, body, content_type="application/json"):
- body = body.encode() if isinstance(body, str) else json.dumps(body, ensure_ascii=False).encode()
- self.send_response(code)
- self.send_header("Content-Type", content_type + "; charset=utf-8")
- self.send_header("Content-Length", str(len(body)))
- self.send_header("Access-Control-Allow-Origin", "*")
- self.end_headers()
- self.wfile.write(body)
-
- def _json(self, code, data):
- self._send(code, data)
-
- def _error(self, code, msg):
- self._json(code, {"error": msg})
-
- def do_OPTIONS(self):
- self.send_response(204)
- self.send_header("Access-Control-Allow-Origin", "*")
- self.send_header("Access-Control-Allow-Methods", "GET,POST,PUT,DELETE,OPTIONS")
- self.send_header("Access-Control-Allow-Headers", "Content-Type")
- self.end_headers()
-
- def do_GET(self):
- path = urlparse(self.path).path
- if path == "/" or path == "/index.html":
- self._send(200, HTML, "text/html")
- return
- if path.startswith("/api/accounts"):
- username = path[len("/api/accounts"):].strip("/")
- if username:
- # GET /api/accounts/ — single account
- users = get_watchlist()
- state = get_state()
- for u in users:
- if u["username"].lower() == username.lower():
- entry = dict(u)
- entry["last_check"] = state.get(u["username"], {}).get("last_check")
- self._json(200, entry)
- return
- self._error(404, "账号不存在")
- return
- # GET /api/accounts — list all
- users = get_watchlist()
- state = get_state()
- accounts = []
- for u in users:
- entry = dict(u)
- sc = state.get(u["username"], {})
- ts = sc.get("last_check")
- if ts:
- try:
- ts = datetime.fromisoformat(ts).strftime("%m-%d %H:%M")
- except Exception:
- pass
- else:
- ts = "从未"
- entry["last_check"] = ts
- accounts.append(entry)
- self._json(200, {"accounts": accounts})
- else:
- self._error(404, "Not Found")
-
- def do_POST(self):
- path = urlparse(self.path).path
- if path != "/api/accounts":
- self._error(404, "Not Found")
- return
- length = int(self.headers.get("Content-Length", 0))
- body = json.loads(self.rfile.read(length)) if length else {}
- url = body.get("url", "").strip()
- if not url:
- self._error(400, "缺少 url 参数")
- return
- try:
- username = extract_username(url)
- except ValueError:
- self._error(400, "无法从输入中提取用户名，请输入 Twitter/X 链接或 @用户名")
- return
-
- users = get_watchlist()
- if any(u["username"].lower() == username.lower() for u in users):
- self._error(409, f"@{username} 已在监控列表中")
- return
-
- display_name = body.get("display_name", "").strip() or username
- users.append({
- "username": username,
- "display_name": display_name,
- "added_at": datetime.now(timezone.utc).isoformat(),
- })
- save_watchlist(users)
- self._json(201, {"message": f"✅ 已添加 @{username}", "username": username})
-
- def do_PUT(self):
- path = urlparse(self.path).path
- username = path[len("/api/accounts"):].strip("/")
- if not username:
- self._error(400, "缺少用户名")
- return
- length = int(self.headers.get("Content-Length", 0))
- body = json.loads(self.rfile.read(length)) if length else {}
- display_name = body.get("display_name", "").strip()
-
- users = get_watchlist()
- for u in users:
- if u["username"].lower() == username.lower():
- if display_name:
- u["display_name"] = display_name
- save_watchlist(users)
- self._json(200, {"message": f"✅ @{username} 已更新"})
- return
- self._error(404, "账号不存在")
-
- def do_DELETE(self):
- path = urlparse(self.path).path
- username = path[len("/api/accounts"):].strip("/")
- if not username:
- self._error(400, "缺少用户名")
- return
- users = get_watchlist()
- before = len(users)
- users = [u for u in users if u["username"].lower() != username.lower()]
- if len(users) < before:
- save_watchlist(users)
- self._json(200, {"message": f"🗑 已移除 @{username}"})
- else:
- self._error(404, "账号不存在")
-
-
-def main():
- print(f"🐦 Twitter 监控管理: http://0.0.0.0:{PORT}")
- server = HTTPServer(("0.0.0.0", PORT), Handler)
- try:
- server.serve_forever()
- except KeyboardInterrupt:
- print("\n已停止")
- server.server_close()
-
-
-if __name__ == "__main__":
- main()