跳到主要内容

压力测试框架

用于测试 AXBlade 后端系统的稳定性、性能和 Hummingbot 兼容性的综合测试框架。

安装​

cd backend/tests/stress
pip install -r requirements.txt

测试模式​

1. 压力测试 (Stress Test)​

测试系统在高负载下的性能表现。

# 快速测试 (5分钟)
python -m stress.run_tests stress quick

# 标准测试 (30分钟)
python -m stress.run_tests stress standard

# 扩展测试 (2小时)
python -m stress.run_tests stress extended

# Hummingbot 配置测试 (1小时)
python -m stress.run_tests stress hummingbot

2. 撮合引擎测试 (Matching Engine Test)​

测试撮合引擎的核心功能和稳定性。

python -m stress.run_tests matching

测试场景:

  • 价格时间优先 (price_time_priority)
  • 部分成交 (partial_fill)
  • 自成交防护 (self_trade_prevention)
  • 订单簿深度 (order_book_depth)
  • 快速取消重下 (rapid_cancel_replace)
  • 大单处理 (large_order_handling)
  • 价格档位合并 (price_level_consolidation)

3. Hummingbot 集成测试​

验证与 Hummingbot 做市机器人的兼容性。

python -m stress.run_tests hummingbot

测试内容:

  • API 兼容性验证
  • 订单生命周期
  • WebSocket 连接
  • 做市策略模拟
  • 高频订单处理
  • 订单簿同步

4. 耐久测试 (Endurance Test)​

长时间运行测试，验证系统长期稳定性。

# 1小时测试
python -m stress.run_tests endurance 1

# 24小时测试
python -m stress.run_tests endurance 24

5. 运行所有测试​

python -m stress.run_tests all

配置​

环境变量​

# API 地址
export ZTDX_API_URL="http://localhost:8080"

# WebSocket 地址
export ZTDX_WS_URL="ws://localhost:8080/ws"

# 测试账户私钥
export TEST_PRIVATE_KEY="0xac0974bec..."

命令行参数​

python -m stress.run_tests <mode> [submode] [options]

选项:
  --url       API 地址 (默认: http://localhost:8080)
  --output    结果输出目录 (默认: ./test_results)
  -v          详细输出

测试结果​

测试结果保存在 ./test_results/ 目录:

test_results/
├── stress_test_quick_20231210_120000.json    # 压力测试报告
├── stress_test_quick_20231210_120000.log     # 测试日志
├── matching_engine_20231210_120000.json      # 撮合引擎报告
├── hummingbot_test_20231210_120000.json      # Hummingbot 测试报告
├── endurance_final_20231210_120000.json      # 耐久测试报告
└── endurance_checkpoint_*.json               # 耐久测试检查点

性能指标​

通过标准​

指标阈值
成功率≥ 95%
错误率≤ 1%
P99 延迟≤ 1000ms
最小吞吐量≥ 30 ops/s

报告内容​

  • 吞吐量统计 (总请求、成功/失败、每秒请求)
  • 延迟统计 (平均、P50、P95、P99、最大)
  • 错误分类和计数
  • 时间序列数据
  • 异常记录

编程接口​

import asyncio
from stress import (
    run_stress_test,
    run_matching_engine_test,
    run_hummingbot_test,
    run_endurance_test,
)

# 快速压力测试
result = asyncio.run(run_stress_test("quick"))

# 撮合引擎测试
result = asyncio.run(run_matching_engine_test())

# Hummingbot 测试
result = asyncio.run(run_hummingbot_test())

# 耐久测试 (2小时)
result = asyncio.run(run_endurance_test(hours=2))

自定义配置​

from stress import StressTestConfig, StressTestRunner

config = StressTestConfig(
    duration=3600,          # 1小时
    concurrent_users=10,    # 10个并发用户
    orders_per_second=50,   # 每秒50个订单
    symbols=["BTC-USD"],    # 测试交易对
    max_error_rate=0.02,    # 2% 错误率阈值
)

runner = StressTestRunner(config)
result = asyncio.run(runner.run())

故障排除​

连接失败​

确保后端服务正在运行:

curl http://localhost:8080/health

认证失败​

检查测试私钥是否有效:

export TEST_PRIVATE_KEY="your_private_key"

余额不足​

测试账户需要有足够的 USDT 余额，通过后端管理接口添加测试余额。

开发​

添加新测试场景​

  1. 在 matching_engine_test.py 中添加测试方法
  2. 在 MatchingEngineTestConfig.test_scenarios 中注册
  3. 运行测试验证

自定义指标​

扩展 MetricsCollector 类添加新的指标收集。