add data manager/downloader

Author: Brian

pytradelib/data.py

@@ -1,10 +1,29 @@
+import os
+from datetime import datetime
+from pytradelib.store import CSVStore
+from pytradelib.quandl.wiki import QuandlDailyWikiProvider
+from pytradelib.settings import DATA_DIR
 
-from pytradelib.utils import get_parse_symbols, csv_to_df
-from pytradelib.quandl import _construct_url as construct_url_quandl
-from pytradelib.quandl import _deconstruct_url as deconstruct_url_quandl
+class DataManager(object):
+    def __init__(self, store=None, data_provider=None):
+        self._store = store or CSVStore()
+        self._provider = data_provider or QuandlDailyWikiProvider()
 
+    def initialize_store(self):
+        raise NotImplementedError
 
-def get_symbols_quandl(symbols, start='2010-01-01', end='2010-08-31', interval=None):
-    if not isinstance(symbols, list):
-        symbols = [symbols]
-    return get_parse_symbols(symbols, start, end, interval, construct_url_quandl, deconstruct_url_quandl, csv_to_df)
+    def update_store(self):
+        symbols = dict([ (symbol, {'start': self._store.get_end_date(symbol),
+                                   'end': datetime.now()} )\
+                         for symbol in self._store.symbols ])
+        self._store.set_dfs(self._provider.download(symbols))
+
+    def analyze(self):
+        results = self._store.analyze()
+        filename = '%s-analysis.csv' % datetime.now().strftime('%Y-%m-%d')
+        results.to_csv(os.path.join(DATA_DIR, filename))
+        return results
+
+if __name__ == '__main__':
+    data_manager = DataManager(CSVStore(), QuandlDailyWikiProvider())
+    data_manager.update_store()

pytradelib/downloader.py

@@ -0,0 +1,74 @@
+from __future__  import print_function
+
+import urllib3.contrib.pyopenssl
+from pytradelib.utils import batch
+
+urllib3.contrib.pyopenssl.inject_into_urllib3()
+
+import requests
+import grequests
+from gevent import monkey
+monkey.patch_all()
+
+from pytradelib.logger import logger
+
+
+class Downloader(object):
+    def __init__(self, batch_size=100, sleep=None):
+        self._batch_size = batch_size
+        self._sleep = sleep
+
+    @property
+    def batch_size(self):
+        return self._batch_size
+
+    @batch_size.setter
+    def batch_size(self, batch_size):
+        self._batch_size = batch_size
+
+    @property
+    def sleep(self):
+        return self._sleep
+
+    @sleep.setter
+    def sleep(self, sleep):
+        self._sleep = sleep
+
+    def download(self, urls):
+        if isinstance(urls, str):
+            return self._download(urls)
+        return self._bulk_download(urls)
+
+    def _download(self, url):
+        logger.info('Download started: ' + url)
+        try:
+            r = requests.get(url)
+            logger.info('Download completed: ' + url)
+            if r.status_code == 200:
+                return r.content
+            r.raise_for_status()
+        except requests.exceptions.Timeout as e:
+            logger.error('Connection timed out: ' + e.__str__())
+        except requests.exceptions.RequestException as e:
+            logger.error('Error downloading: ' + e.__str__())
+        return None
+
+    def _bulk_download(self, urls):
+        results = []
+        for batched_urls in batch(urls, self.batch_size, self.sleep):
+            for r in self.__bulk_download(batched_urls):
+                print('finished downloading ' + r.url)
+                results.append( (r.url, r.content) )
+        return results
+
+    def __bulk_download(self, urls, errors=None):
+        errors = errors or []
+        def exception_handler(req, ex):
+            msg = 'Failed to download ' + req.url
+            if isinstance(ex, requests.exceptions.Timeout):
+                msg = 'Connection timed out: %(ex)s (%(url)s)' % {'ex': ex.__str__(), 'url': req.url}
+            elif isinstance(ex, requests.exceptions.RequestException):
+                msg = 'Error downloading: %(ex)s (%(url)s)' % {'ex': ex, 'url': req.url}
+            errors.append(req.url)
+            logger.error(msg)
+        return grequests.map((grequests.get(url) for url in urls), exception_handler=exception_handler)

pytradelib/logger.py

@@ -0,0 +1,24 @@
+import logging
+from logging.handlers import TimedRotatingFileHandler
+
+from pytradelib.settings import LOG_LEVEL, LOG_FILENAME
+
+LEVELS = {
+    'debug': logging.DEBUG,
+    'info': logging.INFO,
+    'warning': logging.WARNING,
+    'error': logging.ERROR,
+    'critical': logging.CRITICAL,
+}
+
+logger = logging.getLogger('PyTradeLib')
+logger.setLevel(LEVELS.get(LOG_LEVEL, logging.WARNING))
+handler = TimedRotatingFileHandler(LOG_FILENAME, 'midnight')
+handler.setFormatter(logging.Formatter(
+    '%(asctime)s %(levelname)s: pytradelib.%(module)s L%(lineno)s: %(message)s',
+    '%Y-%m-%d %H:%M:%S'
+))
+logger.addHandler(handler)
+
+
+__ALL__ = ('logger',)

