-
-
Save Yoio/d4f40e7d07fdd53f804c0831fcc957a5 to your computer and use it in GitHub Desktop.
talib deque (composition over inheritance)
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import numpy as np | |
from collections import deque | |
from talib.abstract import Function | |
class Bar(object): | |
'''A simplistic holder for bar data with attribute and dict-style lookup.''' | |
open = None | |
high = None | |
low = None | |
close = None | |
volume = None | |
def __init__(self, open_, high, low, close, volume): | |
self['open'] = open_ | |
self['high'] = high | |
self['low'] = low | |
self['close'] = close | |
self['volume'] = volume | |
def __getitem__(self, k): | |
return self.__getattribute__(k) | |
def __setitem__(self, k, v): | |
self.__setattr__(k, v) | |
def __repr__(self): | |
return "Bar(O: %(open)s, H: %(high)s, L: %(low)s, C: %(close)s, "\ | |
"V: %(volume)s)" % self | |
class QFunction(object): | |
def __init__(self, fn_name, *args, **kwargs): | |
self._fn = Function(fn_name, *args, **kwargs) | |
self._maxlen = self._fn.lookback + 1 | |
self._q = deque(maxlen=self._maxlen) | |
def put(self, bar): | |
self._q.append(bar) | |
if len(self._q) < self._maxlen: | |
return None | |
else: | |
d = {'open': np.asarray([b.open for b in self._q]), | |
'high': np.asarray([b.high for b in self._q]), | |
'low': np.asarray([b.low for b in self._q]), | |
'close': np.asarray([b.close for b in self._q]), | |
'volume': np.asarray([b.volume for b in self._q]), | |
} | |
results = self._fn(d) | |
if isinstance(results, (list, tuple)): | |
return tuple([r[-1] for r in results]) | |
else: | |
return results[-1] | |
if __name__ == '__main__': | |
from talib.test_data import ford_2012 | |
bars = [Bar(ford_2012['open'][i], | |
ford_2012['high'][i], | |
ford_2012['low'][i], | |
ford_2012['close'][i], | |
ford_2012['volume'][i] | |
) for i in xrange(len(ford_2012['low']))] | |
fn_name = 'SMA' | |
qfn = QFunction(fn_name, timeperiod=10) | |
results = [qfn.put(bar) for bar in bars] | |
print results | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment