保守公式的择时策略实现 – backtrader中文教程

本文介绍了保守公式方法：Python中的保守公式：量化投资变得容易 

这是许多可能的再平衡方法中的一种，但很容易掌握。方法总结：

• x股票选自Y(100 of 1000)
• 选择标准是
  • 低波动性
  • 高净派息率
  • 高势头
  • 每月重新平衡

考虑到这一点，让我们开始在 backtrader中展示一个可能的实现

数据

即使一个人有一个获胜的策略，如果没有可用于该策略的数据，实际上也不会获胜。这意味着必须考虑数据的外观以及如何加载它。

假设一组CSV（“逗号分隔值”）文件可用，包含以下功能

• ohlcv月度数据
• v在包含Net Payout Yield ( )之后的额外字段npy，以拥有ohlcvn数据集。

因此CSV数据的格式将如下所示

date, open, high, low, close, volume, npy
2001-12-31, 1.0, 1.0, 1.0, 1.0, 0.5, 3.0
2002-01-31, 2.0, 2.5, 1.1, 1.2, 3.0, 5.0
...

即：每月一行。 现在可以准备数据加载器引擎，将创建与backtrader一起交付的通用内置 CSV 加载器的简单扩展。

class NetPayOutData(bt.feeds.GenericCSVData):
    lines = ('npy',)  # add a line containing the net payout yield
    params = dict(
        npy=6,  # npy field is in the 6th column (0 based index)
        dtformat='%Y-%m-%d',  # fix date format a yyyy-mm-dd
        timeframe=bt.TimeFrame.Months,  # fixed the timeframe
        openinterest=-1,  # -1 indicates there is no openinterest field
    )

那就是。ohlcv请注意向数据流中添加一个基本数据点是多么容易 。

1. 通过使用表达式lines=('npy',). 其他常用字段 ( open, high, …) 已经是GenericCSVData
2. 通过用 指示装载位置params = dict(npy=6)。其他字段具有预定义的位置。

参数中的时间范围也已更新，以反映数据的每月性质。

数据加载器必须使用文件名正确实例化，但这是以后的事情，当下面提供标准样板以获得完整的脚本时。

策略

让我们将逻辑放入标准的反向交易者策略中。为了使其尽可能通用和可定制，params将使用与之前处理数据相同的方法。

在深入研究策略之前，让我们考虑一下快速总结中的一点

• x股票选自Y

策略本身并不负责将股票添加到宇宙中，而是负责选择。可能是在只添加了 50 只股票的情况下，如果x并且Y在代码中是固定的，仍然尝试选择 100 只。为应对此类情况，将采取以下措施：

• 有selperc一个值为0.10（即：）的参数10%，以指示要从宇宙中选择的股票数量。这意味着如果存在 1000 只，则只会选择 100 只，如果整个宇宙由 50 只股票组成，则只会选择 5 只。

至于股票排名的公式，它看起来像这样：

• (momentum * net payout) / volatility这意味着那些具有更高动力、更高支出和更低波动性的人将获得更高的分数。

momentum将使用RateOfChange指标 (aka ROC)，它 衡量一段时间内价格变化的比率。

这net payout已经是数据馈送的一部分。

为了计算volatility,股票StandardDeviation的 n-periods回报 ( n-periods, 因为东西将被保存为参数) 将被使用。

有了这些信息，就可以使用正确的参数和指标设置以及计算来初始化策略，这些指标和计算将在以后的每个月迭代中使用。

首先声明和参数

class St(bt.Strategy):
    params = dict(
        selcperc=0.10,  # percentage of stocks to select from the universe
        rperiod=1,  # period for the returns calculation, default 1 period
        vperiod=36,  # lookback period for volatility - default 36 periods
        mperiod=12,  # lookback period for momentum - default 12 periods
        reserve=0.05  # 5% reserve capital
    )

请注意，上面没有提到的东西被添加了，这是一个参数reserve=0.05（即5%），用于计算每只股票的分配百分比，在银行中保留储备资本。尽管对于模拟来说，可以想象要使用 100% 的资金，但这样做可能会遇到常见的问题，例如价格差距、浮点精度并最终错过一些市场入场点。

