用 Python + backtrader 做专业级策略回测

前面几篇文章里，我们用 pandas 手写了回测逻辑：计算信号、乘以收益率、累乘得到净值曲线。这种方式简单直接，适合快速验证一个想法。

但当你的策略开始变复杂——有止损逻辑、有仓位管理、有多标的轮动、有交易费用——纯手写的回测代码就会变得臃肿且容易出错。

这时候你需要一个回测框架。backtrader 是 Python 量化圈最常用的回测框架之一：它支持事件驱动、仓位管理、交易成本、多标的、多时间周期，而且社区生态比较成熟。

这篇文章的目标是：用 AlphaFeed 拿数据，喂给 backtrader 跑回测。两者各做自己擅长的事——AlphaFeed 管数据获取，backtrader 管回测引擎。

1. 安装依赖

pip install alphafeed backtrader pandas matplotlib

2. 把 AlphaFeed 数据转成 backtrader 格式

backtrader 有自己的数据格式要求。最常用的方式是通过 bt.feeds.PandasData 把 DataFrame 喂进去。

AlphaFeed 返回的 DataFrame 字段名和 backtrader 默认的不完全一样，所以需要做一层映射：

import pandas as pd
import backtrader as bt
from alphafeed import AlphaFeed

def load_alphafeed_data(symbol: str, count: int = 500, period: str = "1d") -> bt.feeds.PandasData:
    af = AlphaFeed()

    df = af.klines.get(
        symbol,
        period=period,
        count=count,
        adjust="forward",
        to_dataframe=True,
    )
    df = df.sort_values("trade_date").reset_index(drop=True)

    df["trade_date"] = pd.to_datetime(df["trade_date"])
    df = df.set_index("trade_date")

    df = df.rename(columns={
        "open": "open",
        "high": "high",
        "low": "low",
        "close": "close",
        "volume": "volume",
    })

    data = bt.feeds.PandasData(
        dataname=df,
        open="open",
        high="high",
        low="low",
        close="close",
        volume="volume",
        openinterest=-1,
    )

    return data

这个函数封装了"从 AlphaFeed 获取数据 → 转成 backtrader 能识别的格式"的全过程。后面所有策略都可以复用它。

3. 第一个 backtrader 策略：双均线交叉

我们先用 backtrader 重写第 01 篇里的双均线策略，看看框架化之后的写法：

import backtrader as bt

class DualMA(bt.Strategy):
    params = (
        ("short_period", 20),
        ("long_period", 60),
    )

    def __init__(self):
        self.ma_short = bt.indicators.SimpleMovingAverage(
            self.data.close, period=self.params.short_period,
        )
        self.ma_long = bt.indicators.SimpleMovingAverage(
            self.data.close, period=self.params.long_period,
        )
        self.crossover = bt.indicators.CrossOver(self.ma_short, self.ma_long)

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

和手写 pandas 回测相比，backtrader 的好处是：信号判断和仓位管理分开了。你不需要手动维护 position 列，不需要手动做 shift(1) 来避免未来函数——框架的事件驱动机制自动保证了这一点。

4. 把数据和策略组装起来跑回测

import backtrader as bt
import pandas as pd
from alphafeed import AlphaFeed

def load_alphafeed_data(symbol: str, count: int = 500) -> bt.feeds.PandasData:
    af = AlphaFeed()
    df = af.klines.get(
        symbol, period="1d", count=count,
        adjust="forward", to_dataframe=True,
    )
    df = df.sort_values("trade_date").reset_index(drop=True)
    df["trade_date"] = pd.to_datetime(df["trade_date"])
    df = df.set_index("trade_date")
    return bt.feeds.PandasData(
        dataname=df,
        open="open", high="high", low="low",
        close="close", volume="volume", openinterest=-1,
    )

