Module backtrader.analyzer

Classes

class Analyzer (*args, **kwargs)

Analyzer base class. All analyzers are subclass of this one

An Analyzer instance operates in the frame of a strategy and provides an analysis for that strategy.

Automagically set member attributes:

• self.strategy (giving access to the strategy and anything accessible from it)

• self.datas[x] giving access to the array of data feeds present in the the system, which could also be accessed via the strategy reference

• self.data, giving access to self.datas[0]

• self.dataX -> self.datas[X]

• self.dataX_Y -> self.datas[X].lines[Y]

• self.dataX_name -> self.datas[X].name

• self.data_name -> self.datas[0].name

• self.data_Y -> self.datas[0].lines[Y]

This is not a Lines object, but the methods and operation follow the same design

• __init__ during instantiation and initial setup

• start / stop to signal the begin and end of operations

• prenext / nextstart / next family of methods that follow the calls made to the same methods in the strategy

• notify_trade / notify_order / notify_cashvalue / notify_fund which receive the same notifications as the equivalent methods of the strategy

The mode of operation is open and no pattern is preferred. As such the analysis can be generated with the next calls, at the end of operations during stop and even with a single method like notify_trade

The important thing is to override get_analysis to return a dict-like object containing the results of the analysis (the actual format is implementation dependent)

Expand source code

class Analyzer(with_metaclass(MetaAnalyzer, object)):
    '''Analyzer base class. All analyzers are subclass of this one

    An Analyzer instance operates in the frame of a strategy and provides an
    analysis for that strategy.

    Automagically set member attributes:

      - ``self.strategy`` (giving access to the *strategy* and anything
        accessible from it)

      - ``self.datas[x]`` giving access to the array of data feeds present in
        the the system, which could also be accessed via the strategy reference

      - ``self.data``, giving access to ``self.datas[0]``

      - ``self.dataX`` -> ``self.datas[X]``

      - ``self.dataX_Y`` -> ``self.datas[X].lines[Y]``

      - ``self.dataX_name`` -> ``self.datas[X].name``

      - ``self.data_name`` -> ``self.datas[0].name``

      - ``self.data_Y`` -> ``self.datas[0].lines[Y]``

    This is not a *Lines* object, but the methods and operation follow the same
    design

      - ``__init__`` during instantiation and initial setup

      - ``start`` / ``stop`` to signal the begin and end of operations

      - ``prenext`` / ``nextstart`` / ``next`` family of methods that follow
        the calls made to the same methods in the strategy

      - ``notify_trade`` / ``notify_order`` / ``notify_cashvalue`` /
        ``notify_fund`` which receive the same notifications as the equivalent
        methods of the strategy

    The mode of operation is open and no pattern is preferred. As such the
    analysis can be generated with the ``next`` calls, at the end of operations
    during ``stop`` and even with a single method like ``notify_trade``

    The important thing is to override ``get_analysis`` to return a *dict-like*
    object containing the results of the analysis (the actual format is
    implementation dependent)

    '''
    csv = True

    def __len__(self):
        '''Support for invoking ``len`` on analyzers by actually returning the
        current length of the strategy the analyzer operates on'''
        return len(self.strategy)

    def _register(self, child):
        self._children.append(child)

    def _prenext(self):
        for child in self._children:
            child._prenext()

        self.prenext()

    def _notify_cashvalue(self, cash, value):
        for child in self._children:
            child._notify_cashvalue(cash, value)

        self.notify_cashvalue(cash, value)

    def _notify_fund(self, cash, value, fundvalue, shares):
        for child in self._children:
            child._notify_fund(cash, value, fundvalue, shares)

        self.notify_fund(cash, value, fundvalue, shares)

    def _notify_trade(self, trade):
        for child in self._children:
            child._notify_trade(trade)

        self.notify_trade(trade)

    def _notify_order(self, order):
        for child in self._children:
            child._notify_order(order)

        self.notify_order(order)

    def _nextstart(self):
        for child in self._children:
            child._nextstart()

        self.nextstart()

    def _next(self):
        for child in self._children:
            child._next()

        self.next()

    def _start(self):
        for child in self._children:
            child._start()

        self.start()

    def _stop(self):
        for child in self._children:
            child._stop()

        self.stop()

    def notify_cashvalue(self, cash, value):
        '''Receives the cash/value notification before each next cycle'''
        pass

    def notify_fund(self, cash, value, fundvalue, shares):
        '''Receives the current cash, value, fundvalue and fund shares'''
        pass

    def notify_order(self, order):
        '''Receives order notifications before each next cycle'''
        pass

    def notify_trade(self, trade):
        '''Receives trade notifications before each next cycle'''
        pass

    def next(self):
        '''Invoked for each next invocation of the strategy, once the minum
        preiod of the strategy has been reached'''
        pass

    def prenext(self):
        '''Invoked for each prenext invocation of the strategy, until the minimum
        period of the strategy has been reached

        The default behavior for an analyzer is to invoke ``next``
        '''
        self.next()

    def nextstart(self):
        '''Invoked exactly once for the nextstart invocation of the strategy,
        when the minimum period has been first reached
        '''
        self.next()

    def start(self):
        '''Invoked to indicate the start of operations, giving the analyzer
        time to setup up needed things'''
        pass

    def stop(self):
        '''Invoked to indicate the end of operations, giving the analyzer
        time to shut down needed things'''
        pass

    def create_analysis(self):
        '''Meant to be overriden by subclasses. Gives a chance to create the
        structures that hold the analysis.

        The default behaviour is to create a ``OrderedDict`` named ``rets``
        '''
        self.rets = OrderedDict()

    def get_analysis(self):
        '''Returns a *dict-like* object with the results of the analysis

        The keys and format of analysis results in the dictionary is
        implementation dependent.

        It is not even enforced that the result is a *dict-like object*, just
        the convention

        The default implementation returns the default OrderedDict ``rets``
        created by the default ``create_analysis`` method

        '''
        return self.rets

    def print(self, *args, **kwargs):
        '''Prints the results returned by ``get_analysis`` via a standard
        ``Writerfile`` object, which defaults to writing things to standard
        output
        '''
        writer = bt.WriterFile(*args, **kwargs)
        writer.start()
        pdct = dict()
        pdct[self.__class__.__name__] = self.get_analysis()
        writer.writedict(pdct)
        writer.stop()

    def pprint(self, *args, **kwargs):
        '''Prints the results returned by ``get_analysis`` using the pretty
        print Python module (*pprint*)
        '''
        pp.pprint(self.get_analysis(), *args, **kwargs)

