148 lines
4.4 KiB
Python
148 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试 Bitget 集成功能
|
|
|
|
验证切换到 Bitget 后的各项功能是否正常
|
|
"""
|
|
import sys
|
|
import os
|
|
from pathlib import Path
|
|
|
|
# 添加项目路径
|
|
project_root = Path(__file__).parent.parent
|
|
sys.path.insert(0, str(project_root / "backend"))
|
|
|
|
from app.services.bitget_service import bitget_service
|
|
|
|
|
|
def test_basic_functions():
|
|
"""测试基本功能"""
|
|
print("\n" + "=" * 80)
|
|
print("测试 Bitget 基本功能")
|
|
print("=" * 80)
|
|
|
|
symbol = "BTCUSDT"
|
|
|
|
# 1. 测试获取当前价格
|
|
print(f"\n1. 获取当前价格 ({symbol})...")
|
|
price = bitget_service.get_current_price(symbol)
|
|
if price:
|
|
print(f" ✅ 当前价格: ${price:,.2f}")
|
|
else:
|
|
print(f" ❌ 获取失败")
|
|
return False
|
|
|
|
# 2. 测试获取 K线数据
|
|
print(f"\n2. 获取 K线数据 ({symbol} 5m)...")
|
|
klines = bitget_service.get_klines(symbol, '5m', limit=10)
|
|
if not klines.empty:
|
|
print(f" ✅ 获取 {len(klines)} 根 K线")
|
|
print(f" 最新: 开${klines.iloc[-1]['open']:,.2f} "
|
|
f"高${klines.iloc[-1]['high']:,.2f} "
|
|
f"低${klines.iloc[-1]['low']:,.2f} "
|
|
f"收${klines.iloc[-1]['close']:,.2f}")
|
|
else:
|
|
print(f" ❌ 获取失败")
|
|
return False
|
|
|
|
# 3. 测试获取多周期数据
|
|
print(f"\n3. 获取多周期数据 ({symbol})...")
|
|
multi_data = bitget_service.get_multi_timeframe_data(symbol)
|
|
if multi_data:
|
|
print(f" ✅ 获取成功")
|
|
for interval, df in multi_data.items():
|
|
if not df.empty:
|
|
print(f" {interval}: {len(df)}根, 最新价 ${df.iloc[-1]['close']:,.2f}")
|
|
else:
|
|
print(f" ❌ 获取失败")
|
|
return False
|
|
|
|
# 4. 测试获取资金费率
|
|
print(f"\n4. 获取资金费率 ({symbol})...")
|
|
funding = bitget_service.get_funding_rate(symbol)
|
|
if funding:
|
|
print(f" ✅ 资金费率: {funding['funding_rate_percent']:.4f}%")
|
|
print(f" 标记价格: ${funding['mark_price']:,.2f}")
|
|
print(f" 指数价格: ${funding['index_price']:,.2f}")
|
|
print(f" 市场情绪: {funding['sentiment']}")
|
|
else:
|
|
print(f" ❌ 获取失败")
|
|
return False
|
|
|
|
# 5. 测试获取合约市场数据
|
|
print(f"\n5. 获取合约市场数据 ({symbol})...")
|
|
futures_data = bitget_service.get_futures_market_data(symbol)
|
|
if futures_data:
|
|
print(f" ✅ 获取成功")
|
|
print(f" 溢价率: {futures_data['premium_rate']:.2f}%")
|
|
print(f" 市场情绪: {futures_data['market_sentiment']}")
|
|
else:
|
|
print(f" ❌ 获取失败")
|
|
return False
|
|
|
|
# 6. 测试格式化数据供 LLM
|
|
print(f"\n6. 格式化合约数据供 LLM...")
|
|
formatted = bitget_service.format_futures_data_for_llm(symbol, futures_data)
|
|
if formatted:
|
|
print(f" ✅ 格式化成功")
|
|
print(f" 预览:\n{formatted[:200]}...")
|
|
else:
|
|
print(f" ❌ 格式化失败")
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def test_multiple_symbols():
|
|
"""测试多个交易对"""
|
|
print("\n" + "=" * 80)
|
|
print("测试多个交易对")
|
|
print("=" * 80)
|
|
|
|
symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
|
|
|
|
for symbol in symbols:
|
|
print(f"\n{symbol}:")
|
|
price = bitget_service.get_current_price(symbol)
|
|
if price:
|
|
print(f" ✅ ${price:,.2f}")
|
|
else:
|
|
print(f" ❌ 获取失败")
|
|
|
|
|
|
def main():
|
|
"""主函数"""
|
|
print("\n" + "🚀" * 40)
|
|
print("\nBitget 集成功能测试")
|
|
print(f"测试时间: {pd.Timestamp.now()}")
|
|
|
|
try:
|
|
# 基本功能测试
|
|
if not test_basic_functions():
|
|
print("\n❌ 基本功能测试失败")
|
|
return
|
|
|
|
# 多交易对测试
|
|
test_multiple_symbols()
|
|
|
|
print("\n" + "=" * 80)
|
|
print(" ✅ 所有测试通过!")
|
|
print("=" * 80)
|
|
print("\nBitget 已成功集成，可以正常使用！")
|
|
print("\n切换总结:")
|
|
print(" ✅ crypto_agent.py -> 使用 Bitget")
|
|
print(" ✅ llm_signal_analyzer.py -> 使用 Bitget")
|
|
print(" ✅ paper_trading_service.py -> 使用 Bitget")
|
|
print(" ✅ main.py -> 使用 Bitget")
|
|
print(" ✅ api/paper_trading.py -> 使用 Bitget")
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ 测试出错: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import pandas as pd
|
|
main()
|