trade/engine/backtest

engine/backtest — 回测引擎

事件驱动的历史回测框架，基于 DataService 从 TimescaleDB 读取历史 K 线， 按时间顺序逐根推送给策略，模拟订单成交、跟踪资金曲线、计算绩效指标。

快速开始

import asyncio
from datetime import datetime
from engine.backtest import BacktestEngine, BacktestConfig
from engine.common.config import config

async def main():
    bt_config = BacktestConfig(
        symbol="BTCUSDT",
        interval="1h",
        start_time=datetime(2025, 1, 1),
        end_time=datetime(2025, 6, 1),
        initial_capital=10_000.0,
    )

    engine = BacktestEngine(bt_config, db_config=config.db)
    result = await engine.run(MyStrategy, my_strategy_config)
    print(result.summary())

asyncio.run(main())

核心概念

回测流程

加载历史 K 线 → 预热阶段 → 主循环 → 计算指标 → 输出结果
                     ↓
              逐根 K 线推送：
                1. 执行上根 Bar 产生的买单（在开盘价执行）
                2. 推送 K 线给策略 → 产生信号
                3. 卖出信号立即执行，买入信号延迟到下一根 Bar
                4. 记录资金曲线

避免未来函数

• 买入信号：在当前 K 线收盘生成 → 下一根 K 线开盘价执行
• 卖出信号：在当前 K 线收盘生成 → 当前 K 线收盘价执行

这样可以避免使用已知收盘价来获利的偏差。

交易成本

引擎模拟以下交易成本：

成本项	默认值	说明
手续费	0.1%	按成交额收取
滑点	0.05%	买卖双向滑点

绩效指标

回测完成后自动计算以下指标：

指标	说明
总收益率	(最终权益 - 初始资金) / 初始资金 × 100%
年化收益率	以复利方式年化
夏普比率	(日均收益 / 日收益标准差) × √365
最大回撤	权益从峰值下跌的最大百分比
回撤持续天数	从峰值到恢复（或结束）的最长天数
胜率	盈利交易 / 总交易
盈亏比	总盈利 / 总亏损绝对值
卡尔玛比率	年化收益 / 最大回撤绝对值

API 参考

BacktestConfig

@dataclass
class BacktestConfig:
    symbol: str                          # 交易对
    exchange: str = "binance"            # 交易所
    interval: str = "1h"                 # K 线周期
    start_time: datetime | None = None   # 起始时间
    end_time: datetime | None = None     # 结束时间
    commission_pct: float = 0.001        # 手续费率
    slippage_pct: float = 0.0005         # 滑点率
    min_order_qty: float = 0.001         # 最小下单量
    initial_capital: float = 10_000.0    # 初始资金
    warmup_bars: int = 100               # 预热条数

BacktestEngine

class BacktestEngine:
    def __init__(self, config: BacktestConfig, db_config=None)

    async def run(
        self,
        strategy_cls: Type[BaseStrategy],
        strategy_config: StrategyConfig,
    ) -> BacktestResult

BacktestResult

@dataclass
class BacktestResult:
    config: BacktestConfig          # 回测配置
    strategy_config: dict           # 策略配置
    metrics: BacktestMetrics        # 绩效指标
    trades: list[BacktestTrade]     # 交易记录
    equity_curve: list[dict]        # 资金曲线

    def summary(self) -> str        # 人类可读摘要

编写策略

策略必须继承 BaseStrategy，实现 on_kline() 方法：

from engine.common.base import BaseStrategy, Signal, StrategyConfig
from engine.common.models import Kline

class MyConfig(StrategyConfig):
    param1: int = 10

class MyStrategy(BaseStrategy):
    strategy_type = "my_strategy"

    def __init__(self, config: MyConfig):
        super().__init__(config)
        self._closes = []

    async def on_kline(self, kline: Kline) -> Signal | None:
        self._closes.append(kline.close)
        # 策略逻辑 ...
        if 买入条件:
            return Signal(
                symbol=self.config.symbol,
                side="BUY",
                signal_type="MARKET",
                reason="...",
                timestamp=kline.open_time,
            )
        return None

技术指标库

engine/indicators/ 提供常用的技术指标计算函数，纯 Python 实现，无外部依赖：

from engine.indicators import sma, ema, macd, rsi, bollinger, atr, obv, vwap

closes = [100.0, 101.0, 102.0, ...]
ma = sma(closes, period=20)         # 简单移动平均
ema_vals = ema(closes, period=12)   # 指数移动平均
rsi_vals = rsi(closes, period=14)   # RSI [0, 100]
upper, mid, lower = bollinger(closes, period=20, std=2)  # 布林带
macd_line, signal, hist = macd(closes, fast=12, slow=26, signal=9)  # MACD
atr_vals = atr(highs, lows, closes, period=14)  # ATR

模块	指标	函数
trend	趋势	sma, ema, macd, macd_signal, macd_histogram
momentum	动量	rsi, stoch, stoch_k, stoch_d
volatility	波动率	bollinger, bollinger_upper, bollinger_mid, bollinger_lower, atr
volume	成交量	obv, vwap

所有函数返回与输入等长的 list[float]，不足周期的位置填充为 0.0。

运行示例

cd engine
source .venv/bin/activate
python example/backtest_demo.py