From 8ab3382f4783f0b25e249c53836791af56ef54ef Mon Sep 17 00:00:00 2001 From: Frederic Fortier Date: Fri, 29 Dec 2017 18:35:36 -0500 Subject: [PATCH] BLD: for issue #121, refactored the data portal --- catalyst/exchange/exchange_algorithm.py | 8 + catalyst/exchange/exchange_data_portal.py | 197 ++++++++++------------ 2 files changed, 99 insertions(+), 106 deletions(-) diff --git a/catalyst/exchange/exchange_algorithm.py b/catalyst/exchange/exchange_algorithm.py index 7dfe22b9..fdb0373d 100644 --- a/catalyst/exchange/exchange_algorithm.py +++ b/catalyst/exchange/exchange_algorithm.py @@ -79,6 +79,8 @@ class ExchangeTradingAlgorithmBase(TradingAlgorithm): get_order_attempts=5, get_open_orders_attempts=5, cancel_order_attempts=5, + get_stop_value_attempts=5, + get_history_window_attempts=5, retry_sleeptime=5, ) @@ -289,6 +291,12 @@ class ExchangeTradingAlgorithmBase(TradingAlgorithm): return stats + def run(self, data=None, overwrite_sim_params=True): + data.attempts = self.attempts + return super(ExchangeTradingAlgorithmBase, self).run( + data, overwrite_sim_params + ) + class ExchangeTradingAlgorithmBacktest(ExchangeTradingAlgorithmBase): def __init__(self, *args, **kwargs): diff --git a/catalyst/exchange/exchange_data_portal.py b/catalyst/exchange/exchange_data_portal.py index e9db33ce..1e79e5b3 100644 --- a/catalyst/exchange/exchange_data_portal.py +++ b/catalyst/exchange/exchange_data_portal.py @@ -5,6 +5,7 @@ import numpy as np import pandas as pd from catalyst.assets._assets import TradingPair from logbook import Logger +from redo import retry from catalyst.constants import LOG_LEVEL, AUTO_INGEST from catalyst.data.data_portal import DataPortal @@ -27,6 +28,12 @@ class DataPortalExchangeBase(DataPortal): self.retry_get_spot_value = 5 self.retry_delay = 5 + self.attempts = dict( + get_stop_value_attempts=5, + get_history_window_attempts=5, + retry_sleeptime=5, + ) + super(DataPortalExchangeBase, self).__init__(*args, **kwargs) def _get_history_window(self, @@ -36,33 +43,14 @@ class DataPortalExchangeBase(DataPortal): frequency, field, data_frequency, - ffill=True, - attempt_index=0): - try: - exchange_assets = group_assets_by_exchange(assets) - if len(exchange_assets) > 1: - df_list = [] - for exchange_name in exchange_assets: - assets = exchange_assets[exchange_name] + ffill=True): + exchange_assets = group_assets_by_exchange(assets) + if len(exchange_assets) > 1: + df_list = [] + for exchange_name in exchange_assets: + assets = exchange_assets[exchange_name] - df_exchange = self.get_exchange_history_window( - exchange_name, - assets, - end_dt, - bar_count, - frequency, - field, - data_frequency, - ffill) - - df_list.append(df_exchange) - - # Merging the values values of each exchange - return pd.concat(df_list) - - else: - exchange_name = list(exchange_assets.keys())[0] - return self.get_exchange_history_window( + df_exchange = self.get_exchange_history_window( exchange_name, assets, end_dt, @@ -72,26 +60,22 @@ class DataPortalExchangeBase(DataPortal): data_frequency, ffill) - except ExchangeRequestError as e: - log.warn( - 'get history attempt {}: {}'.format(attempt_index, e) - ) - if attempt_index < self.retry_get_history_window: - sleep(self.retry_delay) - return self._get_history_window(assets, - end_dt, - bar_count, - frequency, - field, - data_frequency, - ffill, - attempt_index + 1) - else: - raise ExchangeBarDataError( - data_type='history', - attempts=attempt_index, - error=e - ) + df_list.append(df_exchange) + + # Merging the values values of each exchange + return pd.concat(df_list) + + else: + exchange_name = list(exchange_assets.keys())[0] + return self.get_exchange_history_window( + exchange_name, + assets, + end_dt, + bar_count, + frequency, + field, + data_frequency, + ffill) def get_history_window(self, assets, @@ -105,13 +89,22 @@ class DataPortalExchangeBase(DataPortal): if field == 'price': field = 'close' - return self._get_history_window(assets, - end_dt, - bar_count, - frequency, - field, - data_frequency, - ffill) + return retry( + action=self._get_history_window, + attempts=self.attempts['get_history_window_attempts'], + sleeptime=self.attempts['retry_sleeptime'], + retry_exceptions=(ExchangeRequestError,), + cleanup=lambda e: log.warn( + 'fetching history again: {}'.format(e) + ), + args=(assets, + end_dt, + bar_count, + frequency, + field, + data_frequency, + ffill) + ) @abc.abstractmethod def get_exchange_history_window(self, @@ -125,69 +118,61 @@ class DataPortalExchangeBase(DataPortal): ffill=True): pass - def _get_spot_value(self, assets, field, dt, data_frequency, - attempt_index=0): - try: - if isinstance(assets, TradingPair): - spot_values = self.get_exchange_spot_value( - assets.exchange, [assets], field, dt, data_frequency) + def _get_spot_value(self, assets, field, dt, data_frequency): + if isinstance(assets, TradingPair): + spot_values = self.get_exchange_spot_value( + assets.exchange, [assets], field, dt, data_frequency) - if not spot_values: - return np.nan + if not spot_values: + return np.nan - return spot_values[0] + return spot_values[0] + + else: + exchange_assets = dict() + for asset in assets: + if asset.exchange not in exchange_assets: + exchange_assets[asset.exchange] = list() + + exchange_assets[asset.exchange].append(asset) + + if len(list(exchange_assets.keys())) == 1: + exchange_name = list(exchange_assets.keys())[0] + return self.get_exchange_spot_value( + exchange_name, assets, field, dt, data_frequency) else: - exchange_assets = dict() - for asset in assets: - if asset.exchange not in exchange_assets: - exchange_assets[asset.exchange] = list() + spot_values = [] + for exchange_name in exchange_assets: + assets = exchange_assets[exchange_name] + exchange_spot_values = self.get_exchange_spot_value( + exchange_name, + assets, + field, + dt, + data_frequency + ) + if len(assets) == 1: + spot_values.append(exchange_spot_values) + else: + spot_values += exchange_spot_values - exchange_assets[asset.exchange].append(asset) - - if len(list(exchange_assets.keys())) == 1: - exchange_name = list(exchange_assets.keys())[0] - return self.get_exchange_spot_value( - exchange_name, assets, field, dt, data_frequency) - - else: - spot_values = [] - for exchange_name in exchange_assets: - assets = exchange_assets[exchange_name] - exchange_spot_values = self.get_exchange_spot_value( - exchange_name, - assets, - field, - dt, - data_frequency - ) - if len(assets) == 1: - spot_values.append(exchange_spot_values) - else: - spot_values += exchange_spot_values - - return spot_values - - except ExchangeRequestError as e: - log.warn( - 'get spot value attempt {}: {}'.format(attempt_index, e) - ) - if attempt_index < self.retry_get_spot_value: - sleep(self.retry_delay) - return self._get_spot_value(assets, field, dt, data_frequency, - attempt_index + 1) - else: - raise ExchangeBarDataError( - data_type='spot', - attempts=attempt_index, - error=e - ) + return spot_values def get_spot_value(self, assets, field, dt, data_frequency): if field == 'price': field = 'close' - return self._get_spot_value(assets, field, dt, data_frequency) + return retry( + action=self._get_spot_value, + attempts=self.attempts['get_stop_value_attempts'], + sleeptime=self.attempts['retry_sleeptime'], + retry_exceptions=(ExchangeRequestError,), + cleanup=lambda e: log.warn( + 'fetching spot value again: {}'.format(e) + ), + args=(assets, field, dt, data_frequency), + ) @abc.abstractmethod def get_exchange_spot_value(self, exchange_name, assets, field, dt,