Backtrader卡尔曼等

Backtrader   |   2022年6月20日

注意

对以下指令的支持从提交开始

发布1.9.30.x 将是包含它的第1个版本 。

backtrader的原始目标之一是成为纯python，即：仅使用标准发行版中可用的软件包。只有一个例外是matplotlib在没有重新发明轮子的情况下进行绘图。虽然是在最晚的可能时刻导入的，以避免中断可能根本不需要绘图的标准操作（如果未安装且不希望，则避免错误）

第二 个例外是 pytz ，随着即时 data feeds 的出现而增加对时区的支持，这些可能位于本地时区之外。同样，导入操作在后台发生，并且仅在可用时才 pytz 发生（用户可以选择传递 pytz 实例）

但是现在完全破例了，因为反向交易者正在使用众所周知的软件包，如numpy，pandasstatsmodel和一些更适度的软件包，如pykalman.或者包含在使用这些软件包的事物的平台中。

来自社区的一些例子：

这个愿望被添加到这里草拟的快速路线图中：

声明性方法

保持 backtrader 的原始精神，同时允许使用这些软件包的关键是不要强迫 纯python用户必须安装这些软件包。

尽管这似乎具有挑战性，并且容易出现多个条件语句，再加上异常处理，但平台内部和外部的用户方法都是依赖于已经用于开发其他概念的相同原则，例如大多数对象的 参数 （命名参数）。

让我们回想一下，如何定义一个Indicator 接受参数并定义 lines：

class MyIndicator(bt.Indicator):
    lines = ('myline',)
    params = (
        ('period', 50),
    )

period参数稍后可寻址为self.params.period或self.p.period如：

def __init__(self):
    print('my period is:', self.p.period)

和 line 中的当前值，如self.lines.myline 或 self.l.myline 如：

def next(self):
    print('mylines[0]:', self.lines.myline[0])

这并不是特别有用，只是展示了params后台机制的声明性方法，该方法还具有对继承（包括多重继承）的适当支持。

正在推出packages

使用相同的声明性技术（有些人称之为元编程），对外来包的支持可以通过以下方式获得：

class MyIndicator(bt.Indicator):
    packages = ('pandas',)
    lines = ('myline',)
    params = (
        ('period', 50),
    )

起泡的藤壶!!!这似乎只是另一个宣言。指针实施者面临的第一个问题是：

  • 我必须手动导入「熊猫」吗？

答案很简单：不。后台机制将导入 pandas 并使其在定义的模块中 MyIndicator 可用。现在可以在 next中运行以下操作：

def next(self):
    print('mylines[0]:', pandas.SomeFunction(self.lines.myline[0]))

该packages 指令还可用于：

  • 在一个声明中导入多个包

  • 将导入分配给别名 alaimport pandas as pd

假设statsmodel也希望sm 完成 pandas.SomeFunction：

class MyIndicator(bt.Indicator):
    packages = ('pandas', ('statsmodel', 'sm'),)
    lines = ('myline',)
    params = (
        ('period', 50),
    )

    def next(self):
        print('mylines[0]:', sm.XX(pandas.SomeFunction(self.lines.myline[0])))

statsmodel 已导入为 sm 且可用。它只需要传递一个可反复运算的（a tuple 是 backtrader 约定），其中包含包的名称和所需的别名。

添加frompackages

Python以不断查找事物而闻名，这是该语言在动态性，内省设施和元编程方面表现出色的原因之一。同时也是无法提供相同性能的原因之一。

通常的加速之一是通过直接从模块中导入符号来删除对模块的查找，以进行本地查找。从这个SomeFunctionpandas看起来像：

from pandas import SomeFunction

或使用别名：

from pandas import SomeFunction as SomeFunc

backtrader 通过指令frompackages 为两者都提供支持。让我们返工 MyIndicator：

class MyIndicator(bt.Indicator):
    frompackages = (('pandas', 'SomeFunction'),)
    lines = ('myline',)
    params = (
        ('period', 50),
    )

    def next(self):
        print('mylines[0]:', SomeFunction(self.lines.myline[0]))

