策略选择的高级应用 – backtrader中文教程
最初的策略选择方法使用两种策略,手动注册和一个简单的[0, 1]
列表来决定哪个是策略的目标。
因为 Python 为元类提供了许多自省的可能性,所以实际上可以自动化该方法。让我们使用一种decorator
在这种情况下可能是侵入性最小的方法(无需为策略定义 元类)
改造策略
现在的策略:
- 在策略之前声明
- 有一个空的
_STRATS
类属性(它之前有返回的策略) - 有一个
register
classmethod 将用作装饰器并接受将添加到_STRATS
- 有一个
COUNT
类方法,它将返回一个迭代器(range
实际上是一个),其中包含要优化的可用策略的计数 - 对实际策略方法没有任何更改:
__new__
,它继续使用idx
参数返回_STRATS
给定索引处的类属性中的任何内容
class StFetcher(object): _STRATS = [] @classmethod def register(cls, target): cls._STRATS.append(target) @classmethod def COUNT(cls): return range(len(cls._STRATS)) def __new__(cls, *args, **kwargs): idx = kwargs.pop('idx') obj = cls._STRATS[idx](*args, **kwargs) return obj
像这样:
StFetcher
策略本身不再包含任何硬编码的策略
装饰待优化策略
示例中的策略不需要重新设计。register
使用 的方法进行装饰 StFetcher
足以将它们添加到选择组合中。
@StFetcher.register class St0(bt.SignalStrategy):
和
@StFetcher.register class St1(bt.SignalStrategy):
利用 COUNT
过去使用 optstrategy 将策略工厂添加到系统时的手动 [0, 1] 列表可以完全替换为对 StFetcher.COUNT() 的透明调用。 硬编码结束了。
cerebro.optstrategy(StFetcher, idx=StFetcher.COUNT())
示例运行
$ ./stselection-revisited.py --optreturn Strat 0 Name OptReturn: - analyzer: OrderedDict([(u'rtot', 0.04847392369449283), (u'ravg', 9.467563221580632e-05), (u'rnorm', 0.02414514457151587), (u'rnorm100', 2.414514457151587)]) Strat 1 Name OptReturn: - analyzer: OrderedDict([(u'rtot', 0.05124714332260593), (u'ravg', 0.00010009207680196471), (u'rnorm', 0.025543999840699633), (u'rnorm100', 2.5543999840699634)])
我们的 2 个策略已经运行并交付(如预期)不同的结果。
Tips:该示例很小,但已在所有可用 CPU 上运行。执行它--maxpcpus=1
会更快。对于更复杂的场景,使用所有 CPU 将很有用。
结论
选择已完全自动化。和以前一样,可以设想像查询数据库以获取可用策略的数量,然后一一获取策略。
示例使用
$ ./stselection-revisited.py --help usage: strategy-selection.py [-h] [--data DATA] [--maxcpus MAXCPUS] [--optreturn] Sample for strategy selection optional arguments: -h, --help show this help message and exit --data DATA Data to be read in (default: ../../datas/2005-2006-day-001.txt) --maxcpus MAXCPUS Limit the numer of CPUs to use (default: None) --optreturn Return reduced/mocked strategy object (default: False)
源码
已包含在 backtrader 的来源中
from __future__ import (absolute_import, division, print_function, unicode_literals) import argparse import backtrader as bt from backtrader.utils.py3 import range class StFetcher(object): _STRATS = [] @classmethod def register(cls, target): cls._STRATS.append(target) @classmethod def COUNT(cls): return range(len(cls._STRATS)) def __new__(cls, *args, **kwargs): idx = kwargs.pop('idx') obj = cls._STRATS[idx](*args, **kwargs) return obj @StFetcher.register class St0(bt.SignalStrategy): def __init__(self): sma1, sma2 = bt.ind.SMA(period=10), bt.ind.SMA(period=30) crossover = bt.ind.CrossOver(sma1, sma2) self.signal_add(bt.SIGNAL_LONG, crossover) @StFetcher.register class St1(bt.SignalStrategy): def __init__(self): sma1 = bt.ind.SMA(period=10) crossover = bt.ind.CrossOver(self.data.close, sma1) self.signal_add(bt.SIGNAL_LONG, crossover) def runstrat(pargs=None): args = parse_args(pargs) cerebro = bt.Cerebro() data = bt.feeds.BacktraderCSVData(dataname=args.data) cerebro.adddata(data) cerebro.addanalyzer(bt.analyzers.Returns) cerebro.optstrategy(StFetcher, idx=StFetcher.COUNT()) results = cerebro.run(maxcpus=args.maxcpus, optreturn=args.optreturn) strats = [x[0] for x in results] # flatten the result for i, strat in enumerate(strats): rets = strat.analyzers.returns.get_analysis() print('Strat {} Name {}:\n - analyzer: {}\n'.format( i, strat.__class__.__name__, rets)) def parse_args(pargs=None): parser = argparse.ArgumentParser( formatter_class=argparse.ArgumentDefaultsHelpFormatter, description='Sample for strategy selection') parser.add_argument('--data', required=False, default='../../datas/2005-2006-day-001.txt', help='Data to be read in') parser.add_argument('--maxcpus', required=False, action='store', default=None, type=int, help='Limit the numer of CPUs to use') parser.add_argument('--optreturn', required=False, action='store_true', help='Return reduced/mocked strategy object') return parser.parse_args(pargs) if __name__ == '__main__': runstrat()
评论被关闭。