update
This commit is contained in:
aaron
parent
5107b0ed09
commit
17995c0a0b
@ -1,217 +1,129 @@
|
||||
"""
|
||||
价格监控服务 - 使用 Binance WebSocket 实时监控价格
|
||||
价格监控服务 - 使用轮询方式获取实时价格（更稳定）
|
||||
"""
|
||||
import threading
|
||||
import time
|
||||
import sys
|
||||
import os
|
||||
import logging
|
||||
import requests
|
||||
from typing import Dict, List, Callable, Optional, Set
|
||||
from app.utils.logger import logger
|
||||
from app.config import get_settings
|
||||
|
||||
# 抑制 binance 库的 WebSocket 错误日志（正确的 logger 名称）
|
||||
logging.getLogger('binance.ws.threaded_stream').setLevel(logging.CRITICAL)
|
||||
logging.getLogger('binance.ws.reconnecting_websocket').setLevel(logging.CRITICAL)
|
||||
logging.getLogger('binance.ws').setLevel(logging.WARNING) # 只显示警告及以上
|
||||
|
||||
|
||||
class SuppressOutput:
|
||||
"""临时抑制 stdout/stderr 输出"""
|
||||
def __init__(self, suppress_stderr=True, suppress_stdout=False):
|
||||
self.suppress_stderr = suppress_stderr
|
||||
self.suppress_stdout = suppress_stdout
|
||||
self._stderr = None
|
||||
self._stdout = None
|
||||
self._devnull = None
|
||||
|
||||
def __enter__(self):
|
||||
self._devnull = open(os.devnull, 'w')
|
||||
if self.suppress_stderr:
|
||||
self._stderr = sys.stderr
|
||||
sys.stderr = self._devnull
|
||||
if self.suppress_stdout:
|
||||
self._stdout = sys.stdout
|
||||
sys.stdout = self._devnull
|
||||
return self
|
||||
|
||||
def __exit__(self, *args):
|
||||
if self._stderr:
|
||||
sys.stderr = self._stderr
|
||||
if self._stdout:
|
||||
sys.stdout = self._stdout
|
||||
if self._devnull:
|
||||
self._devnull.close()
|
||||
|
||||
|
||||
class PriceMonitorService:
|
||||
"""实时价格监控服务"""
|
||||
"""实时价格监控服务（轮询模式）"""
|
||||
|
||||
# Binance API
|
||||
BASE_URL = "https://api.binance.com"
|
||||
|
||||
def __init__(self):
|
||||
"""初始化价格监控服务"""
|
||||
self.settings = get_settings()
|
||||
self.twm = None
|
||||
self.running = False
|
||||
self.subscribed_symbols: Dict[str, str] = {} # symbol -> stream_name
|
||||
self.subscribed_symbols: Set[str] = set()
|
||||
self.price_callbacks: List[Callable[[str, float], None]] = []
|
||||
self.latest_prices: Dict[str, float] = {}
|
||||
self._lock = threading.Lock()
|
||||
self._pending_symbols: List[str] = [] # 待订阅的交易对
|
||||
self._reconnecting = False # 是否正在重连
|
||||
self._desired_symbols: Set[str] = set() # 期望订阅的交易对（用于重连）
|
||||
self._stop_requested = False # 是否请求停止（区分主动停止和意外断开）
|
||||
self._last_message_time: Dict[str, float] = {} # 上次收到消息的时间
|
||||
self._health_check_thread = None
|
||||
self._poll_thread: Optional[threading.Thread] = None
|
||||
self._poll_interval = 3 # 轮询间隔（秒）
|
||||
self._session = requests.Session()
|
||||
|
||||
logger.info("价格监控服务初始化完成")
|
||||
logger.info("价格监控服务初始化完成（轮询模式）")
|
||||
|
||||
def start(self):
|
||||
"""启动 WebSocket 管理器（在独立线程中）"""
|
||||
"""启动价格轮询"""
|
||||
if self.running:
|
||||
logger.warning("价格监控服务已在运行")
|
||||
return
|
||||
|
||||
self._stop_requested = False
|
||||
self.running = True
|
||||
|
||||
def _start_in_thread():
|
||||
try:
|
||||
# 延迟导入，避免在模块加载时就创建事件循环
|
||||
from binance import ThreadedWebsocketManager
|
||||
def _poll_loop():
|
||||
logger.info(f"价格轮询已启动，间隔 {self._poll_interval} 秒")
|
||||
while self.running:
|
||||
try:
|
||||
self._fetch_prices()
|
||||
except Exception as e:
|
||||
logger.error(f"获取价格失败: {e}")
|
||||
|
||||
self.twm = ThreadedWebsocketManager(
|
||||
api_key=self.settings.binance_api_key or "",
|
||||
api_secret=self.settings.binance_api_secret or ""
|
||||
)
|
||||
self.twm.start()
|
||||
self.running = True
|
||||
self._reconnecting = False
|
||||
logger.info("WebSocket 管理器已启动")
|
||||
|
||||
# 等待 WebSocket 完全启动
|
||||
time.sleep(1)
|
||||
|
||||
# 订阅待处理的交易对
|
||||
for symbol in self._pending_symbols:
|
||||
self._do_subscribe(symbol)
|
||||
self._pending_symbols.clear()
|
||||
|
||||
# 重连时恢复之前的订阅
|
||||
for symbol in self._desired_symbols:
|
||||
if symbol not in self.subscribed_symbols:
|
||||
self._do_subscribe(symbol)
|
||||
|
||||
# 启动健康检查
|
||||
self._start_health_check()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"启动 WebSocket 管理器失败: {e}")
|
||||
import traceback
|
||||
logger.error(traceback.format_exc())
|
||||
# 启动失败，尝试重连
|
||||
if not self._stop_requested:
|
||||
self._schedule_reconnect()
|
||||
|
||||
# 在独立线程中启动
|
||||
thread = threading.Thread(target=_start_in_thread, daemon=True)
|
||||
thread.start()
|
||||
|
||||
def _start_health_check(self):
|
||||
"""启动健康检查线程"""
|
||||
def _check_health():
|
||||
while self.running and not self._stop_requested:
|
||||
time.sleep(30) # 每30秒检查一次
|
||||
|
||||
if not self.running or self._stop_requested:
|
||||
break
|
||||
|
||||
# 检查是否有超过60秒没收到消息的交易对
|
||||
now = time.time()
|
||||
for symbol in list(self._desired_symbols):
|
||||
last_time = self._last_message_time.get(symbol, now)
|
||||
if now - last_time > 60:
|
||||
logger.warning(f"{symbol} 超过60秒未收到数据，触发重连")
|
||||
self._schedule_reconnect()
|
||||
# 等待下一次轮询
|
||||
for _ in range(self._poll_interval * 10):
|
||||
if not self.running:
|
||||
break
|
||||
time.sleep(0.1)
|
||||
|
||||
self._health_check_thread = threading.Thread(target=_check_health, daemon=True)
|
||||
self._health_check_thread.start()
|
||||
self._poll_thread = threading.Thread(target=_poll_loop, daemon=True)
|
||||
self._poll_thread.start()
|
||||
|
||||
def stop(self):
|
||||
"""停止 WebSocket 管理器"""
|
||||
# 标记为主动停止
|
||||
self._stop_requested = True
|
||||
|
||||
"""停止价格轮询"""
|
||||
if not self.running:
|
||||
return
|
||||
|
||||
# 先标记为停止，防止回调继续处理
|
||||
self.running = False
|
||||
logger.info("价格监控服务已停止")
|
||||
|
||||
def _fetch_prices(self):
|
||||
"""获取所有订阅交易对的价格"""
|
||||
if not self.subscribed_symbols:
|
||||
return
|
||||
|
||||
symbols = list(self.subscribed_symbols)
|
||||
|
||||
# 如果只有少量交易对，逐个获取
|
||||
if len(symbols) <= 3:
|
||||
for symbol in symbols:
|
||||
self._fetch_single_price(symbol)
|
||||
else:
|
||||
# 批量获取所有价格
|
||||
self._fetch_all_prices(symbols)
|
||||
|
||||
def _fetch_single_price(self, symbol: str):
|
||||
"""获取单个交易对价格"""
|
||||
try:
|
||||
# 抑制关闭时的错误输出（binance 库用 print 输出错误）
|
||||
with SuppressOutput(suppress_stderr=True, suppress_stdout=True):
|
||||
# 先停止所有 socket 订阅
|
||||
if self.twm:
|
||||
for _, stream_name in list(self.subscribed_symbols.items()):
|
||||
try:
|
||||
self.twm.stop_socket(stream_name)
|
||||
except:
|
||||
pass
|
||||
|
||||
# 等待一小段时间让 socket 关闭
|
||||
time.sleep(0.5)
|
||||
|
||||
# 然后停止管理器
|
||||
try:
|
||||
self.twm.stop()
|
||||
except:
|
||||
pass
|
||||
|
||||
self.subscribed_symbols.clear()
|
||||
self._desired_symbols.clear()
|
||||
self._last_message_time.clear()
|
||||
logger.info("WebSocket 管理器已停止")
|
||||
url = f"{self.BASE_URL}/api/v3/ticker/price"
|
||||
response = self._session.get(url, params={'symbol': symbol}, timeout=5)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
price = float(data['price'])
|
||||
self._update_price(symbol, price)
|
||||
except Exception as e:
|
||||
# 忽略关闭时的错误
|
||||
pass
|
||||
logger.debug(f"获取 {symbol} 价格失败: {e}")
|
||||
|
||||
def _schedule_reconnect(self, delay: int = 5):
|
||||
"""安排重连"""
|
||||
if self._stop_requested or self._reconnecting:
|
||||
return
|
||||
def _fetch_all_prices(self, symbols: List[str]):
|
||||
"""批量获取价格"""
|
||||
try:
|
||||
url = f"{self.BASE_URL}/api/v3/ticker/price"
|
||||
response = self._session.get(url, timeout=10)
|
||||
response.raise_for_status()
|
||||
all_prices = response.json()
|
||||
|
||||
self._reconnecting = True
|
||||
logger.warning(f"WebSocket 连接断开，{delay} 秒后尝试重连...")
|
||||
# 过滤出订阅的交易对
|
||||
symbol_set = set(symbols)
|
||||
for item in all_prices:
|
||||
symbol = item['symbol']
|
||||
if symbol in symbol_set:
|
||||
price = float(item['price'])
|
||||
self._update_price(symbol, price)
|
||||
except Exception as e:
|
||||
logger.debug(f"批量获取价格失败: {e}")
|
||||
|
||||
def _reconnect():
|
||||
time.sleep(delay)
|
||||
if not self._stop_requested:
|
||||
self._do_reconnect()
|
||||
def _update_price(self, symbol: str, price: float):
|
||||
"""更新价格并触发回调"""
|
||||
old_price = self.latest_prices.get(symbol)
|
||||
|
||||
thread = threading.Thread(target=_reconnect, daemon=True)
|
||||
thread.start()
|
||||
# 只有价格变化时才触发回调
|
||||
if old_price != price:
|
||||
self.latest_prices[symbol] = price
|
||||
|
||||
def _do_reconnect(self):
|
||||
"""执行重连"""
|
||||
if self._stop_requested:
|
||||
return
|
||||
# 调用所有注册的回调函数
|
||||
with self._lock:
|
||||
callbacks = self.price_callbacks.copy()
|
||||
|
||||
logger.info("正在重新连接 WebSocket...")
|
||||
|
||||
# 清理旧连接（抑制错误输出）
|
||||
with SuppressOutput(suppress_stderr=True, suppress_stdout=True):
|
||||
try:
|
||||
if self.twm:
|
||||
self.twm.stop()
|
||||
except:
|
||||
pass
|
||||
|
||||
self.twm = None
|
||||
self.running = False
|
||||
self.subscribed_symbols.clear()
|
||||
|
||||
# 重新启动
|
||||
self.start()
|
||||
for callback in callbacks:
|
||||
try:
|
||||
callback(symbol, price)
|
||||
except Exception as e:
|
||||
logger.error(f"价格回调执行出错: {e}")
|
||||
|
||||
def subscribe_symbol(self, symbol: str):
|
||||
"""
|
||||
@ -222,52 +134,27 @@ class PriceMonitorService:
|
||||
"""
|
||||
symbol = symbol.upper()
|
||||
|
||||
# 记录期望订阅的交易对（用于重连恢复）
|
||||
self._desired_symbols.add(symbol)
|
||||
|
||||
if symbol in self.subscribed_symbols:
|
||||
logger.debug(f"已订阅 {symbol}")
|
||||
return
|
||||
|
||||
self.subscribed_symbols.add(symbol)
|
||||
logger.info(f"已订阅 {symbol} 价格更新")
|
||||
|
||||
# 如果服务未启动，自动启动
|
||||
if not self.running:
|
||||
# 如果还没启动，先加入待订阅列表，然后启动
|
||||
if symbol not in self._pending_symbols:
|
||||
self._pending_symbols.append(symbol)
|
||||
self.start()
|
||||
return
|
||||
|
||||
self._do_subscribe(symbol)
|
||||
|
||||
def _do_subscribe(self, symbol: str):
|
||||
"""实际执行订阅"""
|
||||
if not self.twm or not self.running:
|
||||
return
|
||||
|
||||
try:
|
||||
stream_name = self.twm.start_symbol_ticker_socket(
|
||||
callback=self._handle_price_update,
|
||||
symbol=symbol
|
||||
)
|
||||
self.subscribed_symbols[symbol] = stream_name
|
||||
self._last_message_time[symbol] = time.time()
|
||||
logger.info(f"已订阅 {symbol} 价格更新")
|
||||
except Exception as e:
|
||||
logger.error(f"订阅 {symbol} 失败: {e}")
|
||||
# 立即获取一次价格
|
||||
self._fetch_single_price(symbol)
|
||||
|
||||
def unsubscribe_symbol(self, symbol: str):
|
||||
"""取消订阅交易对"""
|
||||
symbol = symbol.upper()
|
||||
if symbol not in self.subscribed_symbols:
|
||||
return
|
||||
|
||||
try:
|
||||
stream_name = self.subscribed_symbols[symbol]
|
||||
self.twm.stop_socket(stream_name)
|
||||
del self.subscribed_symbols[symbol]
|
||||
self._desired_symbols.discard(symbol)
|
||||
if symbol in self.subscribed_symbols:
|
||||
self.subscribed_symbols.discard(symbol)
|
||||
self.latest_prices.pop(symbol, None)
|
||||
logger.info(f"已取消订阅 {symbol}")
|
||||
except Exception as e:
|
||||
logger.error(f"取消订阅 {symbol} 失败: {e}")
|
||||
|
||||
def add_price_callback(self, callback: Callable[[str, float], None]):
|
||||
"""
|
||||
@ -286,70 +173,27 @@ class PriceMonitorService:
|
||||
if callback in self.price_callbacks:
|
||||
self.price_callbacks.remove(callback)
|
||||
|
||||
def _handle_price_update(self, msg: Dict):
|
||||
"""处理 WebSocket 价格更新消息"""
|
||||
# 如果服务已停止或正在重连，忽略消息
|
||||
if not self.running or self._reconnecting or self._stop_requested:
|
||||
return
|
||||
|
||||
try:
|
||||
# 检查错误消息
|
||||
if msg.get('e') == 'error':
|
||||
error_type = msg.get('type', '')
|
||||
error_msg = str(msg.get('m', ''))
|
||||
|
||||
# 这些错误通常是正常的连接关闭，不需要记录
|
||||
ignored_errors = ['ReadLoopClosed', 'ConnectionClosed', 'WebSocketClosed', 'read loop']
|
||||
if error_type in ignored_errors or any(e.lower() in error_msg.lower() for e in ignored_errors):
|
||||
# 如果不是主动停止，触发重连
|
||||
if not self._stop_requested and self.running:
|
||||
self.running = False
|
||||
self._schedule_reconnect()
|
||||
return
|
||||
|
||||
# 其他错误记录日志（但不刷屏）
|
||||
if self.running and not self._stop_requested:
|
||||
logger.warning(f"WebSocket 消息: {msg}")
|
||||
return
|
||||
|
||||
symbol = msg.get('s') # 交易对
|
||||
price_str = msg.get('c') # 最新价格
|
||||
|
||||
if not symbol or not price_str:
|
||||
return
|
||||
|
||||
price = float(price_str)
|
||||
|
||||
# 更新最新价格缓存和消息时间
|
||||
self.latest_prices[symbol] = price
|
||||
self._last_message_time[symbol] = time.time()
|
||||
|
||||
# 调用所有注册的回调函数
|
||||
with self._lock:
|
||||
callbacks = self.price_callbacks.copy()
|
||||
|
||||
for callback in callbacks:
|
||||
try:
|
||||
callback(symbol, price)
|
||||
except Exception as e:
|
||||
logger.error(f"价格回调执行出错: {e}")
|
||||
|
||||
except Exception as e:
|
||||
if self.running and not self._stop_requested:
|
||||
logger.error(f"处理价格更新出错: {e}")
|
||||
|
||||
def get_latest_price(self, symbol: str) -> Optional[float]:
|
||||
"""获取交易对的最新缓存价格"""
|
||||
return self.latest_prices.get(symbol.upper())
|
||||
|
||||
def get_all_prices(self) -> Dict[str, float]:
|
||||
"""获取所有订阅交易对的最新价格"""
|
||||
return self.latest_prices.copy()
|
||||
|
||||
def get_subscribed_symbols(self) -> List[str]:
|
||||
"""获取已订阅的交易对列表"""
|
||||
return list(self.subscribed_symbols.keys())
|
||||
return list(self.subscribed_symbols)
|
||||
|
||||
def is_running(self) -> bool:
|
||||
"""检查服务是否在运行"""
|
||||
return self.running
|
||||
|
||||
def set_poll_interval(self, seconds: int):
|
||||
"""设置轮询间隔（秒）"""
|
||||
self._poll_interval = max(1, seconds)
|
||||
logger.info(f"轮询间隔已设置为 {self._poll_interval} 秒")
|
||||
|
||||
|
||||
# 全局单例
|
||||
_price_monitor_service: Optional[PriceMonitorService] = None
|
||||
|
||||
Loading…
Reference in New Issue
Block a user