backtrader 自定义分析器，解决多股回测剖析艰难问题
解决了啥：
- 解决回测后获取要害指标
- 解决多股回测，获取订单剖析
- 解决多股回测交易点可视化标识
效果图
通过自定义分析器 KeyIndicatorAnalyzer，TradeListAnalyzer，获取回测后果数据，通过回测数据能够轻松可视化回测后果。
可视化局部须要本人通过后果数据实现，这部分前面有空再写写。
要害指标分析器：KeyIndicatorAnalyzer 分析器获取
订单分析器：TradeListAnalyzer 分析器获取
个股交易点，TradeListAnalyzer 分析器获取
要害指标分析器
该指标器次要剖析策略：累计收益率,年化收益率, 最大回撤，胜率, 夏普率, 凯利比率,近7天收益率, 近30天收益率,佣金占资产比, 开平仓总次数等，通过这些重要指标反映出策略是否可行。
夏普率: 它的定义是投资收益与无风险收益之差的期望值，再除以投资标准差（即其波动性）。夏普率越高越好，一般来说，夏普率大于 1.0 就是很不错的了。
夏普率的计算公式：(Rp-Rf)/p
其中，Rp为投资组合的预期收益率，Rf为无风险收益率，p为投资组合的标准差。
公认默认无风险收益率为年化3%
公式：sharpe = (回报率均值 - 无风险利率) / 回报率标准差。
凯利公式： 定义：计算每次交易，投入资金占总资金的最优比率的剖析者，宣称每次交易按此比例投入资金失去的回报最大，危险最小。
公式：K = W - [(1 - W) / R]
其中，K为凯利公式，W胜率，R为盈亏比，即均匀盈利除以均匀损失。
解读：如果凯利比率为负，阐明该投资策略不可行，应放弃；如果凯利比率为正，例如 kelly_percent = 0.2，阐明每次交易投入资金占总资金的20%为最优比率。
其余指标： 见字知义。
多说无益，代码表白了所有想说的话。
KeyIndicatorAnalyzer 代码
import backtrader as btimport numpy as npimport pandas as pdclass KeyIndicatorAnalyzer(bt.Analyzer): """ 要害指标分析器 """ def __init__(self): super(KeyIndicatorAnalyzer, self).__init__() # 年 period self.year_period = 252 # 月 period self.month_period = 21 # 周 period self.week_period = 5 # 每日详情 self.daily_details = [] # 佣金 self.commission = 0 # 盈利 self.win_list = [] # 亏损 self.loss_list = [] # 重要指标 self.key_indicators_df = pd.DataFrame( columns=[ '策略', '累计收益率', '年化收益率', '最大回撤', '胜率', '夏普率', '凯利比率', '近7天收益率', '近30天收益率', '佣金占资产比', '开平仓总次数' ]) # 每日详情指标，用于画图，{本策略：DataFrame, 基准名：DataFrame}，其中基准名，是最初传进来的基准名 self.daily_chart_dict = dict() def get_analysis_data(self, benchmark_df, benchmark_name): """ 获取剖析数据，传基准数据过去，比照应用的。 @param benchmark_df: 基准数据 @param benchmark_name: 基准名称 """ self._calculate_benchmark_indicators(benchmark_df, benchmark_name) return self.key_indicators_df, self.daily_chart_dict def _calculate_benchmark_indicators(self, benchmark_df, benchmark_name): """ 计算基准的重要指标 """ series = benchmark_df['close'] total_return = self.total_return(series) annual_return = self.annual_return(series) period = self.week_period recent_7_days_return = self.recent_period_return(series, period) period = self.month_period recent_30_days_return = self.recent_period_return(series, period) max_drawdown = self.max_drawdown(series) sharp_ratio = self.sharp_ratio(series) self.key_indicators_df.loc[len(self.key_indicators_df)] = [ benchmark_name, total_return, annual_return, max_drawdown, None, sharp_ratio, None, recent_7_days_return, recent_30_days_return, None, None ] # 收益率走势 df = pd.DataFrame(index=benchmark_df.index) s = self.yield_curve(series) # 插入一列 df.insert(0, '收益率', s) df.index.name = '日期' self.daily_chart_dict[benchmark_name] = df def next(self): super(KeyIndicatorAnalyzer, self).next() # 以后日期 current_date = self.strategy.data.datetime.date(0) # 总资产 total_value = self.strategy.broker.getvalue() # 现金 cash = self.strategy.broker.getcash() self.daily_details.append({ '日期': current_date, '总资产': total_value, '现金': cash }) def notify_trade(self, trade): # 交易敞开 if trade.isclosed: # 佣金 self.commission += trade.commission # 盈利与亏损 if trade.pnlcomm >= 0: # 盈利退出盈利列表，利润 0 算盈利 self.win_list.append(trade.pnlcomm) else: # 亏损退出亏损列表 self.loss_list.append(trade.pnlcomm) def stop(self): # 胜率 if self._win_times() + self._loss_times() == 0: win_rate = 0 else: win_percent = self._win_times() / (self._win_times() + self._loss_times()) win_rate = f'{round(win_percent * 100, 2)}%' df = pd.DataFrame(self.daily_details) # 累计收益率 total_return = self.total_return(df['总资产']) # 年化收益率 annual_return = self.annual_return(df['总资产']) # 最近7天收益率 period = self.week_period recent_7_days_return = self.recent_period_return(df['总资产'], period) # 最近30天收益率 period = self.month_period recent_30_days_return = self.recent_period_return(df['总资产'], period) # 最大回撤 max_drawdown = self.max_drawdown(df['总资产']) # 计算夏普率 sharp_ratio = self.sharp_ratio(df['总资产']) # 计算凯利比率 kelly_percent = self.kelly_percent() # 佣金占总资产比 commission_percent = self.commission_percent(df['总资产']) # 交易次数 trade_times = self._win_times() + self._loss_times() # 本策略的指标 self.key_indicators_df.loc[len(self.key_indicators_df)] = [ '本策略', total_return, annual_return, max_drawdown, win_rate, sharp_ratio, kelly_percent, recent_7_days_return, recent_30_days_return, commission_percent, trade_times ] # 收益率走势 df['收益率'] = self.yield_curve(df['总资产']) df.set_index('日期', inplace=True) # 每日详情指标输入 self.daily_chart_dict['本策略'] = df def commission_percent(self, series) -> str: """ 佣金比例 """ percent = self.commission / series.iloc[0] return f'{round(percent * 100, 2)}%' def yield_curve(self, series) -> pd.Series: """ 收益率曲线 """ percent = (series - series.iloc[0]) / series.iloc[0] return round(percent * 100, 2) def total_return(self, series) -> str: """ 累计收益率 """ percent = (series.iloc[-1] - series.iloc[0]) / series.iloc[0] return f'{round(percent * 100, 2)}%' def annual_return(self, series) -> str: """ 年化收益率 """ percent = (series.iloc[-1] - series.iloc[0]) / series.iloc[0] / len(series) * self.year_period return f'{round(percent * 100, 2)}%' def recent_period_return(self, series, period) -> str: """ 最近一段时间收益率 """ percent = (series.iloc[-1] - series.iloc[-period]) / series.iloc[-period] return f'{round(percent * 100, 2)}%' def max_drawdown(self, series) -> str: """ 最大回撤 """ s = (series - series.expanding().max()) / series.expanding().max() percent = s.min() return f'{round(percent * 100, 2)}%' def sharp_ratio(self, series) -> float: """ 夏普率 夏普率：它的定义是投资收益与无风险收益之差的期望值，再除以投资标准差（即其波动性） 夏普率越高，代表每接受一单位的危险，会产生较多的超额报酬。 夏普率越低，代表每接受一单位的危险，会产生较少的超额报酬。 夏普率为正，代表该投资报酬率高于无风险收益率，反之则低于无风险收益率。 夏普率为负，代表该投资报酬率为负，亦即投资损失。 夏普率越高越好，一般来说，夏普率大于1.0就是很不错的了。 夏普率的计算公式：(Rp-Rf)/p 其中，Rp为投资组合的预期收益率，Rf为无风险收益率，p为投资组合的标准差。 公认默认无风险收益率为年化3% 公式：sharpe = (回报率均值 - 无风险利率) / 回报率标准差 """ ret_s = series.pct_change().fillna(0) avg_ret_s = ret_s.mean() avg_risk_free = 0.03 / self.year_period sd_ret_s = ret_s.std() sharp = (avg_ret_s - avg_risk_free) / sd_ret_s sharp_year = round(np.sqrt(self.year_period) * sharp, 3) return sharp_year def kelly_percent(self) -> str: """ 凯利公式 定义：计算每次交易，投入资金占总资金的最优比率的剖析者， 宣称每次交易按此比例投入资金失去的回报最大，危险最小。 公式：K = W - [(1 - W) / R] 其中，K为凯利公式，W胜率，R为盈亏比，即均匀盈利除以均匀损失。 解读：如果凯利比率为负，阐明该投资策略不可行，应放弃；如果凯利比率为正， 例如 kelly_percent = 0.2，阐明每次交易投入资金占总资金的20%为最优比率。 未必牢靠，只是个参考 """ win_times = self._win_times() loss_times = self._loss_times() if win_times > 0 and loss_times > 0: avg_win = np.average(self.win_list) # 均匀盈利 avg_loss = abs(np.average(self.loss_list)) # 均匀亏损，取绝对值 win_loss_ratio = avg_win / avg_loss # 盈亏比 if win_loss_ratio == 0: kelly_percent = None else: sum_trades = win_times + loss_times win_percent = win_times / sum_trades # 胜率 # 计算凯利比率 # 即每次交易投入资金占总资金的最优比率 kelly_percent = win_percent - ((1 - win_percent) / win_loss_ratio) else: kelly_percent = None # 信息有余 return f'{round(kelly_percent * 100, 2)}%' if kelly_percent else None def _win_times(self): """ 盈利次数 """ return len(self.win_list) def _loss_times(self): """ 亏损次数 """ return len(self.loss_list)应用
多股回测时，须要遵守规则，规定回测第 0 地位的数据为回测工夫参考，不参加回测，个别是大盘指数。
| 函数 | 参数 | 阐明 |
|---|---|---|
| get_analysis_data() | benchmark_df: 基准参考指数，例如：沪深 300<br/> benchmark_name:参考指数名称 | 返回：self.key_indicators_df：本策略和参考策略的要害指标 dataframe<br/>self.daily_chart_dict: 参考指数和本策略收益走势，key 为策略名，value 为收益走势 dataframe |
| _calculate_benchmark_indicators() | benchmark_df：参考指数, <br/>benchmark_name： 参考名称 | 计算参考指数的要害指标，用于比照本策略 |
订单分析器
记录开平仓所有订单的指标，订单号，股票，买入日期，卖价，卖出日期，买价，收益率，利润，利润总资产比，股数，股本，仓位比，累计收益，持股天数，最大利润，最大亏损。
通过该分析器，岂但能够获取到所有的订单详细情况，还能够获取交易胜利的股票以及对应股票的交易点。
该分析器代码参考了链接[1]
TradeListAnalyzer 代码
import backtrader as btimport pandas as pdclass TradeListAnalyzer(bt.Analyzer): """ 交易列表分析器 https://community.backtrader.com/topic/1274/closed-trade-list-including-mfe-mae-analyzer/2 """ def __init__(self): self.trades = [] self.cum_profit = 0.0 def get_analysis(self) -> tuple: """ 获取剖析数据 @return: 交易订单列表，交易日期 """ trade_list_df = pd.DataFrame(self.trades) return trade_list_df, self._get_trade_date(trade_list_df) def _get_trade_date(self, trade_list_df): """ 获取交易日期 @return: 交易日期，获取某只股票的交易日期， 返回字典，key为股票名，value为(买入日期列表，卖出日期列表) """ trade_dict = dict() if not trade_list_df.empty: # 分组，找出交易日期 grouped = trade_list_df.groupby('股票') for name, group in grouped: buy_date_list = list(group['买入日期']) sell_date_list = list(group['卖出日期']) # 判断是否有交易日期 if trade_dict.get(name) is None: trade_dict[name] = (buy_date_list, sell_date_list) else: trade_dict[name][0].extend(buy_date_list) trade_dict[name][1].extend(sell_date_list) return trade_dict def notify_trade(self, trade): if trade.isclosed: total_value = self.strategy.broker.getvalue() dir = 'short' if trade.history[0].event.size > 0: dir = 'long' pricein = trade.history[len(trade.history) - 1].status.price priceout = trade.history[len(trade.history) - 1].event.price datein = bt.num2date(trade.history[0].status.dt) dateout = bt.num2date(trade.history[len(trade.history) - 1].status.dt) if trade.data._timeframe >= bt.TimeFrame.Days: datein = datein.date() dateout = dateout.date() pcntchange = 100 * priceout / pricein - 100 pnl = trade.history[len(trade.history) - 1].status.pnlcomm pnlpcnt = 100 * pnl / total_value barlen = trade.history[len(trade.history) - 1].status.barlen pbar = pnl / barlen self.cum_profit += pnl size = value = 0.0 for record in trade.history: if abs(size) < abs(record.status.size): size = record.status.size value = record.status.value highest_in_trade = max(trade.data.high.get(ago=0, size=barlen + 1)) lowest_in_trade = min(trade.data.low.get(ago=0, size=barlen + 1)) hp = 100 * (highest_in_trade - pricein) / pricein lp = 100 * (lowest_in_trade - pricein) / pricein if dir == 'long': mfe = hp mae = lp if dir == 'short': mfe = -lp mae = -hp self.trades.append( {'订单': trade.ref, '股票': trade.data._name, # 'dir': dir, '买入日期': datein, '卖价': round(pricein, 2), '卖出日期': dateout, '买价': round(priceout, 2), '收益率%': round(pcntchange, 2), '利润': round(pnl, 2), '利润总资产比%': round(pnlpcnt, 2), '股数': size, '股本': round(value, 2), '仓位比%': round(value / total_value * 100, 2), '累计收益': round(self.cum_profit, 2), '持股天数': barlen, # 以每根 bar 的工夫为单位，这里按天计算 # 'pnl/bar': round(pbar, 2), '最大利润%': round(mfe, 2), '最大亏损%': round(mae, 2)})应用
| 函数 | 参数 | 阐明 |
|---|---|---|
| get_analysis() | trade_list_df：订单交易列表 dataframe<br/>self._get_trade_date(trade_list_df)订单交易列表里获取交易的股票和对应的交易日期 trade_dict 类型 dict 返回字典，key为股票名，value为(买入日期列表，卖出日期列表) | 回测结束后调用获取 |
backtrader 如何应用这两个自定义分析器
增加自定义分析器
关上交易订单记录
获取后果
到这里，能够获取到回测的后果，通过回测的后果，本人能够轻松实现多股回测的可视化。
参考链接
[1] 订单分析器： https://community.backtrader.com/topic/1274/closed-trade-list...
写于 2023 年 06 月 18 日 10:21
本文由mdnice多平台公布