""" Hyperliquid 小资金测试脚本 测试内容： 1. 获取账户状态 2. 市价单测试（小金额） 3. 限价单测试（小金额） 4. 止盈止损设置 5. 查询持仓和挂单 6. 清理测试（平仓、取消挂单） 运行方式： python3 test_hyperliquid_operations.py ⚠️ 请确保 .env 中已配置： - CLAWFI_WALLET_ADDRESS - CLAWFI_PRIVATE_KEY - hyperliquid_trading_enabled=true """ import sys import os import time # 添加项目路径 sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) from app.services.hyperliquid_trading_service import HyperliquidTradingService from app.utils.logger import logger class HyperliquidTester: """Hyperliquid 测试器""" def __init__(self): """初始化测试器""" try: self.service = HyperliquidTradingService() logger.info("✅ Hyperliquid 服务初始化成功") except Exception as e: logger.error(f"❌ 初始化失败: {e}") raise def print_section(self, title: str): """打印分隔线""" print(f"\n{'='*60}") print(f" {title}") print(f"{'='*60}\n") def test_1_get_account_state(self): """测试1：获取账户状态""" self.print_section("测试1: 获取账户状态") state = self.service.get_account_state() print(f"💰 账户价值: ${state['account_value']:,.2f}") print(f"📊 已用保证金: ${state['total_margin_used']:,.2f}") print(f"💵 可用余额: ${state['available_balance']:,.2f}") print(f"📈 持仓数量: {len(state['positions'])}") positions = self.service.get_open_positions() if positions: print(f"\n当前持仓:") for pos in positions: coin = pos['coin'] size = pos['size'] entry = pos['entry_price'] pnl = pos['unrealized_pnl'] print(f" - {coin}: {size:.4f} @ ${entry:,.2f} | PnL: ${pnl:,.2f}") else: print(f" 无持仓") return state def test_2_market_order(self, symbol: str = "BTC", is_buy: bool = True, size: float = 0.001): """测试2：市价单（小金额测试）""" self.print_section(f"测试2: 市价单 - {'买入' if is_buy else '卖出'} {symbol}") # 检查风险限制 risk = self.service.check_risk_limits() print(f"🔍 风险检查:") print(f" 初始余额: ${risk['initial_balance']:,.2f}") print(f" 当前价值: ${risk['current_value']:,.2f}") print(f" 回撤: {risk['drawdown_percent']:.2f}%") print(f" 安全交易: {'✅' if risk['safe_to_trade'] else '❌'}") if not risk['safe_to_trade']: print("⚠️ 风险检查未通过，跳过测试") return None # 更新杠杆 print(f"\n⚙️ 设置杠杆为 2x...") self.service.update_leverage(symbol, 2) # 下市价单 print(f"\n📝 下市价单: {'买入' if is_buy else '卖出'} {size} {symbol}") result = self.service.place_market_order( symbol=symbol, is_buy=is_buy, size=size ) if result.get('success'): print(f"✅ 市价单成功") print(f" 方向: {'买入' if is_buy else '卖出'}") print(f" 数量: {size}") print(f" 结果: {result.get('result')}") # 设置止盈止损 print(f"\n🎯 设置止盈止损...") current_positions = self.service.get_open_positions() if current_positions: pos = current_positions[0] entry_price = pos['entry_price'] is_long = pos['size'] > 0 # 计算止盈止损价格（小范围测试） if is_long: tp_price = entry_price * 1.01 # +1% sl_price = entry_price * 0.995 # -0.5% else: tp_price = entry_price * 0.99 # -1% sl_price = entry_price * 1.005 # +0.5% print(f" 入场价: ${entry_price:,.2f}") print(f" 止盈价: ${tp_price:,.2f}") print(f" 止损价: ${sl_price:,.2f}") tp_sl_result = self.service.set_tp_sl( symbol=symbol, is_long=is_long, size=abs(size), tp_price=tp_price, sl_price=sl_price ) if tp_sl_result.get('success'): print(f"✅ 止盈止损设置成功") else: print(f"❌ 止盈止损设置失败: {tp_sl_result.get('error')}") else: print(f"❌ 市价单失败: {result.get('error')}") # 等待一下查看结果 print(f"\n⏳ 等待2秒查看持仓状态...") time.sleep(2) # 查询挂单 orders = self.service.get_open_orders(symbol) if orders: print(f"\n📋 当前挂单:") for order in orders: print(f" - ID: {order['order_id']}") print(f" 方向: {order['side']}") print(f" 价格: ${order['price']:,.2f}") print(f" 数量: {order['size']}") else: print(f"\n📋 无挂单") return result def test_3_limit_order(self, symbol: str = "BTC", is_buy: bool = True, size: float = 0.001): """测试3：限价单（挂单测试）""" self.print_section(f"测试3: 限价单 - {'买入' if is_buy else '卖出'} {symbol}") # 获取当前价格 import requests try: response = requests.get("https://api.hyperliquid.xyz/info", json={ "type": "meta", "coin": symbol }, timeout=5) data = response.json() # 从响应中获取当前价格（简化处理，使用一个估算值） current_price = 95000 # 假设 BTC 当前价格 except: current_price = 95000 # 设置挂单价格（距离当前价格 1%） if is_buy: limit_price = current_price * 0.99 # 低于市价 1% else: limit_price = current_price * 1.01 # 高于市价 1% print(f"💡 当前价格估算: ${current_price:,.2f}") print(f"📝 挂单价格: ${limit_price:,.2f} ({'低于' if is_buy else '高于'}市价1%)") # 下限价单 result = self.service.place_limit_order( symbol=symbol, is_buy=is_buy, size=size, price=limit_price ) if result.get('success'): print(f"✅ 限价单成功") print(f" 方向: {'买入' if is_buy else '卖出'}") print(f" 数量: {size}") print(f" 价格: ${limit_price:,.2f}") # 查询挂单 orders = self.service.get_open_orders(symbol) if orders: print(f"\n📋 当前挂单:") for order in orders: print(f" - ID: {order['order_id']}") print(f" 方向: {order['side']}") print(f" 价格: ${order['price']:,.2f}") print(f" 数量: {order['size']}") return result else: print(f"❌ 限价单失败: {result.get('error')}") return None def test_4_check_tp_sl(self, symbol: str = "BTC"): """测试4：查询止盈止损""" self.print_section(f"测试4: 查询止盈止损 - {symbol}") tp_sl = self.service.get_tp_sl_prices(symbol) print(f"📊 止盈止损状态:") if tp_sl['take_profit']: print(f" ✅ 止盈: ${tp_sl['take_profit']:,.2f}") else: print(f" ➖ 止盈: 未设置") if tp_sl['stop_loss']: print(f" ✅ 止损: ${tp_sl['stop_loss']:,.2f}") else: print(f" ➖ 止损: 未设置") return tp_sl def test_5_cleanup(self, symbol: str = "BTC"): """测试5：清理（平仓、取消挂单）""" self.print_section(f"测试5: 清理测试 - {symbol}") # 取消所有挂单 print(f"🗑️ 取消所有挂单...") cancel_result = self.service.cancel_all_orders(symbol) if cancel_result.get('success'): print(f"✅ 取消挂单成功") else: print(f"❌ 取消挂单失败: {cancel_result.get('error')}") # 平仓 positions = self.service.get_open_positions() symbol_positions = [p for p in positions if p['coin'] == symbol] if symbol_positions: print(f"\n📤 平仓 {symbol}...") for pos in symbol_positions: size = abs(pos['size']) is_long = pos['size'] > 0 result = self.service.place_market_order( symbol=symbol, is_buy=not is_long, # 平仓方向相反 size=size, reduce_only=True ) if result.get('success'): print(f"✅ 平仓成功: {size} {symbol}") else: print(f"❌ 平仓失败: {result.get('error')}") else: print(f"📊 无持仓需要平仓") # 最终状态 print(f"\n📊 最终状态:") final_positions = self.service.get_open_positions() final_orders = self.service.get_open_orders(symbol) print(f" 持仓: {len(final_positions)} 个") print(f" 挂单: {len(final_orders)} 个") if len(final_positions) == 0 and len(final_orders) == 0: print(f"✅ 清理完成，无持仓和挂单") else: print(f"⚠️ 清理未完全，请检查") def run_all_tests(self, auto_mode: bool = True): """运行所有测试 Args: auto_mode: 自动模式，不等待用户确认（默认True） """ print("\n" + "="*60) print(" Hyperliquid 小资金测试") if auto_mode: print(" 自动模式: 将直接执行所有测试") else: print(" 交互模式: 每步需要确认") print("="*60) def wait_for_confirmation(msg: str): """等待用户确认（仅在非自动模式下）""" if not auto_mode: try: input(f"\n⚠️ {msg}\n按 Enter 继续，Ctrl+C 取消...") except EOFError: print("⚠️ 检测到非交互环境，切换到自动模式") return True return True try: # 测试1：账户状态 self.test_1_get_account_state() # 测试2：市价单 wait_for_confirmation("即将测试市价单（小金额）") self.test_2_market_order(symbol="BTC", is_buy=True, size=0.001) # 测试3：限价单 wait_for_confirmation("即将测试限价单（小金额）") self.test_3_limit_order(symbol="BTC", is_buy=False, size=0.001) # 测试4：查询止盈止损 self.test_4_check_tp_sl(symbol="BTC") # 测试5：清理 wait_for_confirmation("即将清理测试（平仓、取消挂单）") self.test_5_cleanup(symbol="BTC") print("\n" + "="*60) print(" ✅ 所有测试完成") print("="*60) except KeyboardInterrupt: print("\n\n⚠️ 测试被用户中断") print("🧹 执行清理...") # 尝试清理 self.test_5_cleanup(symbol="BTC") except Exception as e: logger.error(f"❌ 测试出错: {e}") import traceback traceback.print_exc() # 尝试清理 print("\n🧹 执行清理...") self.test_5_cleanup(symbol="BTC") def main(): """主函数""" tester = HyperliquidTester() tester.run_all_tests() if __name__ == "__main__": main()