当然，这开始添加更多的括弧。例如，如果要从中pandas导入两（2）个东西，则如下所示：

class MyIndicator(bt.Indicator):
    frompackages = (('pandas', ['SomeFunction', 'SomeFunction2']),)
    lines = ('myline',)
    params = (
        ('period', 50),
    )

    def next(self):
        print('mylines[0]:', SomeFunction2(SomeFunction(self.lines.myline[0])))

为了清楚起见SomeFunction，已经SomeFunction2放在a而不是atuplelist中，以方括弧，[]并且能够更好地阅读它。

也可以别名SomeFunction 例如 SFunc.完整范例：

class MyIndicator(bt.Indicator):
    frompackages = (('pandas', [('SomeFunction', 'SFunc'), 'SomeFunction2']),)
    lines = ('myline',)
    params = (
        ('period', 50),
    )

    def next(self):
        print('mylines[0]:', SomeFunction2(SFunc(self.lines.myline[0])))

从不同的包导入是可能的，但代价是更多的括弧。当然， line 断裂和缩进确实有说明：

class MyIndicator(bt.Indicator):
    frompackages = (
        ('pandas', [('SomeFunction', 'SFunc'), 'SomeFunction2']),
        ('statsmodel', 'XX'),
    )
    lines = ('myline',)
    params = (
        ('period', 50),
    )

    def next(self):
        print('mylines[0]:', XX(SomeFunction2(SFunc(self.lines.myline[0]))))

使用继承

frompackages两者都packages支持（多重）继承。例如，可能有一个基类，它为所有子类添加numpy支持：

class NumPySupport(object):
    packages = ('numpy',)


class MyIndicator(bt.Indicator, NumPySupport):
    packages = ('pandas',)

MyIndicator 将需要从后台机械导入两者 numpy ，并且 pandas 将能够使用它们。

介绍卡尔曼和朋友们

注意

以下两个指针都需要同行评审以确认实施情况。请小心使用。

可以在下面找到实现 的范例KalmanMovingAverage 。这是根据这里的一篇文章建模的： Quantopian讲座系列：卡尔曼滤波器

实现：

class KalmanMovingAverage(bt.indicators.MovingAverageBase):
    packages = ('pykalman',)
    frompackages = (('pykalman', [('KalmanFilter', 'KF')]),)
    lines = ('kma',)
    alias = ('KMA',)
    params = (
        ('initial_state_covariance', 1.0),
        ('observation_covariance', 1.0),
        ('transition_covariance', 0.05),
    )

    plotlines = dict(cov=dict(_plotskip=True))

    def __init__(self):
        self.addminperiod(self.p.period)  # when to deliver values
        self._dlast = self.data(-1)  # get previous day value

    def nextstart(self):
        self._k1 = self._dlast[0]
        self._c1 = self.p.initial_state_covariance

        self._kf = pykalman.KalmanFilter(
            transition_matrices=[1],
            observation_matrices=[1],
            observation_covariance=self.p.observation_covariance,
            transition_covariance=self.p.transition_covariance,
            initial_state_mean=self._k1,
            initial_state_covariance=self._c1,
        )

        self.next()

    def next(self):
        k1, self._c1 = self._kf.filter_update(self._k1, self._c1, self.data[0])
        self.lines.kma[0] = self._k1 = k1

KalmanFilter并在这里发布以下帖子： QSTrader中基于卡尔曼筛检程序的对交易策略

class NumPy(object):
    packages = (('numpy', 'np'),)