class DualMA(bt.Strategy):
    params = (
        ("short_period", 20),
        ("long_period", 60),
    )

    def __init__(self):
        self.ma_short = bt.indicators.SMA(self.data.close, period=self.p.short_period)
        self.ma_long = bt.indicators.SMA(self.data.close, period=self.p.long_period)
        self.crossover = bt.indicators.CrossOver(self.ma_short, self.ma_long)

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


cerebro = bt.Cerebro()

data = load_alphafeed_data("600519.SH", count=800)
cerebro.adddata(data)

cerebro.addstrategy(DualMA)

cerebro.broker.setcash(1_000_000)
cerebro.broker.setcommission(commission=0.001)

cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe", riskfreerate=0.02)
cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown")
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trades")

print(f"初始资金: {cerebro.broker.getvalue():,.0f}")

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

print(f"最终资金: {cerebro.broker.getvalue():,.0f}")
print(f"总收益率: {(cerebro.broker.getvalue() / 1_000_000 - 1) * 100:.2f}%")

sharpe = strategy.analyzers.sharpe.get_analysis()
dd = strategy.analyzers.drawdown.get_analysis()
trades = strategy.analyzers.trades.get_analysis()

print(f"夏普比率: {sharpe.get('sharperatio', 'N/A')}")
print(f"最大回撤: {dd.max.drawdown:.2f}%")
print(f"总交易次数: {trades.total.total}")

cerebro.plot(style="candlestick")

backtrader 自带的 cerebro.plot() 会画出包含 K 线、均线、买卖点标记、资金曲线的完整图表。

5. 加入止损和止盈

纯均线策略没有风控，一旦趋势反转就会坐过山车。加上止损止盈：

import backtrader as bt

class DualMAWithRisk(bt.Strategy):
    params = (
        ("short_period", 20),
        ("long_period", 60),
        ("stop_loss", 0.05),
        ("take_profit", 0.15),
    )

    def __init__(self):
        self.ma_short = bt.indicators.SMA(self.data.close, period=self.p.short_period)
        self.ma_long = bt.indicators.SMA(self.data.close, period=self.p.long_period)
        self.crossover = bt.indicators.CrossOver(self.ma_short, self.ma_long)
        self.entry_price = None

    def next(self):
        if not self.position:
            if self.crossover > 0:
                self.buy()
                self.entry_price = self.data.close[0]
        else:
            current = self.data.close[0]
            pnl_pct = (current - self.entry_price) / self.entry_price

            if pnl_pct <= -self.p.stop_loss:
                self.sell()
                self.log(f"止损平仓 | 亏损 {pnl_pct:.2%}")
            elif pnl_pct >= self.p.take_profit:
                self.sell()
                self.log(f"止盈平仓 | 盈利 {pnl_pct:.2%}")
            elif self.crossover < 0:
                self.sell()
                self.log(f"信号平仓 | 盈亏 {pnl_pct:.2%}")

    def log(self, txt):
        dt = self.data.datetime.date(0)
        print(f"[{dt}] {txt}")

这段代码展示了 backtrader 的一个核心优势：你可以在 next() 方法里自由地写交易逻辑，而不需要像 pandas 回测那样用向量化操作来模拟条件判断。

6. 参数优化

backtrader 支持内置的参数优化（grid search）：

import backtrader as bt

cerebro = bt.Cerebro()

data = load_alphafeed_data("600519.SH", count=800)
cerebro.adddata(data)

cerebro.optstrategy(
    DualMA,
    short_period=range(10, 31, 5),
    long_period=range(40, 81, 10),
)

cerebro.broker.setcash(1_000_000)
cerebro.broker.setcommission(commission=0.001)
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe")

results = cerebro.run()

param_results = []
for run in results:
    for strategy in run:
        sharpe_info = strategy.analyzers.sharpe.get_analysis()
        sr = sharpe_info.get("sharperatio", None)
        param_results.append({
            "short": strategy.params.short_period,
            "long": strategy.params.long_period,
            "sharpe": sr,
            "final_value": strategy.broker.getvalue(),
        })

import pandas as pd
param_df = pd.DataFrame(param_results).sort_values("sharpe", ascending=False, na_position="last")
print("=== 参数优化结果 ===")
print(param_df.head(10).to_string(index=False))

