Strategy Selection Revisited

The original Strategy Selection approach used two strategies, which were manually registered and a simple [0, 1] list to decide which would be the target of the strategy.

Because Python offers a lot of instrospection possibilities with metaclasses, one may actually automate the approach. Let’s do it with a decorator approach which is probably the least invasive in this case (no need to define a metaclass for the strategies)

Reworking the factory

The factory now:

• is declared before the strategies

• has an empty _STRATS class attribute (it had the strategies to return before)

• has a register classmethod which will be used as decorator and which accepts an argument which will be added to _STRATS

• has a COUNT classmethod which will return an iterator (a range actually) with the count of the available strategies to be optimized

• bears no changes to the actual factory method: __new__, which keeps on using the idx parameter to return whatever is in the _STRATS class attribute at the given index

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

As such:

• The StFetcher strategy factory no longer contains any hardcoded strategies in itself

Decorating the to-be-optimized strategies

The strategies in the example don’t need to be reworked. Decoration with the register method of StFetcher is enough to have them added to the selection mix.

@StFetcher.register
class St0(bt.SignalStrategy):

and

@StFetcher.register
class St1(bt.SignalStrategy):

Taking advantage of COUNT

The manual [0, 1] list from the past when adding the strategy factory to the system with optstrategy can be fully replaced with a transparent call to StFetcher.COUNT(). Hardcoding is over.

    cerebro.optstrategy(StFetcher, idx=StFetcher.COUNT())

A sample run

$ ./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)])

Our 2 strategies have been run and deliver (as expected) different results.

Note

The sample is minimal but has been run with all available CPUs. Executing it with --maxpcpus=1 will be faster. For more complex scenarios using all CPUs will be useful.

Conclusion

Selection has been fully automated. As before one could envision something like querying a database for the number of available strategies and then fetch the strategies one by one.

Sample Usage

$ ./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)

The code

Which has been included in the sources of 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()