在量化投资的世界里，“回测”是连接交易灵感与实盘获利之间最重要的桥梁。很多初学者掌握了基础的 Python 语法，却苦于不知道如何将策略逻辑转化为可视化的回测报告。
今天，我们将通过三份依次进阶的代码示例，详细拆解如何使用 Python 最流行的回测框架 Backtrader，结合 Tushare 数据源，搭建一个支持多品种并行、自定义资金分析且拥有专业图表展示的量化系统。
在开始之前，我们需要认识一下支撑这个系统的“四驾马车”：
Backtrader: Python 界的量化回测“瑞士军刀”，功能强大且灵活。
Tushare: 免费且好用的金融数据接口，为策略提供“燃料”。
Pandas: 数据处理神器，用于清洗和准备行情数据。
Matplotlib: 绘图库，负责将枯燥的数据转化为直观的资金曲线。
策略逻辑：反直觉的 RSI 动量策略
通常教科书上的 RSI（相对强弱指标）用法是“高抛低吸”（>70 卖出，<30 买入）。但本代码采用了动量突破的逻辑：
这个策略不再博取震荡收益，而是试图捕捉强劲的单边趋势。
代码拆解
1. 数据的准备与清洗
量化的第一步是获取高质量数据。代码中定义了三个主要指数：中证500、中证1000、创业板指。代码展示了如何处理多只股票/指数。Backtrader 允许同时加载多个数据源（Data Feeds），这对于测试投资组合策略至关重要。
# 数据获取与清洗核心片段for code, name in indices: df = ts.get_k_data(code, start='2014-01-01') # 从 Tushare 获取数据 # ... 数据清洗步骤 ... data_feed = bt.feeds.PandasData(..., dataname=df) cerebro.adddata(data_feed, name=name) # 加载到回测引擎
2. 策略引擎的实现
这是整个系统的“大脑”。我们在 MultiAssetRSIStrategy 类中定义交易规则。
class MultiAssetRSIStrategy(bt.Strategy): def __init__(self): # 为每一个数据源（指数）单独创建一个 RSI 指标 self.rsi_indicators = [] for data in self.datas: rsi = bt.indicators.RSI(data.close, period=14) self.rsi_indicators.append(rsi) def next(self): # 遍历所有资产，独立判断买卖逻辑 for data, rsi in self.asset_indicators: position = self.getposition(data) if not position and rsi[0] > 70: # 强势突破买入 self.buy(data=data, size=100) elif position and rsi[0] < 30: # 趋势走坏卖出 self.sell(data=data, size=100)
zip(self.datas, self.rsi_indicators)的写法非常 Pythonic，优雅地实现了多品种的循环遍历，互不干扰。3. 独家定制：资金分析器 (Analyzer)
Backtrader 自带的绘图虽然方便，但无法灵活定制。为了画出类似基金净值的曲线，代码中手写了一个 CashValueAnalyzer。
class CashValueAnalyzer(bt.Analyzer): def next(self): # 每一个交易日，记录下当天的总资产（现金+持仓市值） current_value = self.strategy.broker.getvalue() self.cash_values.append(current_value)
4. 专业级可视化 (Visualization)
代码最精彩的部分在于最后的绘图优化。为了解决 Matplotlib 默认日期显示杂乱的问题，代码使用了专业的 Locator 和 Formatter。
图表优化细节：
# 核心绘图优化代码ax.xaxis.set_major_locator(MonthLocator(interval=12, bymonth=1)) # 每年一个主刻度ax.xaxis.set_major_formatter(DateFormatter('%Y')) # 格式：2024ax.xaxis.set_minor_locator(MonthLocator(interval=3)) # 每季度一个副刻度
运行效果展示，当你运行这段代码时，你将看到两部分输出：
1. 终端输出，清晰的资金变动情况：
2. 可视化图表
你会得到一张包含两部分的专业图表：
专业级 Python 量化系统：风控、复利与绩效评估在上小节文章中，我们搭建了一个基础的多品种 RSI 策略框架。今天，我们将给这个引擎装上“涡轮增压”和“安全气囊”。
我们将解决新手最容易忽略的两个问题：“买多少？”（仓位管理）和“怎么输得少？”（风控指标）。我们将把这个“玩具级”的回测脚本升级为“实战入门级”的量化系统。
我们将重点进行以下 4项硬核升级：
仓位管理 (Position Sizing)：从傻瓜式的“固定买100股”升级为“按资金比例下单”（例如每次使用20%资金），让复利效应发挥作用。
风险控制 (Risk Management)：新增移动止损 (Trailing Stop) 逻辑。在趋势策略中，止损比止盈更重要，能保住胜利果实。
交易日志 (Logging System)：增加详细的订单打印功能，让你清楚看到每一笔买卖的价格、费用和盈亏，拒绝“黑箱操作”。
专业绩效评估 (Performance Metrics)：除了看最终资金，我们还要计算 夏普比率 (Sharpe Ratio) 和 最大回撤 (Max Drawdown)，这才是专业机构评价策略的核心指标。
升级代码：
import backtrader as btimport tushare as tsimport pandas as pdimport numpy as npimport matplotlib.pyplot as pltfrom matplotlib.dates import DateFormatter, MonthLocator, date2num# --- 配置 Matplotlib 中文显示 ---plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei", "Heiti TC"]plt.rcParams['axes.unicode_minus'] = False# 1. 资金曲线分析器 (保持不变)class CashValueAnalyzer(bt.Analyzer): def __init__(self): self.cash_values = [] self.dates = [] def next(self): self.dates.append(self.strategy.datas[0].datetime.date(0)) self.cash_values.append(self.strategy.broker.getvalue()) def get_analysis(self): return {'dates': self.dates, 'cash_values': self.cash_values}# 2. 升级版策略：增加日志、风控与动态仓位class ProMultiAssetRSIStrategy(bt.Strategy): params = ( ('rsi_period', 14), ('stop_loss_dist', 0.05), # 5% 移动止损 ('printlog', True), # 是否打印日志 ) def log(self, txt, dt=None): """ 日志打印辅助函数 """ if self.params.printlog: dt = dt or self.datas[0].datetime.date(0) print(f'[{dt.isoformat()}] {txt}') def __init__(self): self.rsi_indicators = [] self.orders = {} # 记录每个资产的订单状态 for data in self.datas: rsi = bt.indicators.RSI(data.close, period=self.params.rsi_period) self.rsi_indicators.append(rsi) self.orders[data] = None # 初始化订单状态 self.asset_indicators = list(zip(self.datas, self.rsi_indicators)) def notify_order(self, order): """ 订单状态变化通知 """ if order.status in [order.Submitted, order.Accepted]: return if order.status in [order.Completed]: if order.isbuy(): self.log(f'BUY EXECUTED: {order.data._name}, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}') elif order.issell(): self.log(f'SELL EXECUTED: {order.data._name}, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm: {order.executed.comm:.2f}') elif order.status in [order.Canceled, order.Margin, order.Rejected]: self.log(f'Order Canceled/Margin/Rejected: {order.data._name}') self.orders[order.data] = None def notify_trade(self, trade): """ 交易结算是通知 (计算盈亏) """ if not trade.isclosed: return self.log(f'OPERATION PROFIT: {trade.data._name}, GROSS {trade.pnl:.2f}, NET {trade.pnlcomm:.2f}') def next(self): for data, rsi in self.asset_indicators: # 如果有未完成的订单，不进行新操作 if self.orders[data]: return position = self.getposition(data) # --- 核心交易逻辑 --- # 1. 买入：无持仓 且 RSI > 70 (动量突破) if not position: if rsi[0] > 70: self.log(f'BUY CREATE: {data._name} (RSI: {rsi[0]:.2f})') # 使用 Sizer 自动计算数量，这里只发信号 self.orders[data] = self.buy(data=data) # 2. 卖出：有持仓 且 (RSI < 30 或 触发移动止损) else: # 移动止损逻辑：如果收盘价跌破了 (买入价 * 0.95)，则止损 # 注意：Backtrader有专门的TrailingStop订单，这里为了演示逻辑手动写个简单的 stop_price = position.price * (1.0 - self.params.stop_loss_dist) if rsi[0] < 30: self.log(f'SELL CREATE (Signal): {data._name} (RSI: {rsi[0]:.2f})') self.orders[data] = self.sell(data=data) elif data.close[0] < stop_price: self.log(f'SELL CREATE (Stop Loss): {data._name} (Price: {data.close[0]:.2f} < Stop: {stop_price:.2f})') self.orders[data] = self.sell(data=data)# 3. 数据准备 (保持稳定)indices = [('000905', '中证500'), ('000852', '中证1000'), ('399006', '创业板指')]cerebro = bt.Cerebro()for code, name in indices: # 注意：实际使用建议将数据下载到本地 CSV，避免频繁请求接口 df = ts.get_k_data(code, start='2018-01-01') # 缩短时间范围以加快演示 if df.empty: continue df['date'] = pd.to_datetime(df['date']) df = df.set_index('date', drop=True) data_feed = bt.feeds.PandasData( dataname=df, fromdate=pd.Timestamp('2018-01-01'), todate=pd.Timestamp.now(), open=0, high=1, low=2, close=3, volume=4, openinterest=-1 ) cerebro.adddata(data_feed, name=name)# 4. 升级配置：仓位与分析器cerebro.addstrategy(ProMultiAssetRSIStrategy)cerebro.addanalyzer(CashValueAnalyzer, _name='cash')# [新增] 夏普比率分析器cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe', timeframe=bt.TimeFrame.Years, riskfreerate=0.02)# [新增] 最大回撤分析器cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')# [升级] 仓位管理：每次交易使用当前总资金的 20%cerebro.addsizer(bt.sizers.PercentSizer, percents=20)cerebro.broker.setcash(1000000.0)cerebro.broker.setcommission(commission=0.001)print('--- 回测开始 ---')results = cerebro.run()strat = results[0]# 5. 打印专业绩效报告print('\n--- 绩效报告 ---')print(f'初始资金: {1000000.00:.2f}')print(f'最终资金: {cerebro.broker.getvalue():.2f}')print(f'夏普比率: {strat.analyzers.sharpe.get_analysis()["sharperatio"]:.4f}')print(f'最大回撤: {strat.analyzers.drawdown.get_analysis()["max"]["drawdown"]:.2f}%')print(f'回撤金额: {strat.analyzers.drawdown.get_analysis()["max"]["moneydown"]:.2f}')# 6. 绘图 (只绘制独立的资金曲线，更清晰)cash_data = strat.analyzers.cash.get_analysis()dates_num = date2num(cash_data['dates'])cash_values = cash_data['cash_values']plt.figure(figsize=(12, 6))plt.plot_date(dates_num, cash_values, 'r-', linewidth=1.5, label='策略净值')# 填充回撤区域 (可视化优化)plt.fill_between(dates_num, cash_values, 1000000, where=(np.array(cash_values) < 1000000), facecolor='red', alpha=0.1, label='亏损区域')plt.fill_between(dates_num, cash_values, 1000000, where=(np.array(cash_values) >= 1000000), facecolor='green', alpha=0.1, label='盈利区域')ax = plt.gca()ax.xaxis.set_major_locator(MonthLocator(interval=12, bymonth=1))ax.xaxis.set_major_formatter(DateFormatter('%Y'))ax.xaxis.set_minor_locator(MonthLocator(interval=3))plt.title('多品种 RSI 策略资金曲线 (含仓位管理)', fontsize=14)plt.grid(True, linestyle='--', alpha=0.5)plt.legend()plt.tight_layout()plt.show()
1. 动态仓位管理 (Smart Sizing)
旧版本每次只买 100 股，这在资金量大时毫无意义。
代码：
cerebro.addsizer(bt.sizers.PercentSizer, percents=20)
原理：无论你现在有 100 万还是 200 万，每次开仓都动用总资金的 20%。这能实现**“赢了加仓，输了减仓”**的复利效果。
2. 交易全透明 (Logging)
不再是黑箱。我们在策略中增加了 notify_order 和 notify_trade 方法。
3. 硬核风控 (Risk Control)
RSI 策略最大的弱点是“钝化”（指标在高位一直不下来，或者暴跌时指标虽低但价格还在跌）。
4. 机构级指标 (Metrics)
光看赚了多少钱是不够的，如果赚 50% 却经历过 40% 的回撤，那这个策略很难坚持。
5. 图表升级细节
我们在绘图中增加了一个“盈亏区域填充”功能：
绿色背景区域：代表当前资金高于初始资金（赚钱）。
红色背景区域：代表当前资金低于初始资金（亏钱）。
这种视觉反馈能让你一眼看出策略在哪些年份表现挣扎，哪些年份在躺赚。
在上一版本中，我们加入了仓位管理和止损。但在实战中，单纯的 RSI 动量策略有一个致命弱点：在长期熊市中，RSI 也会偶尔突破 70，诱骗你进场，然后价格继续暴跌。
这一次，我们将把系统升级为“基金经理级”。我们将引入两个量化交易中的核心概念：
趋势过滤 (Trend Filtering)：解决 RSI 在熊市中频繁“假突破”导致亏损的痛点。
基准对比 (Benchmark Comparison)：不跟大盘比收益的策略都是耍流氓。我们将把策略净值与“沪深300”的走势画在同一张图上，一目了然地看到你是否跑赢了市场。
升级核心：
import backtrader as btimport tushare as tsimport pandas as pdimport numpy as npimport matplotlib.pyplot as pltfrom matplotlib.dates import DateFormatter, MonthLocator, date2num# --- 配置 Matplotlib 中文显示 ---plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei", "Heiti TC"]plt.rcParams['axes.unicode_minus'] = False# 1. 资金曲线分析器class CashValueAnalyzer(bt.Analyzer): def __init__(self): self.cash_values = [] self.dates = [] def next(self): self.dates.append(self.strategy.datas[0].datetime.date(0)) self.cash_values.append(self.strategy.broker.getvalue()) def get_analysis(self): return {'dates': self.dates, 'cash_values': self.cash_values}# 2. 高阶策略：带趋势过滤的 RSIclass TrendFilteredRSIStrategy(bt.Strategy): params = ( ('rsi_period', 14), ('filter_period', 200), # 新增：200日均线作为牛熊分界线 ('stop_loss_dist', 0.08), # 放宽止损到 8% ) def __init__(self): self.inds = {} for data in self.datas: self.inds[data] = {} # RSI 指标 self.inds[data]['rsi'] = bt.indicators.RSI(data.close, period=self.params.rsi_period) # SMA 趋势过滤指标 (200日均线) self.inds[data]['sma'] = bt.indicators.SMA(data.close, period=self.params.filter_period) # 记录订单避免重复提交 self.orders = {data: None for data in self.datas} def notify_order(self, order): if order.status in [order.Completed]: # 简化的日志，只看成交 if order.isbuy(): print(f'[买入] {order.data._name} 价格: {order.executed.price:.2f}') elif order.issell(): print(f'[卖出] {order.data._name} 价格: {order.executed.price:.2f}') self.orders[order.data] = None def next(self): for data in self.datas: if self.orders[data]: continue # 有未完成订单则跳过 position = self.getposition(data) rsi = self.inds[data]['rsi'][0] current_price = data.close[0] sma_filter = self.inds[data]['sma'][0] # --- 核心升级：双重条件开仓 --- # 1. RSI > 70 (动量) # 2. 现价 > 200日均线 (趋势向上) if not position: if rsi > 70 and current_price > sma_filter: self.buy(data=data) print(f' >> 信号触发: {data._name} 突破且位于均线之上') # --- 卖出逻辑 --- else: stop_price = position.price * (1.0 - self.params.stop_loss_dist) # 止盈：RSI 回落 if rsi < 40: self.sell(data=data) # 止损：跌破止损线 elif current_price < stop_price: self.sell(data=data) print(f' >> 止损触发: {data._name}')# 3. 数据准备 (包含策略数据 和 基准数据)cerebro = bt.Cerebro()# 策略交易的标的indices = [('399006', '创业板指'), ('000905', '中证500')]start_date = '2019-01-01'for code, name in indices: df = ts.get_k_data(code, start=start_date) if df.empty: continue df['date'] = pd.to_datetime(df['date']) df = df.set_index('date', drop=True) data = bt.feeds.PandasData( dataname=df, fromdate=pd.Timestamp(start_date), open=0, high=1, low=2, close=3, volume=4, openinterest=-1 ) cerebro.adddata(data, name=name)# 4. 获取基准数据 (沪深300) 用于对比# 注意：这部分数据不加入 cerebro，而是用于最后画图benchmark_df = ts.get_k_data('hs300', start=start_date)benchmark_df['date'] = pd.to_datetime(benchmark_df['date'])benchmark_df = benchmark_df.set_index('date')# 计算基准的归一化净值 (从1开始)benchmark_norm = benchmark_df['close'] / benchmark_df['close'].iloc[0] * 1000000# 5. 系统配置cerebro.addstrategy(TrendFilteredRSIStrategy)cerebro.addanalyzer(CashValueAnalyzer, _name='cash')cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe', riskfreerate=0.02, timeframe=bt.TimeFrame.Years)cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown')# 动态仓位：资金分成 2 份，每份用 45% (留点余地防止拒单)cerebro.addsizer(bt.sizers.PercentSizer, percents=45) cerebro.broker.setcash(1000000.0)cerebro.broker.setcommission(commission=0.001)print('--- 回测开始 ---')results = cerebro.run()strat = results[0]# 6. 最终结果与绘图final_value = cerebro.broker.getvalue()print(f'\n策略最终资金: {final_value:.2f}')print(f'基准最终模拟: {benchmark_norm.iloc[-1]:.2f}')print(f'夏普比率: {strat.analyzers.sharpe.get_analysis()["sharperatio"]:.4f}')# --- 绘图升级：策略 vs 基准 ---cash_data = strat.analyzers.cash.get_analysis()dates = cash_data['dates']strategy_values = cash_data['cash_values']plt.figure(figsize=(14, 7))# 绘制策略曲线plt.plot(dates, strategy_values, 'r-', linewidth=2, label='我的量化策略')# 绘制基准曲线 (需要对齐时间轴)# 将 benchmark 数据截取到策略相同的时间段aligned_benchmark = benchmark_norm.reindex(dates).fillna(method='ffill')plt.plot(dates, aligned_benchmark, 'k--', alpha=0.6, linewidth=1.5, label='沪深300 (基准)')# 填充超额收益区域plt.fill_between(dates, strategy_values, aligned_benchmark, where=(np.array(strategy_values) > np.array(aligned_benchmark)), facecolor='red', alpha=0.1, label='跑赢市场')plt.fill_between(dates, strategy_values, aligned_benchmark, where=(np.array(strategy_values) <= np.array(aligned_benchmark)), facecolor='gray', alpha=0.1, label='跑输市场')# 格式化ax = plt.gca()ax.xaxis.set_major_locator(MonthLocator(interval=12, bymonth=1))ax.xaxis.set_major_formatter(DateFormatter('%Y'))plt.title(f'策略 vs 沪深300：是否存在超额收益 (Alpha)?', fontsize=14)plt.ylabel('资金净值')plt.grid(True, linestyle='--', alpha=0.5)plt.legend()plt.tight_layout()plt.show()
深度解析升级内容
1. 增加系统过滤条件
新手死于追高，老手死于抄底，高手死于震荡。 RSI 策略最大的坑在于：在熊市反弹（Dead Cat Bounce）时，RSI 很容易突破 70，发出买入信号，然后市场继续暴跌。
代码逻辑：if rsi > 70 and current_price > sma_filter
白话解释：这就像是在说，“我只在天气晴朗（价格在200日均线之上）的时候，才去冲浪（RSI 突破）”。
效果：这会大幅减少交易次数，可能会让你错过一些机会，但能帮你躲过 2018 年或 2022 年那种持续阴跌的“绞肉机”行情。
2. 基准对比：寻找真实的 Alpha
很多策略号称“年化 20%”，结果一看那年大盘涨了 30%。这种策略是没有价值的，不如直接买指数 ETF。
3. 多标的并行与资金分配
代码中将资金分配逻辑调整为：
cerebro.addsizer(bt.sizers.PercentSizer, percents=45)
因为我们有两个标的（创业板 + 中证500），如果设为 50%，当资金稍微有一点波动或产生佣金时，第二只股票可能因为资金不足（Cash < 50%）而无法买入。设为 45% 是一个安全的工程实践技巧。
运行结果
至此，我们的系统已经具备了择时（RSI）、过滤（均线）、风控（止损）和 归因分析（Benchmark）等功能。文中代码与条件设定均为示例载体，核心目标在于传递解决问题的方法论——建议聚焦于思路的理解与迁移，而非局限于示例结果本身。在掌握基础方法后，可结合自身思考对代码进行迭代优化，构建更健壮、高效的量化策略，真正实现从“理解”到构建策略”的进阶。
如您想获取文章完整代码，请关注数量技术宅公众号，后台回复关键词：组合测试
