286 lines
8.1 KiB
Python
286 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
实盘交易功能测试脚本
|
||
|
||
测试 Bitget 实盘交易 API 的各项功能
|
||
⚠️ 注意: 此脚本会进行真实交易，请确保在测试网环境运行！
|
||
"""
|
||
import sys
|
||
import os
|
||
from pathlib import Path
|
||
|
||
# 添加项目路径
|
||
project_root = Path(__file__).parent.parent
|
||
sys.path.insert(0, str(project_root / "backend"))
|
||
|
||
from app.config import get_settings
|
||
from app.services.bitget_trading_api import BitgetTradingAPI, get_bitget_trading_api
|
||
from app.services.real_trading_service import get_real_trading_service
|
||
|
||
|
||
def print_section(title: str):
|
||
"""打印分节标题"""
|
||
print("\n" + "=" * 80)
|
||
print(f" {title}")
|
||
print("=" * 80)
|
||
|
||
|
||
def test_api_connection():
|
||
"""测试 API 连接"""
|
||
print_section("1. 测试 API 连接")
|
||
|
||
settings = get_settings()
|
||
|
||
print(f"\n配置信息:")
|
||
print(f" API Key: {settings.bitget_api_key[:10]}..." if settings.bitget_api_key else " API Key: 未配置")
|
||
print(f" 测试网: {'是 ✅' if settings.bitget_use_testnet else '否 ⚠️'}")
|
||
|
||
if not settings.bitget_api_key or not settings.bitget_api_secret:
|
||
print("\n❌ API Key 未配置，请在 .env 文件中设置:")
|
||
print(" BITGET_API_KEY=your_api_key")
|
||
print(" BITGET_API_SECRET=your_api_secret")
|
||
print(" BITGET_PASSPHRASE=your_passphrase")
|
||
print(" BITGET_USE_TESTNET=true")
|
||
return False
|
||
|
||
# 创建 API 实例
|
||
api = BitgetTradingAPI(
|
||
api_key=settings.bitget_api_key,
|
||
api_secret=settings.bitget_api_secret,
|
||
passphrase=settings.bitget_passphrase,
|
||
use_testnet=settings.bitget_use_testnet
|
||
)
|
||
|
||
# 测试连接
|
||
if api.test_connection():
|
||
print("\n✅ API 连接成功！")
|
||
return True
|
||
else:
|
||
print("\n❌ API 连接失败！")
|
||
print(" 请检查:")
|
||
print(" 1. API Key 和 Secret 是否正确")
|
||
print(" 2. 是否在测试网环境")
|
||
print(" 3. API 权限是否正确")
|
||
return False
|
||
|
||
|
||
def test_query_account():
|
||
"""测试查询账户"""
|
||
print_section("2. 测试查询账户")
|
||
|
||
api = get_bitget_trading_api()
|
||
if not api:
|
||
print("\n❌ 交易 API 未初始化")
|
||
return False
|
||
|
||
# 查询余额
|
||
print("\n查询账户余额...")
|
||
balance = api.get_balance()
|
||
|
||
if balance:
|
||
usdt_info = balance.get('USDT', {})
|
||
print(f"\n✅ USDT 余额:")
|
||
print(f" 可用: ${float(usdt_info.get('available', 0)):,.2f}")
|
||
print(f" 冻结: ${float(usdt_info.get('frozen', 0)):,.2f}")
|
||
print(f" 锁定: ${float(usdt_info.get('locked', 0)):,.2f}")
|
||
return True
|
||
else:
|
||
print("\n❌ 查询余额失败")
|
||
return False
|
||
|
||
|
||
def test_query_position():
|
||
"""测试查询持仓"""
|
||
print_section("3. 测试查询持仓")
|
||
|
||
api = get_bitget_trading_api()
|
||
if not api:
|
||
print("\n❌ 交易 API 未初始化")
|
||
return False
|
||
|
||
# 查询持仓
|
||
print("\n查询当前持仓...")
|
||
positions = api.get_position()
|
||
|
||
print(f"\n✅ 当前持仓数量: {len(positions)}")
|
||
|
||
if positions:
|
||
for pos in positions:
|
||
symbol = pos.get('symbol', 'N/A')
|
||
total = pos.get('total', 0)
|
||
available = pos.get('available', 0)
|
||
hold_side = pos.get('holdSide', 'N/A')
|
||
avg_price = pos.get('averageOpenPrice', 0)
|
||
|
||
print(f"\n {symbol} ({hold_side})")
|
||
print(f" 持仓: {total}")
|
||
print(f" 可用: {available}")
|
||
print(f" 均价: ${avg_price}")
|
||
else:
|
||
print(" 无持仓")
|
||
|
||
return True
|
||
|
||
|
||
def test_place_small_order():
|
||
"""
|
||
测试下单（小仓位测试）
|
||
|
||
⚠️ 警告: 此函数会进行真实交易！
|
||
建议在测试网运行，并使用最小仓位
|
||
"""
|
||
print_section("4. 测试下单 (⚠️ 实盘交易)")
|
||
|
||
# 确认是否要测试
|
||
print("\n⚠️ 警告: 此测试会进行真实下单！")
|
||
print(" 建议: 1. 确保在测试网环境")
|
||
print(" 2. 使用最小仓位测试")
|
||
print(" 3. 先手动撤销测试订单")
|
||
|
||
api = get_bitget_trading_api()
|
||
if not api:
|
||
print("\n❌ 交易 API 未初始化")
|
||
return False
|
||
|
||
# 使用 BTCUSDT 进行测试
|
||
symbol = "BTCUSDT"
|
||
|
||
# 查询当前价格
|
||
from app.services.bitget_service import bitget_service
|
||
current_price = bitget_service.get_current_price(symbol)
|
||
|
||
if not current_price:
|
||
print(f"\n❌ 无法获取 {symbol} 当前价格")
|
||
return False
|
||
|
||
print(f"\n{symbol} 当前价格: ${current_price:,.2f}")
|
||
|
||
# 使用最小数量（1张，通常约等于1 USDT）
|
||
test_size = 1
|
||
|
||
# 使用市价单测试（快速成交）
|
||
print(f"\n准备下测试单:")
|
||
print(f" 交易对: {symbol}")
|
||
print(f" 方向: 开多 (open_long)")
|
||
print(f" 类型: 市价单")
|
||
print(f" 数量: {test_size} 张 (约 ${test_size})")
|
||
|
||
# 取消注释以实际下单
|
||
print("\n❗ 下单功能已禁用，避免意外交易")
|
||
print(" 如需测试，请手动取消下面的注释并确认:")
|
||
print("")
|
||
|
||
"""
|
||
# 实际下单代码（已注释，需手动取消注释）
|
||
result = api.place_order(
|
||
symbol=symbol,
|
||
side='open_long',
|
||
order_type='market',
|
||
size=test_size
|
||
)
|
||
|
||
if result:
|
||
order_id = result.get('orderId')
|
||
print(f"\n✅ 下单成功!")
|
||
print(f" 订单ID: {order_id}")
|
||
print(f" 请在交易所检查订单状态")
|
||
|
||
# 询问是否立即撤单
|
||
print("\n是否立即撤单？(测试用)")
|
||
# input("按回车键撤单...")
|
||
|
||
# 撤单
|
||
if api.cancel_order(symbol, order_id=order_id):
|
||
print(f"\n✅ 撤单成功")
|
||
else:
|
||
print(f"\n❌ 撤单失败，请手动在交易所撤单")
|
||
|
||
return True
|
||
else:
|
||
print(f"\n❌ 下单失败")
|
||
return False
|
||
"""
|
||
|
||
return False
|
||
|
||
|
||
def test_real_trading_service():
|
||
"""测试实盘交易服务"""
|
||
print_section("5. 测试实盘交易服务")
|
||
|
||
service = get_real_trading_service()
|
||
|
||
if not service:
|
||
print("\n❌ 实盘交易服务未启用")
|
||
print(" 请在 .env 中设置:")
|
||
print(" REAL_TRADING_ENABLED=true")
|
||
return False
|
||
|
||
# 查询账户状态
|
||
print("\n查询账户状态...")
|
||
account = service.get_account_status()
|
||
|
||
print(f"\n✅ 账户状态:")
|
||
print(f" 总余额: ${account['current_balance']:,.2f}")
|
||
print(f" 可用: ${account['available']:,.2f}")
|
||
print(f" 已用: ${account['used_margin']:,.2f}")
|
||
print(f" 持仓: ${account['total_position_value']:,.2f}")
|
||
|
||
# 同步持仓
|
||
print("\n同步交易所持仓...")
|
||
positions = service.sync_positions_from_exchange()
|
||
|
||
print(f"\n✅ 交易所持仓: {len(positions)} 个")
|
||
|
||
return True
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("\n" + "🚀" * 40)
|
||
print("\nBitget 实盘交易功能测试")
|
||
print(f"测试时间: {datetime.now()}")
|
||
|
||
settings = get_settings()
|
||
|
||
print("\n⚠️ 重要提醒:")
|
||
print(f" 当前环境: {'测试网 ✅' if settings.bitget_use_testnet else '生产网 ⚠️'}")
|
||
|
||
try:
|
||
# 1. 测试 API 连接
|
||
if not test_api_connection():
|
||
print("\n❌ API 连接测试失败，请检查配置")
|
||
return
|
||
|
||
# 2. 测试查询账户
|
||
if not test_query_account():
|
||
print("\n❌ 查询账户测试失败")
|
||
return
|
||
|
||
# 3. 测试查询持仓
|
||
if not test_query_position():
|
||
print("\n❌ 查询持仓测试失败")
|
||
return
|
||
|
||
# 4. 测试下单（需要手动取消注释）
|
||
# test_place_small_order()
|
||
|
||
# 5. 测试实盘交易服务
|
||
if not test_real_trading_service():
|
||
print("\n⚠️ 实盘交易服务未启用（可忽略）")
|
||
|
||
print("\n" + "=" * 80)
|
||
print(" ✅ 所有测试通过!")
|
||
print("=" * 80)
|
||
print("\n实盘交易功能已就绪！")
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ 测试出错: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
from datetime import datetime
|
||
main()
|