test_binance.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
KlineProxy - Binance API 功能测试脚本
支持现货和期货市场的全面测试
"""

import argparse
import requests
import sys
from datetime import datetime
from typing import Dict, List, Any, Optional


class Colors:
    """ANSI 颜色代码"""
    HEADER = '\033[95m'
    BLUE = '\033[94m'
    CYAN = '\033[96m'
    GREEN = '\033[92m'
    YELLOW = '\033[93m'
    RED = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'


class TestStats:
    """测试统计"""
    def __init__(self):
        self.total = 0
        self.passed = 0
        self.failed = 0
        self.warnings = 0

    def add_pass(self):
        self.total += 1
        self.passed += 1

    def add_fail(self):
        self.total += 1
        self.failed += 1

    def add_warning(self):
        self.warnings += 1

    def print_summary(self):
        print("\n" + "=" * 80)
        print(f"{Colors.BOLD}测试统计{Colors.ENDC}")
        print("=" * 80)
        print(f"总测试数: {self.total}")
        print(f"{Colors.GREEN}✅ 通过: {self.passed}{Colors.ENDC}")
        print(f"{Colors.RED}❌ 失败: {self.failed}{Colors.ENDC}")
        print(f"{Colors.YELLOW}⚠️  警告: {self.warnings}{Colors.ENDC}")

        success_rate = (self.passed / self.total * 100) if self.total > 0 else 0
        print(f"成功率: {success_rate:.1f}%")
        print("=" * 80)


class BinanceAPITester:
    """Binance API 测试器"""

    def __init__(self, base_url: str = "http://localhost:1888", verbose: bool = False):
        self.base_url = base_url
        self.verbose = verbose
        self.stats = TestStats()

    def log(self, message: str, color: str = ""):
        """输出日志"""
        if color:
            print(f"{color}{message}{Colors.ENDC}")
        else:
            print(message)

    def log_verbose(self, message: str):
        """详细日志（仅在 verbose 模式输出）"""
        if self.verbose:
            print(f"{Colors.CYAN}[详细] {message}{Colors.ENDC}")

    def print_section(self, title: str, test_number: int = None):
        """打印测试章节标题"""
        print("\n" + "=" * 80)
        if test_number:
            print(f"{Colors.BOLD}{test_number}. {title}{Colors.ENDC}")
        else:
            print(f"{Colors.BOLD}{title}{Colors.ENDC}")
        print("=" * 80)

    def make_request(self, endpoint: str, params: Optional[Dict] = None, timeout: int = 10) -> Optional[Any]:
        """发送 HTTP 请求"""
        url = f"{self.base_url}{endpoint}"
        self.log_verbose(f"请求: GET {url}")
        if params:
            self.log_verbose(f"参数: {params}")
        if timeout != 10:
            self.log_verbose(f"超时时间: {timeout}秒")

        try:
            response = requests.get(url, params=params, timeout=timeout)
            response.raise_for_status()
            data = response.json()
            self.log_verbose(f"响应状态: {response.status_code}")
            return data
        except requests.exceptions.RequestException as e:
            self.log(f"❌ 请求失败: {e}", Colors.RED)
            return None

    def test_health(self):
        """测试1: 健康检查"""
        self.print_section("健康检查", 1)

        data = self.make_request("/actuator/health")
        if data and data.get('status') == 'UP':
            self.log(f"✅ 服务状态: {data['status']}", Colors.GREEN)
            self.stats.add_pass()
        else:
            self.log(f"❌ 服务状态异常: {data}", Colors.RED)
            self.stats.add_fail()

    def test_server_time(self):
        """测试2: 服务器时间"""
        self.print_section("服务器时间", 2)

        data = self.make_request("/api/v3/time")
        if data and 'serverTime' in data:
            timestamp = data['serverTime'] / 1000
            beijing_time = datetime.fromtimestamp(timestamp)
            self.log(f"服务器时间戳: {data['serverTime']}")
            self.log(f"北京时间: {beijing_time}")
            self.log("✅ 服务器时间获取成功", Colors.GREEN)
            self.stats.add_pass()
        else:
            self.log("❌ 服务器时间获取失败", Colors.RED)
            self.stats.add_fail()

    def test_exchange_info(self):
        """测试3: 交易所信息"""
        self.print_section("现货交易所信息", 3)

        # exchangeInfo 返回数据量大，使用更长的超时时间
        data = self.make_request("/api/v3/exchangeInfo", timeout=30)
        if data:
            symbols = data.get('symbols', [])
            usdt_pairs = [s['symbol'] for s in symbols if s['symbol'].endswith('USDT')]

            self.log(f"总交易对数: {len(symbols)}")
            self.log(f"USDT 交易对数: {len(usdt_pairs)}")

            if self.verbose and usdt_pairs:
                self.log(f"前10个USDT交易对: {', '.join(usdt_pairs[:10])}")

            self.log("✅ 交易所信息获取成功", Colors.GREEN)
            self.stats.add_pass()
        else:
            self.log("❌ 交易所信息获取失败", Colors.RED)
            self.stats.add_fail()

    def test_btc_spot_price(self):
        """测试4: BTC 现货价格"""
        self.print_section("BTC 现货价格", 4)

        data = self.make_request("/api/v3/ticker/price", {'symbol': 'BTCUSDT'})

        if isinstance(data, dict) and 'symbol' in data:
            self.log(f"交易对: {data['symbol']}")
            self.log(f"价格: ${data['price']}")
            self.log("✅ BTC 现货价格获取成功", Colors.GREEN)
            self.stats.add_pass()
        elif isinstance(data, list) and len(data) > 0:
            self.log(f"交易对: {data[0]['symbol']}")
            self.log(f"价格: ${data[0]['price']}")
            self.log("✅ BTC 现货价格获取成功", Colors.GREEN)
            self.stats.add_pass()
        else:
            self.log("⚠️  数据未加载 (需要等待WebSocket数据)", Colors.YELLOW)
            self.stats.add_warning()
            self.stats.add_fail()

    def test_btc_futures_price(self):
        """测试5: BTC 期货价格"""
        self.print_section("BTC 期货价格", 5)

        data = self.make_request("/fapi/v1/ticker/price", {'symbol': 'BTCUSDT'})

        if isinstance(data, dict) and 'symbol' in data:
            self.log(f"期货交易对: {data['symbol']}")
            self.log(f"期货价格: ${data['price']}")
            self.log("✅ BTC 期货价格获取成功", Colors.GREEN)
            self.stats.add_pass()
        elif isinstance(data, list) and len(data) > 0:
            self.log(f"期货交易对: {data[0]['symbol']}")
            self.log(f"期货价格: ${data[0]['price']}")
            self.log("✅ BTC 期货价格获取成功", Colors.GREEN)
            self.stats.add_pass()
        else:
            self.log("⚠️  数据未加载 (需要等待WebSocket数据)", Colors.YELLOW)
            self.stats.add_warning()
            self.stats.add_fail()

    def test_24hr_ticker_single(self):
        """测试6: BTC 24小时行情（单个）"""
        self.print_section("BTC 24小时行情（单个）", 6)

        data = self.make_request("/api/v3/ticker/24hr", {'symbol': 'BTCUSDT'})

        if isinstance(data, dict) and 'symbol' in data:
            self.log(f"交易对: {data['symbol']}")
            self.log(f"当前价格: ${data['lastPrice']}")
            self.log(f"24h涨跌: {data['priceChange']} ({data['priceChangePercent']}%)")
            self.log(f"24h最高: ${data['highPrice']}")
            self.log(f"24h最低: ${data['lowPrice']}")
            self.log(f"24h成交量: {data['volume']} BTC")
            self.log("✅ 24小时行情获取成功", Colors.GREEN)
            self.stats.add_pass()
        elif isinstance(data, list) and len(data) > 0:
            d = data[0]
            self.log(f"交易对: {d['symbol']}")
            self.log(f"当前价格: ${d['lastPrice']}")
            self.log(f"24h涨跌: {d['priceChange']} ({d['priceChangePercent']}%)")
            self.log(f"24h最高: ${d['highPrice']}")
            self.log(f"24h最低: ${d['lowPrice']}")
            self.log(f"24h成交量: {d['volume']} BTC")
            self.log("✅ 24小时行情获取成功", Colors.GREEN)
            self.stats.add_pass()
        else:
            self.log("⚠️  数据未加载 (需要等待WebSocket数据)", Colors.YELLOW)
            self.stats.add_warning()
            self.stats.add_fail()

    def test_24hr_ticker_all(self):
        """测试7: 所有交易对24小时行情"""
        self.print_section("所有交易对24小时行情", 7)

        data = self.make_request("/api/v3/ticker/24hr")

        if isinstance(data, list) and len(data) > 0:
            self.log(f"总交易对数: {len(data)}")

            # 找出涨幅最大的前5个
            sorted_data = sorted(data, key=lambda x: float(x.get('priceChangePercent', 0)), reverse=True)
            top_gainers = sorted_data[:5]

            self.log("\n涨幅前5名:")
            for i, ticker in enumerate(top_gainers, 1):
                self.log(f"  {i}. {ticker['symbol']}: {ticker['priceChangePercent']}%", Colors.GREEN)

            self.log("✅ 所有交易对行情获取成功", Colors.GREEN)
            self.stats.add_pass()
        else:
            self.log("⚠️  数据未加载 (需要等待WebSocket数据)", Colors.YELLOW)
            self.stats.add_warning()
            self.stats.add_fail()

    def test_klines(self, symbol: str = "BTCUSDT", intervals: List[str] = None):
        """测试8: K线数据（多种时间周期）"""
        if intervals is None:
            intervals = ["15m", "1h", "4h", "1d"]

        self.print_section(f"K线数据测试 - {symbol}", 8)

        for interval in intervals:
            self.log(f"\n--- {interval} K线 ---")
            data = self.make_request("/api/v3/klines", {
                'symbol': symbol,
                'interval': interval,
                'limit': 3
            })

            if data and len(data) > 0:
                self.log(f"获取到 {len(data)} 根K线")
                for i, kline in enumerate(data, 1):
                    timestamp = datetime.fromtimestamp(kline[0] / 1000)
                    self.log(f"  K线{i}: 时间={timestamp}, 开={kline[1]}, 高={kline[2]}, 低={kline[3]}, 收={kline[4]}, 量={kline[5]}")
                self.log(f"✅ {interval} K线获取成功", Colors.GREEN)
                self.stats.add_pass()
            else:
                self.log(f"⚠️  {interval} K线数据为空 (WebSocket未连接)", Colors.YELLOW)
                self.log(f"   这可能是正常的，REST API功能仍然正常", Colors.YELLOW)
                self.stats.add_warning()
                self.stats.add_fail()

    def test_funding_rate(self):
        """测试9: 资金费率"""
        self.print_section("期货资金费率", 9)

        data = self.make_request("/fapi/v1/fundingRate", {'symbol': 'BTCUSDT', 'limit': 3})

        if data and len(data) > 0:
            self.log(f"获取到 {len(data)} 条资金费率记录")
            for i, rate in enumerate(data, 1):
                timestamp = datetime.fromtimestamp(rate['fundingTime'] / 1000)
                self.log(f"  记录{i}: 时间={timestamp}, 费率={rate['fundingRate']}")
            self.log("✅ 资金费率获取成功", Colors.GREEN)
            self.stats.add_pass()
        else:
            self.log("⚠️  资金费率数据为空", Colors.YELLOW)
            self.stats.add_warning()
            self.stats.add_fail()

    def test_premium_index(self):
        """测试10: 溢价指数"""
        self.print_section("期货溢价指数", 10)

        data = self.make_request("/fapi/v1/premiumIndex", {'symbol': 'BTCUSDT'})

        if data and 'symbol' in data:
            self.log(f"交易对: {data['symbol']}")
            self.log(f"标记价格: {data.get('markPrice', 'N/A')}")
            self.log(f"指数价格: {data.get('indexPrice', 'N/A')}")
            self.log(f"预估结算价: {data.get('estimatedSettlePrice', 'N/A')}")
            self.log(f"最新资金费率: {data.get('lastFundingRate', 'N/A')}")
            self.log(f"下次资金费率时间: {datetime.fromtimestamp(data.get('nextFundingTime', 0) / 1000)}")
            self.log("✅ 溢价指数获取成功", Colors.GREEN)
            self.stats.add_pass()
        else:
            self.log("⚠️  溢价指数数据为空", Colors.YELLOW)
            self.stats.add_warning()
            self.stats.add_fail()

    def test_futures_exchange_info(self):
        """测试11: 期货交易所信息"""
        self.print_section("期货交易所信息", 11)

        # exchangeInfo 返回数据量大，使用更长的超时时间
        data = self.make_request("/fapi/v1/exchangeInfo", timeout=30)

        if data:
            symbols = data.get('symbols', [])
            usdt_pairs = [s['symbol'] for s in symbols if s['symbol'].endswith('USDT')]

            self.log(f"总期货合约数: {len(symbols)}")
            self.log(f"USDT 期货合约数: {len(usdt_pairs)}")

            if self.verbose and usdt_pairs:
                self.log(f"前10个USDT期货合约: {', '.join(usdt_pairs[:10])}")

            self.log("✅ 期货交易所信息获取成功", Colors.GREEN)
            self.stats.add_pass()
        else:
            self.log("❌ 期货交易所信息获取失败", Colors.RED)
            self.stats.add_fail()

    def test_all_prices(self):
        """测试12: 所有交易对价格"""
        self.print_section("所有现货交易对价格", 12)

        data = self.make_request("/api/v3/ticker/price")

        if isinstance(data, list) and len(data) > 0:
            self.log(f"总价格记录数: {len(data)}")

            # 显示前5个
            self.log("\n前5个交易对价格:")
            for i, ticker in enumerate(data[:5], 1):
                self.log(f"  {i}. {ticker['symbol']}: ${ticker['price']}")

            self.log("✅ 所有交易对价格获取成功", Colors.GREEN)
            self.stats.add_pass()
        else:
            self.log("⚠️  价格数据为空", Colors.YELLOW)
            self.stats.add_warning()
            self.stats.add_fail()

    def run_all_tests(self):
        """运行所有测试"""
        self.log(f"\n{Colors.BOLD}{'=' * 80}{Colors.ENDC}")
        self.log(f"{Colors.BOLD}KlineProxy - Binance API 功能测试{Colors.ENDC}")
        self.log(f"{Colors.BOLD}测试目标: {self.base_url}{Colors.ENDC}")
        self.log(f"{Colors.BOLD}{'=' * 80}{Colors.ENDC}\n")

        # 基础测试
        self.test_health()
        self.test_server_time()
        self.test_exchange_info()

        # 现货市场测试
        self.test_btc_spot_price()
        self.test_24hr_ticker_single()
        self.test_24hr_ticker_all()
        self.test_all_prices()

        # K线数据测试
        self.test_klines()

        # 期货市场测试
        self.test_futures_exchange_info()
        self.test_btc_futures_price()
        self.test_funding_rate()
        self.test_premium_index()

        # 打印统计
        self.stats.print_summary()


def main():
    """主函数"""
    parser = argparse.ArgumentParser(
        description='KlineProxy Binance API 功能测试脚本',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  %(prog)s                          # 使用默认配置测试
  %(prog)s --url http://localhost:1888  # 指定服务地址
  %(prog)s --verbose                # 详细输出模式
  %(prog)s --url http://localhost:1888 --verbose  # 组合使用
        """
    )

    parser.add_argument(
        '--url',
        default='http://localhost:1888',
        help='KlineProxy 服务地址 (默认: http://localhost:1888)'
    )

    parser.add_argument(
        '--verbose', '-v',
        action='store_true',
        help='详细输出模式'
    )

    args = parser.parse_args()

    # 创建测试器并运行
    tester = BinanceAPITester(base_url=args.url, verbose=args.verbose)

    try:
        tester.run_all_tests()

        # 根据测试结果返回退出码
        if tester.stats.failed > 0:
            sys.exit(1)
        else:
            sys.exit(0)

    except KeyboardInterrupt:
        print(f"\n\n{Colors.YELLOW}测试被用户中断{Colors.ENDC}")
        sys.exit(130)
    except Exception as e:
        print(f"\n\n{Colors.RED}❌ 测试失败: {e}{Colors.ENDC}")
        import traceback
        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    main()