Backtrader 2026: Python回测引擎以100倍速度验证交易策略 —— 完整指南

python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装Backtrader (截至2026年5月为v1.9.81.127)
pip install backtrader

# 可选: 安装TA-Lib封装以扩展指标
pip install TA-Lib

# 可选: 安装matplotlib用于绘图
pip install matplotlib

import backtrader as bt
print(bt.__version__)

# 预期输出: 1.9.81.127 或更高

import backtrader as bt
import datetime

class SmaCross(bt.Strategy):
    params = dict(fast=10, slow=30)

    def __init__(self):
        self.fast_sma = bt.indicators.SMA(period=self.p.fast)
        self.slow_sma = bt.indicators.SMA(period=self.p.slow)
        self.crossover = bt.indicators.CrossOver(self.fast_sma, self.slow_sma)

    def next(self):
        if not self.position:  # 不在市场中
            if self.crossover > 0:  # 快线上穿慢线
                self.buy()
        elif self.crossover < 0:  # 快线下穿慢线
            self.sell()

# 创建Cerebro引擎
cerebro = bt.Cerebro()
cerebro.addstrategy(SmaCross)

# 加载数据
import yfinance as yf
df = yf.download("AAPL", start="2020-01-01", end="2025-01-01")
data = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data)

# 设置初始资金
cerebro.broker.setcash(10000.0)

# 运行回测
print(f"起始组合价值: {cerebro.broker.getvalue():.2f}")
cerebro.run()
print(f"最终组合价值: {cerebro.broker.getvalue():.2f}")

# 绘制结果
cerebro.plot()

class RSIMeanReversion(bt.Strategy):
    params = dict(rsi_period=14, oversold=30, overbought=70)

    def __init__(self):
        self.rsi = bt.indicators.RSI(period=self.p.rsi_period)

    def next(self):
        if not self.position:
            if self.rsi < self.p.oversold:
                self.buy()
        else:
            if self.rsi > self.p.overbought:
                self.sell()

    def notify_order(self, order):
        if order.status in [order.Completed]:
            if order.isbuy():
                print(f"买入执行价格 {order.executed.price:.2f}")
            else:
                print(f"卖出执行价格 {order.executed.price:.2f}")

class BollingerBreakout(bt.Strategy):
    params = dict(period=20, devfactor=2.0)

    def __init__(self):
        self.bbands = bt.indicators.BollingerBands(
            period=self.p.period, devfactor=self.p.devfactor
        )
        self.atr = bt.indicators.ATR(period=14)

    def next(self):
        if not self.position:
            if self.data.close > self.bbands.lines.top:
                # 基于ATR的仓位管理买入突破
                size = int(self.broker.getvalue() * 0.02 / self.atr[0])
                self.buy(size=size)
        else:
            if self.data.close < self.bbands.lines.mid:
                self.sell()

    def notify_trade(self, trade):
        if trade.isclosed:
            print(f"交易盈亏: {trade.pnlcomm:.2f}")

class MultiTimeframeMomentum(bt.Strategy):
    params = dict(daily_period=20, weekly_period=10)

    def __init__(self):
        # 日线SMA
        self.daily_sma = bt.indicators.SMA(self.data0, period=self.p.daily_period)
        # 周线SMA (使用data1作为周线重采样数据)
        self.weekly_sma = bt.indicators.SMA(self.data1, period=self.p.weekly_period)

    def next(self):
        # 仅在日线和周线趋势一致时交易
        if (self.data0.close > self.daily_sma[0] and
            self.data1.close > self.weekly_sma[0] and
            not self.position):
            self.buy()
        elif (self.data0.close < self.daily_sma[0] and
              self.data1.close < self.weekly_sma[0] and
              self.position):
            self.sell()

import backtrader.feeds as btfeeds
import yfinance as yf

# 下载并加载
df = yf.download("SPY", start="2020-01-01", end="2026-01-01")
data = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data)

data = btfeeds.GenericCSVData(
    dataname='btc_usd.csv',
    dtformat='%Y-%m-%d',
    datetime=0, open=1, high=2, low=3, close=4, volume=5,
    fromdate=datetime.datetime(2020, 1, 1),
    todate=datetime.datetime(2026, 1, 1)
)
cerebro.adddata(data)

# 添加SPY和VIX用于波动率过滤交易
spy_data = bt.feeds.PandasData(dataname=spy_df, name="SPY")
vix_data = bt.feeds.PandasData(dataname=vix_df, name="VIX")

cerebro.adddata(spy_data)
cerebro.adddata(vix_data)

# 在策略中: self.datas[0] = SPY, self.datas[1] = VIX

import ccxt

exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv("BTC/USDT", timeframe="1d", since=1577836800000)

# 转换为pandas DataFrame
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)

data = bt.feeds.PandasData(dataname=df)
cerebro.adddata(data)

import backtrader as bt

class SmaCross(bt.Strategy):
    params = dict(fast=10, slow=30)

    def __init__(self):
        self.fast_sma = bt.indicators.SMA(period=self.p.fast)
        self.slow_sma = bt.indicators.SMA(period=self.p.slow)
        self.crossover = bt.indicators.CrossOver(self.fast_sma, self.slow_sma)

    def next(self):
        if not self.position and self.crossover > 0:
            self.buy()
        elif self.position and self.crossover < 0:
            self.sell()