class KalmanFilterInd(bt.Indicator, NumPy):
    _mindatas = 2  # needs at least 2 data feeds

    packages = ('pandas',)
    lines = ('et', 'sqrt_qt')

    params = dict(
        delta=1e-4,
        vt=1e-3,
    )

    def __init__(self):
        self.wt = self.p.delta / (1 - self.p.delta) * np.eye(2)
        self.theta = np.zeros(2)
        self.P = np.zeros((2, 2))
        self.R = None

        self.d1_prev = self.data1(-1)  # data1 yesterday's price

    def next(self):
        F = np.asarray([self.data0[0], 1.0]).reshape((1, 2))
        y = self.d1_prev[0]

        if self.R is not None:  # self.R starts as None, self.C set below
            self.R = self.C + self.wt
        else:
            self.R = np.zeros((2, 2))

        yhat = F.dot(self.theta)
        et = y - yhat

        # Q_t is the variance of the prediction of observations and hence
        # \sqrt{Q_t} is the standard deviation of the predictions
        Qt = F.dot(self.R).dot(F.T) + self.p.vt
        sqrt_Qt = np.sqrt(Qt)

        # The posterior value of the states \theta_t is distributed as a
        # multivariate Gaussian with mean m_t and variance-covariance C_t
        At = self.R.dot(F.T) / Qt
        self.theta = self.theta + At.flatten() * et
        self.C = self.R - At * F.dot(self.R)

        # Fill the lines
        self.lines.et[0] = et
        self.lines.sqrt_qt[0] = sqrt_Qt

为了它，它显示了如何packages 与继承一起工作（pandas 不是真的需要）

范例的运行：

$ ./kalman-things.py --plot

生成此图表

示例用法

$ ./kalman-things.py --help
usage: kalman-things.py [-h] [--data0 DATA0] [--data1 DATA1]
                        [--fromdate FROMDATE] [--todate TODATE]
                        [--cerebro kwargs] [--broker kwargs] [--sizer kwargs]
                        [--strat kwargs] [--plot [kwargs]]

Packages and Kalman

optional arguments:
  -h, --help           show this help message and exit
  --data0 DATA0        Data to read in (default:
                       ../../datas/nvda-1999-2014.txt)
  --data1 DATA1        Data to read in (default:
                       ../../datas/orcl-1995-2014.txt)
  --fromdate FROMDATE  Date[time] in YYYY-MM-DD[THH:MM:SS] format (default:
                       2006-01-01)
  --todate TODATE      Date[time] in YYYY-MM-DD[THH:MM:SS] format (default:
                       2007-01-01)
  --cerebro kwargs     kwargs in key=value format (default: runonce=False)
  --broker kwargs      kwargs in key=value format (default: )
  --sizer kwargs       kwargs in key=value format (default: )
  --strat kwargs       kwargs in key=value format (default: )
  --plot [kwargs]      kwargs in key=value format (default: )

示例代码

from __future__ import (absolute_import, division, print_function,
                        unicode_literals)


import argparse
import datetime

import backtrader as bt


class KalmanMovingAverage(bt.indicators.MovingAverageBase):
    packages = ('pykalman',)
    frompackages = (('pykalman', [('KalmanFilter', 'KF')]),)
    lines = ('kma',)
    alias = ('KMA',)
    params = (
        ('initial_state_covariance', 1.0),
        ('observation_covariance', 1.0),
        ('transition_covariance', 0.05),
    )

    def __init__(self):
        self.addminperiod(self.p.period)  # when to deliver values
        self._dlast = self.data(-1)  # get previous day value

    def nextstart(self):
        self._k1 = self._dlast[0]
        self._c1 = self.p.initial_state_covariance

        self._kf = pykalman.KalmanFilter(
            transition_matrices=[1],
            observation_matrices=[1],
            observation_covariance=self.p.observation_covariance,
            transition_covariance=self.p.transition_covariance,
            initial_state_mean=self._k1,
            initial_state_covariance=self._c1,
        )

        self.next()

    def next(self):
        k1, self._c1 = self._kf.filter_update(self._k1, self._c1, self.data[0])
        self.lines.kma[0] = self._k1 = k1


class NumPy(object):
    packages = (('numpy', 'np'),)


