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()

推薦閱讀

相關文章

Backtrader按日線交易

似乎在世界某個地方有一種权益(Interest)可以總結如下： 使用每日柱線引入訂單，但使用開盤價 這來自工單#105订单执行逻辑与当前数据和#101动态投注计算中的對話 backtrader 嘗試盡可能保持現實，並且在處理每日柱線時適用以下前提： 當每日柱被評估時，柱線已經結束 這是有道理的，

Backtrader教程：指標 - 開發

如果必須開發任何東西（除了一個或多個獲勝策略之外），那麼這個東西就是一個自定義指標。 根據作者的說法，平臺內的這種開發很容易。 需要滿足以下條件： 從指標派生的類（直接或從現有的子類派生） 定義它將保持lines 指標必須至少具有 1 line。

Backtrader教程：安裝

要求和版本 backtrader 是獨立的，沒有外部依賴關係（除非要繪圖） 基本要求是： Python 2.7 Python 3.2 / 3.3/ 3.4 / 3.5 pypy/pypy3 如果需要繪圖，則其他要求： Matplotlib >= 1.4.

Backtrader股票篩選

在尋找其他一些東西時，我在StackOverlow家族網站之一上遇到了一個問題：Quantitative Finance aka Quant StackExchange。問題： 它被標記為Python，因此值得一看的是 backtrader 是否能夠勝任這項任務。 分析儀本身 該問題似乎適合用於簡單的分析器。

Backtrader教程：過濾器 - 參考

工作階段篩檢程式 類 backtrader.filters。

Backtrader數據多時間幀

有時投資決策是使用不同的時間框架做出的： 每周評估趨勢 每天執行條目 或者5分鐘對60分鐘。 這意味著需要將多個時間幀的數據組合在 backtrader 中以支援此類組合。 對它的本機支持已經內置。

Backtrader卡爾曼等

注意 對以下指令的支援從提交開始 發佈1.9.30.x 將是包含它的第1個版本 。 backtrader的原始目標之一是成為純python，即：僅使用標準發行版中可用的軟體包。只有一個例外是matplotlib在沒有重新發明輪子的情況下進行繪圖。

Backtrader教程：數據饋送 - 熊貓

注意 pandas 並且必須安裝其依賴項 支援Pandas Dataframes似乎受到很多人的關注，他們依賴於已經可用的解析代碼來分析不同的數據源（包括CSV）和Pandas提供的其他功能。 數據饋送的重要聲明。 注意 這些只是 聲明。不要盲目複製此代碼。

Backtrader教程：分析儀 - PyFolio

注意 從（至少）2017-07-25pyfolio 開始，API已更改，不再 create_full_tear_sheet 具有 gross_lev 作為命名參數的參數。

Backtrader信貸利息

在某些情況下，真實經紀人的現金金額可能會減少，因為資產操作包括利率。例子： 賣空股票 交易所買賣基金包括多頭和空頭 這意味著不僅交易構成了系統的盈利能力，因為信貸上的利息在帳戶上佔有一席之地。 為了涵蓋這種情況， backtrader 包括（從發佈1.8.8.96開始）功能來考慮這一點。