370 lines
12 KiB
Python
370 lines
12 KiB
Python
"""
|
||
Hyperliquid 真实 API 集成测试
|
||
|
||
⚠️ 警告：此测试会使用真实 API 调用和真实订单！
|
||
- 使用最小下单量（ETH，szDecimals=3）
|
||
- 市价单会立即成交，产生实际盈亏
|
||
- 测试后自动清理所有订单和持仓
|
||
|
||
覆盖接口：
|
||
- 账户状态查询
|
||
- 杠杆设置
|
||
- 持仓查询
|
||
- 市价开仓
|
||
- 止盈止损设置（TP limit 单 + SL trigger 单）
|
||
- 止盈止损验证（读取挂单确认 TP 和 SL 都存在）
|
||
- 市价平仓
|
||
|
||
运行方式：
|
||
cd backend
|
||
python3 tests/test_hyperliquid_live_integration.py
|
||
"""
|
||
import os
|
||
import sys
|
||
import time
|
||
import traceback
|
||
from datetime import datetime
|
||
|
||
# 添加项目路径
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from dotenv import load_dotenv
|
||
load_dotenv(os.path.join(os.path.dirname(__file__), '..', '..', '.env'))
|
||
|
||
|
||
# ==================== 测试配置 ====================
|
||
|
||
TEST_SYMBOL = 'ETH' # ETH 精度好（szDecimals=3），手续费低
|
||
TEST_SIZE = 0.01 # 最小下单量
|
||
TEST_LEVERAGE = 5 # 测试杠杆倍数
|
||
|
||
|
||
class TestResult:
|
||
"""测试结果收集器"""
|
||
|
||
def __init__(self):
|
||
self.results = []
|
||
|
||
def record(self, name: str, passed: bool, detail: str = ""):
|
||
self.results.append((name, passed, detail))
|
||
status = "✅ PASS" if passed else "❌ FAIL"
|
||
print(f" {status}: {name}")
|
||
if detail:
|
||
print(f" {detail}")
|
||
|
||
def summary(self):
|
||
print(f"\n{'='*60}")
|
||
print("测试结果汇总")
|
||
print(f"{'='*60}")
|
||
passed = sum(1 for _, p, _ in self.results if p)
|
||
total = len(self.results)
|
||
for name, p, detail in self.results:
|
||
status = "✅" if p else "❌"
|
||
line = f" {status} {name}"
|
||
if not p and detail:
|
||
line += f" — {detail}"
|
||
print(line)
|
||
print(f"\n 总计: {passed}/{total} 通过")
|
||
print(f"{'='*60}")
|
||
return passed == total
|
||
|
||
|
||
# ==================== 测试函数 ====================
|
||
|
||
def test_account_state(service, r: TestResult):
|
||
"""查询账户状态"""
|
||
try:
|
||
state = service.get_account_state()
|
||
av = state['account_value']
|
||
ab = state['available_balance']
|
||
r.record(
|
||
"查询账户状态",
|
||
av > 0,
|
||
f"权益=${av:,.2f}, 可用=${ab:,.2f}"
|
||
)
|
||
except Exception as e:
|
||
r.record("查询账户状态", False, str(e))
|
||
|
||
|
||
def test_update_leverage(service, r: TestResult):
|
||
"""设置杠杆"""
|
||
try:
|
||
result = service.update_leverage(TEST_SYMBOL, TEST_LEVERAGE)
|
||
r.record("设置杠杆", True, f"{TEST_SYMBOL} → {TEST_LEVERAGE}x")
|
||
except Exception as e:
|
||
r.record("设置杠杆", False, str(e))
|
||
|
||
|
||
def test_get_positions(service, r: TestResult):
|
||
"""查询持仓"""
|
||
try:
|
||
positions = service.get_open_positions()
|
||
count = len(positions)
|
||
r.record("查询持仓", True, f"当前活跃持仓: {count} 个")
|
||
except Exception as e:
|
||
r.record("查询持仓", False, str(e))
|
||
|
||
|
||
def test_market_order_with_tp_sl(service, r: TestResult):
|
||
"""市价开仓 → 设置止盈止损 → 验证 → 平仓"""
|
||
opened = False
|
||
try:
|
||
# 0. 先清理已有持仓和挂单
|
||
try:
|
||
service.cancel_all_orders(TEST_SYMBOL)
|
||
except:
|
||
pass
|
||
try:
|
||
pos = service.get_position_for_symbol(TEST_SYMBOL)
|
||
if pos:
|
||
service.market_close_position(TEST_SYMBOL)
|
||
time.sleep(1)
|
||
except:
|
||
pass
|
||
|
||
# 1. 获取当前价格
|
||
all_mids = service.info.all_mids()
|
||
current_price = float(all_mids.get(TEST_SYMBOL, 0))
|
||
if current_price <= 0:
|
||
r.record("获取当前价格", False, f"无法获取 {TEST_SYMBOL} 价格")
|
||
return
|
||
|
||
print(f"\n 当前 {TEST_SYMBOL}: ${current_price:,.2f}")
|
||
|
||
# 2. 计算最小下单量
|
||
sz_decimals = service.get_sz_decimals(TEST_SYMBOL)
|
||
import math
|
||
size = max(math.floor(TEST_SIZE * (10 ** sz_decimals)) / (10 ** sz_decimals), 1 / (10 ** sz_decimals))
|
||
print(f" 下单量: {size} ({sz_decimals} 位精度)")
|
||
|
||
# 3. 设置杠杆
|
||
service.update_leverage(TEST_SYMBOL, TEST_LEVERAGE)
|
||
|
||
# 4. 市价开多
|
||
result = service.place_market_order(
|
||
symbol=TEST_SYMBOL,
|
||
is_buy=True,
|
||
size=size,
|
||
reduce_only=False
|
||
)
|
||
if not result.get('success'):
|
||
r.record("市价开仓", False, result.get('error', '未知错误'))
|
||
return
|
||
|
||
r.record("市价开仓", True, f"{TEST_SYMBOL} buy {size}")
|
||
opened = True
|
||
|
||
time.sleep(2)
|
||
|
||
# 5. 验证持仓
|
||
position = service.get_position_for_symbol(TEST_SYMBOL)
|
||
r.record("验证持仓存在", position is not None,
|
||
f"size={position['size']}, entry=${position['entry_price']:,.2f}" if position else "未找到持仓")
|
||
|
||
# 6. 设置止盈止损
|
||
tp_price = round(current_price * 1.02, 2) # +2%
|
||
sl_price = round(current_price * 0.98, 2) # -2%
|
||
print(f" 设置 TP=${tp_price:,.2f}, SL=${sl_price:,.2f}")
|
||
|
||
tp_sl_result = service.set_tp_sl(
|
||
symbol=TEST_SYMBOL,
|
||
is_long=True,
|
||
size=size,
|
||
tp_price=tp_price,
|
||
sl_price=sl_price
|
||
)
|
||
|
||
tp_set = tp_sl_result.get('tp_set', False)
|
||
sl_set = tp_sl_result.get('sl_set', False)
|
||
errors = tp_sl_result.get('errors', [])
|
||
|
||
detail = f"TP=${tp_price:,.2f}({'✅' if tp_set else '❌'}), SL=${sl_price:,.2f}({'✅' if sl_set else '❌'})"
|
||
if errors:
|
||
detail += f" errors={errors}"
|
||
r.record("设置止盈止损", tp_set and sl_set, detail)
|
||
|
||
# 7. 验证止盈止损挂单
|
||
time.sleep(1)
|
||
tp_sl_prices = service.get_tp_sl_prices(TEST_SYMBOL)
|
||
has_tp = tp_sl_prices.get('take_profit') is not None
|
||
has_sl = tp_sl_prices.get('stop_loss') is not None
|
||
r.record("验证 TP/SL 挂单", has_tp and has_sl,
|
||
f"TP={tp_sl_prices.get('take_profit')}({'✅' if has_tp else '❌'}), "
|
||
f"SL={tp_sl_prices.get('stop_loss')}({'✅' if has_sl else '❌'})")
|
||
|
||
# 8. 取消止盈止损
|
||
try:
|
||
service.cancel_tp_sl_orders(TEST_SYMBOL)
|
||
except:
|
||
pass
|
||
time.sleep(1)
|
||
|
||
# 9. 市价平仓
|
||
close_result = service.market_close_position(TEST_SYMBOL)
|
||
r.record("市价平仓", close_result.get('success', False),
|
||
close_result.get('error', f"成功"))
|
||
if close_result.get('success'):
|
||
opened = False
|
||
|
||
time.sleep(2)
|
||
|
||
# 10. 验证已平仓
|
||
position_after = service.get_position_for_symbol(TEST_SYMBOL)
|
||
r.record("验证已平仓", position_after is None)
|
||
|
||
except Exception as e:
|
||
r.record("市价单流程异常", False, f"{e}\n{traceback.format_exc()}")
|
||
finally:
|
||
if opened:
|
||
try:
|
||
service.cancel_all_orders(TEST_SYMBOL)
|
||
time.sleep(0.5)
|
||
service.market_close_position(TEST_SYMBOL)
|
||
print(" 🧹 已自动清理残留持仓")
|
||
except Exception as cleanup_err:
|
||
print(f" ⚠️ 清理失败，请手动检查: {cleanup_err}")
|
||
|
||
|
||
def test_set_tp_sl_partial_failure(service, r: TestResult):
|
||
"""测试 set_tp_sl: 第一个失败不影响第二个"""
|
||
# 这个测试验证我们的修复：独立的 try-except
|
||
# 如果 TP 失败（例如价格为 0），SL 应该仍然被设置
|
||
opened = False
|
||
try:
|
||
# 先清理
|
||
try:
|
||
service.cancel_all_orders(TEST_SYMBOL)
|
||
except:
|
||
pass
|
||
pos = service.get_position_for_symbol(TEST_SYMBOL)
|
||
if pos:
|
||
service.market_close_position(TEST_SYMBOL)
|
||
time.sleep(1)
|
||
|
||
# 1. 市价开仓
|
||
sz_decimals = service.get_sz_decimals(TEST_SYMBOL)
|
||
import math
|
||
size = max(math.floor(TEST_SIZE * (10 ** sz_decimals)) / (10 ** sz_decimals), 1 / (10 ** sz_decimals))
|
||
service.update_leverage(TEST_SYMBOL, TEST_LEVERAGE)
|
||
|
||
result = service.place_market_order(
|
||
symbol=TEST_SYMBOL,
|
||
is_buy=True,
|
||
size=size,
|
||
reduce_only=False
|
||
)
|
||
if not result.get('success'):
|
||
r.record("部分失败测试: 开仓", False, result.get('error', '未知'))
|
||
return
|
||
opened = True
|
||
time.sleep(2)
|
||
|
||
# 2. 设置一个有效的 SL（但故意不设置 TP → tp_price=None）
|
||
all_mids = service.info.all_mids()
|
||
current_price = float(all_mids.get(TEST_SYMBOL, 0))
|
||
sl_price = round(current_price * 0.98, 2)
|
||
|
||
tp_sl_result = service.set_tp_sl(
|
||
symbol=TEST_SYMBOL,
|
||
is_long=True,
|
||
size=size,
|
||
tp_price=None, # 不设 TP
|
||
sl_price=sl_price # 只设 SL
|
||
)
|
||
sl_set = tp_sl_result.get('sl_set', False)
|
||
r.record("部分设置测试 (仅 SL)", sl_set, f"sl_set={sl_set}, errors={tp_sl_result.get('errors', [])}")
|
||
|
||
# 3. 验证 SL 挂单存在
|
||
time.sleep(1)
|
||
tp_sl_prices = service.get_tp_sl_prices(TEST_SYMBOL)
|
||
has_sl = tp_sl_prices.get('stop_loss') is not None
|
||
r.record("验证 SL 挂单", has_sl, f"SL={tp_sl_prices.get('stop_loss')}")
|
||
|
||
# 4. 清理
|
||
service.cancel_all_orders(TEST_SYMBOL)
|
||
time.sleep(1)
|
||
service.market_close_position(TEST_SYMBOL)
|
||
opened = False
|
||
|
||
except Exception as e:
|
||
r.record("部分失败测试异常", False, f"{e}\n{traceback.format_exc()}")
|
||
finally:
|
||
if opened:
|
||
try:
|
||
service.cancel_all_orders(TEST_SYMBOL)
|
||
time.sleep(0.5)
|
||
service.market_close_position(TEST_SYMBOL)
|
||
print(" 🧹 已自动清理残留持仓")
|
||
except:
|
||
pass
|
||
|
||
|
||
# ==================== 主入口 ====================
|
||
|
||
def main():
|
||
print(f"\n{'='*60}")
|
||
print(f" Hyperliquid 实盘接口集成测试")
|
||
print(f"{'='*60}")
|
||
print(f" 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f" 交易对: {TEST_SYMBOL}")
|
||
print(f" 下单量: {TEST_SIZE}")
|
||
print(f" 杠杆: {TEST_LEVERAGE}x")
|
||
print(f"{'='*60}")
|
||
|
||
r = TestResult()
|
||
|
||
# 初始化
|
||
try:
|
||
from app.services.hyperliquid_trading_service import HyperliquidTradingService
|
||
service = HyperliquidTradingService()
|
||
print(f" 钱包: {service.wallet_address[:10]}...")
|
||
except Exception as e:
|
||
print(f"\n❌ 初始化失败: {e}")
|
||
traceback.print_exc()
|
||
sys.exit(1)
|
||
|
||
# ---- 基础测试 ----
|
||
print(f"\n{'─'*40}")
|
||
print(" 基础接口")
|
||
print(f"{'─'*40}")
|
||
|
||
test_account_state(service, r)
|
||
time.sleep(0.3)
|
||
|
||
test_update_leverage(service, r)
|
||
time.sleep(0.3)
|
||
|
||
test_get_positions(service, r)
|
||
time.sleep(0.3)
|
||
|
||
# ---- 核心测试: 开仓 → TP/SL → 平仓 ----
|
||
print(f"\n{'─'*40}")
|
||
print(" 开仓 → 止盈止损 → 验证 → 平仓")
|
||
print(f"{'─'*40}")
|
||
|
||
test_market_order_with_tp_sl(service, r)
|
||
time.sleep(1)
|
||
|
||
# ---- 边界测试: 部分设置 ----
|
||
print(f"\n{'─'*40}")
|
||
print(" 部分设置测试")
|
||
print(f"{'─'*40}")
|
||
|
||
test_set_tp_sl_partial_failure(service, r)
|
||
|
||
# ---- 汇总 ----
|
||
all_passed = r.summary()
|
||
sys.exit(0 if all_passed else 1)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
print("\n⚠️ 此测试会产生真实订单和手续费！")
|
||
print(f" 使用 {TEST_SYMBOL} 最小量 {TEST_SIZE}")
|
||
|
||
confirm = input("\n是否继续？(yes/no): ")
|
||
if confirm.strip().lower() != 'yes':
|
||
print("已取消")
|
||
sys.exit(0)
|
||
|
||
main()
|