1#!/usr/bin/env python 2# -*- coding: utf-8; py-indent-offset:4 -*- 3############################################################################### 4# 5# Copyright (C) 2015, 2016, 2017 Daniel Rodriguez 6# 7# This program is free software: you can redistribute it and/or modify 8# it under the terms of the GNU General Public License as published by 9# the Free Software Foundation, either version 3 of the License, or 10# (at your option) any later version. 11# 12# This program is distributed in the hope that it will be useful, 13# but WITHOUT ANY WARRANTY; without even the implied warranty of 14# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 15# GNU General Public License for more details. 16# 17# You should have received a copy of the GNU General Public License 18# along with this program. If not, see <http://www.gnu.org/licenses/>. 19# 20############################################################################### 21from __future__ import (absolute_import, division, print_function, 22 unicode_literals) 23 24 25from datetime import datetime 26 27import backtrader as bt 28from backtrader.utils.py3 import range 29 30 31class MetaChainer(bt.DataBase.__class__): 32 def __init__(cls, name, bases, dct): 33 '''Class has already been created ... register''' 34 # Initialize the class 35 super(MetaChainer, cls).__init__(name, bases, dct) 36 37 def donew(cls, *args, **kwargs): 38 '''Intercept const. to copy timeframe/compression from 1st data''' 39 # Create the object and set the params in place 40 _obj, args, kwargs = super(MetaChainer, cls).donew(*args, **kwargs) 41 42 if args: 43 _obj.p.timeframe = args[0]._timeframe 44 _obj.p.compression = args[0]._compression 45 46 return _obj, args, kwargs 47 48 49class Chainer(bt.with_metaclass(MetaChainer, bt.DataBase)): 50 '''Class that chains datas''' 51 52 def islive(self): 53 '''Returns ``True`` to notify ``Cerebro`` that preloading and runonce 54 should be deactivated''' 55 return True 56 57 def __init__(self, *args): 58 self._args = args 59 60 def start(self): 61 super(Chainer, self).start() 62 for d in self._args: 63 d.setenvironment(self._env) 64 d._start() 65 66 # put the references in a separate list to have pops 67 self._ds = list(self._args) 68 self._d = self._ds.pop(0) if self._ds else None 69 self._lastdt = datetime.min 70 71 def stop(self): 72 super(Chainer, self).stop() 73 for d in self._args: 74 d.stop() 75 76 def get_notifications(self): 77 return [] if self._d is None else self._d.get_notifications() 78 79 def _gettz(self): 80 '''To be overriden by subclasses which may auto-calculate the 81 timezone''' 82 if self._args: 83 return self._args[0]._gettz() 84 return bt.utils.date.Localizer(self.p.tz) 85 86 def _load(self): 87 while self._d is not None: 88 if not self._d.next(): # no values from current data source 89 self._d = self._ds.pop(0) if self._ds else None 90 continue 91 92 # Cannot deliver a date equal or less than an alredy delivered 93 dt = self._d.datetime.datetime() 94 if dt <= self._lastdt: 95 continue 96 97 self._lastdt = dt 98 99 for i in range(self._d.size()): 100 self.lines[i][0] = self._d.lines[i][0] 101 102 return True 103 104 # Out of the loop -> self._d is None, no data feed to return from 105 return False 106