Subclasses

• TimeFrameAnalyzerBase
• AnnualReturn
• DrawDown
• GrossLeverage
• PeriodStats
• PositionsValue
• PyFolio
• SharpeRatio
• SQN
• TradeAnalyzer
• Transactions

Class variables

var csv

var frompackages

var packages

var params

Methods

def create_analysis(self)

Meant to be overriden by subclasses. Gives a chance to create the structures that hold the analysis.

The default behaviour is to create a OrderedDict named rets

Expand source code

def create_analysis(self):
    '''Meant to be overriden by subclasses. Gives a chance to create the
    structures that hold the analysis.

    The default behaviour is to create a ``OrderedDict`` named ``rets``
    '''
    self.rets = OrderedDict()

def get_analysis(self)

Returns a dict-like object with the results of the analysis

The keys and format of analysis results in the dictionary is implementation dependent.

It is not even enforced that the result is a dict-like object, just the convention

The default implementation returns the default OrderedDict rets created by the default create_analysis method

Expand source code

def get_analysis(self):
    '''Returns a *dict-like* object with the results of the analysis

    The keys and format of analysis results in the dictionary is
    implementation dependent.

    It is not even enforced that the result is a *dict-like object*, just
    the convention

    The default implementation returns the default OrderedDict ``rets``
    created by the default ``create_analysis`` method

    '''
    return self.rets

def next(self)

Invoked for each next invocation of the strategy, once the minum preiod of the strategy has been reached

Expand source code

def next(self):
    '''Invoked for each next invocation of the strategy, once the minum
    preiod of the strategy has been reached'''
    pass

def nextstart(self)

Invoked exactly once for the nextstart invocation of the strategy, when the minimum period has been first reached

Expand source code

def nextstart(self):
    '''Invoked exactly once for the nextstart invocation of the strategy,
    when the minimum period has been first reached
    '''
    self.next()

def notify_cashvalue(self, cash, value)

Receives the cash/value notification before each next cycle

Expand source code

def notify_cashvalue(self, cash, value):
    '''Receives the cash/value notification before each next cycle'''
    pass

def notify_fund(self, cash, value, fundvalue, shares)

Receives the current cash, value, fundvalue and fund shares

Expand source code

def notify_fund(self, cash, value, fundvalue, shares):
    '''Receives the current cash, value, fundvalue and fund shares'''
    pass

def notify_order(self, order)

