lead-lag-scripts/test_database.py

#!/usr/bin/env python3
"""
测试数据库功能
"""

import sys
import time
import random
from pathlib import Path

# 添加项目路径到Python路径
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root / "src"))

from leadlag.database import TradingDatabase

def test_database():
    """测试数据库功能"""
    print("🧪 开始测试数据库功能...")
    
    # 创建测试数据库
    db_path = "test_trading_data.db"
    db = TradingDatabase(db_path)
    
    print(f"✅ 数据库创建成功: {db_path}")
    
    # 测试价格数据记录
    print("\n📊 测试价格数据记录...")
    
    base_price = 50000.0
    for i in range(10):
        # 模拟价格波动
        binance_price = base_price + random.uniform(-100, 100)
        lighter_price = binance_price + random.uniform(-5, 5)
        spread_bps = (lighter_price - binance_price) / binance_price * 10000
        
        db.record_price_data(
            symbol="BTCUSDT",
            binance_price=binance_price,
            lighter_price=lighter_price,
            lighter_bid=lighter_price - 0.5,
            lighter_ask=lighter_price + 0.5,
            lighter_bid_size=random.uniform(0.1, 1.0),
            lighter_ask_size=random.uniform(0.1, 1.0),
            spread_bps=spread_bps,
            raw_data={"test": True, "index": i}
        )
        
        print(f"  记录 {i+1}: Binance={binance_price:.2f}, Lighter={lighter_price:.2f}, Spread={spread_bps:.2f}bps")
        time.sleep(0.1)  # 模拟时间间隔
    
    # 测试交易事件记录
    print("\n🔄 测试交易事件记录...")
    
    events = [
        ("open_long", 49950.0, 0.1, "long", True),
        ("open_long_success", 49950.0, 0.1, "long", True),
        ("close_long", 50050.0, 0.1, "long", True),
        ("close_long_success", 50050.0, 0.1, "long", True),
        ("open_short", 50100.0, 0.1, "short", False),
        ("open_short_failed", 50100.0, 0.1, "short", False)
    ]
    
    for event_type, price, quantity, side, success in events:
        db.record_trading_event(
            symbol="BTCUSDT",
            event_type=event_type,
            price=price,
            quantity=quantity,
            side=side,
            strategy_state="EXECUTING_OPEN" if "open" in event_type else "EXECUTING_CLOSE",
            spread_bps=random.uniform(-10, 10),
            lighter_price=price + random.uniform(-1, 1),
            binance_price=price + random.uniform(-2, 2),
            success=success,
            tx_hash="0x123456789abcdef" if success else None,
            error_message="Test error" if not success else None,
            metadata={"test": True, "event": event_type}
        )
        
        status = "✅ 成功" if success else "❌ 失败"
        print(f"  事件: {event_type} - 价格={price:.2f}, 数量={quantity}, {status}")
    
    # 测试数据查询
    print("\n📈 测试数据查询...")
    
    # 查询价格数据 (最近1小时)
    current_time = time.time()
    one_hour_ago = current_time - 3600
    
    price_data = db.get_price_data(start_time=one_hour_ago, limit=100)
    print(f"  价格数据记录数: {len(price_data)}")
    
    # 查询交易事件
    trading_events = db.get_trading_events(start_time=one_hour_ago, limit=100)
    print(f"  交易事件记录数: {len(trading_events)}")
    
    # 查询统计信息
    stats = db.get_statistics()
    print(f"  统计信息: {stats}")
    
    print("\n🎉 数据库功能测试完成！")
    print(f"📁 测试数据库文件: {Path(db_path).absolute()}")
    
    return db_path

def main():
    """主函数"""
    try:
        test_db_path = test_database()
        
        print("\n" + "="*60)
        print("🚀 现在可以启动Web Dashboard查看测试数据:")
        print(f"   python run_dashboard.py")
        print("   或者修改 run_dashboard.py 中的数据库路径为:")
        print(f"   {Path(test_db_path).absolute()}")
        print("="*60)
        
    except Exception as e:
        print(f"❌ 测试失败: {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)

if __name__ == '__main__':
    main()