title: 测试指南 description: Backtrader 开发者测试实践和指南


测试指南

本指南介绍为 Backtrader 框架贡献代码时的测试实践和指南。

目录

测试框架

Backtrader 使用 pytest作为测试框架。主要特性:

  • pytest: 测试运行器和断言库

  • pytest-cov: 覆盖率报告

  • pytest-xdist: 并行测试执行

安装

pip install pytest pytest-cov pytest-xdist

```bash

### 配置

测试配置在 `pytest.ini` 中定义:

```ini
[pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*

markers =
    priority_p0: 关键测试 - 核心功能
    priority_p1: 高优先级测试 - 核心用户流程
    priority_p2: 中优先级测试 - 次要功能
    priority_p3: 低优先级测试 - 很少使用的功能
    integration: 需要实时交易所连接的集成测试
    websocket: WebSocket 专用集成测试
    trading: 在模拟交易所进行真实订单的测试

filterwarnings =
    ignore::RuntimeWarning
    ignore::DeprecationWarning

```bash

## 测试组织

### 目录结构

```bash
tests/
├── conftest.py              # 共享夹具和配置

├── datas/                   # 测试数据文件

├── original_tests/          # 核心功能测试

├── add_tests/               # 额外测试覆盖

├── strategies/              # 策略专用测试

├── base_functions/          # 基础函数测试

└── integration/             # 集成测试

```bash

### 测试文件命名

- 测试文件必须以 `test_` 开头:`test_indicator.py``test_strategy.py`
- 测试类必须以 `Test` 开头:`TestSMA``TestBroker`
- 测试函数必须以 `test_` 开头:`test_sma_calculation()`

## 测试分类

### 单元测试

单元测试独立测试各个组件。它们应该:

- 使用模拟数据(无外部依赖)
- 快速执行(每个测试 < 1 秒)
- 每个测试只测试一个行为

```python
def test_sma_calculation():
    """测试 SMA 指标计算正确。"""

# 创建测试数据
    data = [1, 2, 3, 4, 5]
    period = 3

# 预期结果
    expected = 3.0  # (3 + 4 + 5) / 3

# 运行测试
    result = calculate_sma(data, period)

# 断言
    assert result == expected

```bash

### 集成测试

集成测试验证多个组件协同工作。它们:

- 使用真实数据源或测试网络连接
- 标记为 `@pytest.mark.integration`
- 可能需要 API 密钥或外部服务

```python
import backtrader as bt
import pytest

@pytest.mark.integration
def test_ib_connection():
    """测试 Interactive Brokers 连接(需要测试网络)。"""
    cerebro = bt.Cerebro()
    store = bt.stores.IBStore(port=7497)  # 模拟交易端口
    data = store.getdata(dataname='AAPL')

    cerebro.adddata(data)
    result = cerebro.run()
    assert len(result) > 0

```bash

### 优先级

测试按优先级分类:

| 优先级 | 描述 | 使用场景 |

|--------|------|----------|

| `priority_p0` | 关键 - 核心功能 | 必要功能、数据加载、订单执行 |

| `priority_p1` |  - 经常使用 | 常见指标、标准策略 |

| `priority_p2` |  - 次要功能 | 较少见的指标、边界情况 |

| `priority_p3` |  - 很少使用 | 冷门功能、遗留代码 |

## 测试标记

### 使用标记

```python
import pytest

@pytest.mark.priority_p0
def test_data_feed_loading():
    """关键测试 - 数据源必须正确加载。"""
    pass

@pytest.mark.priority_p1
@pytest.mark.integration
def test_live_api_connection():
    """高优先级集成测试。"""
    pass

@pytest.mark.websocket
async def test_websocket_feed():
    """WebSocket 专用测试。"""
    pass

```bash

### 使用标记运行

```bash

# 仅运行关键测试

pytest tests/ -m "priority_p0"

# 跳过集成测试

pytest tests/ -m "not integration"

# 运行多个标记

pytest tests/ -m "priority_p0 or priority_p1"

# 跳过 WebSocket 测试

pytest tests/ -m "not websocket"

```bash

## 编写测试

### 测试结构

良好的测试遵循 Arrange-Act-Assert 模式:

```python
def test_indicator_calculation():
    """测试指标计算预期值。"""

# Arrange - 设置测试数据和条件
    cerebro = bt.Cerebro()
    data = create_test_data()
    cerebro.adddata(data)
    cerebro.addstrategy(TestStrategy)

# Act - 执行被测试的代码
    result = cerebro.run()

# Assert - 验证预期结果
    assert len(result) == 1
    assert result[0].analyzers.sharpe.get_analysis()['sharperatio'] > 0

```bash

### 完整测试示例

以下是测试指标的完整示例:

```python

# !/usr/bin/env python

"""测试简单移动平均线指标。"""

import backtrader as bt
import pytest
import datetime

class TestStrategy(bt.Strategy):
    """用于 SMA 验证的测试策略。"""

    params = (
        ('period', 15),
    )

    def __init__(self):
        """初始化指标和测试参数。"""
        self.sma = bt.indicators.SMA(self.data, period=self.p.period)
        self.expected_values = []
        self.actual_values = []

    def next(self):
        """记录 SMA 值以供验证。"""
        if len(self.data) > self.p.period:
            self.actual_values.append(self.sma[0])

    def stop(self):
        """验证 SMA 计算。"""
        assert len(self.actual_values) > 0, "没有计算 SMA 值"

# 此处添加其他断言


@pytest.mark.priority_p0
def test_sma_basic_calculation():
    """测试 SMA 使用基本数据正确计算。"""
    cerebro = bt.Cerebro()

# 创建简单测试数据
    data = bt.feeds.BacktraderCSVData(
        dataname='tests/datas/2006-day-001.txt',
        fromdate=datetime.datetime(2006, 1, 1),
        todate=datetime.datetime(2006, 1, 31),  # 一个月
    )

    cerebro.adddata(data)
    cerebro.addstrategy(TestStrategy, period=15)
    cerebro.broker.setcash(10000.0)

    results = cerebro.run()

# 验证策略已运行
    assert len(results) == 1
    strat = results[0]

# 验证 SMA 已计算
    assert hasattr(strat, 'sma')
    assert len(strat.sma) > 0


@pytest.mark.priority_p1
@pytest.mark.parametrize("period", [5, 10, 15, 20, 30])
def test_sma_different_periods(period):
    """测试不同周期值的 SMA。"""
    cerebro = bt.Cerebro()

    data = bt.feeds.BacktraderCSVData(
        dataname='tests/datas/2006-day-001.txt',
        fromdate=datetime.datetime(2006, 1, 1),
        todate=datetime.datetime(2006, 2, 28),
    )

    cerebro.adddata(data)
    cerebro.addstrategy(TestStrategy, period=period)

    results = cerebro.run()
    assert len(results) == 1


@pytest.mark.priority_p2
def test_sma_with_multiple_data_feeds():
    """测试多数据源的 SMA。"""
    cerebro = bt.Cerebro()

    data1 = bt.feeds.BacktraderCSVData(
        dataname='tests/datas/2006-day-001.txt',
        fromdate=datetime.datetime(2006, 1, 1),
        todate=datetime.datetime(2006, 3, 31),
    )

    data2 = bt.feeds.BacktraderCSVData(
        dataname='tests/datas/2006-day-002.txt',
        fromdate=datetime.datetime(2006, 1, 1),
        todate=datetime.datetime(2006, 3, 31),
    )

    cerebro.adddata(data1)
    cerebro.adddata(data2)
    cerebro.addstrategy(TestStrategy, period=10)

    results = cerebro.run()
    assert len(results) == 1

```bash

### 测试策略

测试交易策略时,重点关注:

1. **订单执行**:验证订单正确下达
2. **持仓管理**:检查持仓按预期开/平仓
3. **指标使用**:确保指标正确初始化

```python
def test_strategy_buy_signal():
    """测试策略在信号时执行买入。"""
    cerebro = bt.Cerebro()

    class BuyTestStrategy(bt.Strategy):
        def __init__(self):
            self.buy_executed = False
            self.sma_fast = bt.indicators.SMA(period=5)
            self.sma_slow = bt.indicators.SMA(period=15)

        def next(self):
            if not self.buy_executed:
                if self.sma_fast[0] > self.sma_slow[0]:
                    self.buy()
                    self.buy_executed = True

    data = bt.feeds.BacktraderCSVData(
        dataname='tests/datas/2006-day-001.txt',
        fromdate=datetime.datetime(2006, 1, 1),
        todate=datetime.datetime(2006, 6, 30),
    )

    cerebro.adddata(data)
    cerebro.addstrategy(BuyTestStrategy)

    results = cerebro.run()
    strat = results[0]

# 验证至少执行了一个买入订单
    assert strat.buy_executed

```bash

### 使用模拟数据测试

对于独立的单元测试,创建模拟数据:

```python
import backtrader as bt

class MockData(bt.feeds.PandasData):
    """用于测试的模拟数据源。"""
    params = (
        ('datetime', None),
        ('open', -1),
        ('high', -1),
        ('low', -1),
        ('close', -1),
        ('volume', -1),
        ('openinterest', -1),
    )

def create_mock_data():
    """创建简单的模拟数据源。"""
    import pandas as pd

    dates = pd.date_range('2023-01-01', periods=100, freq='D')
    data = pd.DataFrame({
        'datetime': dates,
        'open': 100 + range(100),
        'high': 102 + range(100),
        'low': 99 + range(100),
        'close': 101 + range(100),
        'volume': 1000,
    })

    return MockData(dataname=data)

def test_with_mock_data():
    """使用模拟数据测试指标。"""
    cerebro = bt.Cerebro()
    data = create_mock_data()
    cerebro.adddata(data)
    cerebro.addstrategy(bt.Strategy)

    result = cerebro.run()
    assert len(result) > 0

```bash

## 夹具和辅助工具

### 内置夹具

`conftest.py` 文件提供共享夹具:

```python
@pytest.fixture
def sample_data(datas_path):
    """为测试提供标准样本数据源。"""
    datapath = datas_path / "2006-day-001.txt"
    data = bt.feeds.BacktraderCSVData(
        dataname=str(datapath),
        fromdate=datetime.datetime(2006, 1, 1),
        todate=datetime.datetime(2006, 12, 31),
    )
    return data

@pytest.fixture
def cerebro_engine():
    """提供基本 Cerebro 引擎实例。"""
    cerebro = bt.Cerebro()
    yield cerebro

# 清理
    cerebro = None

@pytest.fixture
def cerebro_with_cash(cerebro_engine):
    """提供设置了初始资金的 Cerebro。"""
    cerebro_engine.broker.setcash(10000.0)
    return cerebro_engine

```bash

### 使用夹具

```python
def test_with_fixture(sample_data, cerebro_engine):
    """使用 conftest.py 中的夹具进行测试。"""
    cerebro_engine.adddata(sample_data)
    cerebro_engine.addstrategy(bt.Strategy)

    result = cerebro_engine.run()
    assert len(result) > 0

```bash

### 创建自定义夹具

```python

# 在您的测试文件或 conftest.py 中

@pytest.fixture
def macd_indicator():
    """创建带有标准参数的 MACD 指标。"""
    class MACDStrategy(bt.Strategy):
        def __init__(self):
            self.macd = bt.indicators.MACD(period_me1=12,
                                           period_me2=26,
                                           period_signal=9)

    return MACDStrategy

```bash

## 覆盖率要求

### 覆盖率配置

覆盖率在 `pyproject.toml` 中配置:

```toml
[tool.coverage.run]
source = ["backtrader"]
omit = [
    "*/tests/*",
    "*/test_*",
    "setup.py",
    "*/crypto_tests/*"
]

[tool.coverage.report]
exclude_lines = [
    "pragma: no cover",
    "def __repr__",
    "if self.debug:",
    "raise NotImplementedError",
    "if __name__ == .__main__.:",
]

```bash

### 运行覆盖率

```bash

# 生成覆盖率报告

pytest tests/ --cov=backtrader --cov-report=term-missing

# 生成 HTML 报告

pytest tests/ --cov=backtrader --cov-report=html

# 结合标记使用

pytest tests/ -m "not integration" --cov=backtrader

```bash

### 覆盖率目标

- **新代码**:目标是 90%+ 的覆盖率
- **关键路径**:100% 覆盖率(P0 测试)
- **现有代码**:保持当前覆盖率水平

## 运行测试

### 基本命令

```bash

# 运行所有测试

pytest tests/ -v

# 并行运行(4 个工作进程)

pytest tests/ -n 4 -v

# 运行特定测试文件

pytest tests/add_tests/test_sma.py -v

# 运行特定测试函数

pytest tests/add_tests/test_sma.py::test_sma_calculation -v

# 首次失败时停止

pytest tests/ -x

# 失败时显示局部变量

pytest tests/ -l

# 显示详细输出和打印语句

pytest tests/ -s

```bash

### 按类别运行

```bash

# 指标测试

pytest tests/add_tests/test_ind*.py tests/original_tests/test_ind*.py -v

# 策略测试

pytest tests/add_tests/test_strategy*.py tests/original_tests/test_strategy*.py -v

# 分析器测试

pytest tests/add_tests/test_analyzer*.py tests/original_tests/test_analyzer*.py -v

# 经纪人测试

pytest tests/add_tests/test_broker.py -v

```bash

### 使用 Make 运行

```bash

# 运行所有测试

make test

# 运行带覆盖率

make test-coverage

# 运行特定测试文件

make test-file TEST=tests/add_tests/test_sma.py

```bash

### 持续测试

开发时使用 pytest-watch 自动运行测试:

```bash
pip install pytest-watch
ptw tests/ -- -v

```bash

## 最佳实践

### 应该做的:

1. **先写测试**(尽可能采用 TDD 方法)
2. **使用描述性测试名称**:`test_sma_calculates_correctly()` 而不是 `test_1()`
3. **保持测试独立**- 测试之间无共享状态

4.**使用夹具**处理通用设置

1. **模拟外部依赖**- API 调用、文件 I/O

6.**测试边界情况**- 空数据、最小周期、边界条件
7.**添加文档字符串**说明测试内容

1. **使用标记**进行测试分类

### 不应该做的:

1. **不要硬编码路径**- 使用夹具或相对路径

2.**不要测试实现细节**- 测试行为
3.**不要编写庞大的测试**- 一个概念一个断言
4.**不要忽略测试**- 修复或标记为预期失败
5.**不要在单元测试中使用实时数据**

1. **不要提交注释掉的代码**

## 调试测试

### 使用 pdb

```python
def test_failing_case():
    """需要调试的失败测试。"""
    import pdb; pdb.set_trace()

    cerebro = bt.Cerebro()

# ... 其余测试代码

```bash

### 使用 pytest 的 pdb

```bash

# 失败时进入调试器

pytest tests/ --pdb

# 出错时进入调试器(不仅仅是失败)

pytest tests/ --pdb --trace

```bash

### 打印测试输出

```bash

# 显示打印语句

pytest tests/ -s -v

# 捕获输出但在失败时显示

pytest tests/ --capture=no

```bash

## 测试数据

### 测试数据文件

测试数据位于 `tests/datas/` 目录:

```bash
tests/datas/
├── 2006-day-001.txt      # 2006 年日线数据

├── 2006-day-002.txt      # 2006 年日线数据(第二组)

├── 2006-week-001.txt     # 2006 年周线数据

└── ...

```bash

### 创建测试数据

```python
def create_test_csv(filename, num_bars=100):
    """创建测试 CSV 数据文件。"""
    import csv
    from datetime import datetime, timedelta

    with open(filename, 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(['datetime', 'open', 'high', 'low', 'close', 'volume', 'openinterest'])

        dt = datetime(2023, 1, 1)
        for i in range(num_bars):
            base_price = 100.0 + i * 0.1
            writer.writerow([
                dt.strftime('%Y-%m-%d %H:%M:%S'),
                base_price,        # open
                base_price + 0.5,  # high
                base_price - 0.5,  # low
                base_price + 0.2,  # close
                1000,              # volume
                0                  # openinterest
            ])
            dt += timedelta(days=1)

```bash

## 常见测试场景

### 测试指标注册

```python
def test_indicator_registration():
    """测试指标正确注册到策略。"""
    cerebro = bt.Cerebro()

    class TestStrategy(bt.Strategy):
        def __init__(self):
            self.sma = bt.indicators.SMA(period=10)

    data = bt.feeds.BacktraderCSVData(
        dataname='tests/datas/2006-day-001.txt',
    )

    cerebro.adddata(data)
    cerebro.addstrategy(TestStrategy)
    cerebro.run()

# 验证策略已运行
    assert len(cerebro.runstrats[0]) > 0

```bash

### 测试分析器

```python
def test_sharpe_analyzer():
    """测试夏普比率分析器。"""
    cerebro = bt.Cerebro()

    data = bt.feeds.BacktraderCSVData(
        dataname='tests/datas/2006-day-001.txt',
        fromdate=datetime.datetime(2006, 1, 1),
        todate=datetime.datetime(2006, 12, 31),
    )

    cerebro.adddata(data)
    cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')

    results = cerebro.run()
    strat = results[0]

# 验证分析器存在并返回有效值
    assert hasattr(strat.analyzers, 'sharpe')
    analysis = strat.analyzers.sharpe.get_analysis()
    assert 'sharperatio' in analysis

```bash

### 测试观察器

```python
def test_drawdown_observer():
    """测试回撤观察器。"""
    cerebro = bt.Cerebro()

    data = bt.feeds.BacktraderCSVData(
        dataname='tests/datas/2006-day-001.txt',
    )

    cerebro.adddata(data)
    cerebro.addobserver(bt.observers.DrawDown)

    results = cerebro.run()
    strat = results[0]

# 验证观察器已附加
    assert len(strat.observers) > 0

```bash

## 另请参阅

- [开发环境设置](setup_zh.md)
- [代码风格](style_zh.md)
- [贡献指南](contributing_zh.md)