首先，创建一个小的日志记录方法，它允许记录投资组合是如何重新平衡的。

def log(self, arg):
    print('{} {}'.format(self.datetime.date(), arg))

在 __init__方法开始时，计算要排名的股票数量，并应用储备资本参数来确定银行的每只股票百分比。

def __init__(self):
    # calculate 1st the amount of stocks that will be selected
    self.selnum = int(len(self.datas) * self.p.selcperc)

    # allocation perc per stock
    # reserve kept to make sure orders are not rejected due to
    # margin. Prices are calculated when known (close), but orders can only
    # be executed next day (opening price). Price can gap upwards
    self.perctarget = (1.0 - self.p.reserve) % self.selnum

最后初始化结束，计算每只股票的波动率和动量指标，然后将其应用于每只股票排名公式计算。

# returns, volatilities and momentums
rs = [bt.ind.PctChange(d, period=self.p.rperiod) for d in self.datas]
vs = [bt.ind.StdDev(ret, period=self.p.vperiod) for ret in rs]
ms = [bt.ind.ROC(d, period=self.p.mperiod) for d in self.datas]

# simple rank formula: (momentum * net payout) / volatility
# the highest ranked: low vol, large momentum, large payout
self.ranks = {d: d.npy * m / v for d, v, m in zip(self.datas, vs, ms)}

现在是每个月进行迭代的时候了。排名可以在 self.ranks字典中找到。每次迭代都必须对键/值对进行排序，以获取哪些项目必须离开，哪些项目必须成为投资组合的一部分（保留或添加）

def next(self):
    # sort data and current rank
    ranks = sorted(
        self.ranks.items(),  # get the (d, rank), pair
        key=lambda x: x[1][0],  # use rank (elem 1) and current time "0"
        reverse=True,  # highest ranked 1st ... please
    )

iterable 以相反的顺序排序，因为排名公式为排名最高的股票提供更高的分数。

重新平衡现在到期了。

再平衡1：排名靠前和持仓股票

# put top ranked in dict with data as key to test for presence
rtop = dict(ranks[:self.selnum])

# For logging purposes of stocks leaving the portfolio
rbot = dict(ranks[self.selnum:])

这里发生了一些 Python 技巧，因为dict正在使用 a 。原因是，如果将排名靠前的股票放入 a中， Python 将在内部使用该list运算符来检查该运算符是否存在 。虽然不太可能，但两只股票在同一天有相同的价值是可能的。当使用散列值检查作为键的一部分的项目是否存在时。==indict

注意：出于记录目的rbot（排名底部）也使用不存在的库存创建rtop。

为了稍后区分必须离开投资组合的股票、只需重新平衡的股票和新排名靠前的股票，准备了投资组合中的当前股票列表。

# prepare quick lookup list of stocks currently holding a position
posdata = [d for d, pos in self.getpositions().items() if pos]

再平衡 2：出售不再排名靠前的产品

就像在现实世界中一样，在反向交易者生态系统中，必须先卖出再买，以确保有足够的现金。

# remove those no longer top ranked
# do this first to issue sell orders and free cash
for d in (d for d in posdata if d not in rtop):
    self.log('Exit {} - Rank {:.2f}'.format(d._name, rbot[d][0]))
    self.order_target_percent(d, target=0.0)

当前有未平仓头寸且不再排名靠前的股票被卖出（即target=0.0）。

笔记

一个简单的self.close(data)就足够了，而不是明确说明目标百分比。

再平衡3：对所有排名靠前的股票下达目标订单

投资组合的总价值会随着时间的推移而变化，投资组合中的那些股票可能不得不略微增加/减少当前头寸以匹配预期百分比。order_target_percent是进入市场的理想

# rebalance those already top ranked and still there
for d in (d for d in posdata if d in rtop):
    self.log('Rebal {} - Rank {:.2f}'.format(d._name, rtop[d][0]))
    self.order_target_percent(d, target=self.perctarget)
    del rtop[d]  # remove it, to simplify next iteration

