huice/test_unified_player.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Test script for Unified Futures Data Player
验证统一价格轴版本的期货数据播放器
"""

import sys
import os

def test_imports():
    """测试必需的库"""
    print("Testing required libraries...")

    try:
        import pandas as pd
        print("+ pandas imported successfully")
    except ImportError as e:
        print(f"- pandas import failed: {e}")
        return False

    try:
        import numpy as np
        print("+ numpy imported successfully")
    except ImportError as e:
        print(f"- numpy import failed: {e}")
        return False

    try:
        import matplotlib.pyplot as plt
        print("+ matplotlib imported successfully")
    except ImportError as e:
        print(f"- matplotlib import failed: {e}")
        return False

    try:
        import tkinter as tk
        print("+ tkinter imported successfully")
    except ImportError as e:
        print(f"- tkinter import failed: {e}")
        return False

    return True

def test_data_file():
    """测试数据文件"""
    print("\nTesting data file...")

    data_path = 'data/au2512_20251013.parquet'
    if os.path.exists(data_path):
        print(f"+ Data file found: {data_path}")

        try:
            import pandas as pd
            df = pd.read_parquet(data_path)
            print(f"+ Data loaded successfully: {len(df)} rows")

            # Test tick analysis
            columns_mapping = {
                '买1价': 'bid1_price', '卖1价': 'ask1_price',
                '买2价': 'bid2_price', '卖2价': 'ask2_price',
                '买3价': 'bid3_price', '卖3价': 'ask3_price',
                '买4价': 'bid4_price', '卖4价': 'ask4_price',
                '买5价': 'bid5_price', '卖5价': 'ask5_price'
            }
            df = df.rename(columns=columns_mapping)

            all_prices = []
            for i in range(1, 6):
                all_prices.extend(df[f'bid{i}_price'].dropna().tolist())
                all_prices.extend(df[f'ask{i}_price'].dropna().tolist())

            unique_prices = sorted(list(set(all_prices)))
            print(f"+ Unique price levels: {len(unique_prices)}")
            print(f"+ Price range: {min(unique_prices):.2f} - {max(unique_prices):.2f}")

            return True
        except Exception as e:
            print(f"- Failed to load data: {e}")
            return False
    else:
        print(f"- Data file not found: {data_path}")
        return False

def test_unified_functionality():
    """测试统一版本功能"""
    print("\nTesting Unified Futures Player...")

    try:
        from futures_player_unified import UnifiedFuturesPlayer
        print("+ UnifiedFuturesPlayer class imported successfully")

        print("+ Creating unified player instance...")
        player = UnifiedFuturesPlayer()

        if player.df is not None and len(player.df) > 0:
            print(f"+ Unified player data loaded: {len(player.df)} rows")

            # Test display update
            player.update_display()
            print("+ Display update test passed")

            # Test market depth data processing
            test_row = player.df.iloc[0]
            bid_prices, bid_volumes, ask_prices, ask_volumes = player.get_market_depth_data(test_row)
            print(f"+ Market depth test - Bids: {len(bid_prices)}, Asks: {len(ask_prices)}")

            # Clean up
            player.on_closing()
            return True
        else:
            print("- Unified player data not loaded properly")
            return False

    except Exception as e:
        print(f"- Unified player test failed: {e}")
        return False

def main():
    """主测试函数"""
    print("=== Unified Futures Player Test Suite ===\n")

    # Test imports
    if not test_imports():
        print("\n- Import test failed. Please install required libraries:")
        print("pip install pandas numpy matplotlib")
        return False

    # Test data file
    if not test_data_file():
        print("\n- Data file test failed. Please check data path.")
        return False

    # Test unified functionality
    if not test_unified_functionality():
        print("\n- Unified functionality test failed.")
        return False

    print("\n+ All tests passed! The unified futures player is ready.")
    print("\nKey improvements in this version:")
    print("+ Unified price axis showing bid/ask orders together")
    print("+ True minimum tick spacing (0.02 yuan)")
    print("+ Visual price gaps between bid and ask")
    print("+ Real-time spread visualization")
    print("+ Enhanced market depth analysis")

    print("\nTo run the unified application:")
    print("python futures_player_unified.py")

    return True

if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)