注意：参数优化的结果仅供参考。夏普最高的参数组合可能是过拟合的。你应该把数据分成训练集和测试集，在训练集上选参数，在测试集上验证。

7. 多标的回测

backtrader 支持同时加载多只股票的数据，在一个策略里交叉操作：

import backtrader as bt
import pandas as pd
from alphafeed import AlphaFeed

def load_data(symbol: str, count: int = 500) -> bt.feeds.PandasData:
    af = AlphaFeed()
    df = af.klines.get(
        symbol, period="1d", count=count,
        adjust="forward", to_dataframe=True,
    )
    df = df.sort_values("trade_date").reset_index(drop=True)
    df["trade_date"] = pd.to_datetime(df["trade_date"])
    df = df.set_index("trade_date")
    return bt.feeds.PandasData(
        dataname=df,
        open="open", high="high", low="low",
        close="close", volume="volume", openinterest=-1,
    )


class MultiStockMomentum(bt.Strategy):
    params = (
        ("momentum_period", 20),
        ("rebalance_interval", 20),
        ("top_n", 2),
    )

    def __init__(self):
        self.counter = 0
        self.momentums = {}
        for d in self.datas:
            self.momentums[d._name] = bt.indicators.RateOfChange(
                d.close, period=self.p.momentum_period,
            )

    def next(self):
        self.counter += 1
        if self.counter % self.p.rebalance_interval != 0:
            return

        rankings = []
        for d in self.datas:
            mom = self.momentums[d._name][0]
            rankings.append((d, mom))

        rankings.sort(key=lambda x: x[1], reverse=True)

        top_stocks = set(d._name for d, _ in rankings[:self.p.top_n])

        for d in self.datas:
            if d._name in top_stocks:
                target_pct = 1.0 / self.p.top_n
                self.order_target_percent(d, target=target_pct)
            else:
                if self.getposition(d).size > 0:
                    self.close(d)


cerebro = bt.Cerebro()

symbols = {
    "maotai": "600519.SH",
    "pingan": "601318.SH",
    "byd": "002594.SZ",
    "catl": "300750.SZ",
    "wuliangye": "000858.SZ",
}

for name, sym in symbols.items():
    data = load_data(sym, count=500)
    data._name = name
    cerebro.adddata(data, name=name)

cerebro.addstrategy(MultiStockMomentum)
cerebro.broker.setcash(1_000_000)
cerebro.broker.setcommission(commission=0.001)

cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe")
cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown")

print(f"初始资金: {cerebro.broker.getvalue():,.0f}")

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

print(f"最终资金: {cerebro.broker.getvalue():,.0f}")
print(f"总收益率: {(cerebro.broker.getvalue() / 1_000_000 - 1) * 100:.2f}%")

sharpe = st.analyzers.sharpe.get_analysis()
dd = st.analyzers.drawdown.get_analysis()
print(f"夏普比率: {sharpe.get('sharperatio', 'N/A')}")
print(f"最大回撤: {dd.max.drawdown:.2f}%")

这个策略每 20 天做一次再平衡：计算所有标的的 20 日动量，选动量最强的 2 只等权持有，其余清仓。

8. 自定义 AlphaFeed 数据源类

如果你经常用 AlphaFeed + backtrader，可以封装一个更方便的数据源类：

import pandas as pd
import backtrader as bt
from alphafeed import AlphaFeed


class AlphaFeedData(bt.feeds.PandasData):
    """AlphaFeed 数据源封装，直接传 symbol 即可"""

    @classmethod
    def create(
        cls,
        symbol: str,
        period: str = "1d",
        count: int = 500,
        adjust: str = "forward",
        api_key: str = None,
    ):
        af = AlphaFeed(api_key=api_key) if api_key else AlphaFeed()

        df = af.klines.get(
            symbol,
            period=period,
            count=count,
            adjust=adjust,
            to_dataframe=True,
        )
        df = df.sort_values("trade_date").reset_index(drop=True)
        df["trade_date"] = pd.to_datetime(df["trade_date"])
        df = df.set_index("trade_date")

        return cls(
            dataname=df,
            open="open",
            high="high",
            low="low",
            close="close",
            volume="volume",
            openinterest=-1,
        )