在将新股票添加到投资组合之前，重新平衡已有头寸的股票，因为新股票只会发出buy订单并消耗现金。rtop[data].pop()在重新平衡后从其中删除现有股票，剩余的股票rtop是那些将新添加到投资组合中的股票。

# issue a target order for the newly top ranked stocks
# do this last, as this will generate buy orders consuming cash
for d in rtop:
    self.log('Enter {} - Rank {:.2f}'.format(d._name, rtop[d][0]))
    self.order_target_percent(d, target=self.perctarget)

运行它并评估它！

拥有一个数据加载器类和策略是不够的。就像任何其他框架一样，需要一些样板。下面的代码使它成为可能。

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

    cerebro = bt.Cerebro()

    # Data feed kwargs
    dkwargs = dict(**eval('dict(' + args.dargs + ')'))

    # Parse from/to-date
    dtfmt, tmfmt = '%Y-%m-%d', 'T%H:%M:%S'
    if args.fromdate:
        fmt = dtfmt + tmfmt * ('T' in args.fromdate)
        dkwargs['fromdate'] = datetime.datetime.strptime(args.fromdate, fmt)

    if args.todate:
        fmt = dtfmt + tmfmt * ('T' in args.todate)
        dkwargs['todate'] = datetime.datetime.strptime(args.todate, fmt)

    # add all the data files available in the directory datadir
    for fname in glob.glob(os.path.join(args.datadir, '*')):
        data = NetPayOutData(dataname=fname, **dkwargs)
        cerebro.adddata(data)

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

    # set the cash
    cerebro.broker.setcash(args.cash)

    cerebro.run()  # execute it all

    # Basic performance evaluation ... final value ... minus starting cash
    pnl = cerebro.broker.get_value() - args.cash
    print('Profit ... or Loss: {:.2f}'.format(pnl))

执行以下操作的地方：

• 解析参数并使其可用（这显然是可选的，因为一切都可以硬编码，但好的做法就是好的做法）
• 创建cerebro引擎实例。是的，这是西班牙语中“大脑”的意思，是负责在黑暗中协调管弦乐动作的框架的一部分。尽管它可以接受多个选项，但默认值应该足以满足大多数用例。
• 加载数据文件，这是通过简单的目录扫描 args.datadir完成的，所有文件都加载NetPayOutData并添加到cerebro实例中
• 添加策略
• 设置现金，默认为1,000,000. 鉴于用例是100针对500. 这也是一个可以改变的论点。
• 并打电话cerebro.run()
• 最后评估性能

为了能够直接从命令行运行具有不同参数的东西，argparse下面提供了一个启用的样板，其中包含整个代码

绩效评估

以最终结果值的形式添加的幼稚绩效评估，即：最终净资产值减去起始现金。

backtrader生态系统提供了一组内置的性能分析器，这些分析器也可以使用，例如：SharpeRatio、Variability-Weighted Return等SQN 。

完整的源码

最后，将大部分工作作为整体呈现。享受！

import argparse
import datetime
import glob
import os.path

import backtrader as bt


class NetPayOutData(bt.feeds.GenericCSVData):
    lines = ('npy',)  # add a line containing the net payout yield
    params = dict(
        npy=6,  # npy field is in the 6th column (0 based index)
        dtformat='%Y-%m-%d',  # fix date format a yyyy-mm-dd
        timeframe=bt.TimeFrame.Months,  # fixed the timeframe
        openinterest=-1,  # -1 indicates there is no openinterest field
    )


