把backtrader改造成金融强化学习回测引擎

原创文章第119篇，专注“个人成长与财富自由、世界运作的逻辑， AI量化投资”。

继续强化学习应用于金融投资。

我们的AI量化平台，针对传统规则量化策略，进行了“积木式”的拆分，这种拆分的好处，就是最大化复用代码逻辑，这样开发策略又快且不容易出错。

针对强化学习环境，我们也打算这么做。看到有些平台，股票一个环境，加密货币一个环境，期货又是另一个环境，甚至把数据源处理都耦合到环境中，这是不对的。维护起来特别麻烦且容易出错。

01 强化学习环境的构成

一个完整的金融强化学习环境，包括数据（通常是OHLC以及特征数据）、交易动作（做多，平仓或者权重），回测系统（回测系统能够对交易动作做出反馈），激励指标（收益率、夏普比等）。

当下很多开源包在实现这个环境，存在一些问题。它们大篇幅做数据特征处理，实现了简单的量化回测，激励指标比较单一。这些环境扩展性不好，更谈不上接入实盘。

前文我们对传统量化与机器学习量化实现了统一的框架，数据管理及自动化标注，接入到backtrader的回测系统中。

同样，我们也希望可以最大化复用之前的成果，把强化学习环境也整合到一起，我们就不必关心数据处理，回测系统等环节的处理。

然而，backtrader本身为是传统量化而生，它的run直接就遍历了整个时间序列，而强化学习是由环境来调用回测引擎，一步一步往前走。

对于backtrader要进行一些改造。

一、重写_run_next

Backtrader在执行run的时间，在_runnext函数里整个顺序进行遍历。我们重写这个函数，只做第一步初始化的运算。

def _runnext(self, runstrats):
    '''
    Actual implementation of run in full next mode. All objects have its
    ``next`` method invoke on each data arrival
    '''
    self.runstrats_container = runstrats
    self._init_run()

def _init_run(self):
    datas = sorted(self.datas,
                   key=lambda x: (x._timeframe, x._compression))
    datas1 = datas[1:]
    data0 = datas[0]
    d0ret = True
    rsonly = [i for i, x in enumerate(datas)
              if x.resampling and not x.replaying]
    onlyresample = len(datas) == len(rsonly)
    noresample = not rsonly

    clonecount = sum(d._clone for d in datas)
    ldatas = len(datas)
    ldatas_noclones = ldatas - clonecount
    dt0 = date2num(datetime.datetime.max) - 2  # default at max

    self.bt_state_container = {"datas": datas,
                               "datas1": datas1,
                               "data0": data0,
                               "d0ret": d0ret,
                               "rsonly": rsonly,
                               "onlyresample": onlyresample,
                               "noresample": noresample,
                               "ldatas_noclones": ldatas_noclones,
                               "dt0": dt0,
                               }

二、提供一个step单步执行函数

这个函数的代码大多可以从cerebro里查到，这里就是展开描述。

