Module `backtrader.store`

Expand source code

#!/usr/bin/env python
# -*- coding: utf-8; py-indent-offset:4 -*-
###############################################################################
#
# Copyright (C) 2015-2023 Daniel Rodriguez
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
###############################################################################
from __future__ import (absolute_import, division, print_function,
                        unicode_literals)

import collections

from backtrader.metabase import MetaParams
from backtrader.utils.py3 import with_metaclass


class MetaSingleton(MetaParams):
    '''Metaclass to make a metaclassed class a singleton'''
    def __init__(cls, name, bases, dct):
        super(MetaSingleton, cls).__init__(name, bases, dct)
        cls._singleton = None

    def __call__(cls, *args, **kwargs):
        if cls._singleton is None:
            cls._singleton = (
                super(MetaSingleton, cls).__call__(*args, **kwargs))

        return cls._singleton


class Store(with_metaclass(MetaSingleton, object)):
    '''Base class for all Stores'''

    _started = False

    params = ()

    def getdata(self, *args, **kwargs):
        '''Returns ``DataCls`` with args, kwargs'''
        data = self.DataCls(*args, **kwargs)
        data._store = self
        return data

    @classmethod
    def getbroker(cls, *args, **kwargs):
        '''Returns broker with *args, **kwargs from registered ``BrokerCls``'''
        broker = cls.BrokerCls(*args, **kwargs)
        broker._store = cls
        return broker

    BrokerCls = None  # broker class will autoregister
    DataCls = None  # data class will auto register

    def start(self, data=None, broker=None):
        if not self._started:
            self._started = True
            self.notifs = collections.deque()
            self.datas = list()
            self.broker = None

        if data is not None:
            self._cerebro = self._env = data._env
            self.datas.append(data)

            if self.broker is not None:
                if hasattr(self.broker, 'data_started'):
                    self.broker.data_started(data)

        elif broker is not None:
            self.broker = broker

    def stop(self):
        pass

    def put_notification(self, msg, *args, **kwargs):
        self.notifs.append((msg, args, kwargs))

    def get_notifications(self):
        '''Return the pending "store" notifications'''
        self.notifs.append(None)  # put a mark / threads could still append
        return [x for x in iter(self.notifs.popleft, None)]

Classes

class MetaSingleton (name, bases, dct)

Metaclass to make a metaclassed class a singleton

Expand source code

class MetaSingleton(MetaParams):
    '''Metaclass to make a metaclassed class a singleton'''
    def __init__(cls, name, bases, dct):
        super(MetaSingleton, cls).__init__(name, bases, dct)
        cls._singleton = None

    def __call__(cls, *args, **kwargs):
        if cls._singleton is None:
            cls._singleton = (
                super(MetaSingleton, cls).__call__(*args, **kwargs))

        return cls._singleton

Ancestors

MetaParams
MetaBase
builtins.type

class Store (*args, **kwargs)

Base class for all Stores

Expand source code

class Store(with_metaclass(MetaSingleton, object)):
    '''Base class for all Stores'''

    _started = False

    params = ()

    def getdata(self, *args, **kwargs):
        '''Returns ``DataCls`` with args, kwargs'''
        data = self.DataCls(*args, **kwargs)
        data._store = self
        return data

    @classmethod
    def getbroker(cls, *args, **kwargs):
        '''Returns broker with *args, **kwargs from registered ``BrokerCls``'''
        broker = cls.BrokerCls(*args, **kwargs)
        broker._store = cls
        return broker

    BrokerCls = None  # broker class will autoregister
    DataCls = None  # data class will auto register

    def start(self, data=None, broker=None):
        if not self._started:
            self._started = True
            self.notifs = collections.deque()
            self.datas = list()
            self.broker = None

        if data is not None:
            self._cerebro = self._env = data._env
            self.datas.append(data)

            if self.broker is not None:
                if hasattr(self.broker, 'data_started'):
                    self.broker.data_started(data)

        elif broker is not None:
            self.broker = broker

    def stop(self):
        pass

    def put_notification(self, msg, *args, **kwargs):
        self.notifs.append((msg, args, kwargs))

    def get_notifications(self):
        '''Return the pending "store" notifications'''
        self.notifs.append(None)  # put a mark / threads could still append
        return [x for x in iter(self.notifs.popleft, None)]

Subclasses

VChartFile

Class variables

var BrokerCls
var DataCls
var frompackages
var packages
var params

Static methods

def getbroker(*args, **kwargs)

Returns broker with args, *kwargs from registered BrokerCls

Expand source code

@classmethod
def getbroker(cls, *args, **kwargs):
    '''Returns broker with *args, **kwargs from registered ``BrokerCls``'''
    broker = cls.BrokerCls(*args, **kwargs)
    broker._store = cls
    return broker

Methods

def get_notifications(self)

Return the pending "store" notifications

Expand source code

def get_notifications(self):
    '''Return the pending "store" notifications'''
    self.notifs.append(None)  # put a mark / threads could still append
    return [x for x in iter(self.notifs.popleft, None)]

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

Returns DataCls with args, kwargs

Expand source code

def getdata(self, *args, **kwargs):
    '''Returns ``DataCls`` with args, kwargs'''
    data = self.DataCls(*args, **kwargs)
    data._store = self
    return data

def put_notification(self, msg, *args, **kwargs)

Expand source code

def put_notification(self, msg, *args, **kwargs):
    self.notifs.append((msg, args, kwargs))

def start(self, data=None, broker=None)

Expand source code

def start(self, data=None, broker=None):
    if not self._started:
        self._started = True
        self.notifs = collections.deque()
        self.datas = list()
        self.broker = None

    if data is not None:
        self._cerebro = self._env = data._env
        self.datas.append(data)

        if self.broker is not None:
            if hasattr(self.broker, 'data_started'):
                self.broker.data_started(data)

    elif broker is not None:
        self.broker = broker

def stop(self)

Expand source code

def stop(self):
    pass