forked from mementum/backtrader
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathstore.py
More file actions
94 lines (74 loc) · 3.06 KB
/
store.py
File metadata and controls
94 lines (74 loc) · 3.06 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
#!/usr/bin/env python
# -*- coding: utf-8; py-indent-offset:4 -*-
###############################################################################
#
# Copyright (C) 2015-2020 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)]