Problem in parallel

[weibudda]

Following the documention sample code, "Strategy Selection Revisited", https://www.backtrader.com/blog/posts/2017-05-16-stsel-revisited/stsel-revisited/

when --maxcpus 1, everything is OK, but when --maxcpus 2, or anything else but 1, it reports :

multiprocessing.pool.MaybeEncodingError: Error sending result: '[<backtrader.cerebro.OptReturn object at 0x0000000015921988>]'. Reason: 'PicklingError("Can't pickle <class
'mp_main.St0'>: it's not the same object as mp_main.St0")'

what's the problem?

the related details:

(backtrader37) D:\Python\Jupyter\BackTrader\Test>python ./stselection-revisited.py --optreturn --maxcpus 2
Traceback (most recent call last):
File "./stselection-revisited.py", line 89, in <module>
runstrat()
File "./stselection-revisited.py", line 59, in runstrat
results = cerebro.run(maxcpus=args.maxcpus, optreturn=args.optreturn)
File "C:\Users\WEI.conda\envs\backtrader37\lib\site-packages\backtrader\cerebro.py", line 1143, in run
for r in pool.imap(self, iterstrats):
File "C:\Users\WEI.conda\envs\backtrader37\lib\multiprocessing\pool.py", line 748, in next
raise value
multiprocessing.pool.MaybeEncodingError: Error sending result: '[<backtrader.cerebro.OptReturn object at 0x0000000015921988>]'. Reason: 'PicklingError("Can't pickle <class
'mp_main.St0'>: it's not the same object as mp_main.St0")'

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()

[vladisld]

It seems you're right - the sample in the blog article could not work.

The problem with it is the @StFetcher.register attribute that binds the St0 and St1 strategy classes with the StFetcher class in the same module. Running such code in multiprocessor environment, each worker process will try to instantiate the St0 and St1 strategy classes in its own process - however the binding will now be done with the StFetcher class in the working process and not with the main process - resulting in different class definition from pickle serialization perspective.

Getting rid of such attribute and explicitly registering the strategies instead - solves the problem:

from __future__ import (absolute_import, division, print_function,
                        unicode_literals)

import argparse

import backtrader as bt
import threading
from backtrader.utils.py3 import range


class StFetcher(object):
    _STRATS = []

    @classmethod
    def register(cls, target):
        print(f"register {threading.get_ident()}:{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


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)


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)

StFetcher.register(St0)
StFetcher.register(St1)


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=False)

    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='w:/backtrader/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', default=False,
                        help='Return reduced/mocked strategy object')

    return parser.parse_args(pargs)


if __name__ == '__main__':
    runstrat()

producing the following output:

python.exe C:/Users/Vlad/PycharmProjects/test/test_strategy_selection_revisited.py --maxcpus=2 --optreturn
register 5288:<class '__main__.St0'>
register 5288:<class '__main__.St1'>
register 18368:<class '__mp_main__.St0'>
register 18368:<class '__mp_main__.St1'>
register 10612:<class '__mp_main__.St0'>
register 10612:<class '__mp_main__.St1'>
Strat 0 Name St0:
  - analyzer: OrderedDict([('rtot', 0.04847392369449283), ('ravg', 9.467563221580632e-05), ('rnorm', 0.02414514457151587), ('rnorm100', 2.414514457151587)])

Strat 1 Name St1:
  - analyzer: OrderedDict([('rtot', 0.05124714332260593), ('ravg', 0.00010009207680196471), ('rnorm', 0.025543999840699633), ('rnorm100', 2.5543999840699634)])

[vladisld]

unfortunately manually registering the strategies neutralize all the advantages of the idea expressed in the original blog post - being it the automatic/implicit strategy registration :-(