410 lines
15 KiB
Python
410 lines
15 KiB
Python
"""
|
||
币安客户端封装 - 提供异步交易接口
|
||
"""
|
||
import asyncio
|
||
import logging
|
||
from typing import Dict, List, Optional, Any
|
||
from binance import AsyncClient, BinanceSocketManager
|
||
from binance.exceptions import BinanceAPIException
|
||
from unicorn_websocket import UnicornWebSocketManager
|
||
import config
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class BinanceClient:
|
||
"""币安客户端封装类"""
|
||
|
||
def __init__(self, api_key: str = None, api_secret: str = None, testnet: bool = False):
|
||
"""
|
||
初始化币安客户端
|
||
|
||
Args:
|
||
api_key: API密钥
|
||
api_secret: API密钥
|
||
testnet: 是否使用测试网
|
||
"""
|
||
self.api_key = api_key or config.BINANCE_API_KEY
|
||
self.api_secret = api_secret or config.BINANCE_API_SECRET
|
||
self.testnet = testnet or config.USE_TESTNET
|
||
self.client: Optional[AsyncClient] = None
|
||
self.socket_manager: Optional[BinanceSocketManager] = None
|
||
self.unicorn_manager: Optional[UnicornWebSocketManager] = None
|
||
self.use_unicorn = config.TRADING_CONFIG.get('USE_UNICORN_WEBSOCKET', True)
|
||
|
||
async def connect(self, timeout: int = None, retries: int = None):
|
||
"""
|
||
连接币安API
|
||
|
||
Args:
|
||
timeout: 连接超时时间（秒），默认从config读取
|
||
retries: 重试次数，默认从config读取
|
||
"""
|
||
timeout = timeout or config.CONNECTION_TIMEOUT
|
||
retries = retries or config.CONNECTION_RETRIES
|
||
|
||
last_error = None
|
||
for attempt in range(retries):
|
||
try:
|
||
logger.info(
|
||
f"尝试连接币安API (第 {attempt + 1}/{retries} 次, "
|
||
f"测试网: {self.testnet}, 超时: {timeout}秒)..."
|
||
)
|
||
|
||
# 创建客户端
|
||
self.client = await AsyncClient.create(
|
||
api_key=self.api_key,
|
||
api_secret=self.api_secret,
|
||
testnet=self.testnet
|
||
)
|
||
|
||
# 测试连接（带超时）
|
||
try:
|
||
await asyncio.wait_for(self.client.ping(), timeout=timeout)
|
||
except asyncio.TimeoutError:
|
||
await self.client.close_connection()
|
||
raise asyncio.TimeoutError(f"ping超时 ({timeout}秒)")
|
||
|
||
self.socket_manager = BinanceSocketManager(self.client)
|
||
logger.info(f"✓ 币安客户端连接成功 (测试网: {self.testnet})")
|
||
|
||
# 启动Unicorn WebSocket（如果启用）
|
||
if self.use_unicorn:
|
||
try:
|
||
self.unicorn_manager = UnicornWebSocketManager(testnet=self.testnet)
|
||
if self.unicorn_manager.start():
|
||
logger.info("✓ Unicorn WebSocket管理器启动成功")
|
||
# 启动异步数据处理
|
||
asyncio.create_task(self.unicorn_manager.process_stream_data())
|
||
else:
|
||
logger.warning("Unicorn WebSocket管理器启动失败，将使用标准WebSocket")
|
||
self.unicorn_manager = None
|
||
except Exception as e:
|
||
logger.warning(f"启动Unicorn WebSocket失败: {e}，将使用标准WebSocket")
|
||
self.unicorn_manager = None
|
||
|
||
# 验证API密钥权限
|
||
await self._verify_api_permissions()
|
||
|
||
return
|
||
|
||
except asyncio.TimeoutError as e:
|
||
last_error = f"连接超时: {e}"
|
||
logger.warning(f"连接超时，剩余 {retries - attempt - 1} 次重试机会")
|
||
if attempt < retries - 1:
|
||
await asyncio.sleep(2) # 等待2秒后重试
|
||
except Exception as e:
|
||
last_error = str(e)
|
||
logger.warning(f"连接失败: {e}，剩余 {retries - attempt - 1} 次重试机会")
|
||
if self.client:
|
||
try:
|
||
await self.client.close_connection()
|
||
except:
|
||
pass
|
||
if attempt < retries - 1:
|
||
await asyncio.sleep(2)
|
||
|
||
error_msg = f"连接币安API失败 (已重试 {retries} 次): {last_error}"
|
||
logger.error("=" * 60)
|
||
logger.error(error_msg)
|
||
logger.error("=" * 60)
|
||
logger.error("故障排查建议：")
|
||
logger.error("1. 检查网络连接是否正常")
|
||
logger.error("2. 检查API密钥是否正确")
|
||
logger.error("3. 如果在中国大陆，可能需要使用代理或VPN")
|
||
if self.testnet:
|
||
logger.error("4. 测试网地址可能无法访问，尝试设置 USE_TESTNET=False")
|
||
logger.error("5. 检查防火墙设置")
|
||
logger.error("=" * 60)
|
||
raise ConnectionError(error_msg)
|
||
|
||
async def _verify_api_permissions(self):
|
||
"""
|
||
验证API密钥权限
|
||
"""
|
||
try:
|
||
# 尝试获取账户信息来验证权限
|
||
await self.client.futures_account()
|
||
logger.info("✓ API密钥权限验证通过")
|
||
except BinanceAPIException as e:
|
||
error_code = e.code if hasattr(e, 'code') else None
|
||
if error_code == -2015:
|
||
logger.warning("⚠ API密钥权限验证失败，可能无法进行合约交易")
|
||
logger.warning("请检查API密钥是否启用了合约交易权限")
|
||
else:
|
||
logger.warning(f"⚠ API密钥验证时出现错误: {e}")
|
||
|
||
async def disconnect(self):
|
||
"""断开连接"""
|
||
# 停止Unicorn WebSocket
|
||
if self.unicorn_manager:
|
||
self.unicorn_manager.stop()
|
||
self.unicorn_manager = None
|
||
|
||
if self.client:
|
||
await self.client.close_connection()
|
||
logger.info("币安客户端已断开连接")
|
||
|
||
async def get_all_usdt_pairs(self) -> List[str]:
|
||
"""
|
||
获取所有USDT交易对
|
||
|
||
Returns:
|
||
USDT交易对列表
|
||
"""
|
||
try:
|
||
# 获取合约市场信息
|
||
exchange_info = await self.client.futures_exchange_info()
|
||
usdt_pairs = [
|
||
symbol['symbol']
|
||
for symbol in exchange_info['symbols']
|
||
if symbol['symbol'].endswith('USDT')
|
||
and symbol['status'] == 'TRADING'
|
||
and symbol.get('contractType') == 'PERPETUAL' # U本位永续合约
|
||
]
|
||
logger.info(f"获取到 {len(usdt_pairs)} 个USDT永续合约交易对")
|
||
return usdt_pairs
|
||
except BinanceAPIException as e:
|
||
logger.error(f"获取交易对失败: {e}")
|
||
return []
|
||
|
||
async def get_klines(self, symbol: str, interval: str = '5m', limit: int = 2) -> List[List]:
|
||
"""
|
||
获取K线数据（合约市场）
|
||
|
||
Args:
|
||
symbol: 交易对
|
||
interval: K线周期
|
||
limit: 获取数量
|
||
|
||
Returns:
|
||
K线数据列表
|
||
"""
|
||
try:
|
||
klines = await self.client.futures_klines(symbol=symbol, interval=interval, limit=limit)
|
||
return klines
|
||
except BinanceAPIException as e:
|
||
logger.error(f"获取 {symbol} K线数据失败: {e}")
|
||
return []
|
||
|
||
async def get_ticker_24h(self, symbol: str) -> Optional[Dict]:
|
||
"""
|
||
获取24小时行情数据（合约市场）
|
||
|
||
Args:
|
||
symbol: 交易对
|
||
|
||
Returns:
|
||
24小时行情数据
|
||
"""
|
||
try:
|
||
ticker = await self.client.futures_symbol_ticker(symbol=symbol)
|
||
stats = await self.client.futures_ticker(symbol=symbol)
|
||
return {
|
||
'symbol': symbol,
|
||
'price': float(ticker['price']),
|
||
'volume': float(stats.get('quoteVolume', 0)),
|
||
'changePercent': float(stats.get('priceChangePercent', 0))
|
||
}
|
||
except BinanceAPIException as e:
|
||
logger.error(f"获取 {symbol} 24小时行情失败: {e}")
|
||
return None
|
||
|
||
async def get_account_balance(self) -> Dict[str, float]:
|
||
"""
|
||
获取U本位合约账户余额
|
||
|
||
Returns:
|
||
账户余额字典 {'available': 可用余额, 'total': 总余额}
|
||
"""
|
||
try:
|
||
account = await self.client.futures_account()
|
||
assets = account.get('assets', [])
|
||
usdt_asset = next((a for a in assets if a['asset'] == 'USDT'), None)
|
||
|
||
if usdt_asset:
|
||
return {
|
||
'available': float(usdt_asset['availableBalance']),
|
||
'total': float(usdt_asset['walletBalance']),
|
||
'margin': float(usdt_asset['marginBalance'])
|
||
}
|
||
return {'available': 0.0, 'total': 0.0, 'margin': 0.0}
|
||
except BinanceAPIException as e:
|
||
error_code = e.code if hasattr(e, 'code') else None
|
||
error_msg = str(e)
|
||
|
||
logger.error("=" * 60)
|
||
logger.error(f"获取账户余额失败: {error_msg}")
|
||
|
||
if error_code == -2015:
|
||
logger.error("=" * 60)
|
||
logger.error("API密钥权限错误 (错误代码: -2015)")
|
||
logger.error("可能的原因：")
|
||
logger.error("1. API密钥无效或已过期")
|
||
logger.error("2. API密钥没有合约交易权限")
|
||
logger.error("3. IP地址未添加到API密钥白名单")
|
||
logger.error("4. 测试网/生产网环境不匹配")
|
||
logger.error("=" * 60)
|
||
logger.error("解决方案：")
|
||
logger.error("1. 登录币安账户，检查API密钥状态")
|
||
logger.error("2. 确保API密钥已启用'合约交易'权限")
|
||
logger.error("3. 如果设置了IP白名单，请添加当前服务器IP")
|
||
logger.error("4. 检查 USE_TESTNET 配置是否正确")
|
||
logger.error(f" 当前配置: USE_TESTNET = {self.testnet}")
|
||
logger.error("=" * 60)
|
||
elif error_code == -1022:
|
||
logger.error("签名错误，请检查API密钥和密钥是否正确")
|
||
elif error_code == -2010:
|
||
logger.error("账户余额不足")
|
||
else:
|
||
logger.error(f"错误代码: {error_code}")
|
||
|
||
return {'available': 0.0, 'total': 0.0, 'margin': 0.0}
|
||
|
||
async def get_open_positions(self) -> List[Dict]:
|
||
"""
|
||
获取当前持仓
|
||
|
||
Returns:
|
||
持仓列表
|
||
"""
|
||
try:
|
||
positions = await self.client.futures_position_information()
|
||
open_positions = [
|
||
{
|
||
'symbol': pos['symbol'],
|
||
'positionAmt': float(pos['positionAmt']),
|
||
'entryPrice': float(pos['entryPrice']),
|
||
'markPrice': float(pos.get('markPrice', 0)),
|
||
'unRealizedProfit': float(pos['unRealizedProfit']),
|
||
'leverage': int(pos['leverage'])
|
||
}
|
||
for pos in positions
|
||
if float(pos['positionAmt']) != 0
|
||
]
|
||
return open_positions
|
||
except BinanceAPIException as e:
|
||
logger.error(f"获取持仓信息失败: {e}")
|
||
return []
|
||
|
||
async def place_order(
|
||
self,
|
||
symbol: str,
|
||
side: str,
|
||
quantity: float,
|
||
order_type: str = 'MARKET',
|
||
price: Optional[float] = None
|
||
) -> Optional[Dict]:
|
||
"""
|
||
下单
|
||
|
||
Args:
|
||
symbol: 交易对
|
||
side: 方向 'BUY' 或 'SELL'
|
||
quantity: 数量
|
||
order_type: 订单类型 'MARKET' 或 'LIMIT'
|
||
price: 限价单价格
|
||
|
||
Returns:
|
||
订单信息
|
||
"""
|
||
try:
|
||
if order_type == 'MARKET':
|
||
order = await self.client.futures_create_order(
|
||
symbol=symbol,
|
||
side=side,
|
||
type='MARKET',
|
||
quantity=quantity
|
||
)
|
||
else:
|
||
if price is None:
|
||
raise ValueError("限价单必须指定价格")
|
||
order = await self.client.futures_create_order(
|
||
symbol=symbol,
|
||
side=side,
|
||
type='LIMIT',
|
||
timeInForce='GTC',
|
||
quantity=quantity,
|
||
price=price
|
||
)
|
||
|
||
logger.info(f"下单成功: {symbol} {side} {quantity} @ {order_type}")
|
||
return order
|
||
except BinanceAPIException as e:
|
||
logger.error(f"下单失败 {symbol} {side}: {e}")
|
||
return None
|
||
|
||
async def cancel_order(self, symbol: str, order_id: int) -> bool:
|
||
"""
|
||
取消订单
|
||
|
||
Args:
|
||
symbol: 交易对
|
||
order_id: 订单ID
|
||
|
||
Returns:
|
||
是否成功
|
||
"""
|
||
try:
|
||
await self.client.futures_cancel_order(symbol=symbol, orderId=order_id)
|
||
logger.info(f"取消订单成功: {symbol} {order_id}")
|
||
return True
|
||
except BinanceAPIException as e:
|
||
logger.error(f"取消订单失败: {e}")
|
||
return False
|
||
|
||
async def set_leverage(self, symbol: str, leverage: int = 10) -> bool:
|
||
"""
|
||
设置杠杆倍数
|
||
|
||
Args:
|
||
symbol: 交易对
|
||
leverage: 杠杆倍数
|
||
|
||
Returns:
|
||
是否成功
|
||
"""
|
||
try:
|
||
await self.client.futures_change_leverage(symbol=symbol, leverage=leverage)
|
||
logger.info(f"设置杠杆成功: {symbol} {leverage}x")
|
||
return True
|
||
except BinanceAPIException as e:
|
||
logger.error(f"设置杠杆失败: {e}")
|
||
return False
|
||
|
||
def subscribe_realtime_prices(self, symbols: List[str], callback) -> bool:
|
||
"""
|
||
订阅实时价格流（使用Unicorn）
|
||
|
||
Args:
|
||
symbols: 交易对列表
|
||
callback: 价格更新回调函数 callback(symbol, price, price_data)
|
||
|
||
Returns:
|
||
是否成功
|
||
"""
|
||
if not self.unicorn_manager:
|
||
logger.warning("Unicorn WebSocket未启用，无法订阅实时价格")
|
||
return False
|
||
|
||
try:
|
||
self.unicorn_manager.subscribe_ticker(symbols, callback)
|
||
logger.info(f"订阅 {len(symbols)} 个交易对的实时价格流")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"订阅实时价格流失败: {e}")
|
||
return False
|
||
|
||
def get_realtime_price(self, symbol: str) -> Optional[float]:
|
||
"""
|
||
获取实时价格（从Unicorn WebSocket）
|
||
|
||
Args:
|
||
symbol: 交易对
|
||
|
||
Returns:
|
||
实时价格，如果未订阅则返回None
|
||
"""
|
||
if self.unicorn_manager:
|
||
return self.unicorn_manager.get_realtime_price(symbol)
|
||
return None |