class KalmanFilterInd(bt.Indicator, NumPy):
    _mindatas = 2  # needs at least 2 data feeds

    packages = ('pandas',)
    lines = ('et', 'sqrt_qt')

    params = dict(
        delta=1e-4,
        vt=1e-3,
    )

    def __init__(self):
        self.wt = self.p.delta / (1 - self.p.delta) * np.eye(2)
        self.theta = np.zeros(2)
        self.R = None

        self.d1_prev = self.data1(-1)  # data1 yesterday's price

    def next(self):
        F = np.asarray([self.data0[0], 1.0]).reshape((1, 2))
        y = self.d1_prev[0]

        if self.R is not None:  # self.R starts as None, self.C set below
            self.R = self.C + self.wt
        else:
            self.R = np.zeros((2, 2))

        yhat = F.dot(self.theta)
        et = y - yhat

        # Q_t is the variance of the prediction of observations and hence
        # \sqrt{Q_t} is the standard deviation of the predictions
        Qt = F.dot(self.R).dot(F.T) + self.p.vt
        sqrt_Qt = np.sqrt(Qt)

        # The posterior value of the states \theta_t is distributed as a
        # multivariate Gaussian with mean m_t and variance-covariance C_t
        At = self.R.dot(F.T) / Qt
        self.theta = self.theta + At.flatten() * et
        self.C = self.R - At * F.dot(self.R)

        # Fill the lines
        self.lines.et[0] = et
        self.lines.sqrt_qt[0] = sqrt_Qt


class KalmanSignals(bt.Indicator):
    _mindatas = 2  # needs at least 2 data feeds

    lines = ('long', 'short',)

    def __init__(self):
        kf = KalmanFilterInd()
        et, sqrt_qt = kf.lines.et, kf.lines.sqrt_qt

        self.lines.long = et < -1.0 * sqrt_qt
        # longexit is et > -1.0 * sqrt_qt ... the opposite of long
        self.lines.short = et > sqrt_qt
        # shortexit is et < sqrt_qt ... the opposite of short


class St(bt.Strategy):
    params = dict(
        ksigs=False,  # attempt trading
        period=30,
    )

    def __init__(self):
        if self.p.ksigs:
            self.ksig = KalmanSignals()
            KalmanFilter()

        KalmanMovingAverage(period=self.p.period)
        bt.ind.SMA(period=self.p.period)
        if True:
            kf = KalmanFilterInd()
            kf.plotlines.sqrt_qt._plotskip = True

    def next(self):
        if not self.p.ksigs:
            return

        size = self.position.size
        if not size:
            if self.ksig.long:
                self.buy()
            elif self.ksig.short:
                self.sell()

        elif size > 0:
            if not self.ksig.long:
                self.close()
        elif not self.ksig.short:  # implicit size < 0
            self.close()


def runstrat(args=None):
    args = parse_args(args)

    cerebro = bt.Cerebro()

    # Data feed kwargs
    kwargs = dict()

    # Parse from/to-date
    dtfmt, tmfmt = '%Y-%m-%d', 'T%H:%M:%S'
    for a, d in ((getattr(args, x), x) for x in ['fromdate', 'todate']):
        if a:
            strpfmt = dtfmt + tmfmt * ('T' in a)
            kwargs[d] = datetime.datetime.strptime(a, strpfmt)

    # Data feed
    data0 = bt.feeds.YahooFinanceCSVData(dataname=args.data0, **kwargs)
    cerebro.adddata(data0)

    data1 = bt.feeds.YahooFinanceCSVData(dataname=args.data1, **kwargs)
    data1.plotmaster = data0
    cerebro.adddata(data1)

    # Broker
    cerebro.broker = bt.brokers.BackBroker(**eval('dict(' + args.broker + ')'))

    # Sizer
    cerebro.addsizer(bt.sizers.FixedSize, **eval('dict(' + args.sizer + ')'))

    # Strategy
    cerebro.addstrategy(St, **eval('dict(' + args.strat + ')'))

    # Execute
    cerebro.run(**eval('dict(' + args.cerebro + ')'))

    if args.plot:  # Plot if requested to
        cerebro.plot(**eval('dict(' + args.plot + ')'))


