a22dfe459c
- Add hyperliquid_trading_service.py with position management and TP/SL - Implement dual-track trading (paper trading + Hyperliquid) - Add position size calculation based on available margin - Support net position mode (Hyperliquid) vs order mode (paper) - Add risk controls: 10% circuit breaker, 10x max leverage - Add test script for Hyperliquid SDK validation
308 lines
9.8 KiB
Python
308 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Hyperliquid SDK 测试脚本
|
||
用于验证 Hyperliquid 集成是否正常工作
|
||
|
||
⚠️ 警告：此脚本仅执行查询操作，不会执行任何交易
|
||
"""
|
||
import os
|
||
import sys
|
||
from typing import Dict, Any
|
||
|
||
# 添加项目路径
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
|
||
def test_sdk_import():
|
||
"""测试 SDK 导入"""
|
||
print("\n" + "="*60)
|
||
print("📦 测试 1: SDK 导入")
|
||
print("="*60)
|
||
|
||
try:
|
||
from hyperliquid.info import Info
|
||
from hyperliquid.exchange import Exchange
|
||
from eth_account import Account
|
||
print("✅ SDK 导入成功")
|
||
return True
|
||
except ImportError as e:
|
||
print(f"❌ SDK 导入失败: {e}")
|
||
print("\n请运行以下命令安装 SDK:")
|
||
print(" npx clawfi-hyperliquid-skill --wallet=YOUR_WALLET --key=YOUR_KEY")
|
||
return False
|
||
|
||
|
||
def test_env_variables():
|
||
"""测试环境变量"""
|
||
print("\n" + "="*60)
|
||
print("🔑 测试 2: 环境变量")
|
||
print("="*60)
|
||
|
||
wallet = os.getenv("CLAWFI_WALLET_ADDRESS")
|
||
private_key = os.getenv("CLAWFI_PRIVATE_KEY")
|
||
|
||
if not wallet:
|
||
print("❌ CLAWFI_WALLET_ADDRESS 未设置")
|
||
print("\n请运行以下命令设置环境变量:")
|
||
print(" npx clawfi-hyperliquid-skill --wallet=YOUR_WALLET --key=YOUR_KEY")
|
||
return False
|
||
|
||
if not private_key:
|
||
print("❌ CLAWFI_PRIVATE_KEY 未设置")
|
||
print("\n请运行以下命令设置环境变量:")
|
||
print(" npx clawfi-hyperliquid-skill --wallet=YOUR_WALLET --key=YOUR_KEY")
|
||
return False
|
||
|
||
print(f"✅ CLAWFI_WALLET_ADDRESS: {wallet}")
|
||
print(f"✅ CLAWFI_PRIVATE_KEY: {private_key[:10]}...{private_key[-4:]}")
|
||
return True
|
||
|
||
|
||
def test_connection(wallet: str, private_key: str):
|
||
"""测试连接和基础查询"""
|
||
print("\n" + "="*60)
|
||
print("🔗 测试 3: 连接 Hyperliquid")
|
||
print("="*60)
|
||
|
||
try:
|
||
from hyperliquid.info import Info
|
||
from hyperliquid.exchange import Exchange
|
||
from eth_account import Account
|
||
|
||
# 初始化
|
||
account = Account.from_key(private_key)
|
||
info = Info(base_url="https://api.hyperliquid.xyz", skip_ws=True)
|
||
exchange = Exchange(account, base_url="https://api.hyperliquid.xyz",
|
||
account_address=wallet)
|
||
|
||
print(f"✅ Agent 地址: {account.address}")
|
||
print(f"✅ 目标钱包: {wallet}")
|
||
|
||
# 测试查询账户状态
|
||
print("\n📊 查询账户状态...")
|
||
user_state = info.user_state(wallet)
|
||
|
||
if not user_state:
|
||
print("❌ 无法获取账户状态")
|
||
return False
|
||
|
||
# 账户价值
|
||
margin_summary = user_state.get("marginSummary", {})
|
||
account_value = float(margin_summary.get("accountValue", 0))
|
||
withdrawable = float(margin_summary.get("withdrawable", 0))
|
||
total_margin_used = float(margin_summary.get("totalMarginUsed", 0))
|
||
|
||
print(f"✅ 账户价值: ${account_value:,.2f}")
|
||
print(f"✅ 可提取: ${withdrawable:,.2f}")
|
||
print(f"✅ 已用保证金: ${total_margin_used:,.2f}")
|
||
|
||
# 查询持仓
|
||
print("\n📈 查询持仓...")
|
||
positions = user_state.get("assetPositions", [])
|
||
|
||
if not positions:
|
||
print("✅ 无持仓")
|
||
else:
|
||
open_positions = []
|
||
for pos in positions:
|
||
p = pos.get("position", {})
|
||
size = float(p.get("szi", 0))
|
||
if size != 0:
|
||
open_positions.append({
|
||
"coin": p.get("coin"),
|
||
"size": size,
|
||
"entry_px": float(p.get("entryPx", 0)),
|
||
"unrealized_pnl": float(p.get("unrealizedPnl", 0))
|
||
})
|
||
|
||
if not open_positions:
|
||
print("✅ 无活跃持仓")
|
||
else:
|
||
print(f"✅ 活跃持仓数: {len(open_positions)}")
|
||
for pos in open_positions:
|
||
side = "做多" if pos["size"] > 0 else "做空"
|
||
print(f" {pos['coin']}: {side} {abs(pos['size'])} @ ${pos['entry_px']:,.2f} | "
|
||
f"PnL: ${pos['unrealized_pnl']:,.2f}")
|
||
|
||
# 查询挂单
|
||
print("\n📋 查询挂单...")
|
||
open_orders = user_state.get("openOrders", [])
|
||
|
||
if not open_orders:
|
||
print("✅ 无挂单")
|
||
else:
|
||
print(f"✅ 挂单数: {len(open_orders)}")
|
||
for order in open_orders:
|
||
coin = order.get("coin")
|
||
side = order.get("side")
|
||
size = float(order.get("totalSz", 0))
|
||
limit_px = float(order.get("limitPx", 0))
|
||
print(f" {coin}: {side} {size} @ ${limit_px:,.2f}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 连接失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
def test_tick_size():
|
||
"""测试获取 tick size"""
|
||
print("\n" + "="*60)
|
||
print("📏 测试 4: 获取 Tick Size")
|
||
print("="*60)
|
||
|
||
try:
|
||
from hyperliquid.info import Info
|
||
|
||
info = Info(base_url="https://api.hyperliquid.xyz", skip_ws=True)
|
||
|
||
# 获取元数据
|
||
meta = info.meta()
|
||
universe = meta.get("universe", [])
|
||
|
||
print(f"✅ 可交易币种数: {len(universe)}")
|
||
|
||
# 显示前几个币种的 tick size
|
||
for asset in universe[:5]:
|
||
name = asset.get("name")
|
||
tick_size = float(asset.get("tickSize", 1.0))
|
||
sz_decimals = asset.get("szDecimals", 5)
|
||
print(f" {name}: tick_size={tick_size}, sz_decimals={sz_decimals}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 获取 tick size 失败: {e}")
|
||
return False
|
||
|
||
|
||
def test_service_initialization():
|
||
"""测试服务初始化"""
|
||
print("\n" + "="*60)
|
||
print("🔧 测试 5: 服务初始化")
|
||
print("="*60)
|
||
|
||
try:
|
||
from app.services.hyperliquid_trading_service import get_hyperliquid_service
|
||
|
||
# 尝试获取服务（如果未启用会返回 None）
|
||
service = get_hyperliquid_service()
|
||
|
||
if service is None:
|
||
print("⚠️ Hyperliquid 服务未启用（hyperliquid_trading_enabled=False）")
|
||
print(" 要启用服务，请在 .env 文件中设置: hyperliquid_trading_enabled=true")
|
||
return False
|
||
|
||
print("✅ Hyperliquid 服务初始化成功")
|
||
print(f" 钱包地址: {service.wallet_address}")
|
||
print(f" 最大杠杆: {service.max_total_leverage}x")
|
||
print(f" 熔断阈值: {service.circuit_breaker_drawdown * 100}%")
|
||
print(f" 单笔最大持仓: ${service.max_single_position}")
|
||
|
||
# 测试获取账户状态
|
||
print("\n📊 测试获取账户状态...")
|
||
state = service.get_account_state()
|
||
print(f"✅ 账户价值: ${state['account_value']:,.2f}")
|
||
print(f"✅ 可用余额: ${state['available_balance']:,.2f}")
|
||
|
||
# 测试获取持仓
|
||
print("\n📈 测试获取持仓...")
|
||
positions = service.get_open_positions()
|
||
if positions:
|
||
print(f"✅ 活跃持仓: {len(positions)} 个")
|
||
for pos in positions:
|
||
side = "做多" if pos["size"] > 0 else "做空"
|
||
print(f" {pos['coin']}: {side} {abs(pos['size'])}")
|
||
else:
|
||
print("✅ 无活跃持仓")
|
||
|
||
# 测试获取止盈止损价格
|
||
print("\n🛡️ 测试获取止盈止损...")
|
||
if positions:
|
||
for pos in positions:
|
||
tp_sl = service.get_tp_sl_prices(pos["coin"])
|
||
print(f" {pos['coin']}: TP={tp_sl['take_profit']}, SL={tp_sl['stop_loss']}")
|
||
else:
|
||
# 测试 BTC 的止盈止损（即使没有持仓）
|
||
tp_sl = service.get_tp_sl_prices("BTC")
|
||
print(f" BTC: TP={tp_sl['take_profit']}, SL={tp_sl['stop_loss']}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 服务初始化失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
def main():
|
||
"""主测试流程"""
|
||
print("\n" + "🦅"*30)
|
||
print(" Hyperliquid 集成测试")
|
||
print("🦅"*30)
|
||
|
||
results = {}
|
||
|
||
# 测试 1: SDK 导入
|
||
results["sdk_import"] = test_sdk_import()
|
||
if not results["sdk_import"]:
|
||
print("\n❌ SDK 导入失败，无法继续测试")
|
||
return False
|
||
|
||
# 测试 2: 环境变量
|
||
results["env_vars"] = test_env_variables()
|
||
if not results["env_vars"]:
|
||
print("\n❌ 环境变量未设置，无法继续测试")
|
||
return False
|
||
|
||
# 获取环境变量
|
||
wallet = os.getenv("CLAWFI_WALLET_ADDRESS")
|
||
private_key = os.getenv("CLAWFI_PRIVATE_KEY")
|
||
|
||
# 测试 3: 连接
|
||
results["connection"] = test_connection(wallet, private_key)
|
||
if not results["connection"]:
|
||
print("\n⚠️ 连接测试失败，但继续测试其他功能")
|
||
|
||
# 测试 4: Tick size
|
||
results["tick_size"] = test_tick_size()
|
||
|
||
# 测试 5: 服务初始化
|
||
results["service"] = test_service_initialization()
|
||
|
||
# 总结
|
||
print("\n" + "="*60)
|
||
print("📊 测试结果总结")
|
||
print("="*60)
|
||
|
||
for test_name, result in results.items():
|
||
status = "✅ 通过" if result else "❌ 失败"
|
||
print(f" {test_name}: {status}")
|
||
|
||
all_passed = all(results.values())
|
||
|
||
if all_passed:
|
||
print("\n🎉 所有测试通过！Hyperliquid 集成正常工作")
|
||
else:
|
||
print("\n⚠️ 部分测试失败，请检查上述错误信息")
|
||
|
||
return all_passed
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
success = main()
|
||
sys.exit(0 if success else 1)
|
||
except KeyboardInterrupt:
|
||
print("\n\n⚠️ 测试被用户中断")
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(f"\n\n❌ 测试过程中发生异常: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
sys.exit(1)
|