diff --git a/catalyst/constants.py b/catalyst/constants.py index cde29914..172de34e 100644 --- a/catalyst/constants.py +++ b/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 \ No newline at end of file diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index da7908e2..25eaab86 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/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, diff --git a/catalyst/exchange/exchange_utils.py b/catalyst/exchange/exchange_utils.py index d4119491..3546ed28 100644 --- a/catalyst/exchange/exchange_utils.py +++ b/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 diff --git a/tests/exchange/test_bundle.py b/tests/exchange/test_bundle.py index cd5f2bde..eed7af4f 100644 --- a/tests/exchange/test_bundle.py +++ b/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