BLD: implementing issue #65, adding local symbols definition

catalyst/constants.py

@@ -5,5 +5,6 @@ import logbook
 LOG_LEVEL = logbook.INFO
 
 DATE_TIME_FORMAT = '%Y-%m-%d %H:%M'
+DATE_FORMAT = '%Y-%m-%d'
 
 AUTO_INGEST = False

catalyst/exchange/exchange_bundle.py

@@ -1,3 +1,4 @@
+import json
 import os
 import shutil
 from datetime import datetime, timedelta
@@ -26,8 +27,9 @@ from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \
 from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \
     TempBundleNotFoundError, \
     NoDataAvailableOnExchange, \
-    PricingDataNotLoadedError, DataCorruptionError
-from catalyst.exchange.exchange_utils import get_exchange_folder
+    PricingDataNotLoadedError, DataCorruptionError, ExchangeSymbolsNotFound
+from catalyst.exchange.exchange_utils import get_exchange_folder, \
+    get_exchange_symbols, perf_serial, symbols_serial
 from catalyst.utils.cli import maybe_show_progress
 from catalyst.utils.paths import ensure_directory
 
@@ -661,13 +663,73 @@ class ExchangeBundle:
 
         """
         log.info('ingesting csv file: {}'.format(path))
-        problems = []
+        try:
+            symbols_def = get_exchange_symbols(
+                self.exchange_name, is_local=True
+            )
+        except ExchangeSymbolsNotFound:
+            symbols_def = dict()
 
+        problems = []
         df = pd.read_csv(
             path,
-            names=['symbol', 'last_traded', 'open', 'high', 'close', 'volume'],
-            parse_dates=[1]
+            header=0,
+            sep=',',
+            dtype=dict(
+                symbol=np.object_,
+                last_traded=np.object_,
+                open=np.float64,
+                high=np.float64,
+                close=np.float64,
+                volume=np.float64
+            ),
+            parse_dates=['last_traded'],
+            index_col=None
         )
+
+        symbols = df['symbol'].unique()
+        trading_pairs = dict()
+        for symbol in symbols:
+            start_dt = df['last_traded'].min()
+            end_dt = df['last_traded'].max()
+            end_dt_key = 'end_{}'.format(data_frequency)
+
+            if symbol is symbols_def:
+                symbol_def = symbols_def[symbol]
+
+                start_dt = symbol_def['start_date'] \
+                    if symbol_def['start_date'] < start_dt else start_dt
+
+                end_dt = symbol_def[end_dt_key] \
+                    if symbol_def[end_dt_key] > end_dt else end_dt
+
+                end_daily = end_dt \
+                    if data_frequency == 'daily' else symbol_def['end_daily']
+
+                end_minute = end_dt \
+                    if data_frequency == 'minute' else symbol_def['end_minute']
+
+            else:
+                end_daily = end_dt if data_frequency == 'daily' else None
+                end_minute = end_dt if data_frequency == 'minute' else None
+
+            trading_pair = TradingPair(
+                symbol=symbol,
+                exchange=self.exchange_name,
+                start_date=start_dt,
+                end_date=end_dt,
+                leverage=0,  # TODO: add as an optional column
+                asset_name=symbol,
+                min_trade_size=0,  # TODO: add as an optional column
+                end_daily=end_daily,
+                end_minute=end_minute,
+                exchange_symbol=symbol
+            )
+            trading_pairs[symbol] = trading_pair.to_dict()
+
+        symbols_def_json = json.dumps(trading_pairs, default=symbols_serial)
+        df.set_index(['symbol', 'last_traded'], drop=True, inplace=True)
+        df.tz_localize('UTC', level=1)
         # problems += self.ingest_df(
         #     ohlcv_df=df,
         #     data_frequency=data_frequency,

catalyst/exchange/exchange_utils.py

@@ -9,6 +9,7 @@ import pandas as pd
 from catalyst.assets._assets import TradingPair
 from six.moves.urllib import request
 
+from catalyst.constants import DATE_TIME_FORMAT, DATE_FORMAT
 from catalyst.exchange.exchange_errors import ExchangeSymbolsNotFound, \
     InvalidHistoryFrequencyError, InvalidHistoryFrequencyAlias
 from catalyst.utils.paths import data_root, ensure_directory, \
@@ -42,7 +43,7 @@ def get_exchange_folder(exchange_name, environ=None):
     return exchange_folder
 
 
-def get_exchange_symbols_filename(exchange_name, environ=None):
+def get_exchange_symbols_filename(exchange_name, is_local=False, environ=None):
     """
     The absolute path of the exchange's symbol.json file.
 
@@ -56,8 +57,9 @@ def get_exchange_symbols_filename(exchange_name, environ=None):
     str
 
     """
+    name = 'symbols.json' if not is_local else 'symbols_local.json'
     exchange_folder = get_exchange_folder(exchange_name, environ)
-    return os.path.join(exchange_folder, 'symbols.json')
+    return os.path.join(exchange_folder, name)
 
 
 def download_exchange_symbols(exchange_name, environ=None):
@@ -80,13 +82,14 @@ def download_exchange_symbols(exchange_name, environ=None):
     return response
 
 
-def get_exchange_symbols(exchange_name, environ=None):
+def get_exchange_symbols(exchange_name, is_local=False, environ=None):
     """
     The de-serialized content of the exchange's symbols.json.
 
     Parameters
     ----------
     exchange_name: str
+    is_local: bool
     environ:
 
     Returns
@@ -94,12 +97,11 @@ def get_exchange_symbols(exchange_name, environ=None):
     Object
 
     """
-    filename = get_exchange_symbols_filename(exchange_name)
+    filename = get_exchange_symbols_filename(exchange_name, is_local)
 
-    if not os.path.isfile(filename) or \
-                    pd.Timedelta(pd.Timestamp('now',
-                                              tz='UTC') - last_modified_time(
-                        filename)).days > 1:
+    if not is_local and (not os.path.isfile(filename) or pd.Timedelta(
+                pd.Timestamp('now', tz='UTC') - last_modified_time(
+                filename)).days > 1):
         download_exchange_symbols(exchange_name, environ)
 
     if os.path.isfile(filename):
@@ -363,6 +365,25 @@ def get_exchange_bundles_folder(exchange_name, environ=None):
     return temp_bundles
 
 
+def symbols_serial(obj):
+    """
+    JSON serializer for objects not serializable by default json code
+
+    Parameters
+    ----------
+    obj: Object
+
+    Returns
+    -------
+    str
+
+    """
+    if isinstance(obj, (datetime, date)):
+        return obj.floor('1D').strftime(DATE_FORMAT)
+
+    raise TypeError("Type %s not serializable" % type(obj))
+
+
 def perf_serial(obj):
     """
     JSON serializer for objects not serializable by default json code

tests/exchange/test_bundle.py

@@ -525,3 +525,13 @@ class TestExchangeBundle:
         print('TAIL\n{}'.format(df.tail(10)))
         df.to_csv(path)
         pass
+
+    def test_ingest_csv(self):
+        data_frequency = 'minute'
+        exchange_name = 'bittrex'
+        path = '/Users/fredfortier/Dropbox/Enigma/Data/bat_eth.csv'
+
+        exchange_bundle = ExchangeBundle(exchange_name)
+        exchange_bundle.ingest_csv(path, data_frequency)
+
+        pass