最初的策略選擇方法使用兩個策略，手動註冊和一個簡單的[0, 1]列表來決定哪個是策略的目標。
因為 Python 為元類提供了許多自省的可能性，所以實際上可以自動化該方法。讓我們使用decorator方法來完成，這在這種情況下可能是侵入性最小的（無需為策略定義元類）
現在的工廠：
在策略之前聲明
有一個空的
_STRATS類屬性（它之前有返回的策略）有一個
register類方法，它將用作裝飾器並接受將添加到_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策略工廠本身不再包含任何硬編碼的策略
示例中的策略不需要重新設計。使用StFetcher的register方法進行裝飾足以將它們添加到選擇組合中。
@StFetcher.register class St0(bt.SignalStrategy):
和
@StFetcher.register class St1(bt.SignalStrategy):
過去使用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 個策略已經運行並交付（如預期）不同的結果。
筆記
該示例很小，但已在所有可用 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()