算法

概述

bt的核心构建模块之一是Algo和密切相关的AlgoStack

algo stack

算法

一个算法本质上是一个返回True或False的函数。它接受一个参数,即正在测试的Strategy。一个算法理想情况下应该只服务于一个特定的目的。这个目的可以控制执行流程,可以控制证券选择,证券分配等。例如,你可以有一个算法来检查月份是否已经改变(例如bt.algos.RunMonthly)。如果已经改变,这个算法返回True,如果没有,返回False。

算法栈

一个 AlgoStack 是一个将多个 Algo 组合在一起的类,并在每个 Algo 返回 True 时依次运行它们。一旦某个 Algo 返回 False,AlgoStack 就会停止执行并返回 False(毕竟 AlgoStack 也是一个 Algo)。这使我们能够将不同的 Algo 组合在一起,并通过 Algo 的返回值来控制执行流程。如果需要,许多 AlgoStack 本身也可以包含在另一个 AlgoStack 中。

通过将策略逻辑分解为这些小代码块,我们实现了可测试性和可重用性——这两个在软件开发中非常有吸引力的特性。

数据传递

为了在不同的算法之间传递数据,策略有两个属性: tempperm。它们都是字典,用于存储由算法生成的数据。 临时数据在每次数据变化时都会刷新,而永久数据则不会被更改。

算法通常会在临时或永久对象中设置和/或需要值。例如, bt.algos.WeighEqually 算法会在临时对象中设置 'weights' 键,并且 它需要临时对象中的 'selected' 键。

例如,让我们来看一个简单的选择 -> 权重 -> 分配逻辑链。我们将这个策略分解为3个算法:

在这种情况下,选择算法可以在策略的临时字典中设置‘selected’键,而权重算法可以读取这些值并依次在临时字典中设置‘weights’键。然后,分配算法将读取‘weights’并相应地采取行动。

为了扩展简单的选择 -> 权重 -> 分配逻辑链以包括额外的风险/暴露计算步骤,可以通过实现特定的算法来实现这一目的。这些算法可以在权重之前(用于基于风险的投资组合构建)或之后(用于报告)使用。参见,例如 UpdateRisk

注意

为了保持最大的灵活性,目前没有检查来确保AlgoStack是有效的。因此,用户和Algos的创建者有责任确保需求和副作用被充分记录并正确使用(顺便说一下,这可能不是解决这个问题的最佳方式。如果你有更好的想法,请告诉我!)。

开发者应在文档字符串中添加一个部分,概述“集合”和“要求”。请参阅bt.algos.WeighEqually的文档字符串以获取示例。

实现

在大多数情况下,算法必须保持某种状态。在这种情况下,将它们实现为类并定义__call__方法会更容易,如下所示:

class MyAlgo(bt.Algo):

    def __init__(self, arg1, arg2):
        self.arg1 = arg1
        self.arg2 = arg2

    def __call__(self, target):
        # my logic goes here

        # accessing/storing variables through target.temp['key']

        # remember to return a bool - True in most cases
        return True

请注意,类上的属性不应特定于任何特定目标。

然而,对于不需要保留任何状态的算法,您可以简单地将其实现为一个基本函数,该函数接受一个参数 - 策略:

def MyAlgo2(target):
    # all the logic

    return True

最佳实践

可重用性

回想一下,算法应该在不同的回测中可重复使用(包括在不同基础证券组合或不同时间范围内的回测),而回测是策略和数据的逻辑组合。然而,在某些情况下,算法需要使用一些额外的数据,这些数据确实依赖于证券组合或时间范围(即预先计算好的信号数据框)。

处理这个问题的最佳方法是使用数据的名称来构建Algo,并使用这个命名的数据来实例化回测:

class MyAlgo(bt.Algo):

    def __init__(self, signal_name ):
        self.signal_name = signal_name

    def __call__(self, target):
        # my logic goes here

        # accessing data via target.get_data( self.signal_name )

        # remember to return a bool - True in most cases
        return True

# create the strategy
s = bt.Strategy('s1', [bt.algos.MyAlgo( 'my_signal' )])

# create a backtest and run it
test = bt.Backtest(s, data, additional_data={'my_signal':signal_df})
res = bt.run(test)

# Run the same strategy on different data without changing MyAlgo
test = bt.Backtest(s, data2, additional_data={'my_signal':signal_df2})
res = bt.run(test)

请注意,框架本身使用了一些额外的数据键来支持附加功能(即 bidoffer, coupon, cost_longcost_short)。这些在 setup 函数中有文档记录,具体在 SecurityCouponPayingSecurity 中。

调试

调试算法的最简单方法是利用现有的调试算法之一或编写自己的调试算法!只需将它们插入到算法堆栈的适当位置,并添加断点以检查传递策略的状态。

分支和控制流

虽然Algo的设置可能看起来过于简单(一个返回True或False的函数列表),但这是一个强大的构造,允许复杂的分支和条件结构。特别是,分支是通过Or Algo实现的。

例如,下面的代码说明了策略表现的打印可以与投资组合的再平衡在不同的时间线上进行。可以通过将这些算法放在相关堆栈的头部来添加额外的条件。

import bt

data = bt.get('spy,agg', start='2010-01-01')

# create two separate algo stacks and combine the branches
logging_stack = bt.AlgoStack(
                    bt.algos.RunWeekly(),
                    bt.algos.PrintInfo('{name}:{now}. Value:{_value:0.0f}, Price:{_price:0.4f}')
                    )
trading_stack = bt.AlgoStack(
                    bt.algos.RunMonthly(),
                    bt.algos.SelectAll(),
                    bt.algos.WeighEqually(),
                    bt.algos.Rebalance()
                    )
branch_stack =  bt.AlgoStack(
                    # Upstream algos could go here...
                    bt.algos.Or( [ logging_stack, trading_stack ] )
                    # Downstream algos could go here...
                    )

s = bt.Strategy('strategy', branch_stack, ['spy', 'agg'])
t = bt.Backtest(s, data)
r = bt.run(t)