Receives order notifications before each next cycle

Expand source code

def notify_order(self, order):
    '''Receives order notifications before each next cycle'''
    pass

def notify_trade(self, trade)

Receives trade notifications before each next cycle

Expand source code

def notify_trade(self, trade):
    '''Receives trade notifications before each next cycle'''
    pass

def pprint(self, *args, **kwargs)

Prints the results returned by get_analysis using the pretty print Python module (pprint)

Expand source code

def pprint(self, *args, **kwargs):
    '''Prints the results returned by ``get_analysis`` using the pretty
    print Python module (*pprint*)
    '''
    pp.pprint(self.get_analysis(), *args, **kwargs)

def prenext(self)

Invoked for each prenext invocation of the strategy, until the minimum period of the strategy has been reached

The default behavior for an analyzer is to invoke next

Expand source code

def prenext(self):
    '''Invoked for each prenext invocation of the strategy, until the minimum
    period of the strategy has been reached

    The default behavior for an analyzer is to invoke ``next``
    '''
    self.next()

def print(self, *args, **kwargs)

Prints the results returned by get_analysis via a standard Writerfile object, which defaults to writing things to standard output

Expand source code

def print(self, *args, **kwargs):
    '''Prints the results returned by ``get_analysis`` via a standard
    ``Writerfile`` object, which defaults to writing things to standard
    output
    '''
    writer = bt.WriterFile(*args, **kwargs)
    writer.start()
    pdct = dict()
    pdct[self.__class__.__name__] = self.get_analysis()
    writer.writedict(pdct)
    writer.stop()

def start(self)

Invoked to indicate the start of operations, giving the analyzer time to setup up needed things

Expand source code

def start(self):
    '''Invoked to indicate the start of operations, giving the analyzer
    time to setup up needed things'''
    pass

def stop(self)

Invoked to indicate the end of operations, giving the analyzer time to shut down needed things

Expand source code

def stop(self):
    '''Invoked to indicate the end of operations, giving the analyzer
    time to shut down needed things'''
    pass

class MetaAnalyzer (*args, **kwargs)

type(object) -> the object's type type(name, bases, dict, **kwds) -> a new type

Expand source code

class MetaAnalyzer(bt.MetaParams):
    def donew(cls, *args, **kwargs):
        '''
        Intercept the strategy parameter
        '''
        # Create the object and set the params in place
        _obj, args, kwargs = super(MetaAnalyzer, cls).donew(*args, **kwargs)

        _obj._children = list()

        _obj.strategy = strategy = bt.metabase.findowner(_obj, bt.Strategy)
        _obj._parent = bt.metabase.findowner(_obj, Analyzer)

        # Register with a master observer if created inside one
        masterobs = bt.metabase.findowner(_obj, bt.Observer)
        if masterobs is not None:
            masterobs._register_analyzer(_obj)

        _obj.datas = strategy.datas

        # For each data add aliases: for first data: data and data0
        if _obj.datas:
            _obj.data = data = _obj.datas[0]

            for l, line in enumerate(data.lines):
                linealias = data._getlinealias(l)
                if linealias:
                    setattr(_obj, 'data_%s' % linealias, line)
                setattr(_obj, 'data_%d' % l, line)

            for d, data in enumerate(_obj.datas):
                setattr(_obj, 'data%d' % d, data)

                for l, line in enumerate(data.lines):
                    linealias = data._getlinealias(l)
                    if linealias:
                        setattr(_obj, 'data%d_%s' % (d, linealias), line)
                    setattr(_obj, 'data%d_%d' % (d, l), line)

        _obj.create_analysis()

        # Return to the normal chain
        return _obj, args, kwargs

    def dopostinit(cls, _obj, *args, **kwargs):
        _obj, args, kwargs = \
            super(MetaAnalyzer, cls).dopostinit(_obj, *args, **kwargs)

        if _obj._parent is not None:
            _obj._parent._register(_obj)

        # Return to the normal chain
        return _obj, args, kwargs

Ancestors

• MetaParams
• MetaBase
• builtins.type

Subclasses

• MetaTimeFrameAnalyzerBase

Methods

def donew(cls, *args, **kwargs)

Intercept the strategy parameter

Expand source code