pytradelib/utils.py

@@ -1,73 +1,77 @@
+from __future__ import print_function
+
 import os
+import time
+import pytz
 import pandas as pd
 
-pd.set_option('io.hdf.default_format', 'table')
-
-from pandas import HDFStore
 from pandas.compat import StringIO, bytes_to_str
 
-import grequests
-from gevent import monkey
-monkey.patch_all()
-
-DATA_DIR = os.environ['HOME'] + '/.pytradelib'
-__STORE = None
+import datetime as dt
+
+def batch(list_, size, sleep=None):
+    list_ = list(list_)
+    len_ = len(list_)
+    for i in xrange((len_ / size) + 1):
+        start_idx = i * size
+        end_idx = (i + 1) * size
+        if end_idx > len_:
+            end_idx = len_
+        yield list_[start_idx:end_idx]
+        if sleep:
+            print('Sleeping for %d seconds' % sleep)
+            time.sleep(sleep)
+
+
+def _sanitize_dates(start, end):
+    from pandas.core.datetools import to_datetime
+    start = to_datetime(start)
+    end = to_datetime(end)
+    if start is None:
+        start = dt.datetime(2010, 1, 1)
+    if end is None:
+        end = dt.datetime.today()
+    return start, end
 
 
-def _bulk_download(urls):
-    return grequests.imap((grequests.get(url) for url in urls))
-
-def get_parse_symbols(symbols, start, end, interval, symbol_to_url, url_to_symbol, data_to_df):
-    urls = (symbol_to_url(symbol.upper(), start, end, interval) for symbol in symbols)
-    def parse_response_to_symbol_and_df(r):
-        return url_to_symbol(r.url), data_to_df(r.text)
-    data = map(parse_response_to_symbol_and_df, _bulk_download(urls))
-    bulk_persist(data)
-    return data
-
 def csv_to_df(text):
-    rs = pd.read_csv(StringIO(bytes_to_str(text)), index_col=0,
-                     parse_dates=True, na_values='-')[::-1]
+    df = pd.read_csv(StringIO(bytes_to_str(text)), index_col=0,
+                     parse_dates=True, infer_datetime_format=True,
+                     na_values='-')[::-1]
 
     # Yahoo! Finance sometimes does this awesome thing where they
     # return 2 rows for the most recent business day
-    if len(rs) > 2 and rs.index[-1] == rs.index[-2]: # pragma: no cover
-        rs = rs[:-1]
+    if len(df) > 2 and df.index[-1] == df.index[-2]: # pragma: no cover
+        df = df[:-1]
 
-    # Get rid of unicode characters in index name.
+    # Get rid of unicode charactedf in index name.
     try:
-        rs.index.name = rs.index.name.decode('unicode_escape').encode('ascii', 'ignore')
+        df.index.name = df.index.name.decode('unicode_escape').encode('ascii', 'ignore')
     except AttributeError:
         # Python 3 string has no decode method.
-        rs.index.name = rs.index.name.encode('ascii', 'ignore').decode()
-    return rs
-
-def get_store():
-    global __STORE
-    if not __STORE:
-        if not os.path.exists(DATA_DIR):
-            os.mkdir(DATA_DIR)
-        __STORE = HDFStore(DATA_DIR + '/store.hdf5')
-    return __STORE
-
-def store_path(symbol, interval):
-    return '/symbols/%s/%s' % (symbol.upper(), interval.lower())
-
-def exists(symbol, interval):
-    store = get_store()
-    return store_path(symbol, interval) in store.keys()
-
-def persist(symbol, interval, df):
-    store = get_store()
-    if exists(symbol, interval):
-        store.append(store_path(symbol, interval), df)
-    else:
-        store.put(store_path(symbol, interval), df)
-
-def bulk_persist(data):
-    for symbol_data, df in data:
-        persist(symbol_data['symbol'], symbol_data['interval'], df)
-
-def most_recent_datetime(symbol, interval):
-    store = get_store()
-    return store.get(store_path(symbol, interval)).tail(1).index[0].to_datetime()
+        df.index.name = df.index.name.encode('ascii', 'ignore').decode()
+
+    column_renames = {'Adj. Open': 'Adj Open', 'Adj. High': 'Adj High',
+                      'Adj. Low': 'Adj Low', 'Adj. Close': 'Adj Close',
+                      'Adj. Volume': 'Adj Volume'}
+    df.rename(columns=column_renames, inplace=True)
+    return df.tz_localize(pytz.UTC)
+
+
+def percent_change(from_val, to_val):
+    # coerce to float for decimal division
+    diff = float(to_val) - from_val
+    return (diff / from_val) * 100
+
+
+def crossed(value, yesterday, today, use_adjusted=True):
+    def key(price_key):
+        return 'Adj ' + price_key if use_adjusted else price_key
+    crossed_over = yesterday[key('Close')] < value < today[key('Close')]
+    crossed_under = yesterday[key('Close')] > value > today[key('Close')]
+    return crossed_over or crossed_under
+
+
+def within_percent_of_value(price, value, percent=1):
+    diff = percent * 0.01 * 0.5 * value
+    return (value - diff) < price < (value + diff)