def parse_args(pargs=None):
    parser = argparse.ArgumentParser(
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
        description=(
            'Packages and Kalman'
        )
    )

    parser.add_argument('--data0', default='../../datas/nvda-1999-2014.txt',
                        required=False, help='Data to read in')

    parser.add_argument('--data1', default='../../datas/orcl-1995-2014.txt',
                        required=False, help='Data to read in')

    # Defaults for dates
    parser.add_argument('--fromdate', required=False, default='2006-01-01',
                        help='Date[time] in YYYY-MM-DD[THH:MM:SS] format')

    parser.add_argument('--todate', required=False, default='2007-01-01',
                        help='Date[time] in YYYY-MM-DD[THH:MM:SS] format')

    parser.add_argument('--cerebro', required=False, default='runonce=False',
                        metavar='kwargs', help='kwargs in key=value format')

    parser.add_argument('--broker', required=False, default='',
                        metavar='kwargs', help='kwargs in key=value format')

    parser.add_argument('--sizer', required=False, default='',
                        metavar='kwargs', help='kwargs in key=value format')

    parser.add_argument('--strat', required=False, default='',
                        metavar='kwargs', help='kwargs in key=value format')

    parser.add_argument('--plot', required=False, default='',
                        nargs='?', const='{}',
                        metavar='kwargs', help='kwargs in key=value format')

    return parser.parse_args(pargs)


if __name__ == '__main__':
    runstrat()

推荐阅读

相关文章

BacktraderPython Hidden Powers 2

让我们进一步讨论一下Python的隐藏功能如何在 backtrader 中使用，以及如何实现它以尝试实现主要目标：易用性 这些定义是什么？ 例如指针： import backtrader as bt class MyIndicator(bt.

Backtrader教程：数据馈送 - 扩展(Extending DataFeed)

GitHub 中的问题实际上是在推动文档部分的完成，或者説明我了解我是否backtrader 具有我从一开始就设想的易用性和灵活性以及在此过程中做出的决定。 在本例中为问题 #9。

BacktraderPyFolio 集成

注意 2017年2月 pyfolio API 已更改，不再 create_full_tear_sheet 具有 gross_lev 作为命名参数的参数。

Backtrader教程：数据馈送 - 展期交割

并非每个供应商都为可以交易的工具提供连续的未来。有时提供的数据是仍然有效的到期日期的数据，即：仍在交易的日期 这在回溯测试方面并不是很有帮助，因为数据分散在几个不同的仪器上，这些仪器另外...时间重叠。 能够正确地将这些仪器的数据从过去连接到连续的流中，可以减轻疼痛。

Backtrader跨越数字

《backtrader》的发布1.9.27.105纠正了一个疏忽。这是一个疏忽，因为拼图的所有部分都已到位，但启动并不是在所有角落都进行的。 该机制使用一个名为的属性_mindatas，因此让我们将其称为： mindatas。 社区问了这个问题，答案并不是很到位。

Backtrader混合时间帧

1.3.0.92版本带来了混合来自不同时间帧的数据（来自 data feeds 和/或指针）的可能性。 到版本：https://github.com/mementum/backtrader/发布/标签/1.3.0.92 背景：指示器是智能哑对象。 他们很聪明，因为他们可以进行复杂的计算。

Backtrader蟒蛇隐藏的力量3

Last，但并非最不重要的一点是，在这个系列中，关于如何在 backtrader 中使用Python的隐藏功能是一些神奇变量是如何出现的。

Backtrader教程：数据馈送 - 熊猫

注意 pandas 并且必须安装其依赖项 支持Pandas Dataframes似乎受到很多人的关注，他们依赖于已经可用的解析代码来分析不同的数据源（包括CSV）和Pandas提供的其他功能。 数据馈送的重要声明。 注意 这些只是 声明。不要盲目拷贝此代码。

Backtrader动量策略

在另一篇伟大的文章中，泰迪·科克（Teddy Koker）再次展示了算法交易策略的发展之路： 研究优先应用 pandas 回溯测试，然后使用 backtrader 荣誉!!! 该帖子可以在以下位置找到： 泰迪·科克（Teddy Koker）给我留言，问我是否可以评论 backtrader的用法。

Backtrader终极振荡器

backtrader开发启动时的目标之一是使开发新的指针变得非常容易（至少对作者本人而言），以在数学和视觉上测试想法。 门票#102 是关于将 UltimateOscillator 添加到 backtrader 注意 它将在下一个版本中添加，同时可以使用下面的代码使用它。 票证中所示的参考： 以及： 无需在这里重复。