def donew(cls, *args, **kwargs):
    '''
    Intercept the strategy parameter
    '''
    # Create the object and set the params in place
    _obj, args, kwargs = super(MetaAnalyzer, cls).donew(*args, **kwargs)

    _obj._children = list()

    _obj.strategy = strategy = bt.metabase.findowner(_obj, bt.Strategy)
    _obj._parent = bt.metabase.findowner(_obj, Analyzer)

    # Register with a master observer if created inside one
    masterobs = bt.metabase.findowner(_obj, bt.Observer)
    if masterobs is not None:
        masterobs._register_analyzer(_obj)

    _obj.datas = strategy.datas

    # For each data add aliases: for first data: data and data0
    if _obj.datas:
        _obj.data = data = _obj.datas[0]

        for l, line in enumerate(data.lines):
            linealias = data._getlinealias(l)
            if linealias:
                setattr(_obj, 'data_%s' % linealias, line)
            setattr(_obj, 'data_%d' % l, line)

        for d, data in enumerate(_obj.datas):
            setattr(_obj, 'data%d' % d, data)

            for l, line in enumerate(data.lines):
                linealias = data._getlinealias(l)
                if linealias:
                    setattr(_obj, 'data%d_%s' % (d, linealias), line)
                setattr(_obj, 'data%d_%d' % (d, l), line)

    _obj.create_analysis()

    # Return to the normal chain
    return _obj, args, kwargs

def dopostinit(cls, _obj, *args, **kwargs)

Expand source code

def dopostinit(cls, _obj, *args, **kwargs):
    _obj, args, kwargs = \
        super(MetaAnalyzer, cls).dopostinit(_obj, *args, **kwargs)

    if _obj._parent is not None:
        _obj._parent._register(_obj)

    # Return to the normal chain
    return _obj, args, kwargs

class MetaTimeFrameAnalyzerBase (*args, **kwargs)

type(object) -> the object's type type(name, bases, dict, **kwds) -> a new type

Expand source code

class MetaTimeFrameAnalyzerBase(Analyzer.__class__):
    def __new__(meta, name, bases, dct):
        # Hack to support original method name
        if '_on_dt_over' in dct:
            dct['on_dt_over'] = dct.pop('_on_dt_over')  # rename method

        return super(MetaTimeFrameAnalyzerBase, meta).__new__(meta, name,
                                                              bases, dct)

Ancestors

• MetaAnalyzer
• MetaParams
• MetaBase
• builtins.type

Inherited members

• MetaAnalyzer:
  • donew

class TimeFrameAnalyzerBase (*args, **kwargs)

Analyzer base class. All analyzers are subclass of this one

An Analyzer instance operates in the frame of a strategy and provides an analysis for that strategy.

Automagically set member attributes:

• self.strategy (giving access to the strategy and anything accessible from it)

• self.datas[x] giving access to the array of data feeds present in the the system, which could also be accessed via the strategy reference

• self.data, giving access to self.datas[0]

• self.dataX -> self.datas[X]

• self.dataX_Y -> self.datas[X].lines[Y]

• self.dataX_name -> self.datas[X].name

• self.data_name -> self.datas[0].name

• self.data_Y -> self.datas[0].lines[Y]

This is not a Lines object, but the methods and operation follow the same design

• __init__ during instantiation and initial setup

• start / stop to signal the begin and end of operations

• prenext / nextstart / next family of methods that follow the calls made to the same methods in the strategy

• notify_trade / notify_order / notify_cashvalue / notify_fund which receive the same notifications as the equivalent methods of the strategy

The mode of operation is open and no pattern is preferred. As such the analysis can be generated with the next calls, at the end of operations during stop and even with a single method like notify_trade

The important thing is to override get_analysis to return a dict-like object containing the results of the analysis (the actual format is implementation dependent)

Expand source code