cerebro = bt.Cerebro()
cerebro.adddata(AlphaFeedData.create("600519.SH", count=800))
cerebro.addstrategy(DualMA)
cerebro.broker.setcash(1_000_000)
cerebro.run()

一行 AlphaFeedData.create("600519.SH") 就搞定数据加载，后续所有策略都可以直接用。

9. 回测结果的完整报告

backtrader 内置了丰富的分析器，可以生成专业级的回测报告：

import backtrader as bt

cerebro = bt.Cerebro()

data = load_alphafeed_data("600519.SH", count=800)
cerebro.adddata(data)
cerebro.addstrategy(DualMAWithRisk, stop_loss=0.05, take_profit=0.15)

cerebro.broker.setcash(1_000_000)
cerebro.broker.setcommission(commission=0.001)

cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe", riskfreerate=0.02)
cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown")
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trades")
cerebro.addanalyzer(bt.analyzers.Returns, _name="returns")
cerebro.addanalyzer(bt.analyzers.VWR, _name="vwr")

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

print("=" * 50)
print("回测报告")
print("=" * 50)

returns = st.analyzers.returns.get_analysis()
print(f"总收益率:   {returns.get('rtot', 0) * 100:.2f}%")
print(f"年化收益率: {returns.get('rnorm', 0) * 100:.2f}%")

sharpe = st.analyzers.sharpe.get_analysis()
print(f"夏普比率:   {sharpe.get('sharperatio', 'N/A')}")

vwr = st.analyzers.vwr.get_analysis()
print(f"变异加权收益: {vwr.get('vwr', 'N/A')}")

dd = st.analyzers.drawdown.get_analysis()
print(f"最大回撤:   {dd.max.drawdown:.2f}%")
print(f"最长回撤期: {dd.max.len} 天")

trades = st.analyzers.trades.get_analysis()
print(f"\n交易统计:")
print(f"  总交易: {trades.total.total}")
if hasattr(trades, 'won') and hasattr(trades.won, 'total'):
    print(f"  盈利: {trades.won.total}")
    print(f"  亏损: {trades.lost.total}")
    win_rate = trades.won.total / trades.total.total * 100 if trades.total.total > 0 else 0
    print(f"  胜率: {win_rate:.1f}%")

10. pandas 手写 vs backtrader：怎么选

维度	pandas 手写	backtrader
上手速度	快，几十行就能跑	需要学框架概念
适合场景	快速验证想法、因子研究	完整策略开发、仓位管理
止损/止盈	手写复杂	内置支持
多标的	需要自己管理	框架原生支持
交易费用	手动扣减	配置一次自动计算
参数优化	自己写循环	内置optstrategy
可视化	自己画 matplotlib	cerebro.plot() 一键出图
实盘对接	无	支持（需要写 broker 适配）

建议的路线是：先用 pandas 快速验证 10 个想法，选出 2–3 个有潜力的，再用 backtrader 做完整回测和风控优化。

结语

AlphaFeed 和 backtrader 是"数据层"和"回测层"的自然搭配。

AlphaFeed 把数据获取压缩到一行代码，backtrader 把回测逻辑组织成专业框架。两者组合起来，你可以快速迭代策略、精确计算交易成本、做参数优化、跑多标的轮动——而不需要从零造轮子。

这篇文章提供了一个完整的"数据 → 策略 → 回测 → 报告"流程模板。你可以在此基础上替换策略逻辑、增加标的数量、调整风控参数，逐步构建自己的策略库。

参考链接：

• AlphaFeed 官网：https://alphafeed.org/
• Python SDK 快速开始：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart
• backtrader 文档：https://www.backtrader.com/docu/