关于程序员:backtrader-自定义分析器解决多股回测难分析困难问题

backtrader 自定义分析器，解决多股回测剖析艰难问题

解决了啥：

1. 解决回测后获取要害指标
2. 解决多股回测，获取订单剖析
3. 解决多股回测交易点可视化标识

效果图

通过自定义分析器 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 bt
import numpy as np
import pandas as pd


class 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 地位的数据为回测工夫参考，不参加回测，个别是大盘指数。

订单分析器

记录开平仓所有订单的指标，订单号，股票，买入日期，卖价，卖出日期，买价，收益率，利润，利润总资产比，股数，股本，仓位比，累计收益，持股天数，最大利润，最大亏损。

通过该分析器，岂但能够获取到所有的订单详细情况，还能够获取交易胜利的股票以及对应股票的交易点。

该分析器代码参考了链接 [1]

TradeListAnalyzer 代码

import backtrader as bt
import pandas as pd


class 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)})

应用

backtrader 如何应用这两个自定义分析器

增加自定义分析器

关上交易订单记录

获取后果

到这里，能够获取到回测的后果，通过回测的后果，本人能够轻松实现多股回测的可视化。

参考链接

[1] 订单分析器：https://community.backtrader.com/topic/1274/closed-trade-list…

写于 2023 年 06 月 18 日 10:21

本文由 mdnice 多平台公布