#!/usr/bin/env python # -*- coding: utf-8 -*- """ 交易策略模块 实现基于Lighter和Binance价差的交易策略 """ import logging from enum import Enum import os import lighter import time # 配置日志 logs_dir = "logs" if not os.path.exists(logs_dir): os.makedirs(logs_dir) logger = logging.getLogger("strategy") logger.setLevel(logging.INFO) # 显式设置logger级别为INFO class StrategyState(Enum): """策略状态枚举""" WAITING_INIT = 1 # 等待初始化 IDLE_MONITORING = 2 # 空闲状态监听价差 EXECUTING_OPEN = 3 # 执行开仓操作 CHECKING_OPEN = 4 # 检查开仓结果 WAITING_CONVERGENCE = 5 # 交易完成等待价差收敛 EXECUTING_CLOSE = 6 # 执行平仓操作 CHECKING_CLOSE = 7 # 检查平仓结果 POSITION_CLOSED = 8 # 平仓完成 class TradingStrategy: """交易策略类""" def __init__(self): """初始化策略""" self.state = StrategyState.WAITING_INIT self.current_position = None # 当前持仓信息 self.entry_price_bps = 5 # 入场时的价差（单位：bps） self.target_symbol = "DOGE" # 目标交易对 self.trade_quantity = 1 # 交易数量（买卖数量） self.account_info = None # 存储账户信息 self.last_account_update_time = 0 # 上次更新账户信息的时间戳 self.last_trade_time = 0 # 上次交易时间戳（开仓或平仓） self.position_side = None # 持仓方向：'long' 或 'short' self.account_index = 318163 self.api_key_index = 0 self.api_client = lighter.ApiClient() self.account_api = lighter.AccountApi(self.api_client) self.transaction_api = lighter.TransactionApi(self.api_client) self.signer_client = lighter.SignerClient( url='https://mainnet.zklighter.elliot.ai', private_key='72aafa0426f7ff2806c68625ca5c88de153e34fcb23567f3b872cd56334d2848fb223466efff9c21', account_index=self.account_index, api_key_index=self.api_key_index ) # Check client connection err = self.signer_client.check_client() if err is not None: logger.error(f"SignerClient CheckClient error: {err}") return logger.info("策略初始化完成，当前状态: WAITING_INIT") async def do_strategy(self, market_data): """ 执行策略逻辑 Args: market_data: 包含市场数据的字典，格式: { 'symbol': str, 'binance_mark_price': float, 'binance_price': float, 'lighter_mark_price': float, 'lighter_price': float, 'timestamp': int, 'orderbook': dict # 市场信息(market_info)，不是深度信息 } """ if not market_data: return # 更新账户信息，但至少间隔1秒 current_time = time.time() if current_time - self.last_account_update_time >= 1.0: # 确保至少间隔1秒 try: account_response = await self.account_api.account(by="index", value=f"{self.account_index}") if account_response.code == 200: self.account_info = account_response self.last_account_update_time = current_time # 更新时间戳 # logger.info(f"账户信息更新成功: 可用余额={account_response.accounts[0].available_balance}, 总资产={account_response.accounts[0].total_asset_value}") else: logger.warning(f"账户信息查询失败: code={account_response.code}, message={account_response.message}") except Exception as e: logger.error(f"查询账户信息时出错: {str(e)}") return symbol = market_data.get('symbol') # 如果是DOGE交易对，打印实时行情 if symbol == self.target_symbol: await self._print_market_data(market_data) # 根据当前状态执行相应逻辑 if self.state == StrategyState.WAITING_INIT: await self._handle_waiting_init() elif self.state == StrategyState.IDLE_MONITORING: await self._handle_idle_monitoring(market_data) elif self.state == StrategyState.EXECUTING_OPEN: await self._handle_executing_open(market_data) elif self.state == StrategyState.CHECKING_OPEN: await self._handle_checking_open(market_data) elif self.state == StrategyState.WAITING_CONVERGENCE: await self._handle_waiting_convergence(market_data) elif self.state == StrategyState.EXECUTING_CLOSE: await self._handle_executing_close(market_data) elif self.state == StrategyState.CHECKING_CLOSE: await self._handle_checking_close(market_data) elif self.state == StrategyState.POSITION_CLOSED: await self._handle_position_closed() async def _print_market_data(self, market_data): """打印市场数据""" symbol = market_data.get('symbol') # binance_mark = market_data.get('binance_mark_price') binance_price = market_data.get('binance_price') # lighter_mark = market_data.get('lighter_mark_price') lighter_price = market_data.get('lighter_price') # 计算价差，转换为bps单位 if binance_price and lighter_price: # 确保两个价格都是浮点数 binance_price_float = float(binance_price) if isinstance(binance_price, str) else binance_price lighter_price_float = float(lighter_price) if isinstance(lighter_price, str) else lighter_price # 计算价差并转换为bps (1bps = 0.01%) price_diff_bps = int((lighter_price_float - binance_price_float) / binance_price_float * 10000) if binance_price_float else 0 else: price_diff_bps = None # 格式化输出 price_diff_str = f"{price_diff_bps}bps" if price_diff_bps is not None else "N/A" logger.info(f"[{symbol}] Binance: 最新价={binance_price} | Lighter: 最新价={lighter_price} | 价差={price_diff_str}") async def _handle_waiting_init(self): """处理等待初始化状态""" # 初始化完成后转到空闲监听状态 self.state = StrategyState.IDLE_MONITORING logger.info("状态转换: WAITING_INIT -> IDLE_MONITORING") async def _handle_idle_monitoring(self, market_data): """处理空闲监听状态 - 监控价差""" symbol = market_data.get('symbol') if symbol != self.target_symbol: return binance_price = market_data.get('binance_price') lighter_price = market_data.get('lighter_price') orderbook = market_data.get('orderbook') if not binance_price or not lighter_price or not orderbook: return # 计算价差（单位：bps） binance_price_float = float(binance_price) if isinstance(binance_price, str) else binance_price lighter_price_float = float(lighter_price) if isinstance(lighter_price, str) else lighter_price price_diff_bps = (lighter_price_float - binance_price_float) / binance_price_float * 10000 # 检查是否触发开仓条件 if price_diff_bps > self.entry_price_bps: # 做空：价差过大，lighter价格高于binance，卖出lighter logger.info(f"触发做空条件：价差={price_diff_bps:.2f}bps > {self.entry_price_bps}bps") self.position_side = 'short' self.state = StrategyState.EXECUTING_OPEN logger.info(f"状态转换: IDLE_MONITORING -> EXECUTING_OPEN") elif price_diff_bps < -self.entry_price_bps: # 做多：价差过小（负值），lighter价格低于binance，买入lighter logger.info(f"触发做多条件：价差={price_diff_bps:.2f}bps < -{self.entry_price_bps}bps") self.position_side = 'long' self.state = StrategyState.EXECUTING_OPEN logger.info(f"状态转换: IDLE_MONITORING -> EXECUTING_OPEN") async def _open_position(self, orderbook, binance_price): """开仓""" # 确定开仓方向和价格 if self.position_side == 'long': # 做多：在Lighter买入（使用ask价格） price = binance_price is_ask = False side_desc = '做多' else: # short # 做空：在Lighter卖出（使用bid价格） price = binance_price is_ask = True side_desc = '做空' logger.info(f"开始开仓：方向={side_desc}，数量={self.trade_quantity}，价格={price}") tx_hash, error = await self.create_order_and_send_tx( orderbook=orderbook, quantity=self.trade_quantity, price=price, is_ask=is_ask, reduce_only=False ) if error: logger.error(f"开仓失败: {error}") logger.info(f"状态转换: EXECUTING_OPEN -> IDLE_MONITORING") self.state = StrategyState.IDLE_MONITORING # 开仓失败，保持在 EXECUTING_OPEN 状态，等待重试 return # 记录开仓时间 self.last_trade_time = time.time() # 转换状态到检查开仓 self.state = StrategyState.CHECKING_OPEN logger.info(f"状态转换: EXECUTING_OPEN -> CHECKING_OPEN，交易哈希={tx_hash}") logger.info(f"等待1秒后检查持仓...") async def _handle_executing_open(self, market_data): """处理执行开仓状态 - 执行开仓操作""" symbol = market_data.get('symbol') if symbol != self.target_symbol: return orderbook = market_data.get('orderbook') if not orderbook: logger.warning("缺少市场信息，无法执行开仓") return binance_price = market_data.get('binance_price') lighter_price = market_data.get('lighter_price') if binance_price is None or lighter_price is None: logger.warning("价格数据不完整，无法执行开仓") return # 执行开仓操作 await self._open_position(orderbook, binance_price) async def _handle_checking_open(self, market_data): """处理检查开仓状态 - 等待1秒后检查持仓""" # 检查是否已经等待了至少1秒 if time.time() - self.last_trade_time < 1.0: return # 检查持仓 symbol = market_data.get('symbol') if symbol != self.target_symbol: return if not self.account_info or not self.account_info.accounts: logger.warning("账户信息不可用，无法检查持仓") return # 查找目标交易对的持仓 position = None for pos in self.account_info.accounts[0].positions: if pos.symbol == self.target_symbol: position = pos break if position and int(position.position) != 0: # 有持仓，转换到等待价差回归状态 self.current_position = position self.state = StrategyState.WAITING_CONVERGENCE logger.info(f"检测到持仓：方向={'做多' if position.sign == 1 else '做空'}，数量={position.position}") logger.info(f"状态转换: CHECKING_OPEN -> WAITING_CONVERGENCE") else: # 没有持仓，回到检测状态 logger.warning(f"开仓后未检测到持仓，回到检测状态") self.state = StrategyState.IDLE_MONITORING logger.info(f"状态转换: CHECKING_OPEN -> IDLE_MONITORING") async def _handle_waiting_convergence(self, market_data): """处理等待收敛状态 - 等待价差回归到0轴""" symbol = market_data.get('symbol') if symbol != self.target_symbol: return binance_price = market_data.get('binance_price') lighter_price = market_data.get('lighter_price') orderbook = market_data.get('orderbook') if not binance_price or not lighter_price or not orderbook: return # 计算价差（单位：bps） binance_price_float = float(binance_price) if isinstance(binance_price, str) else binance_price lighter_price_float = float(lighter_price) if isinstance(lighter_price, str) else lighter_price price_diff_bps = (lighter_price_float - binance_price_float) / binance_price_float * 10000 # 检查是否触发平仓条件 should_close = False if self.position_side == 'long': # 做多：价差需要往上回归（从负值回到0或正值） if price_diff_bps >= 0: should_close = True logger.info(f"做多平仓条件触发：价差={price_diff_bps:.2f}bps >= 0") elif self.position_side == 'short': # 做空：价差需要往下回归（从正值回到0或负值） if price_diff_bps <= 0: should_close = True logger.info(f"做空平仓条件触发：价差={price_diff_bps:.2f}bps <= 0") if should_close: # 转换到执行平仓状态 self.state = StrategyState.EXECUTING_CLOSE logger.info(f"状态转换: WAITING_CONVERGENCE -> EXECUTING_CLOSE") async def _close_position(self, orderbook, binance_price, lighter_price): """平仓""" # 确定平仓价格：使用不利方向的价格 if self.position_side == 'short': # 做空平仓（买入）：取两者较高的价格 close_price = max(binance_price, lighter_price) is_ask = False # 买入 else: # long # 做多平仓（卖出）：取两者较低的价格 close_price = min(binance_price, lighter_price) is_ask = True # 卖出 # 获取实际持仓数量 position_quantity = abs(int(self.current_position.position)) if self.current_position else self.trade_quantity logger.info(f"开始平仓：方向={'做空' if self.position_side == 'short' else '做多'}，价格={close_price}，数量={position_quantity}") tx_hash, error = await self.create_order_and_send_tx( orderbook=orderbook, quantity=position_quantity, price=close_price, is_ask=is_ask, reduce_only=True ) if error: logger.error(f"平仓失败: {error}") # 平仓失败，保持在执行平仓状态，等待下次重试 logger.info(f"平仓失败，保持在 EXECUTING_CLOSE 状态等待重试") return # 记录平仓时间 self.last_trade_time = time.time() # 转换状态到检查平仓 self.state = StrategyState.CHECKING_CLOSE logger.info(f"状态转换: EXECUTING_CLOSE -> CHECKING_CLOSE，交易哈希={tx_hash}") logger.info(f"等待1秒后检查平仓是否生效...") async def _handle_executing_close(self, market_data): """处理执行平仓状态 - 执行平仓操作""" symbol = market_data.get('symbol') if symbol != self.target_symbol: return binance_price = market_data.get('binance_price') lighter_price = market_data.get('lighter_price') orderbook = market_data.get('orderbook') if not binance_price or not lighter_price or not orderbook: return # 计算价格 binance_price_float = float(binance_price) if isinstance(binance_price, str) else binance_price lighter_price_float = float(lighter_price) if isinstance(lighter_price, str) else lighter_price await self._close_position(orderbook, binance_price_float, lighter_price_float) async def _handle_checking_close(self, market_data): """处理检查平仓状态 - 等待1秒后检查持仓是否为0""" # 检查是否已经等待了至少1秒 if time.time() - self.last_trade_time < 1.0: return # 检查持仓 symbol = market_data.get('symbol') if symbol != self.target_symbol: return if not self.account_info or not self.account_info.accounts: logger.warning("账户信息不可用，无法检查平仓状态") return # 查找目标交易对的持仓 position = None for pos in self.account_info.accounts[0].positions: if pos.symbol == self.target_symbol: position = pos break if not position or int(position.position) == 0: # 平仓成功，回到空闲状态 logger.info(f"平仓成功，当前持仓为0") self.state = StrategyState.IDLE_MONITORING self.position_side = None self.current_position = None logger.info(f"状态转换: CHECKING_CLOSE -> IDLE_MONITORING") else: # 平仓未生效，重新执行平仓 logger.warning(f"平仓未生效，当前持仓={position.position}，重新执行平仓") self.state = StrategyState.EXECUTING_CLOSE logger.info(f"状态转换: CHECKING_CLOSE -> EXECUTING_CLOSE") async def _handle_position_closed(self): """处理平仓完成状态""" # 平仓完成后回到空闲监听状态 self.state = StrategyState.IDLE_MONITORING logger.info("状态转换: POSITION_CLOSED -> IDLE_MONITORING") async def create_order_and_send_tx(self, orderbook, quantity, price, is_ask=True, reduce_only=False): """ 创建订单接口 Args: orderbook: 市场信息，等价于之前的doge_market quantity: 下单数量 price: 下单价格 is_ask: 是否为卖单，True为卖，False为买 reduce_only: 是否为只减仓单 Returns: tuple: (tx_info, error) 交易信息和错误信息 """ try: # 计算实际下单数量和价格（根据精度转换） base_amount = int(quantity * (10 ** orderbook.get('supported_size_decimals', 0))) formatted_price = int(price * (10 ** orderbook.get('supported_price_decimals', 6))) # 生成客户端订单ID client_order_index = int(time.time() * 1000) # 记录下单参数 logger.info(f"创建订单 - 市场: {orderbook.get('symbol')}(ID:{orderbook.get('market_id')})") logger.info(f"订单参数 - 数量: {quantity}, 价格: {price}, 方向: {'卖出' if is_ask else '买入'}, 只减仓: {reduce_only}") logger.info(f"格式化参数 - base_amount: {base_amount}, price: {formatted_price}, client_order_index: {client_order_index}") # 签名创建订单 tx_info, error = self.signer_client.sign_create_order( market_index=orderbook.get("market_id"), client_order_index=client_order_index, base_amount=base_amount, price=formatted_price, is_ask=is_ask, order_type=self.signer_client.ORDER_TYPE_MARKET, time_in_force=self.signer_client.ORDER_TIME_IN_FORCE_IMMEDIATE_OR_CANCEL, reduce_only=reduce_only, trigger_price=0, order_expiry=0, # 所有市价单（包括减仓市价单）都必须使用NilOrderExpiry (0) ) if error is not None: logger.error(f"订单签名失败: {error}") return None, error logger.info(f"订单签名成功，准备发送交易: {tx_info}") # 发送交易 tx_response = await self.transaction_api.send_tx(self.signer_client.TX_TYPE_CREATE_ORDER, tx_info) # 检查返回状态码 if tx_response.code != 200: error_msg = f"交易发送失败: code={tx_response.code}, message={tx_response.message}, tx_hash={tx_response.tx_hash}" logger.error(error_msg) raise Exception(error_msg) tx_hash = tx_response.tx_hash logger.info(f"交易发送成功: tx_hash={tx_hash}") return tx_hash, None except Exception as e: logger.error(f"创建订单时发生错误: {str(e)}") return None, str(e) async def main(): strategy = TradingStrategy() # account = await strategy.account_api.account(by="index", value=f"{strategy.account_index}") # [AccountPosition(market_id=3, symbol='DOGE', initial_margin_fraction='10.00', open_order_count=0, pending_order_count=0, position_tied_order_count=0, sign=1, position='1', avg_entry_price='0.194368', position_value='0.194360', unrealized_pnl='-0.000008', realized_pnl='0.000000', liquidation_price='0', total_funding_paid_out=None, margin_mode=0, allocated_margin='0.000000', additional_properties={})] print(account.accounts[0].positions) # doge_market = { # "symbol": "DOGE", # "market_id": 3, # "status": "active", # "taker_fee": "0.0000", # "maker_fee": "0.0000", # "liquidation_fee": "1.0000", # "min_base_amount": "10", # "min_quote_amount": "10.000000", # "order_quote_limit": "", # "supported_size_decimals": 0, # "supported_price_decimals": 6, # "supported_quote_decimals": 6 # } # tx_response, error = await strategy.create_order_and_send_tx(doge_market, 1, 0.1, is_ask=True, reduce_only=True) # if error is not None: # print(f"Error sending first order (first batch): {error}") # return # print(tx_response) if __name__ == '__main__': import asyncio asyncio.run(main())