class St(bt.Strategy):
    params = dict(
        selcperc=0.10,  # percentage of stocks to select from the universe
        rperiod=1,  # period for the returns calculation, default 1 period
        vperiod=36,  # lookback period for volatility - default 36 periods
        mperiod=12,  # lookback period for momentum - default 12 periods
        reserve=0.05  # 5% reserve capital
    )

    def log(self, arg):
        print('{} {}'.format(self.datetime.date(), arg))

    def __init__(self):
        # calculate 1st the amount of stocks that will be selected
        self.selnum = int(len(self.datas) * self.p.selcperc)

        # allocation perc per stock
        # reserve kept to make sure orders are not rejected due to
        # margin. Prices are calculated when known (close), but orders can only
        # be executed next day (opening price). Price can gap upwards
        self.perctarget = (1.0 - self.p.reserve) / self.selnum

        # returns, volatilities and momentums
        rs = [bt.ind.PctChange(d, period=self.p.rperiod) for d in self.datas]
        vs = [bt.ind.StdDev(ret, period=self.p.vperiod) for ret in rs]
        ms = [bt.ind.ROC(d, period=self.p.mperiod) for d in self.datas]

        # simple rank formula: (momentum * net payout) / volatility
        # the highest ranked: low vol, large momentum, large payout
        self.ranks = {d: d.npy * m / v for d, v, m in zip(self.datas, vs, ms)}

    def next(self):
        # sort data and current rank
        ranks = sorted(
            self.ranks.items(),  # get the (d, rank), pair
            key=lambda x: x[1][0],  # use rank (elem 1) and current time "0"
            reverse=True,  # highest ranked 1st ... please
        )

        # put top ranked in dict with data as key to test for presence
        rtop = dict(ranks[:self.selnum])

        # For logging purposes of stocks leaving the portfolio
        rbot = dict(ranks[self.selnum:])

        # prepare quick lookup list of stocks currently holding a position
        posdata = [d for d, pos in self.getpositions().items() if pos]

        # remove those no longer top ranked
        # do this first to issue sell orders and free cash
        for d in (d for d in posdata if d not in rtop):
            self.log('Leave {} - Rank {:.2f}'.format(d._name, rbot[d][0]))
            self.order_target_percent(d, target=0.0)

        # rebalance those already top ranked and still there
        for d in (d for d in posdata if d in rtop):
            self.log('Rebal {} - Rank {:.2f}'.format(d._name, rtop[d][0]))
            self.order_target_percent(d, target=self.perctarget)
            del rtop[d]  # remove it, to simplify next iteration

        # issue a target order for the newly top ranked stocks
        # do this last, as this will generate buy orders consuming cash
        for d in rtop:
            self.log('Enter {} - Rank {:.2f}'.format(d._name, rtop[d][0]))
            self.order_target_percent(d, target=self.perctarget)


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

    cerebro = bt.Cerebro()

    # Data feed kwargs
    dkwargs = dict(**eval('dict(' + args.dargs + ')'))

    # Parse from/to-date
    dtfmt, tmfmt = '%Y-%m-%d', 'T%H:%M:%S'
    if args.fromdate:
        fmt = dtfmt + tmfmt * ('T' in args.fromdate)
        dkwargs['fromdate'] = datetime.datetime.strptime(args.fromdate, fmt)

    if args.todate:
        fmt = dtfmt + tmfmt * ('T' in args.todate)
        dkwargs['todate'] = datetime.datetime.strptime(args.todate, fmt)

    # add all the data files available in the directory datadir
    for fname in glob.glob(os.path.join(args.datadir, '*')):
        data = NetPayOutData(dataname=fname, **dkwargs)
        cerebro.adddata(data)

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

    # set the cash
    cerebro.broker.setcash(args.cash)

    cerebro.run()  # execute it all

    # Basic performance evaluation ... final value ... minus starting cash
    pnl = cerebro.broker.get_value() - args.cash
    print('Profit ... or Loss: {:.2f}'.format(pnl))


def parse_args(pargs=None):
    parser = argparse.ArgumentParser(
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
        description=('Rebalancing with the Conservative Formula'),
    )

    parser.add_argument('--datadir', required=True,
                        help='Directory with data files')

    parser.add_argument('--dargs', default='',
                        metavar='kwargs', help='kwargs in k1=v1,k2=v2 format')

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

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

    parser.add_argument('--cerebro', required=False, default='',
                        metavar='kwargs', help='kwargs in k1=v1,k2=v2 format')

    parser.add_argument('--cash', default=1000000.0, type=float,
                        metavar='kwargs', help='kwargs in k1=v1,k2=v2 format')

    parser.add_argument('--strat', required=False, default='',
                        metavar='kwargs', help='kwargs in k1=v1,k2=v2 format')

    return parser.parse_args(pargs)


if __name__ == '__main__':
    run()