def _step(self, runstrats, datas, datas1, data0, d0ret, rsonly,
          onlyresample, noresample, ldatas_noclones, dt0):

    # if any has live data in the buffer, no data will wait anything
    newqcheck = not any(d.haslivedata() for d in datas)
    if not newqcheck:
        # If no data has reached the live status or all, wait for
        # the next incoming data
        livecount = sum(d._laststatus == d.LIVE for d in datas)
        newqcheck = not livecount or livecount == ldatas_noclones

    lastret = False
    # Notify anything from the store even before moving datas
    # because datas may not move due to an error reported by the store
    self._storenotify()
    if self._event_stop:  # stop if requested
        return True
    self._datanotify()
    if self._event_stop:  # stop if requested
        return True

    # record starting time and tell feeds to discount the elapsed time
    # from the qcheck value
    drets = []
    qstart = datetime.datetime.utcnow()
    for d in datas:
        qlapse = datetime.datetime.utcnow() - qstart
        d.do_qcheck(newqcheck, qlapse.total_seconds())
        drets.append(d.next(ticks=False))

    d0ret = any((dret for dret in drets))
    if not d0ret and any((dret is None for dret in drets)):
        d0ret = None

    if d0ret:
        dts = []
        for i, ret in enumerate(drets):
            dts.append(datas[i].datetime[0] if ret else None)

        # Get index to minimum datetime
        if onlyresample or noresample:
            dt0 = min((d for d in dts if d is not None))
        else:
            dt0 = min((d for i, d in enumerate(dts)
                       if d is not None and i not in rsonly))

        dmaster = datas[dts.index(dt0)]  # and timemaster
        self._dtmaster = dmaster.num2date(dt0)
        self._udtmaster = num2date(dt0)

        # slen = len(runstrats[0])
        # Try to get something for those that didn't return
        for i, ret in enumerate(drets):
            if ret:  # dts already contains a valid datetime for this i
                continue

            # try to get a data by checking with a master
            d = datas[i]
            d._check(forcedata=dmaster)  # check to force output
            if d.next(datamaster=dmaster, ticks=False):  # retry
                dts[i] = d.datetime[0]  # good -> store
                # self._plotfillers2[i].append(slen)  # mark as fill
            else:
                # self._plotfillers[i].append(slen)  # mark as empty
                pass

        # make sure only those at dmaster level end up delivering
        for i, dti in enumerate(dts):
            if dti is not None:
                di = datas[i]
                rpi = False and di.replaying  # to check behavior
                if dti > dt0:
                    if not rpi:  # must see all ticks ...
                        di.rewind()  # cannot deliver yet
                    # self._plotfillers[i].append(slen)
                elif not di.replaying:
                    # Replay forces tick fill, else force here
                    di._tick_fill(force=True)

                # self._plotfillers2[i].append(slen)  # mark as fill

    elif d0ret is None:
        # meant for things like live feeds which may not produce a bar
        # at the moment but need the loop to run for notifications and
        # getting resample and others to produce timely bars
        for data in datas:
            data._check()
    else:
        lastret = data0._last()
        for data in datas1:
            lastret += data._last(datamaster=data0)

        if not lastret:
            # Only go extra round if something was changed by "lasts"
            return True  # return somethin signaling the end

    # Datas may have generated a new notification after next
    self._datanotify()
    if self._event_stop:  # stop if requested
        return True

    if d0ret or lastret:  # if any bar, check timers before broker
        self._check_timers(runstrats, dt0, cheat=True)
        if self.p.cheat_on_open:
            for strat in runstrats:
                strat._next_open()
                if self._event_stop:  # stop if requested
                    return True

    self._brokernotify()
    if self._event_stop:  # stop if requested
        return True

    if d0ret or lastret:  # bars produced by data or filters
        self._check_timers(runstrats, dt0, cheat=False)
        for strat in runstrats:
            strat._next()
            if self._event_stop:  # stop if requested
                return True

            self._next_writers(runstrats)

    self.bt_state_container = {"datas": datas,
                               "datas1": datas1,
                               "data0": data0,
                               "d0ret": d0ret,
                               "rsonly": rsonly,
                               "onlyresample": onlyresample,
                               "noresample": noresample,
                               "ldatas_noclones": ldatas_noclones,
                               "dt0": dt0,
                               }

    return False

如此，我们的backtrader就可以实现单步循环，为我们的环境提供回测能力了。

一些思考：

疫情还在持续，不知要多久，不知以何种方式结束。

目前大家看到的，可预见到的，肯定不是大家所期待的。

我们很多时候决定不了什么，只能耐心等待，保护好自己，积蓄能力。

三年后，也许这些都是故事 ，也许一个超级疫苗就研发出来，从此人间皆安。

我们要做的事情是，如果再有类似的“黑天鹅”事件发生，我们是否不那么被动，有更多的选择的权利？！

小时候，受到不公正的待遇，私下默默努力。但现代毕竟不是武侠里的快意恩愁。我们能做的事情是升级自己的环境，有能力远离那些不喜欢的人，不喜欢的事。尽管不好的事情哪里都有，但越往上走，会越发文明，越发自由。

努力的意义是自由，财务自由就是你能离得开职场，进而不受地点的约束，而同样可以过上有品质的生活，你就是自由的。

创业九死一生，若成会带来大自由，不成则更加不自由。读书，写作也许是很好的一条路。读书，写作是在疫情封控这样的情境下都可以做的事情。

代码细节，请前往星球微信群交流。

强化学习框架stable-baseline3以及pandas datareader

ETF轮动+RSRS择时，加上卡曼滤波：年化48.41%，夏普比1.89