626 lines
22 KiB
Python
626 lines
22 KiB
Python
"""
|
||
Bitget 真实 API 集成测试
|
||
|
||
⚠️ 警告：此测试会使用真实 API 调用和真实订单！
|
||
- 确保使用测试网或接受小额手续费
|
||
- 市价单会立即成交，产生实际盈亏
|
||
- 测试后检查是否有残留订单/持仓
|
||
|
||
运行方式：
|
||
cd backend
|
||
source venv/bin/activate
|
||
python3 tests/test_bitget_live_integration.py
|
||
"""
|
||
import os
|
||
import sys
|
||
import time
|
||
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('/Users/aaron/source_code/Stock_Agent/.env')
|
||
|
||
from app.services.bitget_trading_api_sdk import BitgetTradingAPI
|
||
from app.services.bitget_live_trading_service import BitgetLiveTradingService
|
||
|
||
|
||
class BitgetIntegrationTest:
|
||
"""Bitget 真实 API 集成测试"""
|
||
|
||
def __init__(self):
|
||
self.api_key = os.getenv('BITGET_API_KEY')
|
||
self.api_secret = os.getenv('BITGET_API_SECRET')
|
||
self.passphrase = os.getenv('BITGET_PASSPHRASE')
|
||
self.use_testnet = os.getenv('BITGET_USE_TESTNET', 'true').lower() == 'true'
|
||
|
||
if not all([self.api_key, self.api_secret, self.passphrase]):
|
||
raise ValueError("❌ 请在 .env 中配置 BITGET_API_KEY, BITGET_API_SECRET, BITGET_PASSPHRASE")
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f"Bitget 真实 API 集成测试")
|
||
print(f"{'='*60}")
|
||
print(f"API Key: {self.api_key[:10]}...")
|
||
print(f"测试网模式: {self.use_testnet}")
|
||
print(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f"{'='*60}\n")
|
||
|
||
# 初始化 API
|
||
self.api = BitgetTradingAPI(
|
||
api_key=self.api_key,
|
||
api_secret=self.api_secret,
|
||
passphrase=self.passphrase,
|
||
use_testnet=self.use_testnet
|
||
)
|
||
|
||
# 初始化 Service
|
||
self.service = BitgetLiveTradingService.__new__(BitgetLiveTradingService)
|
||
self.service.trading_api = self.api
|
||
self.service.max_single_position = 100
|
||
self.service.max_total_leverage = 5
|
||
self.service.circuit_breaker_drawdown = 0.10
|
||
self.service._initial_account_state = None
|
||
self.service.initial_balance = None # 添加这个属性
|
||
|
||
# 测试状态
|
||
self.limit_order_id = None
|
||
self.limit_order_symbol = None
|
||
self.market_entry_price = 0.0
|
||
|
||
# ==================== 基础连接测试 ====================
|
||
|
||
def test_01_connection(self):
|
||
"""测试 1: API 连接"""
|
||
print("\n📡 测试 1: API 连接")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
# 获取服务器时间
|
||
server_time = self.api.exchange.fetch_time()
|
||
print(f"✅ 服务器时间: {datetime.fromtimestamp(server_time/1000)}")
|
||
|
||
# 加载市场信息
|
||
markets = self.api.exchange.load_markets()
|
||
btc_market = markets.get('BTC/USDT:USDT', {})
|
||
print(f"✅ BTC 市场ID: {btc_market.get('id', 'N/A')}")
|
||
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ 连接失败: {e}")
|
||
return False
|
||
|
||
def test_02_get_balance(self):
|
||
"""测试 2: 获取账户余额"""
|
||
print("\n💰 测试 2: 获取账户余额")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
balance = self.api.get_balance()
|
||
if balance:
|
||
usdt = balance.get('USDT', {})
|
||
print(f"✅ USDT 可用: {usdt.get('free', 0):.2f}")
|
||
print(f"✅ USDT 冻结: {usdt.get('used', 0):.2f}")
|
||
print(f"✅ USDT 总额: {usdt.get('total', 0):.2f}")
|
||
return True
|
||
else:
|
||
print("❌ 余额为空")
|
||
return False
|
||
except Exception as e:
|
||
print(f"❌ 获取余额失败: {e}")
|
||
return False
|
||
|
||
def test_03_get_positions(self):
|
||
"""测试 3: 获取当前持仓"""
|
||
print("\n📊 测试 3: 获取当前持仓")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
positions = self.api.get_position()
|
||
if positions:
|
||
active_positions = [p for p in positions if float(p.get('contracts', 0)) > 0]
|
||
print(f"✅ 当前持仓数: {len(active_positions)}")
|
||
for pos in active_positions:
|
||
symbol = pos.get('symbol', 'N/A')
|
||
contracts = float(pos.get('contracts', 0))
|
||
side = pos.get('side', 'N/A')
|
||
entry = float(pos.get('entryPrice', 0))
|
||
print(f" - {symbol}: {side} {contracts} 张 @ ${entry:,.2f}")
|
||
else:
|
||
print("✅ 当前无持仓")
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ 获取持仓失败: {e}")
|
||
return False
|
||
|
||
def test_04_set_leverage(self):
|
||
"""测试 4: 设置杠杆"""
|
||
print("\n⚙️ 测试 4: 设置杠杆")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
symbol = 'BTCUSDT'
|
||
leverage = 5
|
||
self.api.set_leverage(symbol, leverage)
|
||
print(f"✅ {symbol} 杠杆已设置为 {leverage}x")
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ 设置杠杆失败: {e}")
|
||
return False
|
||
|
||
# ==================== 限价单测试 ====================
|
||
|
||
def test_05_place_limit_order(self):
|
||
"""测试 5: 下限价单（挂单，预期不成交）"""
|
||
print("\n📝 测试 5: 下限价单")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
symbol = 'BTCUSDT'
|
||
|
||
# 获取当前价格
|
||
ticker = self.api.exchange.fetch_ticker('BTC/USDT:USDT')
|
||
current_price = ticker['last']
|
||
print(f" 当前 BTC 价格: ${current_price:,.2f}")
|
||
|
||
# 计算挂单价格（当前价格 - 5%，预期不会成交）
|
||
limit_price = current_price * 0.95
|
||
print(f" 挂单价格: ${limit_price:,.2f} (低 5%，预期不成交)")
|
||
|
||
# 下 1 张多单（最小单位）
|
||
order = self.api.place_order(
|
||
symbol=symbol,
|
||
side='buy',
|
||
order_type='limit',
|
||
size=1, # 1 张 = 0.01 BTC
|
||
price=limit_price
|
||
)
|
||
|
||
if order:
|
||
self.limit_order_id = order.get('id')
|
||
self.limit_order_symbol = symbol
|
||
print(f"✅ 限价单已下: {self.limit_order_id}")
|
||
print(f" 订单状态: {order.get('status')}")
|
||
return True
|
||
else:
|
||
print("❌ 下单返回空")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 下限价单失败: {e}")
|
||
return False
|
||
|
||
def test_06_get_open_orders(self):
|
||
"""测试 6: 查询挂单"""
|
||
print("\n📋 测试 6: 查询挂单")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
symbol = self.limit_order_symbol or 'BTCUSDT'
|
||
orders = self.api.get_open_orders(symbol)
|
||
|
||
if orders:
|
||
print(f"✅ 当前挂单数: {len(orders)}")
|
||
for order in orders:
|
||
order_id = order.get('id', 'N/A')
|
||
side = order.get('side', 'N/A')
|
||
price = order.get('price', 0)
|
||
amount = order.get('amount', 0)
|
||
print(f" - {order_id}: {side} {amount} @ ${price:,.2f}")
|
||
else:
|
||
print("⚠️ 无挂单")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 查询挂单失败: {e}")
|
||
return False
|
||
|
||
def test_07_cancel_order(self):
|
||
"""测试 7: 撤销挂单"""
|
||
print("\n❌ 测试 7: 撤销挂单")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
if not self.limit_order_id:
|
||
print("⚠️ 没有需要撤销的订单")
|
||
return True
|
||
|
||
# 使用 cancel_all_orders 撤销该 symbol 所有挂单
|
||
success = self.api.cancel_all_orders(self.limit_order_symbol)
|
||
|
||
if success:
|
||
print(f"✅ 已撤销 {self.limit_order_symbol} 的所有挂单")
|
||
self.limit_order_id = None
|
||
return True
|
||
else:
|
||
print(f"❌ 撤单失败")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 撤销订单失败: {e}")
|
||
return False
|
||
|
||
# ==================== 市价单测试（真实成交！）====================
|
||
|
||
def test_08_market_order_open(self):
|
||
"""测试 8: 市价开多单（真实成交！）"""
|
||
print("\n⚡ 测试 8: 市价开多单")
|
||
print("-" * 40)
|
||
print("⚠️ 警告：此测试会产生真实订单和手续费！")
|
||
|
||
try:
|
||
symbol = 'BTCUSDT'
|
||
|
||
# 获取当前价格
|
||
ticker = self.api.exchange.fetch_ticker('BTC/USDT:USDT')
|
||
current_price = ticker['last']
|
||
print(f" 当前 BTC 价格: ${current_price:,.2f}")
|
||
|
||
# 下 1 张市价多单（最小单位 = 0.01 BTC）
|
||
print(f" 下市价多单: 1 张 (0.01 BTC)...")
|
||
order = self.api.place_order(
|
||
symbol=symbol,
|
||
side='buy',
|
||
order_type='market',
|
||
size=1 # 1 张 = 0.01 BTC
|
||
)
|
||
|
||
if order:
|
||
order_id = order.get('id')
|
||
status = order.get('status')
|
||
# 安全获取 average 价格
|
||
avg_price = order.get('average') or order.get('price') or 0
|
||
if avg_price:
|
||
avg_price = float(avg_price)
|
||
print(f"✅ 市价单已成交: {order_id}")
|
||
print(f" 订单状态: {status}")
|
||
print(f" 成交均价: ${avg_price:,.2f}")
|
||
print(f" 成交数量: {order.get('filled', 0)} 张")
|
||
|
||
# 保存入场价
|
||
self.market_entry_price = avg_price
|
||
return True
|
||
else:
|
||
print("❌ 下单返回空")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 市价开仓失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
def test_09_verify_position_opened(self):
|
||
"""测试 9: 验证持仓已开启"""
|
||
print("\n🔍 测试 9: 验证持仓")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
positions = self.api.get_position('BTCUSDT')
|
||
|
||
if positions:
|
||
for pos in positions:
|
||
contracts = float(pos.get('contracts', 0))
|
||
if contracts > 0:
|
||
side = pos.get('side', 'N/A')
|
||
entry_price = float(pos.get('entryPrice', 0))
|
||
unrealized_pnl = float(pos.get('unrealizedPnl', 0))
|
||
print(f"✅ 持仓已确认:")
|
||
print(f" 方向: {side}")
|
||
print(f" 数量: {contracts} 张")
|
||
print(f" 开仓价: ${entry_price:,.2f}")
|
||
print(f" 未实现盈亏: ${unrealized_pnl:,.2f}")
|
||
return True
|
||
|
||
print("❌ 未找到持仓")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 验证持仓失败: {e}")
|
||
return False
|
||
|
||
def test_10_set_tp_sl(self):
|
||
"""测试 10: 设置止盈止损"""
|
||
print("\n🎯 测试 10: 设置止盈止损")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
# 获取当前价格
|
||
ticker = self.api.exchange.fetch_ticker('BTC/USDT:USDT')
|
||
current_price = ticker['last']
|
||
|
||
# 设置止盈（+1%）和止损（-0.5%）
|
||
tp_price = current_price * 1.01 # +1%
|
||
sl_price = current_price * 0.995 # -0.5%
|
||
|
||
print(f" 当前价格: ${current_price:,.2f}")
|
||
print(f" 设置止盈: ${tp_price:,.2f} (+1%)")
|
||
print(f" 设置止损: ${sl_price:,.2f} (-0.5%)")
|
||
|
||
result = self.api.modify_sl_tp(
|
||
symbol='BTCUSDT',
|
||
stop_loss=sl_price,
|
||
take_profit=tp_price
|
||
)
|
||
|
||
if result:
|
||
print(f"✅ 止盈止损已设置")
|
||
else:
|
||
print(f"⚠️ 止盈止损设置返回 False")
|
||
|
||
return True # 无论成功与否都继续测试
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ 设置止盈止损失败: {e}")
|
||
return True # 不算失败，继续测试
|
||
|
||
def test_11_market_order_close(self):
|
||
"""测试 11: 市价平仓"""
|
||
print("\n🔒 测试 11: 市价平仓")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
# 获取当前价格
|
||
ticker = self.api.exchange.fetch_ticker('BTC/USDT:USDT')
|
||
current_price = ticker['last']
|
||
print(f" 当前 BTC 价格: ${current_price:,.2f}")
|
||
|
||
# 先撤销止盈止损单
|
||
try:
|
||
self.api.cancel_all_orders('BTCUSDT')
|
||
print(" 已撤销止盈止损单")
|
||
except:
|
||
pass
|
||
|
||
# 使用 close_position 方法平仓
|
||
print(f" 下市价平仓单...")
|
||
order = self.api.close_position(
|
||
symbol='BTCUSDT',
|
||
side='sell', # 平多
|
||
size=0.01, # 0.01 张
|
||
)
|
||
|
||
if order:
|
||
order_id = order.get('id')
|
||
avg_price = order.get('average') or order.get('price') or 0
|
||
if avg_price:
|
||
avg_price = float(avg_price)
|
||
print(f"✅ 平仓成功: {order_id}")
|
||
print(f" 平仓均价: ${avg_price:,.2f}")
|
||
|
||
# 计算盈亏
|
||
if self.market_entry_price > 0 and avg_price > 0:
|
||
pnl = (avg_price - self.market_entry_price) * 0.01 # 1张 = 0.01 BTC
|
||
pnl_pct = (avg_price - self.market_entry_price) / self.market_entry_price * 100
|
||
print(f" 实现盈亏: ${pnl:,.2f} ({pnl_pct:+.2f}%)")
|
||
|
||
return True
|
||
else:
|
||
print("❌ 平仓返回空")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 市价平仓失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
def test_12_verify_position_closed(self):
|
||
"""测试 12: 验证持仓已平"""
|
||
print("\n✅ 测试 12: 验证持仓已平")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
positions = self.api.get_position('BTCUSDT')
|
||
|
||
has_position = False
|
||
if positions:
|
||
for pos in positions:
|
||
if float(pos.get('contracts', 0)) > 0:
|
||
has_position = True
|
||
print(f"⚠️ 仍有持仓: {pos.get('contracts')} 张")
|
||
|
||
if not has_position:
|
||
print("✅ 持仓已全部平仓")
|
||
return True
|
||
else:
|
||
print("❌ 持仓未完全平仓")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 验证平仓失败: {e}")
|
||
return False
|
||
|
||
# ==================== Service 层测试 ====================
|
||
|
||
def test_13_service_get_account_state(self):
|
||
"""测试 13: Service 层获取账户状态"""
|
||
print("\n🏦 测试 13: Service 获取账户状态")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
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}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ 获取账户状态失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
def test_14_service_get_positions(self):
|
||
"""测试 14: Service 层获取持仓"""
|
||
print("\n📈 测试 14: Service 获取持仓")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
positions = self.service.get_open_positions()
|
||
if positions:
|
||
print(f"✅ 当前持仓数: {len(positions)}")
|
||
for pos in positions:
|
||
# 安全获取字段
|
||
symbol = pos.get('symbol', 'N/A')
|
||
side = pos.get('side', 'N/A')
|
||
size = pos.get('size', 0)
|
||
entry = pos.get('entry_price', 0)
|
||
print(f" - {symbol}: {side} {size} @ ${entry:,.2f}")
|
||
else:
|
||
print("✅ 当前无持仓")
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ 获取持仓失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
def test_15_service_contract_size(self):
|
||
"""测试 15: Service 合约面值"""
|
||
print("\n📐 测试 15: 合约面值换算")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
# 测试已知币种
|
||
btc_size = self.service.get_contract_size('BTC')
|
||
eth_size = self.service.get_contract_size('ETH')
|
||
sol_size = self.service.get_contract_size('SOL')
|
||
|
||
print(f"✅ BTC 合约面值: {btc_size} BTC/张")
|
||
print(f"✅ ETH 合约面值: {eth_size} ETH/张")
|
||
print(f"✅ SOL 合约面值: {sol_size} SOL/张")
|
||
|
||
# 测试币数转合约数
|
||
contracts = self.service.coins_to_contracts('BTC', 0.05) # 0.05 BTC
|
||
print(f"✅ 0.05 BTC = {contracts} 张")
|
||
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ 合约面值测试失败: {e}")
|
||
return False
|
||
|
||
def test_16_risk_check(self):
|
||
"""测试 16: 风控检查"""
|
||
print("\n⚠️ 测试 16: 风控检查")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
result = self.service.check_risk_limits()
|
||
if result['allowed']:
|
||
print(f"✅ 风控检查通过")
|
||
else:
|
||
print(f"⚠️ 风控检查未通过: {result['reason']}")
|
||
return True
|
||
except Exception as e:
|
||
print(f"⚠️ 风控检查异常: {e}")
|
||
return True # 不算失败
|
||
|
||
# ==================== 清理和运行 ====================
|
||
|
||
def cleanup(self):
|
||
"""清理测试产生的订单和持仓"""
|
||
print("\n🧹 清理测试订单...")
|
||
print("-" * 40)
|
||
|
||
try:
|
||
# 撤销所有 BTC 挂单
|
||
success = self.api.cancel_all_orders('BTCUSDT')
|
||
if success:
|
||
print("✅ 已撤销所有 BTC 挂单")
|
||
else:
|
||
print("⚠️ 撤销 BTC 挂单返回 False（可能无挂单）")
|
||
|
||
# 检查是否有残留持仓
|
||
positions = self.api.get_position('BTCUSDT')
|
||
if positions:
|
||
for pos in positions:
|
||
contracts = float(pos.get('contracts', 0))
|
||
if contracts > 0:
|
||
symbol = pos.get('symbol', '')
|
||
print(f"⚠️ 发现残留持仓: {symbol} {contracts} 张")
|
||
print(f" 请手动平仓！")
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ 清理失败: {e}")
|
||
|
||
def run_all_tests(self):
|
||
"""运行所有测试"""
|
||
tests = [
|
||
self.test_01_connection,
|
||
self.test_02_get_balance,
|
||
self.test_03_get_positions,
|
||
self.test_04_set_leverage,
|
||
self.test_05_place_limit_order,
|
||
self.test_06_get_open_orders,
|
||
self.test_07_cancel_order,
|
||
self.test_08_market_order_open,
|
||
self.test_09_verify_position_opened,
|
||
self.test_10_set_tp_sl,
|
||
self.test_11_market_order_close,
|
||
self.test_12_verify_position_closed,
|
||
self.test_13_service_get_account_state,
|
||
self.test_14_service_get_positions,
|
||
self.test_15_service_contract_size,
|
||
self.test_16_risk_check,
|
||
]
|
||
|
||
results = []
|
||
for test in tests:
|
||
try:
|
||
result = test()
|
||
results.append((test.__name__, result))
|
||
time.sleep(0.5) # 避免触发频率限制
|
||
except Exception as e:
|
||
print(f"\n❌ 测试异常: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
results.append((test.__name__, False))
|
||
|
||
# 清理
|
||
self.cleanup()
|
||
|
||
# 汇总
|
||
print("\n" + "=" * 60)
|
||
print("测试结果汇总")
|
||
print("=" * 60)
|
||
|
||
passed = sum(1 for _, r in results if r)
|
||
total = len(results)
|
||
|
||
for name, result in results:
|
||
status = "✅ PASS" if result else "❌ FAIL"
|
||
print(f"{status}: {name}")
|
||
|
||
print(f"\n总计: {passed}/{total} 通过")
|
||
print("=" * 60)
|
||
|
||
return passed == total
|
||
|
||
|
||
if __name__ == '__main__':
|
||
print("\n" + "!" * 60)
|
||
print("⚠️ 警告：此测试将使用真实 API 调用！")
|
||
print("!" * 60)
|
||
print("请确认：")
|
||
print("1. 你已了解这会产生真实 API 调用")
|
||
print("2. 你使用的是测试网或接受小额手续费")
|
||
print("3. 市价单会立即成交，产生实际盈亏")
|
||
print("4. 测试后请检查是否有残留订单/持仓")
|
||
print("!" * 60)
|
||
|
||
confirm = input("\n是否继续？(yes/no): ")
|
||
|
||
if confirm.lower() != 'yes':
|
||
print("已取消测试")
|
||
sys.exit(0)
|
||
|
||
try:
|
||
tester = BitgetIntegrationTest()
|
||
success = tester.run_all_tests()
|
||
sys.exit(0 if success else 1)
|
||
except Exception as e:
|
||
print(f"\n❌ 测试失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
sys.exit(1)
|