Backtrader版本“积木式”AI量化回测系统（全部源代码+数据下载）

解析一下我们支持Backtrader引擎的策略代码：

核心代码继承自bt.Strategy，这是与backtrader内核交互的基础代码。

nofify_order, notify_trade可以打印订单、交易等信息。

在next函数里执行algo_list。

我们的框架基本上实现与底层内核解耦，底层换成pybroker或者bt都是比较容易的事情。

下面是backtrader的策略模板，这个模板是可以复用的，网上很多代码把这块贴了一遍又一遍，其实与真正的策略逻辑没多大关系。

我们的系统，没有使用backtrader的指标体系，数据都是使用我们的“因子表达式”来计算的。因此，与引擎的耦合很小。

我们仅使用backtrader的下单函数，比如buy, sell, order_target_percent这样的调仓，交易函数。

另外还会读取backtrader的持仓情况等。

import backtrader as bt
import pandas as pd
from loguru import logger


class StrategyBase(bt.Strategy):
    def log(self, txt, dt=None):
        dt = dt or self.datas[0].datetime.date(0)
        logger.info('%s, %s' % (dt.isoformat(), txt))

    # 取当前的日期
    def get_current_dt(self):
        #print(self.datas[0].datetime)
        dt = self.datas[0].datetime.date(0).strftime('%Y-%m-%d')
        #print(dt)
        return dt

    # 取当前持仓的data列表
    def get_current_holding_datas(self):
        holdings = []
        for data in self.datas:
            if self.getposition(data).size > 0:
                holdings.append(data)
        return holdings

    # 打印订单日志
    def notify_order(self, order):

        order_status = ['Created', 'Submitted', 'Accepted', 'Partial',
                        'Completed', 'Canceled', 'Expired', 'Margin', 'Rejected']
        # 未被处理的订单
        if order.status in [order.Submitted, order.Accepted]:
            return
            self.log('未处理订单：订单号:%.0f, 标的: %s, 状态状态: %s' % (order.ref,
                                                           order.data._name,
                                                           order_status[order.status]))
            return
        # 已经处理的订单
        if order.status in [order.Partial, order.Completed]:

            if order.isbuy():
                self.log(
                    'BUY EXECUTED, 状态: %s, 订单号:%.0f, 标的: %s, 数量: %.2f, 价格: %.2f, 成本: %.2f, 手续费 %.2f' %
                    (order_status[order.status],  # 订单状态
                     order.ref,  # 订单编号
                     order.data._name,  # 股票名称
                     order.executed.size,  # 成交量
                     order.executed.price,  # 成交价
                     order.executed.value,  # 成交额
                     order.executed.comm))  # 佣金
            else:  # Sell
                self.log(
                    'SELL EXECUTED, status: %s, ref:%.0f, name: %s, Size: %.2f, Price: %.2f, Cost: %.2f, Comm %.2f' %
                    (order_status[order.status],
                     order.ref,
                     order.data._name,
                     order.executed.size,
                     order.executed.price,
                     order.executed.value,
                     order.executed.comm))

        elif order.status in [order.Canceled, order.Margin, order.Rejected, order.Expired]:
            # order.Margin资金不足，订单无法成交
            # 订单未完成
            self.log('未完成订单，订单号:%.0f, 标的 : %s, 订单状态: %s' % (
                order.ref, order.data._name, order_status[order.status]))

        self.order = None

    def notify_trade(self, trade):
        logger.debug('trade......', trade.status)
        # 交易刚打开时
        if trade.justopened:
            self.log('开仓, 标的: %s, 股数: %.2f,价格: %.2f' % (
                trade.getdataname(), trade.size, trade.price))
        # 交易结束
        elif trade.isclosed:
            self.log('平仓, 标的: %s, 股数: %.2f,价格: %.2f, GROSS %.2f, NET %.2f, 手续费 %.2f' % (
                trade.getdataname(), trade.size, trade.price, trade.pnl, trade.pnlcomm, trade.commission))
        # 更新交易状态
        else:
            self.log('交易更新, 标的: %s, 仓位: %.2f,价格: %.2f' % (
                trade.getdataname(), trade.size, trade.price))


class StratgeyAlgo(StrategyBase):
    def __init__(self, algo_list, engine):
        self.algos = algo_list
        self.df_data = engine.df_data
        self.temp = {}
        self.perm = {}
        self.index = -1
        self.dates = list(self.df_data.index.unique())

    def next(self):
        self.index += 1
        self.now = self.dates[self.index]


        self.df_bar = self.df_data.loc[self.now]
        if type(self.df_bar) is pd.Series:
            self.df_bar = self.df_bar.to_frame().T
        self.df_bar.set_index('symbol', inplace=True)

        for algo in self.algos:
            if algo(self) is False:  # 如果algo返回False,直接不运行
                return

吾日三省吾身

“凡事发生必有利于我”。

适当的焦虑与担忧，会激发身体内的斗志，当然不能影响休息，更不能影响身心健康为宜。身边出现了不好的人与事，那它一定是来提醒你，最近可能太“岁月静好”了，需要加快速度，加快脚步成长。

寻宝电影里，主人公背后总有一个坏人团队在“协助”推进剧情，制造紧张气氛。那应该怎么办呢？

正确的做法，集中力量，专注自己的成长。

