From b47602461262f2633d4f57e554469dab8de75c80 Mon Sep 17 00:00:00 2001 From: fredfortier Date: Fri, 3 Nov 2017 20:59:26 -0400 Subject: [PATCH] BUG: work around for issue #53 and possibly fixed issue #47 --- catalyst/examples/simple_loop.py | 32 ++++++------ catalyst/exchange/bundle_utils.py | 6 +++ catalyst/exchange/exchange_bundle.py | 78 ++++++++++++++-------------- catalyst/exchange/exchange_utils.py | 3 +- tests/exchange/test_bundle.py | 68 ++++++++++++++++++------ 5 files changed, 114 insertions(+), 73 deletions(-) diff --git a/catalyst/examples/simple_loop.py b/catalyst/examples/simple_loop.py index 9ebf93b1..45fcbce8 100644 --- a/catalyst/examples/simple_loop.py +++ b/catalyst/examples/simple_loop.py @@ -29,25 +29,25 @@ def handle_data(context, data): print(e) -# run_algorithm( -# capital_base=250, -# start=pd.to_datetime('2017-1-1', utc=True), -# end=pd.to_datetime('2017-10-22', utc=True), -# data_frequency='daily', -# initialize=initialize, -# handle_data=handle_data, -# analyze=None, -# exchange_name='bitfinex', -# algo_namespace='simple_loop', -# base_currency='btc' -# ) run_algorithm( + capital_base=250, + start=pd.to_datetime('2017-1-1', utc=True), + end=pd.to_datetime('2017-10-22', utc=True), + data_frequency='daily', initialize=initialize, handle_data=handle_data, analyze=None, - exchange_name='poloniex', - live=True, + exchange_name='bitfinex', algo_namespace='simple_loop', - base_currency='eth', - live_graph=False + base_currency='btc' ) +# run_algorithm( +# initialize=initialize, +# handle_data=handle_data, +# analyze=None, +# exchange_name='poloniex', +# live=True, +# algo_namespace='simple_loop', +# base_currency='eth', +# live_graph=False +# ) diff --git a/catalyst/exchange/bundle_utils.py b/catalyst/exchange/bundle_utils.py index 4510bce4..ad3fbfd1 100644 --- a/catalyst/exchange/bundle_utils.py +++ b/catalyst/exchange/bundle_utils.py @@ -223,6 +223,9 @@ def get_month_start_end(dt, first_day=None, last_day=None): dt.year, dt.month, month_range[1], 23, 59, 0, 0 ), utc=True) + if month_end > pd.Timestamp.utcnow(): + month_end = pd.Timestamp.utcnow().floor('1D') + return month_start, month_end @@ -247,6 +250,9 @@ def get_year_start_end(dt, first_day=None, last_day=None): year_end = last_day if last_day \ else pd.to_datetime(date(dt.year, 12, 31), utc=True) + if year_end > pd.Timestamp.utcnow(): + year_end = pd.Timestamp.utcnow().floor('1D') + return year_start, year_end diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index f5df0f8e..d25b26ec 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/catalyst/exchange/exchange_bundle.py @@ -303,7 +303,7 @@ class ExchangeBundle: :return: """ - + # Download and extract the bundle path = get_bcolz_chunk( exchange_name=self.exchange.name, symbol=asset.symbol, @@ -335,7 +335,7 @@ class ExchangeBundle: )) if not arrays: - return path + return reader._rootdir periods = self.get_calendar_periods_range( start_dt, end_dt, data_frequency @@ -351,11 +351,12 @@ class ExchangeBundle: if cleanup: log.debug( - 'removing bundle folder following ingestion: {}'.format(path) + 'removing bundle folder following ingestion: {}'.format( + reader._rootdir) ) - shutil.rmtree(path) + shutil.rmtree(reader._rootdir) - return path + return reader._rootdir def get_adj_dates(self, start, end, assets, data_frequency): """ @@ -429,9 +430,7 @@ class ExchangeBundle: get_start_end = get_month_start_end \ if data_frequency == 'minute' else get_year_start_end - start_dt, _ = get_start_end(start_dt) - _, end_dt = get_start_end(end_dt) - + # Get a reader for the main bundle to verify if data exists reader = self.get_reader(data_frequency) chunks = dict() @@ -462,7 +461,6 @@ class ExchangeBundle: chunks[asset] = [] for index, dt in enumerate(dates): - period_start, period_end = get_start_end( dt=dt, first_day=dt if index == 0 else None, @@ -481,17 +479,17 @@ class ExchangeBundle: asset, range_start, period_end, reader ) if not has_data: - chunks[asset].append( - dict( - asset=asset, - period_start=period_start, - period_end=period_end, - period=get_period_label(dt, data_frequency) - ) + period = get_period_label(dt, data_frequency) + chunk = dict( + asset=asset, + period=period, ) + chunks[asset].append(chunk) # We sort the chunks by end date to ingest most recent data first - chunks[asset].sort(key=lambda chunk: chunk['period_end']) + chunks[asset].sort( + key=lambda chunk: pd.to_datetime(chunk['period']) + ) return chunks @@ -503,20 +501,26 @@ class ExchangeBundle: Parameters ---------- assets: list[TradingPair] + data_frequency: str start_dt: datetime end_dt: datetime + show_progress: bool + asset_chunks: bool """ - if start_dt is None: start_dt = self.calendar.first_session if end_dt is None: end_dt = pd.Timestamp.utcnow() - start_dt, end_dt = self.get_adj_dates( - start_dt, end_dt, assets, data_frequency - ) + get_start_end = get_month_start_end \ + if data_frequency == 'minute' else get_year_start_end + + # Assign the first and last day of the period + start_dt, _ = get_start_end(start_dt) + _, end_dt = get_start_end(end_dt) + chunks = self.prepare_chunks( assets=assets, data_frequency=data_frequency, @@ -524,19 +528,9 @@ class ExchangeBundle: end_dt=end_dt ) - # Since chunks are either monthly or yearly, it is possible that - # our ingestion data range is greater than specified. We adjust - # the boundaries to ensure that the writer can write all data. - all_chunks = list(chain.from_iterable(itervalues(chunks))) - for chunk in all_chunks: - if chunk['period_start'] < start_dt: - start_dt = chunk['period_start'] - - if chunk['period_end'] > end_dt: - end_dt = chunk['period_end'] - + # This is the common writer for the entire exchange bundle + # we want to give an end_date far in time writer = self.get_writer(start_dt, end_dt, data_frequency) - if asset_chunks: for asset in chunks: with maybe_show_progress( @@ -558,6 +552,12 @@ class ExchangeBundle: cleanup=True ) else: + all_chunks = list(chain.from_iterable(itervalues(chunks))) + + # We sort the chunks by end date to ingest most recent data first + all_chunks.sort( + key=lambda chunk: pd.to_datetime(chunk['period']) + ) with maybe_show_progress( all_chunks, show_progress, @@ -600,12 +600,12 @@ class ExchangeBundle: show_progress) def get_history_window_series_and_load(self, - assets, # type: List[TradingPair] - end_dt, # type: Timestamp - bar_count, # type: int - field, # type: str - data_frequency, # type: str - algo_end_dt=None # type: Timestamp + assets, + end_dt, + bar_count, + field, + data_frequency, + algo_end_dt=None ): """ Retrieve price data history, ingest missing data. diff --git a/catalyst/exchange/exchange_utils.py b/catalyst/exchange/exchange_utils.py index 4db41d9e..b4f828fe 100644 --- a/catalyst/exchange/exchange_utils.py +++ b/catalyst/exchange/exchange_utils.py @@ -284,13 +284,12 @@ def save_algo_df(algo_name, key, df, environ=None, rel_path=None): ---------- algo_name: str key: str - df: DataFrame + df: pd.DataFrame environ: rel_path: str """ folder = get_algo_folder(algo_name, environ) - if rel_path is not None: folder = os.path.join(folder, rel_path) ensure_directory(folder) diff --git a/tests/exchange/test_bundle.py b/tests/exchange/test_bundle.py index c5e5feb9..6cdd45ca 100644 --- a/tests/exchange/test_bundle.py +++ b/tests/exchange/test_bundle.py @@ -122,18 +122,20 @@ class TestExchangeBundle: pass def test_ingest_daily(self): - # exchange_name = 'bitfinex' - # data_frequency = 'daily' - # include_symbols = 'neo_btc,bch_btc,eth_btc' - - exchange_name = 'poloniex' + exchange_name = 'bitfinex' data_frequency = 'daily' - include_symbols = 'eth_btc' + include_symbols = 'btc_usd' - start = pd.to_datetime('2017-1-1', utc=True) - end = pd.to_datetime('2017-10-16', utc=True) - periods = get_periods_range(start, end, data_frequency) + # exchange_name = 'poloniex' + # data_frequency = 'daily' + # include_symbols = 'eth_btc' + # start = pd.to_datetime('2017-1-1', utc=True) + # end = pd.to_datetime('2017-10-16', utc=True) + # periods = get_periods_range(start, end, data_frequency) + + start = None + end = None exchange = get_exchange(exchange_name) exchange_bundle = ExchangeBundle(exchange) @@ -153,12 +155,18 @@ class TestExchangeBundle: assets.append(exchange.get_asset(pair_symbol)) reader = exchange_bundle.get_reader(data_frequency) + start_dt = reader.first_trading_day + end_dt = reader.last_available_dt + + if data_frequency == 'daily': + end_dt = end_dt - pd.Timedelta(hours=23, minutes=59) + for asset in assets: arrays = reader.load_raw_arrays( sids=[asset.sid], fields=['close'], - start_dt=start, - end_dt=end + start_dt=start_dt, + end_dt=end_dt ) print('found {} rows for {} ingestion\n{}'.format( len(arrays[0]), asset.symbol, arrays[0]) @@ -430,22 +438,50 @@ class TestExchangeBundle: print('\n' + df_to_string(df)) pass - def bundle_to_csv(self): + def main_bundle_to_csv(self): exchange_name = 'bitfinex' - data_frequency = 'daily' - period = '2016' + data_frequency = 'minute' exchange = get_exchange(exchange_name) - bundle = ExchangeBundle(exchange) asset = exchange.get_asset('eth_btc') + self._bundle_to_csv( + asset=asset, + exchange=exchange, + data_frequency=data_frequency, + filename='{}_{}_{}'.format( + exchange_name, data_frequency, asset.symbol + ) + ) + + def bundle_to_csv(self): + exchange_name = 'bitfinex' + data_frequency = 'minute' + period = '2017-06' + symbol = 'etc_btc' + + exchange = get_exchange(exchange_name) + asset = exchange.get_asset(symbol) + path = get_bcolz_chunk( exchange_name=exchange.name, symbol=asset.symbol, data_frequency=data_frequency, period=period ) + self._bundle_to_csv( + asset=asset, + exchange=exchange, + data_frequency=data_frequency, + path=path, + filename=period + ) + + def _bundle_to_csv(self, asset, exchange, data_frequency, filename, + path=None): + bundle = ExchangeBundle(exchange) reader = bundle.get_reader(data_frequency, path=path) + start_dt = reader.first_trading_day end_dt = reader.last_available_dt @@ -475,7 +511,7 @@ class TestExchangeBundle: ) ensure_directory(folder) - path = os.path.join(folder, period + '.csv') + path = os.path.join(folder, filename + '.csv') log.info('creating csv file: {}'.format(path)) print('HEAD\n{}'.format(df.head(10)))