cerebro = bt.Cerebro()

# 参数网格搜索: fast=[5,10,15], slow=[20,30,40]
cerebro.optstrategy(
    SmaCross,
    fast=range(5, 20, 5),
    slow=range(20, 50, 10)
)

# 添加数据和经纪商
data = bt.feeds.PandasData(dataname=yf.download("AAPL", start="2020-01-01", end="2025-01-01"))
cerebro.adddata(data)
cerebro.broker.setcash(10000.0)
cerebro.broker.setcommission(commission=0.001)  # 每笔交易0.1%

# 添加分析器
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')
cerebro.addanalyzer(bt.analyzers.Returns, _name='returns')

# 运行优化 (使用所有CPU核心)
results = cerebro.run(maxcpus=4)

# 按夏普比率提取最佳结果
best = max(results, key=lambda r: r[0].analyzers.sharpe.get_analysis()['sharperatio'] or 0)
print(f"最佳夏普比率: {best[0].analyzers.sharpe.get_analysis()['sharperatio']:.2f}")
print(f"最佳参数: fast={best[0].params.fast}, slow={best[0].params.slow}")

import backtrader as bt
import ccxt

class LiveStrategy(bt.Strategy):
    def __init__(self):
        self.rsi = bt.indicators.RSI(period=14)

    def next(self):
        if not self.position and self.rsi < 30:
            self.buy(size=0.001)  # 0.001 BTC
        elif self.position and self.rsi > 70:
            self.sell(size=0.001)

# 配置实盘交易
cerebro = bt.Cerebro()
cerebro.addstrategy(LiveStrategy)

# 使用CCXT经纪商封装
from ccxtbt import CCXTStore
store = CCXTStore(exchange='binance', currency='USDT',
                  config={'apiKey': 'YOUR_KEY', 'secret': 'YOUR_SECRET'})
broker = store.getbroker()
cerebro.setbroker(broker)

# 获取实时数据
data = store.getdata(dataname='BTC/USDT', timeframe=bt.TimeFrame.Minutes, compression=60)
cerebro.adddata(data)

cerebro.run()

FROM python:3.12-slim

WORKDIR /app
RUN pip install --no-cache-dir backtrader pandas numpy matplotlib yfinance

COPY strategy.py .
COPY data/ ./data/

CMD ["python", "strategy.py"]

# docker-compose.yml
version: '3.8'
services:
  backtrader:
    build: .
    volumes:
      - ./results:/app/results
    environment:
      - INITIAL_CASH=100000
    restart: unless-stopped

# 添加到crontab以在每天早6点运行回测
0 6 * * * cd /path/to/strategy && /path/to/venv/bin/python run_backtest.py >> logs/backtest.log 2>&1

# 真实佣金: 每股$0.01, 最低$1
commission_info = bt.CommissionInfo(
    commission=0.01,
    mult=1.0,
    margin=None,
    commtype=bt.CommissionInfo.COMM_FIXED,
    mincom=1.0
)
cerebro.broker.addcommissioninfo(commission_info)

# 滑点: 0.1%成交价影响
cerebro.broker.set_slippage_perc(perc=0.001)

def walk_forward_analysis(data, train_days=252, test_days=63):
    """运行滚动训练/测试分割以验证稳健性。"""
    results = []
    total_bars = len(data)
    start = 0

    while start + train_days + test_days < total_bars:
        train_data = data[start:start + train_days]
        test_data = data[start + train_days:start + train_days + test_days]

        # 在训练集上优化，在未见数据上测试
        cerebro = bt.Cerebro()
        cerebro.optstrategy(MyStrategy, param1=range(5, 20))
        cerebro.adddata(train_data)
        best_params = cerebro.run()

        cerebro2 = bt.Cerebro()
        cerebro2.addstrategy(MyStrategy, **best_params)
        cerebro2.adddata(test_data)
        result = cerebro2.run()
        results.append(result)

        start += test_days

    return results

class EquityCurve(bt.observer.Observer):
    lines = ('equity',)
    plotinfo = dict(plot=True, subplot=True)

    def next(self):
        self.lines.equity[0] = self._owner.broker.getvalue()

# 添加到cerebro
cerebro.addobserver(EquityCurve)

import logging

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class RiskManagedStrategy(bt.Strategy):
    params = dict(max_risk_per_trade=0.02, max_drawdown=0.15)

    def __init__(self):
        self.peak_value = self.broker.getvalue()

    def next(self):
        current_value = self.broker.getvalue()
        self.peak_value = max(self.peak_value, current_value)
        drawdown = (self.peak_value - current_value) / self.peak_value

        if drawdown > self.p.max_drawdown:
            logger.warning(f"触及最大回撤: {drawdown:.2%}. 平掉所有仓位。")
            self.close()
            return

        # ... 策略逻辑的其余部分

class CustomIndicator(bt.Indicator):
    lines = ('myline',)
    params = dict(period=20)

    def __init__(self):
        self.addminperiod(self.p.period)

    def next(self):
        # 你的自定义计算
        self.lines.myline[0] = sum(self.data.get(size=self.p.period)) / self.p.period