class TimeFrameAnalyzerBase(with_metaclass(MetaTimeFrameAnalyzerBase,
                                           Analyzer)):
    params = (
        ('timeframe', None),
        ('compression', None),
        ('_doprenext', True),
    )

    def _start(self):
        # Override to add specific attributes
        self.timeframe = self.p.timeframe or self.data._timeframe
        self.compression = self.p.compression or self.data._compression

        self.dtcmp, self.dtkey = self._get_dt_cmpkey(datetime.datetime.min)
        super(TimeFrameAnalyzerBase, self)._start()

    def _prenext(self):
        for child in self._children:
            child._prenext()

        if self._dt_over():
            self.on_dt_over()

        if self.p._doprenext:
            self.prenext()

    def _nextstart(self):
        for child in self._children:
            child._nextstart()

        if self._dt_over() or not self.p._doprenext:  # exec if no prenext
            self.on_dt_over()

        self.nextstart()

    def _next(self):
        for child in self._children:
            child._next()

        if self._dt_over():
            self.on_dt_over()

        self.next()

    def on_dt_over(self):
        pass

    def _dt_over(self):
        if self.timeframe == TimeFrame.NoTimeFrame:
            dtcmp, dtkey = MAXINT, datetime.datetime.max
        else:
            # With >= 1.9.x the system datetime is in the strategy
            dt = self.strategy.datetime.datetime()
            dtcmp, dtkey = self._get_dt_cmpkey(dt)

        if self.dtcmp is None or dtcmp > self.dtcmp:
            self.dtkey, self.dtkey1 = dtkey, self.dtkey
            self.dtcmp, self.dtcmp1 = dtcmp, self.dtcmp
            return True

        return False

    def _get_dt_cmpkey(self, dt):
        if self.timeframe == TimeFrame.NoTimeFrame:
            return None, None

        if self.timeframe == TimeFrame.Years:
            dtcmp = dt.year
            dtkey = datetime.date(dt.year, 12, 31)

        elif self.timeframe == TimeFrame.Months:
            dtcmp = dt.year * 100 + dt.month
            _, lastday = calendar.monthrange(dt.year, dt.month)
            dtkey = datetime.datetime(dt.year, dt.month, lastday)

        elif self.timeframe == TimeFrame.Weeks:
            isoyear, isoweek, isoweekday = dt.isocalendar()
            dtcmp = isoyear * 100 + isoweek
            sunday = dt + datetime.timedelta(days=7 - isoweekday)
            dtkey = datetime.datetime(sunday.year, sunday.month, sunday.day)

        elif self.timeframe == TimeFrame.Days:
            dtcmp = dt.year * 10000 + dt.month * 100 + dt.day
            dtkey = datetime.datetime(dt.year, dt.month, dt.day)

        else:
            dtcmp, dtkey = self._get_subday_cmpkey(dt)

        return dtcmp, dtkey

    def _get_subday_cmpkey(self, dt):
        # Calculate intraday position
        point = dt.hour * 60 + dt.minute

        if self.timeframe < TimeFrame.Minutes:
            point = point * 60 + dt.second

        if self.timeframe < TimeFrame.Seconds:
            point = point * 1e6 + dt.microsecond

        # Apply compression to update point position (comp 5 -> 200 // 5)
        point = point // self.compression

        # Move to next boundary
        point += 1

        # Restore point to the timeframe units by de-applying compression
        point *= self.compression

        # Get hours, minutes, seconds and microseconds
        if self.timeframe == TimeFrame.Minutes:
            ph, pm = divmod(point, 60)
            ps = 0
            pus = 0
        elif self.timeframe == TimeFrame.Seconds:
            ph, pm = divmod(point, 60 * 60)
            pm, ps = divmod(pm, 60)
            pus = 0
        elif self.timeframe == TimeFrame.MicroSeconds:
            ph, pm = divmod(point, 60 * 60 * 1e6)
            pm, psec = divmod(pm, 60 * 1e6)
            ps, pus = divmod(psec, 1e6)

        extradays = 0
        if ph > 23:  # went over midnight:
            extradays = ph // 24
            ph %= 24

        # moving 1 minor unit to the left to be in the boundary
        # pm -= self.timeframe == TimeFrame.Minutes
        # ps -= self.timeframe == TimeFrame.Seconds
        # pus -= self.timeframe == TimeFrame.MicroSeconds

        tadjust = datetime.timedelta(
            minutes=self.timeframe == TimeFrame.Minutes,
            seconds=self.timeframe == TimeFrame.Seconds,
            microseconds=self.timeframe == TimeFrame.MicroSeconds)

        # Add extra day if present
        if extradays:
            dt += datetime.timedelta(days=extradays)

        # Replace intraday parts with the calculated ones and update it
        dtcmp = dt.replace(hour=ph, minute=pm, second=ps, microsecond=pus)
        dtcmp -= tadjust
        dtkey = dtcmp

        return dtcmp, dtkey

Ancestors

• Analyzer

Subclasses

• Calmar
• TimeDrawDown
• LogReturnsRolling
• Returns
• TimeReturn
• VWR

Class variables

var frompackages

var packages

var params

Methods

def on_dt_over(self)

Expand source code

def on_dt_over(self):
    pass

Inherited members

• Analyzer:
  • create_analysis
  • get_analysis
  • next
  • nextstart
  • notify_cashvalue
  • notify_fund
  • notify_order
  • notify_trade
  • pprint
  • prenext
  • print
  • start
  • stop