""" Binance Futures API 客户端 封装所有与Binance Futures API的交互: - 账户信息查询（余额、持仓） - 下单（限价/市价/止损） - 改单/撤单 - 持仓查询 安全设计: - API Key只开交易权限，不开提币权限 - 所有请求带签名（HMAC SHA256） - 支持测试网（testnet）模式 """ import os import time import hmac import hashlib import logging from typing import Optional, Dict, List from urllib.parse import urlencode import requests logger = logging.getLogger(__name__) # Binance Futures endpoints MAINNET_BASE = "https://fapi.binance.com" TESTNET_BASE = "https://testnet.binancefuture.com" class BinanceClient: """ Binance Futures API客户端 用法: client = BinanceClient(api_key, api_secret) client = BinanceClient.from_env() # 从环境变量 client = BinanceClient.testnet(api_key, api_secret) # 测试网 # 查询 balance = client.get_balance() positions = client.get_positions() # 下单 order = client.market_buy('BTCUSDT', quantity=0.01) order = client.limit_buy('BTCUSDT', quantity=0.01, price=65000) order = client.stop_loss('BTCUSDT', 'SELL', stop_price=64000, quantity=0.01) """ def __init__( self, api_key: str, api_secret: str, testnet: bool = False, recv_window: int = 5000, ): self.api_key = api_key self.api_secret = api_secret self.base_url = TESTNET_BASE if testnet else MAINNET_BASE self.recv_window = recv_window self.testnet = testnet self.session = requests.Session() self.session.headers.update({ 'X-MBX-APIKEY': self.api_key, }) logger.info(f"BinanceClient initialized ({'testnet' if testnet else 'mainnet'})") @classmethod def from_env(cls, testnet: bool = False) -> 'BinanceClient': """从环境变量创建客户端""" key = os.environ.get('BINANCE_API_KEY', '') secret = os.environ.get('BINANCE_API_SECRET', '') if not key or not secret: raise ValueError( "需设置环境变量 BINANCE_API_KEY 和 BINANCE_API_SECRET" ) return cls(key, secret, testnet=testnet) @classmethod def testnet(cls, api_key: str = '', api_secret: str = '') -> 'BinanceClient': """创建测试网客户端""" key = api_key or os.environ.get('BINANCE_TESTNET_KEY', '') secret = api_secret or os.environ.get('BINANCE_TESTNET_SECRET', '') return cls(key, secret, testnet=True) # ==================== 签名 ==================== def _sign(self, params: dict) -> dict: """为请求参数添加签名""" params['timestamp'] = int(time.time() * 1000) params['recvWindow'] = self.recv_window query_string = urlencode(params) signature = hmac.new( self.api_secret.encode('utf-8'), query_string.encode('utf-8'), hashlib.sha256 ).hexdigest() params['signature'] = signature return params def _request(self, method: str, path: str, params: dict = None, signed: bool = True) -> dict: """发送API请求""" url = f"{self.base_url}{path}" params = params or {} if signed: params = self._sign(params) try: if method == 'GET': resp = self.session.get(url, params=params, timeout=30) elif method == 'POST': resp = self.session.post(url, params=params, timeout=30) elif method == 'DELETE': resp = self.session.delete(url, params=params, timeout=30) else: raise ValueError(f"Unknown method: {method}") data = resp.json() if resp.status_code != 200: error_msg = data.get('msg', str(data)) error_code = data.get('code', resp.status_code) logger.error(f"API error {error_code}: {error_msg}") raise BinanceAPIError(error_code, error_msg) return data except requests.exceptions.RequestException as e: logger.error(f"Request failed: {e}") raise # ==================== 账户信息 ==================== def get_account(self) -> dict: """获取账户信息（余额、持仓、杠杆等）""" return self._request('GET', '/fapi/v2/account') def get_balance(self) -> float: """获取USDT可用余额""" account = self.get_account() for asset in account.get('assets', []): if asset['asset'] == 'USDT': return float(asset['availableBalance']) return 0.0 def get_total_equity(self) -> float: """获取总权益（含未实现盈亏）""" account = self.get_account() return float(account.get('totalWalletBalance', 0)) def get_positions(self) -> List[dict]: """获取所有活跃持仓""" data = self._request('GET', '/fapi/v2/positionRisk') # 只返回有仓位的 active = [] for p in data: amt = float(p.get('positionAmt', 0)) if amt != 0: active.append({ 'symbol': p['symbol'], 'side': 'LONG' if amt > 0 else 'SHORT', 'quantity': abs(amt), 'entry_price': float(p['entryPrice']), 'unrealized_pnl': float(p['unRealizedProfit']), 'leverage': int(p['leverage']), 'notional': abs(float(p.get('notional', 0))), 'margin': float(p.get('isolatedWallet', 0)), 'liquidation_price': float(p.get('liquidationPrice', 0)), }) return active def get_position(self, symbol: str) -> Optional[dict]: """获取单个币种的持仓""" positions = self.get_positions() for p in positions: if p['symbol'] == symbol: return p return None # ==================== 杠杆设置 ==================== def set_leverage(self, symbol: str, leverage: int) -> dict: """设置杠杆倍数""" return self._request('POST', '/fapi/v1/leverage', { 'symbol': symbol, 'leverage': leverage, }) def set_margin_type(self, symbol: str, margin_type: str = 'ISOLATED') -> dict: """设置保证金模式（ISOLATED/CROSSED）""" try: return self._request('POST', '/fapi/v1/marginType', { 'symbol': symbol, 'marginType': margin_type, }) except BinanceAPIError as e: if e.code == -4046: # "No need to change margin type" return {'msg': 'already set'} raise # ==================== 下单 ==================== def market_open(self, symbol: str, side: str, quantity: float) -> dict: """ 市价开仓 Args: symbol: 'BTCUSDT' side: 'BUY' (做多) or 'SELL' (做空) quantity: 数量（币数，如BTC=0.001） """ params = { 'symbol': symbol, 'side': side, 'type': 'MARKET', 'quantity': self._format_quantity(symbol, quantity), } order = self._request('POST', '/fapi/v1/order', params) logger.info(f"Market {side} {symbol}: qty={quantity}, orderId={order.get('orderId')}") return order def limit_open(self, symbol: str, side: str, quantity: float, price: float, time_in_force: str = 'GTC') -> dict: """ 限价开仓 Args: price: 限价价格 time_in_force: 'GTC'(Good Till Cancel) / 'IOC' / 'FOK' """ params = { 'symbol': symbol, 'side': side, 'type': 'LIMIT', 'quantity': self._format_quantity(symbol, quantity), 'price': self._format_price(symbol, price), 'timeInForce': time_in_force, } order = self._request('POST', '/fapi/v1/order', params) logger.info(f"Limit {side} {symbol}: qty={quantity} @ {price}") return order def stop_loss_order(self, symbol: str, side: str, stop_price: float, quantity: float, close_position: bool = False) -> dict: """ 止损单（触发后市价执行） Args: side: 'SELL'(做多的止损) or 'BUY'(做空的止损) stop_price: 触发价格 quantity: 平仓数量 close_position: True则平掉全部仓位 """ params = { 'symbol': symbol, 'side': side, 'type': 'STOP_MARKET', 'stopPrice': self._format_price(symbol, stop_price), 'workingType': 'MARK_PRICE', # 用标记价格触发，避免插针 } if close_position: params['closePosition'] = 'true' else: params['quantity'] = self._format_quantity(symbol, quantity) order = self._request('POST', '/fapi/v1/order', params) logger.info(f"Stop loss {side} {symbol}: trigger @ {stop_price}") return order def market_close(self, symbol: str, side: str, quantity: float) -> dict: """ 市价平仓 Args: side: 'SELL'(平多) or 'BUY'(平空) quantity: 平仓数量 """ params = { 'symbol': symbol, 'side': side, 'type': 'MARKET', 'quantity': self._format_quantity(symbol, quantity), 'reduceOnly': 'true', } order = self._request('POST', '/fapi/v1/order', params) logger.info(f"Close {side} {symbol}: qty={quantity}") return order # ==================== 订单管理 ==================== def cancel_order(self, symbol: str, order_id: int) -> dict: """撤销单个订单""" return self._request('DELETE', '/fapi/v1/order', { 'symbol': symbol, 'orderId': order_id, }) def cancel_all_orders(self, symbol: str) -> dict: """撤销某币种所有挂单""" return self._request('DELETE', '/fapi/v1/allOpenOrders', { 'symbol': symbol, }) def get_open_orders(self, symbol: Optional[str] = None) -> List[dict]: """获取所有挂单""" params = {} if symbol: params['symbol'] = symbol return self._request('GET', '/fapi/v1/openOrders', params) def get_order(self, symbol: str, order_id: int) -> dict: """查询单个订单状态""" return self._request('GET', '/fapi/v1/order', { 'symbol': symbol, 'orderId': order_id, }) # ==================== 市场数据 ==================== def get_price(self, symbol: str) -> float: """获取最新价格""" data = self._request('GET', '/fapi/v1/ticker/price', {'symbol': symbol}, signed=False) return float(data['price']) def get_exchange_info(self, symbol: str) -> dict: """获取交易对信息（精度、最小下单量等）""" data = self._request('GET', '/fapi/v1/exchangeInfo', signed=False) for s in data.get('symbols', []): if s['symbol'] == symbol: return s return {} # ==================== 工具方法 ==================== # 精度缓存 _precision_cache: Dict[str, dict] = {} def _get_precision(self, symbol: str) -> dict: """获取交易对精度""" if symbol not in self._precision_cache: info = self.get_exchange_info(symbol) price_precision = info.get('pricePrecision', 2) qty_precision = info.get('quantityPrecision', 3) filters = {f['filterType']: f for f in info.get('filters', [])} min_qty = float(filters.get('LOT_SIZE', {}).get('minQty', 0.001)) min_notional = float(filters.get('MIN_NOTIONAL', {}).get('notional', 5)) self._precision_cache[symbol] = { 'price_precision': price_precision, 'qty_precision': qty_precision, 'min_qty': min_qty, 'min_notional': min_notional, } return self._precision_cache[symbol] def _format_price(self, symbol: str, price: float) -> str: """格式化价格到正确精度""" prec = self._get_precision(symbol) return f"{price:.{prec['price_precision']}f}" def _format_quantity(self, symbol: str, quantity: float) -> str: """格式化数量到正确精度""" prec = self._get_precision(symbol) formatted = round(quantity, prec['qty_precision']) if formatted < prec['min_qty']: formatted = prec['min_qty'] return f"{formatted:.{prec['qty_precision']}f}" def calc_quantity(self, symbol: str, notional: float) -> float: """ 从名义持仓金额计算下单数量 Args: symbol: 交易对 notional: 名义持仓金额（美元） Returns: 币数量 """ price = self.get_price(symbol) if price <= 0: return 0 return notional / price class BinanceAPIError(Exception): """Binance API错误""" def __init__(self, code: int, msg: str): self.code = code self.msg = msg super().__init__(f"Binance API Error {code}: {msg}")