迭代 138 测试策略¶
版本: v1.0 | 日期: 2026-02-28
1. 测试目标¶
1.1 质量目标¶
| 指标 | 目标 | 说明 |
|——|——|——|
| 单元测试覆盖率 | >= 80% | 核心代码必须覆盖 |
| 集成测试通过率 | 100% | 所有集成测试必须通过 |
| 回归测试通过率 | 100% | 现有 1020 个测试全通过 |
| 性能测试达标率 | 100% | 所有性能指标达标 |
1.2 零回归承诺¶
关键原则*：新功能不能破坏现有功能
现有 1020 个测试必须全部通过
纯 K 线模式（BAR）行为完全不变
所有现有 API 保持向后兼容
2. 测试金字塔¶
/\
/E2E\ 10% - 端到端测试（完整回测场景）
/------\
/集成测试\ 30% - 集成测试（组件协作）
/----------\
/ 单元测试 \ 60% - 单元测试（独立组件）
/--------------\
```bash
### 2.1 单元测试（60%）
- *范围**：独立组件的功能验证
| 组件 | 测试文件 | 重点 |
|------|---------|------|
| StreamingEventQueue | `test_streaming_queue.py` | 事件排序、内存控制 |
| DataChannel | `test_data_channel.py` | 数据验证、缓冲管理 |
| TickChannel | `test_tick_channel.py` | Tick 数据加载、验证 |
| OrderBookChannel | `test_orderbook_channel.py` | OB 数据结构、验证 |
| TickBroker | `test_tick_broker.py` | Tick 撮合逻辑 |
| MixBroker | `test_mix_broker.py` | 混合撮合逻辑 |
| Strategy 回调 | `test_strategy_callbacks.py` | 回调触发、参数传递 |
- *示例**：
```python
# tests/unit/test_streaming_queue.py
import pytest
from backtrader.channel import StreamingEventQueue, Event
class TestStreamingEventQueue:
def test_event_ordering_by_timestamp(self):
"""测试事件按时间戳排序"""
queue = StreamingEventQueue([], [])
# 乱序插入
queue._push_event(Event(timestamp=100.0), 'tick', 'BTC')
queue._push_event(Event(timestamp=50.0), 'tick', 'BTC')
queue._push_event(Event(timestamp=75.0), 'tick', 'BTC')
# 验证弹出顺序
assert queue.pop().timestamp == 50.0
assert queue.pop().timestamp == 75.0
assert queue.pop().timestamp == 100.0
def test_event_ordering_by_priority(self):
"""测试同时间戳按优先级排序"""
queue = StreamingEventQueue([], [])
# 同时间戳，不同优先级
queue._push_event(Event(timestamp=100.0, priority=30), 'tick', 'BTC')
queue._push_event(Event(timestamp=100.0, priority=10), 'funding', 'BTC')
queue._push_event(Event(timestamp=100.0, priority=20), 'orderbook', 'BTC')
# 验证优先级顺序
assert queue.pop().priority == 10 # funding
assert queue.pop().priority == 20 # orderbook
assert queue.pop().priority == 30 # tick
def test_adaptive_window_shrink(self):
"""测试自适应窗口缩小"""
queue = StreamingEventQueue(
[], [],
preload_window=300.0,
max_memory_mb=10,
adaptive=True
)
# 模拟内存压力
# ... 填充大量事件 ...
# 验证窗口缩小
assert queue._window < 300.0
assert queue._adjustment_count > 0
```bash
- --
### 2.2 集成测试（30%）
- *范围**：组件间协作验证
| 场景 | 测试文件 | 重点 |
|------|---------|------|
| Tick 模式回测 | `test_tick_mode_integration.py` | EventQueue + TickBroker + Strategy |
| 混合模式回测 | `test_mixed_mode_integration.py` | Bar + Tick 协作 |
| 多 Channel 协作 | `test_multi_channel.py` | Tick + OB + Funding |
| 异常处理 | `test_error_handling.py` | 数据异常、策略异常 |
| 通知机制 | `test_notification_flow.py` | Broker 通知 + 策略回调 |
- *示例**：
```python
# tests/integration/test_tick_mode_integration.py
import backtrader as bt
from backtrader.channels.tick import TickChannel
class TickStrategy(bt.Strategy):
def __init__(self):
self.tick_count = 0
self.order_count = 0
def on_tick(self, channel, tick):
self.tick_count += 1
# 每 100 个 tick 买入
if self.tick_count % 100 == 0:
self.buy(size=0.1)
def notify_order(self, order):
if order.status == order.Completed:
self.order_count += 1
def test_tick_mode_basic_flow():
"""测试 Tick 模式基本流程"""
cerebro = bt.Cerebro()
# 添加 Tick 通道
cerebro.add_channel(
TickChannel,
symbol='BTC/USDT',
dataname='tests/data/btc_ticks_1day.csv'
)
# 添加策略
cerebro.addstrategy(TickStrategy)
# 运行（Tick 模式）
results = cerebro.run(mode=bt.RunMode.TICK)
strat = results[0]
# 验证
assert strat.tick_count > 0, "应该处理了 tick 数据"
assert strat.order_count > 0, "应该有订单成交"
# 验证 Broker 使用 TickBroker
assert isinstance(cerebro._broker, bt.brokers.TickBroker)
def test_tick_and_orderbook_coordination():
"""测试 Tick 和 OrderBook 协作"""
cerebro = bt.Cerebro()
cerebro.add_channel(TickChannel, symbol='BTC/USDT', dataname='...')
cerebro.add_channel(OrderBookChannel, symbol='BTC/USDT', dataname='...')
# ... 策略使用 OB 辅助 Tick 交易决策 ...
results = cerebro.run(mode=bt.RunMode.TICK)
# 验证事件顺序：OB 优先级高于 Tick
# ...
```bash
- --
### 2.3 端到端测试（10%）
- *范围**：完整业务场景验证
| 场景 | 测试文件 | 重点 |
|------|---------|------|
| 完整 Tick 回测 | `test_e2e_tick_backtest.py` | 数据加载→回测→分析 |
| 混合模式回测 | `test_e2e_mixed_backtest.py` | Bar 主时钟 + Tick 辅助 |
| 实盘模拟 | `test_e2e_live_simulation.py` | WebSocket 数据 + 实时撮合 |
| 性能场景 | `test_e2e_performance.py` | 大数据量、长时间 |
- *示例**：
```python
# tests/e2e/test_e2e_tick_backtest.py
def test_complete_tick_backtest_workflow():
"""测试完整 Tick 回测流程"""
# 1. 准备数据
tick_data = prepare_tick_data('BTC/USDT', days=7)
ob_data = prepare_orderbook_data('BTC/USDT', days=7)
# 2. 配置 Cerebro
cerebro = bt.Cerebro()
cerebro.broker.setcash(100000)
# 3. 添加通道
cerebro.add_channel(TickChannel, symbol='BTC/USDT', dataname=tick_data)
cerebro.add_channel(OrderBookChannel, symbol='BTC/USDT', dataname=ob_data)
# 4. 添加策略
cerebro.addstrategy(AdvancedTickStrategy)
# 5. 添加分析器
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades')
# 6. 运行
results = cerebro.run(mode=bt.RunMode.TICK)
# 7. 验证结果
strat = results[0]
sharpe = strat.analyzers.sharpe.get_analysis()
trades = strat.analyzers.trades.get_analysis()
assert sharpe['sharperatio'] is not None
assert trades['total']['total'] > 0
# 8. 验证性能
# 7 天 tick 数据应该在合理时间内完成
# ...
```bash
- --
## 3. 回归测试策略
### 3.1 现有测试保护
- *目标**：确保现有 1020 个测试全部通过
- *策略**：
1. 每次提交前运行完整回归测试
2. CI/CD 自动运行回归测试
3. 任何回归失败立即修复
- *命令**：
```bash
# 运行所有现有测试
pytest tests/ -v --tb=short
# 排除新增测试
pytest tests/ -v --ignore=tests/phase0 --ignore=tests/phase1 \
- -ignore=tests/integration/tick --ignore=tests/e2e/tick
```bash
### 3.2 兼容性测试
- *测试矩阵**：
| 场景 | 预期行为 | 测试方法 |
|------|---------|---------|
| 纯 K 线策略 | 完全不变 | 运行现有策略示例 |
| 不添加 Channel | BAR 模式 | 验证默认行为 |
| 现有 Broker | BackBroker 不变 | 单元测试 |
| 现有 Analyzer | 正常工作 | 集成测试 |
- *示例**：
```python
# tests/regression/test_backward_compatibility.py
def test_existing_strategy_unchanged():
"""测试现有策略行为不变"""
# 使用现有的示例策略
cerebro = bt.Cerebro()
cerebro.adddata(bt.feeds.GenericCSVData(dataname='...'))
cerebro.addstrategy(bt.strategies.SMA_CrossOver)
# 不添加 Channel，应该自动使用 BAR 模式
results = cerebro.run()
# 验证使用 BackBroker
assert isinstance(cerebro._broker, bt.brokers.BackBroker)
# 验证结果与历史一致
# ...
def test_no_channel_defaults_to_bar_mode():
"""测试不添加 Channel 时默认 BAR 模式"""
cerebro = bt.Cerebro()
cerebro.adddata(...)
cerebro.addstrategy(...)
results = cerebro.run()
# 验证运行模式
assert cerebro._run_mode == bt.RunMode.BAR
```bash
- --
## 4. 性能测试
### 4.1 性能基准
| 指标 | 目标 | 测试方法 |
|------|------|---------|
| 1 天 Tick 回测 | < 10 秒 | benchmark_tick.py |
| 内存使用 | < 200MB | memory_profiler |
| 事件处理吞吐 | > 10K events/s | performance_test.py |
| CPU 使用率 | < 80% | psutil 监控 |
### 4.2 性能测试用例
```python
# tests/performance/benchmark_tick.py
import time
import psutil
from backtrader.channels.tick import TickChannel
def benchmark_tick_backtest():
"""Tick 回测性能基准"""
process = psutil.Process()
# 准备 1 天 tick 数据（约 17M 条）
cerebro = bt.Cerebro()
cerebro.add_channel(
TickChannel,
symbol='BTC/USDT',
dataname='tests/data/btc_ticks_1day.csv'
)
cerebro.addstrategy(SimpleTickStrategy)
# 记录初始状态
start_time = time.time()
start_memory = process.memory_info().rss / 1024 / 1024
# 运行
results = cerebro.run(mode=bt.RunMode.TICK)
# 记录结束状态
end_time = time.time()
end_memory = process.memory_info().rss / 1024 / 1024
elapsed = end_time - start_time
memory_used = end_memory - start_memory
print(f"\n=== Performance Benchmark ===")
print(f"Time: {elapsed:.2f}s")
print(f"Memory: {memory_used:.1f}MB")
print(f"Target: < 10s, < 200MB")
# 验证性能目标
assert elapsed < 10, f"Time {elapsed:.2f}s exceeds 10s target"
assert memory_used < 200, f"Memory {memory_used:.1f}MB exceeds 200MB target"
def benchmark_event_throughput():
"""事件处理吞吐量基准"""
queue = StreamingEventQueue(...)
start = time.time()
count = 0
while queue:
event = queue.pop()
count += 1
elapsed = time.time() - start
throughput = count / elapsed
print(f"Throughput: {throughput:.0f} events/sec")
assert throughput > 10000, f"Throughput {throughput:.0f} below 10K target"
```bash
### 4.3 压力测试
```python
# tests/performance/stress_test.py
def test_long_duration_stability():
"""长时间运行稳定性测试"""
# 30 天 tick 数据
cerebro = bt.Cerebro()
cerebro.add_channel(TickChannel, dataname='30days_ticks.csv')
cerebro.addstrategy(...)
# 监控内存增长
memory_samples = []
def memory_monitor():
while True:
memory_samples.append(psutil.Process().memory_info().rss)
time.sleep(10)
# 运行
results = cerebro.run(mode=bt.RunMode.TICK)
# 验证内存不持续增长（排除内存泄漏）
growth_rate = calculate_memory_growth_rate(memory_samples)
assert growth_rate < 0.01, "Potential memory leak detected"
def test_high_frequency_events():
"""高频事件处理测试"""
# 模拟极高频 tick（1000/秒）
# 验证系统能否处理
# ...
```bash
- --
## 5. 数据驱动测试
### 5.1 测试数据准备
- *数据集**：
| 数据集 | 说明 | 用途 |
|--------|------|------|
| `btc_ticks_1day.csv` | 1 天 BTC tick | 基础功能测试 |
| `btc_ticks_7days.csv` | 7 天 BTC tick | 集成测试 |
| `btc_ticks_30days.csv` | 30 天 BTC tick | 压力测试 |
| `btc_ob_1day.jsonl` | 1 天 OrderBook | OB 功能测试 |
| `corrupted_ticks.csv` | 损坏的 tick 数据 | 异常处理测试 |
| `out_of_order_ticks.csv` | 乱序 tick 数据 | 数据验证测试 |
### 5.2 数据生成工具
```python
# tests/data/generate_test_data.py
def generate_tick_data(symbol, days, ticks_per_second=200):
"""生成模拟 tick 数据"""
import pandas as pd
import numpy as np
total_ticks = days *86400*ticks_per_second
timestamps = np.arange(total_ticks) / ticks_per_second
prices = 50000 + np.cumsum(np.random.randn(total_ticks)*10)
volumes = np.random.exponential(0.5, total_ticks)
directions = np.random.choice(['buy', 'sell'], total_ticks)
df = pd.DataFrame({
'timestamp': timestamps,
'price': prices,
'volume': volumes,
'direction': directions,
'trade_id': [f'T{i}' for i in range(total_ticks)]
})
return df
def generate_corrupted_data():
"""生成包含错误的测试数据"""
df = generate_tick_data('BTC/USDT', days=1)
# 注入各种错误
df.loc[100, 'price'] = -50000 # 负价格
df.loc[200, 'volume'] = -1.0 # 负成交量
df.loc[300, 'direction'] = 'invalid' # 无效方向
df.loc[400, 'timestamp'] = df.loc[399, 'timestamp'] - 1 # 乱序
return df
```bash
- --
## 6. 测试自动化
### 6.1 CI/CD 集成
- *GitHub Actions 配置**：
```yaml
# .github/workflows/test.yml
name: Test Suite
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Set up Python
uses: actions/setup-python@v2
with:
python-version: 3.8
- name: Install dependencies
run: |
pip install -r requirements.txt
pip install pytest pytest-cov
- name: Run unit tests
run: pytest tests/unit -v --cov=backtrader
- name: Run integration tests
run: pytest tests/integration -v
- name: Run regression tests
run: pytest tests/ --ignore=tests/phase* -v
- name: Upload coverage
uses: codecov/codecov-action@v2
```bash
### 6.2 Pre-commit Hooks
```bash
# .pre-commit-config.yaml
repos:
- repo: local
hooks:
- id: pytest-unit
name: Run unit tests
entry: pytest tests/unit -v
language: system
pass_filenames: false
always_run: true
```bash
- --
## 7. 测试报告
### 7.1 报告模板
```markdown
# 测试报告 - Phase X
## 测试概况
- 测试日期：YYYY-MM-DD
- 测试人员：XXX
- 测试环境：Python 3.8, Ubuntu 20.04
## 测试结果
### 单元测试
- 总数：XXX
- 通过：XXX
- 失败：XXX
- 覆盖率：XX%
### 集成测试
- 总数：XXX
- 通过：XXX
- 失败：XXX
### 回归测试
- 总数：1020
- 通过：XXX
- 失败：XXX
### 性能测试
- 1 天 Tick 回测：X.XX 秒 (目标<10 秒)
- 内存使用：XXX MB (目标<200MB)
- 事件吞吐：XXXXX events/s (目标>10K)
## 失败用例分析
### 失败用例 1
- 用例名称：test_xxx
- 失败原因：...
- 修复计划：...
## 风险评估
- 高风险问题：X 个
- 中风险问题：X 个
- 低风险问题：X 个
## 结论
- [ ] 通过验收标准
- [ ] 需要修复后重测
```bash
- --
## 8. 测试工具
### 8.1 推荐工具
| 工具 | 用途 | 安装 |
|------|------|------|
| pytest | 测试框架 | `pip install pytest` |
| pytest-cov | 覆盖率 | `pip install pytest-cov` |
| pytest-xdist | 并行测试 | `pip install pytest-xdist` |
| memory_profiler | 内存分析 | `pip install memory_profiler` |
| psutil | 系统监控 | `pip install psutil` |
| faker | 测试数据生成 | `pip install faker` |
### 8.2 自定义测试工具
```python
# tests/utils/test_helpers.py
import backtrader as bt
from contextlib import contextmanager
@contextmanager
def assert_no_regression():
"""确保测试不影响现有功能"""
# 运行现有测试套件
before = run_regression_tests()
yield
# 再次运行
after = run_regression_tests()
assert before == after, "Regression detected!"
def create_test_cerebro(**kwargs):
"""创建测试用 Cerebro"""
cerebro = bt.Cerebro()
cerebro.broker.setcash(100000)
cerebro.broker.setcommission(commission=0.001)
return cerebro
class MockTickChannel:
"""Mock Tick 通道用于测试"""
def __init__(self, events):
self._events = events
self._index = 0
def load(self):
for event in self._events:
yield event
```bash
- --
## 9. 测试检查清单
### 9.1 Phase 完成前检查
- [ ] 所有单元测试通过
- [ ] 所有集成测试通过
- [ ] 回归测试 100%通过
- [ ] 代码覆盖率 >= 80%
- [ ] 性能测试达标
- [ ] 内存泄漏检查通过
- [ ] 文档更新完成
- [ ] 代码审查通过
### 9.2 发布前检查
- [ ] 所有 Phase 测试通过
- [ ] E2E 测试通过
- [ ] 压力测试通过
- [ ] 兼容性测试通过
- [ ] 性能基准达标
- [ ] 安全审计通过
- [ ] 文档完整
- [ ] 示例代码可运行
- --
## 10. 附录
### A. 测试命令速查
```bash
# 运行所有测试
pytest tests/ -v
# 运行单元测试
pytest tests/unit -v
# 运行集成测试
pytest tests/integration -v
# 运行回归测试
pytest tests/ --ignore=tests/phase* -v
# 运行性能测试
pytest tests/performance -v
# 生成覆盖率报告
pytest tests/ --cov=backtrader --cov-report=html
# 并行运行测试
pytest tests/ -n auto
# 运行特定测试
pytest tests/unit/test_streaming_queue.py::test_event_ordering -v
```bash
### B. 测试数据路径
```bash
tests/
├── data/
│ ├── btc_ticks_1day.csv
│ ├── btc_ticks_7days.csv
│ ├── btc_ticks_30days.csv
│ ├── btc_ob_1day.jsonl
│ ├── corrupted_ticks.csv
│ └── out_of_order_ticks.csv
```bash