算法¶
概述¶
bt的核心构建模块之一是Algo
和密切相关的AlgoStack
。

算法¶
一个算法本质上是一个返回True或False的函数。它接受一个参数,即正在测试的Strategy
。一个算法理想情况下应该只服务于一个特定的目的。这个目的可以控制执行流程,可以控制证券选择,证券分配等。例如,你可以有一个算法来检查月份是否已经改变(例如bt.algos.RunMonthly
)。如果已经改变,这个算法返回True,如果没有,返回False。
数据传递¶
为了在不同的算法之间传递数据,策略有两个属性: temp 和 perm。它们都是字典,用于存储由算法生成的数据。 临时数据在每次数据变化时都会刷新,而永久数据则不会被更改。
算法通常会在临时或永久对象中设置和/或需要值。例如,
bt.algos.WeighEqually
算法会在临时对象中设置 'weights' 键,并且
它需要临时对象中的 'selected' 键。
例如,让我们来看一个简单的选择 -> 权重 -> 分配逻辑链。我们将这个策略分解为3个算法:
选择 在所有的可投资资产中,我想将资本分配到哪些证券?参见,例如
SelectAll
,SelectThese
,SelectWhere
,SelectN
, 等权重 在目标投资组合中,每个选定的证券应该有多少权重?参见,例如
WeighEqually
,WeighRandomly
WeighSpecified
,WeighTarget
,WeighInvVol
,WeighMeanVar
, 等等分配 关闭不再需要的仓位,并将资金分配给那些被选中并给予目标权重的仓位。参见,例如
Rebalance
在这种情况下,选择算法可以在策略的临时字典中设置‘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_long
和 cost_short
)。这些在 setup
函数中有文档记录,具体在 Security
和 CouponPayingSecurity
中。
分支和控制流¶
虽然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)