小时候的经历令人印象深刻。农村的邻里关系极期复杂，而且流动性极低，几代人就生活在这方寸之间，看存量，这近乎无解。我的解法就是努力学习，离开小镇，“离开”这些“纯朴”的人们。后来安排父母也远离这些是非。反而，这些远房亲戚反倒由于距离显得亲密与尊重。偶尔回去，听他们聊聊事非，发现尽管物是人非，但是非还在，只是，作为看客。

把因归于内，把行动归于内，这才是我们可以掌控的事情。

而且我还试图把backtrader改造成适合机器学习和强化学习。后觉得代码不好阅读，改成自己的引擎。

现在想来，越接近实盘，backtrader可以执行的精细化策略尤为重要。

止盈损，做多空等。

当然，经过一年的理解提升，要找回原来的代码，结合这一年的进展，挺容易的。

下面是backtrader的一个封装：

# encoding:utf8
from datetime import datetime

import backtrader as bt
import pandas as pd

from datafeed.dataloader import Dataloader
from engine.strategy import StratgeyAlgo


class BacktraderEngine:
    def __init__(self, df_data, init_cash=1000000.0, benchmark='000300.SH', start=datetime(2010, 1, 1),
                 end=datetime.now().date()):
        self.init_cash = init_cash
        self.start = start
        self.end = end
        self.benchmark = benchmark
        cerebro = bt.Cerebro()
        cerebro.broker.setcash(init_cash)

        # 设置手续费
        cerebro.broker.setcommission(0.0001)
        # 滑点：双边各 0.0001
        cerebro.broker.set_slippage_perc(perc=0.0001)

        self.cerebro = cerebro
        self.cerebro.addanalyzer(bt.analyzers.PyFolio, _name='_PyFolio')

        self.df_data = df_data
        self.symbols = list(set(self.df_data['symbol']))
        self._add_symbols_data()

    def _init_analyzers(self):
        '''
        self.cerebro.addanalyzer(bt.analyzers.Returns, _name='_Returns')
        self.cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='_TradeAnalyzer')
        self.cerebro.addanalyzer(bt.analyzers.AnnualReturn, _name='_AnnualReturn')
        self.cerebro.addanalyzer(bt.analyzers.SharpeRatio, riskfreerate=0.0, annualize=True, _name='_SharpeRatio')
        self.cerebro.addanalyzer(bt.analyzers.DrawDown, _name='_DrawDown')
         '''
        self.cerebro.addanalyzer(bt.analyzers.PyFolio, _name='_PyFolio')

    def _add_symbols_data(self):
        # 加载数据
        for s in self.symbols:
            df_symbol = self.df_data[self.df_data['symbol'] == s]
            df = to_backtrader_dataframe(df_symbol)
            data = bt.feeds.PandasData(dataname=df, name=s, fromdate=self.start, todate=self.end)

            self.cerebro.adddata(data)
            self.cerebro.addobserver(bt.observers.Benchmark,
                                     data=data)
            self.cerebro.addobserver(bt.observers.TimeReturn)

    def run_algo_strategy(self, algo_list):
        self.cerebro.addstrategy(StratgeyAlgo, algo_list=algo_list, engine=self)
        self.results = self.cerebro.run()

    def _bokeh_plot(self):
        from backtrader_plotting import Bokeh
        from backtrader_plotting.schemes import Tradimo
        plotconfig = {
            'id:ind#0': dict(
                subplot=True,
            ),
        }
        b = Bokeh(style='line', scheme=Tradimo(), plotconfig=plotconfig)
        self.cerebro.plot(b)

    def show_result_empyrical(self, returns):
        import empyrical

        print('累计收益：', round(empyrical.cum_returns_final(returns), 3))
        print('年化收益：', round(empyrical.annual_return(returns), 3))
        print('最大回撤：', round(empyrical.max_drawdown(returns), 3))
        print('夏普比', round(empyrical.sharpe_ratio(returns), 3))
        print('卡玛比', round(empyrical.calmar_ratio(returns), 3))
        print('omega', round(empyrical.omega_ratio(returns)), 3)

    def analysis(self, pyfolio=False):
        portfolio_stats = self.results[0].analyzers.getbyname('_PyFolio')
        returns, positions, transactions, _ = portfolio_stats.get_pf_items()
        returns.index = returns.index.tz_convert(None)
        self.show_result_empyrical(returns)

        if pyfolio:
            from pyfolio.tears import create_full_tear_sheet
            create_full_tear_sheet(returns, positions=positions, transactions=transactions)
        else:
            import quantstats
            # df = self.feed.get_df(self.benchmark)
            # df['rate'] = df['close'].pct_change()
            # df = df[['rate']]
            quantstats.reports.html(returns, download_filename='stats.html', output='stats.html',
                                    title='AI量化平台')
            import webbrowser
            webbrowser.open('stats.html')

        '''

        import pyfolio as pf
        pf.create_full_tear_sheet(
            returns,
            positions=positions,
            transactions=transactions)
        '''
        # self.cerebro.plot(volume=False)


def to_backtrader_dataframe(df):
    df.index = pd.to_datetime(df.index)
    df['openinterest'] = 0
    df = df[['open', 'high', 'low', 'close', 'volume', 'openinterest']]
    for c in ['open', 'high', 'low', 'close']:
        df.loc[:, c] = df[c] / df[c][0]
    return df