auto_trade_sys/trading_system/binance_client.py
薇薇安 f99f508b09 a
2026-01-14 13:43:06 +08:00

622 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
币安客户端封装 - 提供异步交易接口
"""
import asyncio
import logging
from typing import Dict, List, Optional, Any
from binance import AsyncClient, BinanceSocketManager
from binance.exceptions import BinanceAPIException
try:
from . import config
except ImportError:
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._symbol_info_cache: Dict[str, Dict] = {} # 缓存交易对信息
self._last_request_time = {} # 记录每个API端点的最后请求时间
self._request_delay = 0.1 # 请求间隔(秒),避免频率限制
self._semaphore = asyncio.Semaphore(10) # 限制并发请求数
self._price_cache: Dict[str, Dict] = {} # WebSocket价格缓存 {symbol: {price, volume, changePercent, timestamp}}
self._price_cache_ttl = 60 # 价格缓存有效期(秒)
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})")
# 验证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):
"""断开连接"""
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 _rate_limited_request(self, endpoint: str, coro):
"""
带速率限制的API请求
Args:
endpoint: API端点标识用于记录请求时间
coro: 异步协程
"""
async with self._semaphore:
# 检查是否需要等待(避免请求过快)
if endpoint in self._last_request_time:
elapsed = asyncio.get_event_loop().time() - self._last_request_time[endpoint]
if elapsed < self._request_delay:
await asyncio.sleep(self._request_delay - elapsed)
self._last_request_time[endpoint] = asyncio.get_event_loop().time()
return await coro
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._rate_limited_request(
f'klines_{symbol}_{interval}',
self.client.futures_klines(symbol=symbol, interval=interval, limit=limit)
)
return klines
except BinanceAPIException as e:
error_code = e.code if hasattr(e, 'code') else None
if error_code == -1003:
logger.warning(f"获取 {symbol} K线数据失败: API请求频率过高建议使用WebSocket或增加扫描间隔")
else:
logger.error(f"获取 {symbol} K线数据失败: {e}")
return []
async def get_ticker_24h(self, symbol: str) -> Optional[Dict]:
"""
获取24小时行情数据合约市场
优先从WebSocket缓存读取如果缓存不可用或过期则使用REST API
Args:
symbol: 交易对
Returns:
24小时行情数据
"""
import time
# 优先从WebSocket缓存读取
if symbol in self._price_cache:
cached = self._price_cache[symbol]
cache_age = time.time() - cached.get('timestamp', 0)
if cache_age < self._price_cache_ttl:
logger.debug(f"从缓存获取 {symbol} 价格: {cached['price']:.8f} (缓存年龄: {cache_age:.1f}秒)")
return {
'symbol': symbol,
'price': cached['price'],
'volume': cached.get('volume', 0),
'changePercent': cached.get('changePercent', 0)
}
else:
logger.debug(f"{symbol} 缓存已过期 ({cache_age:.1f}秒 > {self._price_cache_ttl}秒)使用REST API")
# 如果缓存不可用或过期使用REST APIfallback
logger.debug(f"{symbol} 未在价格缓存中使用REST API获取")
try:
ticker = await self._rate_limited_request(
f'ticker_{symbol}',
self.client.futures_symbol_ticker(symbol=symbol)
)
stats = await self._rate_limited_request(
f'stats_{symbol}',
self.client.futures_ticker(symbol=symbol)
)
result = {
'symbol': symbol,
'price': float(ticker['price']),
'volume': float(stats.get('quoteVolume', 0)),
'changePercent': float(stats.get('priceChangePercent', 0))
}
# 更新缓存
import time
self._price_cache[symbol] = {
**result,
'timestamp': time.time()
}
return result
except BinanceAPIException as e:
error_code = e.code if hasattr(e, 'code') else None
if error_code == -1003:
logger.warning(f"获取 {symbol} 24小时行情失败: API请求频率过高建议使用WebSocket或增加扫描间隔")
else:
logger.error(f"获取 {symbol} 24小时行情失败: {e}")
return None
async def get_all_tickers_24h(self) -> Dict[str, Dict]:
"""
批量获取所有交易对的24小时行情数据更高效
Returns:
交易对行情数据字典 {symbol: {price, volume, changePercent}}
"""
try:
# 使用批量API一次获取所有交易对的数据
tickers = await self._rate_limited_request(
'all_tickers',
self.client.futures_ticker()
)
result = {}
for ticker in tickers:
symbol = ticker['symbol']
if symbol.endswith('USDT'):
result[symbol] = {
'symbol': symbol,
'price': float(ticker.get('lastPrice', 0)),
'volume': float(ticker.get('quoteVolume', 0)),
'changePercent': float(ticker.get('priceChangePercent', 0))
}
logger.debug(f"批量获取到 {len(result)} 个交易对的24小时行情数据")
return result
except BinanceAPIException as e:
error_code = e.code if hasattr(e, 'code') else None
if error_code == -1003:
logger.warning(f"批量获取24小时行情失败: API请求频率过高建议使用WebSocket或增加扫描间隔")
else:
logger.error(f"批量获取24小时行情失败: {e}")
return {}
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 get_symbol_info(self, symbol: str) -> Optional[Dict]:
"""
获取交易对的精度和限制信息
Args:
symbol: 交易对
Returns:
交易对信息字典,包含 quantityPrecision, minQty, stepSize 等
"""
# 先检查缓存
if symbol in self._symbol_info_cache:
return self._symbol_info_cache[symbol]
try:
exchange_info = await self.client.futures_exchange_info()
for s in exchange_info['symbols']:
if s['symbol'] == symbol:
# 提取数量精度信息
quantity_precision = s.get('quantityPrecision', 8)
# 从filters中提取minQty、stepSize和minNotional
min_qty = None
step_size = None
min_notional = None
for f in s.get('filters', []):
if f['filterType'] == 'LOT_SIZE':
min_qty = float(f.get('minQty', 0))
step_size = float(f.get('stepSize', 0))
elif f['filterType'] == 'MIN_NOTIONAL':
min_notional = float(f.get('notional', 0))
# 如果没有从filters获取到minNotional使用默认值5 USDT
if min_notional is None or min_notional == 0:
min_notional = 5.0
info = {
'quantityPrecision': quantity_precision,
'minQty': min_qty or 0,
'stepSize': step_size or 0,
'minNotional': min_notional
}
# 缓存信息
self._symbol_info_cache[symbol] = info
logger.debug(f"获取 {symbol} 精度信息: {info}")
return info
logger.warning(f"未找到交易对 {symbol} 的信息")
return None
except Exception as e:
logger.error(f"获取 {symbol} 交易对信息失败: {e}")
return None
def _adjust_quantity_precision(self, quantity: float, symbol_info: Dict) -> float:
"""
调整数量精度,使其符合币安要求
Args:
quantity: 原始数量
symbol_info: 交易对信息
Returns:
调整后的数量
"""
if not symbol_info:
# 如果没有交易对信息使用默认精度3位小数
return round(quantity, 3)
quantity_precision = symbol_info.get('quantityPrecision', 8)
step_size = symbol_info.get('stepSize', 0)
min_qty = symbol_info.get('minQty', 0)
# 如果有stepSize按照stepSize调整
if step_size > 0:
# 向下取整到stepSize的倍数使用浮点数除法
adjusted = float(int(quantity / step_size)) * step_size
else:
# 否则按照精度调整
adjusted = round(quantity, quantity_precision)
# 确保不小于最小数量
if min_qty > 0 and adjusted < min_qty:
# 如果小于最小数量,尝试向上取整到最小数量
if step_size > 0:
adjusted = min_qty
else:
adjusted = round(min_qty, quantity_precision)
logger.warning(f"数量 {quantity} 小于最小数量 {min_qty},调整为 {adjusted}")
# 最终精度调整
adjusted = round(adjusted, quantity_precision)
if adjusted != quantity:
logger.info(f"数量精度调整: {quantity} -> {adjusted} (精度: {quantity_precision}, stepSize: {step_size}, minQty: {min_qty})")
return adjusted
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:
# 获取交易对精度信息并调整数量
symbol_info = await self.get_symbol_info(symbol)
adjusted_quantity = self._adjust_quantity_precision(quantity, symbol_info)
if adjusted_quantity <= 0:
logger.error(f"调整后的数量无效: {adjusted_quantity} (原始: {quantity})")
return None
# 获取当前价格以计算名义价值
if price is None:
ticker = await self.get_ticker_24h(symbol)
if not ticker:
logger.error(f"无法获取 {symbol} 的价格信息")
return None
current_price = ticker['price']
else:
current_price = price
# 计算订单名义价值
notional_value = adjusted_quantity * current_price
min_notional = symbol_info.get('minNotional', 5.0) if symbol_info else 5.0
logger.info(f"下单检查: {symbol} {side} {adjusted_quantity} (原始: {quantity}) @ {order_type}")
logger.info(f" 当前价格: {current_price:.4f} USDT")
logger.info(f" 订单名义价值: {notional_value:.2f} USDT")
logger.info(f" 最小名义价值: {min_notional:.2f} USDT")
# 检查名义价值是否满足最小要求
if notional_value < min_notional:
logger.warning(
f"{symbol} 订单名义价值不足: {notional_value:.2f} USDT < "
f"最小要求: {min_notional:.2f} USDT"
)
logger.warning(f" 需要增加数量或提高仓位大小")
return None
if order_type == 'MARKET':
order = await self.client.futures_create_order(
symbol=symbol,
side=side,
type='MARKET',
quantity=adjusted_quantity
)
else:
if price is None:
raise ValueError("限价单必须指定价格")
order = await self.client.futures_create_order(
symbol=symbol,
side=side,
type='LIMIT',
timeInForce='GTC',
quantity=adjusted_quantity,
price=price
)
logger.info(f"下单成功: {symbol} {side} {adjusted_quantity} @ {order_type} (名义价值: {notional_value:.2f} USDT)")
return order
except BinanceAPIException as e:
error_code = e.code if hasattr(e, 'code') else None
if error_code == -1111:
logger.error(f"下单失败 {symbol} {side}: 精度错误 - {e}")
logger.error(f" 原始数量: {quantity}")
if symbol_info:
logger.error(f" 交易对精度: {symbol_info}")
elif error_code == -4164:
logger.error(f"下单失败 {symbol} {side}: 订单名义价值不足 - {e}")
logger.error(f" 订单名义价值必须至少为 5 USDT (除非选择 reduce only)")
if symbol_info:
logger.error(f" 最小名义价值: {symbol_info.get('minNotional', 5.0)} USDT")
else:
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 get_realtime_price(self, symbol: str) -> Optional[float]:
"""
获取实时价格(从缓存)
Args:
symbol: 交易对
Returns:
实时价格如果缓存中有则返回否则返回None
"""
import time
if symbol in self._price_cache:
cached = self._price_cache[symbol]
cache_age = time.time() - cached.get('timestamp', 0)
if cache_age < self._price_cache_ttl:
return cached.get('price')
return None