miracle/auto_trade_sys
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

a

Browse Source
This commit is contained in:
薇薇安 2026-01-14 21:08:32 +08:00
parent d883b2057f
commit cd74e96c22
2 changed files with 80 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
trading_system/market_scanner.py
View File

@ -280,21 +280,39 @@ class MarketScanner:
""" """
# 使用标准WebSocket # 使用标准WebSocket
try: try:
if self.client and self.client.socket_manager: import aiohttp
# 使用 BinanceSocketManager 的 futures_symbol_ticker_socket 方法 import json
async with self.client.socket_manager.futures_symbol_ticker_socket(symbol.lower()) as stream:
async for msg in stream: # 直接使用 aiohttp 连接 Binance 期货 WebSocket API
try: # 根据文档：https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams
# WebSocket 返回的数据格式：{'e': '24hrTicker', 's': 'BTCUSDT', 'c': '50000.00', ...} # 端点：wss://fstream.binance.com/ws/<symbol>@ticker
if 'c' in msg: # 'c' 是当前价格 ws_url = f"wss://fstream.binance.com/ws/{symbol.lower()}@ticker"
price = float(msg['c'])
await callback(symbol, price) async with aiohttp.ClientSession() as session:
elif 'data' in msg and 'c' in msg['data']: async with session.ws_connect(ws_url) as ws:
price = float(msg['data']['c']) async for msg in ws:
await callback(symbol, price) if msg.type == aiohttp.WSMsgType.TEXT:
except (KeyError, ValueError, TypeError) as e: try:
logger.debug(f"解析 {symbol} 价格数据失败: {e}") # 解析 JSON 消息
continue data = json.loads(msg.data)
# WebSocket 返回的数据格式：{'e': '24hrTicker', 's': 'BTCUSDT', 'c': '50000.00', ...}
if isinstance(data, dict):
if 'c' in data: # 'c' 是当前价格
price = float(data['c'])
await callback(symbol, price)
elif 'data' in data and isinstance(data['data'], dict) and 'c' in data['data']:
price = float(data['data']['c'])
await callback(symbol, price)
except (KeyError, ValueError, TypeError, json.JSONDecodeError) as e:
logger.debug(f"解析 {symbol} 价格数据失败: {e}")
continue
elif msg.type == aiohttp.WSMsgType.ERROR:
logger.warning(f"监控 {symbol} WebSocket错误: {ws.exception()}")
break
elif msg.type == aiohttp.WSMsgType.CLOSE:
logger.info(f"监控 {symbol} WebSocket连接关闭")
break
except Exception as e: except Exception as e:
logger.error(f"监控 {symbol} 价格失败: {e}") logger.error(f"监控 {symbol} 价格失败: {e}")

77
trading_system/position_manager.py
View File

@ -3,6 +3,8 @@
""" """
import asyncio import asyncio
import logging import logging
import json
import aiohttp
from typing import Dict, List, Optional from typing import Dict, List, Optional
try: try:
from .binance_client import BinanceClient from .binance_client import BinanceClient
@ -745,8 +747,9 @@ class PositionManager:
logger.info("实时监控已禁用，跳过启动") logger.info("实时监控已禁用，跳过启动")
return return
if not self.client or not self.client.client: # WebSocket 现在直接使用 aiohttp，不需要检查 socket_manager
logger.warning("WebSocket未初始化，无法启动实时监控") if not self.client:
logger.warning("客户端未初始化，无法启动实时监控")
return return
# 获取当前所有持仓 # 获取当前所有持仓
@ -824,8 +827,9 @@ class PositionManager:
logger.debug(f"{symbol} 监控任务已存在，跳过") logger.debug(f"{symbol} 监控任务已存在，跳过")
return return
if not self.client or not self.client.client: # WebSocket 现在直接使用 aiohttp，不需要检查 socket_manager
logger.warning(f"{symbol} WebSocket未初始化，无法启动监控") if not self.client:
logger.warning(f"{symbol} 客户端未初始化，无法启动监控")
return return
try: try:
@ -873,34 +877,47 @@ class PositionManager:
break break
# 使用WebSocket订阅价格流 # 使用WebSocket订阅价格流
# 使用 BinanceSocketManager 的 futures_symbol_ticker_socket 方法 # 直接使用 aiohttp 连接 Binance 期货 WebSocket API
# 注意：BinanceSocketManager 需要 AsyncClient 实例 # 根据文档：https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams
if not self.client.socket_manager: # 端点：wss://fstream.binance.com/ws/<symbol>@ticker
raise ValueError("WebSocket管理器未初始化") ws_url = f"wss://fstream.binance.com/ws/{symbol.lower()}@ticker"
async with self.client.socket_manager.futures_symbol_ticker_socket(symbol.lower()) as stream: async with aiohttp.ClientSession() as session:
logger.debug(f"{symbol} WebSocket连接已建立，开始接收价格更新") async with session.ws_connect(ws_url) as ws:
retry_count = 0 # 连接成功，重置重试计数 logger.debug(f"{symbol} WebSocket连接已建立，开始接收价格更新")
retry_count = 0 # 连接成功，重置重试计数
async for msg in stream:
if symbol not in self.active_positions:
logger.info(f"{symbol} 持仓已不存在，停止监控")
break
try: async for msg in ws:
# WebSocket 返回的数据格式：{'e': '24hrTicker', 's': 'BTCUSDT', 'c': '50000.00', ...} if symbol not in self.active_positions:
if 'c' in msg: # 'c' 是当前价格 logger.info(f"{symbol} 持仓已不存在，停止监控")
current_price = float(msg['c']) break
# 立即检查止损止盈
await self._check_single_position(symbol, current_price) if msg.type == aiohttp.WSMsgType.TEXT:
elif 'data' in msg: try:
# 兼容嵌套的数据格式 # 解析 JSON 消息
if 'c' in msg['data']: data = json.loads(msg.data)
current_price = float(msg['data']['c'])
await self._check_single_position(symbol, current_price) # WebSocket 返回的数据格式：{'e': '24hrTicker', 's': 'BTCUSDT', 'c': '50000.00', ...}
except (KeyError, ValueError, TypeError) as e: # 根据文档，ticker 流包含 'c' 字段（最后价格）
logger.debug(f"{symbol} 解析价格数据失败: {e}, 消息: {msg}") if isinstance(data, dict):
continue if 'c' in data: # 'c' 是当前价格
current_price = float(data['c'])
# 立即检查止损止盈
await self._check_single_position(symbol, current_price)
elif 'data' in data:
# 兼容组合流格式（如果使用 /stream 端点）
if isinstance(data['data'], dict) and 'c' in data['data']:
current_price = float(data['data']['c'])
await self._check_single_position(symbol, current_price)
except (KeyError, ValueError, TypeError, json.JSONDecodeError) as e:
logger.debug(f"{symbol} 解析价格数据失败: {e}, 消息: {msg.data[:100] if hasattr(msg, 'data') else 'N/A'}")
continue
elif msg.type == aiohttp.WSMsgType.ERROR:
logger.warning(f"{symbol} WebSocket错误: {ws.exception()}")
break
elif msg.type == aiohttp.WSMsgType.CLOSE:
logger.info(f"{symbol} WebSocket连接关闭")
break
except asyncio.CancelledError: except asyncio.CancelledError:
logger.info(f"{symbol} 监控任务已取消") logger.info(f"{symbol} 监控任务已取消")