diff --git a/catalyst/examples/buy_low_sell_high.py b/catalyst/examples/buy_low_sell_high.py index d481d2e2..f1ec25d7 100644 --- a/catalyst/examples/buy_low_sell_high.py +++ b/catalyst/examples/buy_low_sell_high.py @@ -37,10 +37,6 @@ def initialize(context): context.PROFIT_TARGET = 0.1 context.SLIPPAGE_ALLOWED = 0.05 - context.retry_check_open_orders = 10 - context.retry_update_portfolio = 10 - context.retry_order = 5 - context.swallow_errors = True context.errors = [] diff --git a/catalyst/examples/buy_low_sell_high_live.py b/catalyst/examples/buy_low_sell_high_live.py index 3b884063..3cfc1051 100644 --- a/catalyst/examples/buy_low_sell_high_live.py +++ b/catalyst/examples/buy_low_sell_high_live.py @@ -24,10 +24,6 @@ def initialize(context): context.PROFIT_TARGET = 0.1 context.SLIPPAGE_ALLOWED = 0.02 - context.retry_check_open_orders = 10 - context.retry_update_portfolio = 10 - context.retry_order = 5 - context.errors = [] pass diff --git a/catalyst/exchange/ccxt/ccxt_exchange.py b/catalyst/exchange/ccxt/ccxt_exchange.py index acd1e114..59d7bbf9 100644 --- a/catalyst/exchange/ccxt/ccxt_exchange.py +++ b/catalyst/exchange/ccxt/ccxt_exchange.py @@ -377,7 +377,6 @@ class CCXT(Exchange): candles = dict() for asset in assets: - # try: ohlcvs = self.api.fetch_ohlcv( symbol=symbols[0], timeframe=timeframe, @@ -399,9 +398,6 @@ class CCXT(Exchange): volume=ohlcv[5] )) - # except Exception as e: - # raise ExchangeRequestError(error=e) - if is_single: return six.next(six.itervalues(candles)) diff --git a/catalyst/exchange/exchange_algorithm.py b/catalyst/exchange/exchange_algorithm.py index 3e26c4d2..509c00e8 100644 --- a/catalyst/exchange/exchange_algorithm.py +++ b/catalyst/exchange/exchange_algorithm.py @@ -17,10 +17,10 @@ import sys from datetime import timedelta from os import listdir from os.path import isfile, join -from time import sleep import logbook import pandas as pd +from redo import retry import catalyst.protocol as zp from catalyst.algorithm import TradingAlgorithm @@ -28,7 +28,6 @@ from catalyst.constants import LOG_LEVEL from catalyst.exchange.exchange_blotter import ExchangeBlotter from catalyst.exchange.exchange_errors import ( ExchangeRequestError, - ExchangePortfolioDataError, OrderTypeNotSupported) from catalyst.exchange.exchange_execution import ExchangeLimitOrder from catalyst.exchange.live_graph_clock import LiveGraphClock @@ -72,12 +71,22 @@ class ExchangeTradingAlgorithmBase(TradingAlgorithm): and self.sim_params.arena == 'backtest': self.simulate_orders = True + # Operations with retry features + self.attempts = dict( + get_transactions_attempts=5, + order_attempts=5, + synchronize_portfolio_attempts=5, + get_open_orders_attempts=5, + retry_sleeptime=5, + ) + self.blotter = ExchangeBlotter( data_frequency=self.data_frequency, # Default to NeverCancel in catalyst cancel_policy=self.cancel_policy, simulate_orders=self.simulate_orders, - exchanges=self.exchanges + exchanges=self.exchanges, + attempts=self.attempts, ) @staticmethod @@ -345,12 +354,6 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): self.is_running = True - self.retry_check_open_orders = 5 - self.retry_synchronize_portfolio = 5 - self.retry_get_open_orders = 5 - self.retry_order = 2 - self.retry_delay = 5 - self.stats_minutes = 1 super(ExchangeTradingAlgorithmLive, self).__init__(*args, **kwargs) @@ -470,7 +473,7 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): def updated_account(self): return self.perf_tracker.get_account(False) - def synchronize_portfolio(self, attempt_index=0): + def synchronize_portfolio(self): """ Synchronizes the portfolio tracked by the algorithm to refresh its current value. @@ -498,60 +501,45 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): total_cash = 0.0 total_positions_value = 0.0 - try: - # Position keys correspond to assets - positions = self.portfolio.positions - assets = list(positions) - exchange_assets = group_assets_by_exchange(assets) - for exchange_name in self.exchanges: - assets = exchange_assets[exchange_name] \ - if exchange_name in exchange_assets else [] + # Position keys correspond to assets + positions = self.portfolio.positions + assets = list(positions) + exchange_assets = group_assets_by_exchange(assets) + for exchange_name in self.exchanges: + assets = exchange_assets[exchange_name] \ + if exchange_name in exchange_assets else [] - exchange_positions = copy.deepcopy( - [positions[asset] for asset in assets] - ) - - exchange = self.exchanges[exchange_name] # Type: Exchange - - if base_currency is None: - base_currency = exchange.base_currency - - cash, positions_value = exchange.sync_positions( - positions=exchange_positions, - check_balances=check_balances, - cash=self.portfolio.cash, - ) - total_cash += cash - total_positions_value += positions_value - - # Applying modifications to the original positions - for position in exchange_positions: - tracker.update_position( - asset=position.asset, - amount=position.amount, - last_sale_date=position.last_sale_date, - last_sale_price=position.last_sale_price, - ) - - if not check_balances: - total_cash = self.portfolio.cash - - return total_cash, total_positions_value - - except ExchangeRequestError as e: - log.warn( - 'update portfolio attempt {}: {}'.format(attempt_index, e) + exchange_positions = copy.deepcopy( + [positions[asset] for asset in assets] ) - if attempt_index < self.retry_synchronize_portfolio: - sleep(self.retry_delay) - return self.synchronize_portfolio(attempt_index + 1) - else: - raise ExchangePortfolioDataError( - data_type='update-portfolio', - attempts=attempt_index, - error=e + + exchange = self.exchanges[exchange_name] # Type: Exchange + + if base_currency is None: + base_currency = exchange.base_currency + + cash, positions_value = exchange.sync_positions( + positions=exchange_positions, + check_balances=check_balances, + cash=self.portfolio.cash, + ) + total_cash += cash + total_positions_value += positions_value + + # Applying modifications to the original positions + for position in exchange_positions: + tracker.update_position( + asset=position.asset, + amount=position.amount, + last_sale_date=position.last_sale_date, + last_sale_price=position.last_sale_price, ) + if not check_balances: + total_cash = self.portfolio.cash + + return total_cash, total_positions_value + def add_pnl_stats(self, period_stats): """ Save p&l stats. @@ -652,7 +640,15 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): if len(new_transactions) > 0: self.perf_tracker.update_performance() - cash, positions_value = self.synchronize_portfolio() + cash, positions_value = retry( + action=self.synchronize_portfolio, + attempts=self.attempts['synchronize_portfolio_attempts'], + sleeptime=self.attempts['retry_sleeptime'], + retry_exceptions=(ExchangeRequestError,), + cleanup=lambda e: log.warn( + 'ordering again: {}'.format(e) + ), + ) log.info( 'got totals from exchanges, cash: {} positions: {}'.format( cash, positions_value @@ -762,33 +758,19 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): def batch_market_order(self, share_counts): raise NotImplementedError() - def _get_open_orders(self, asset=None, attempt_index=0): - try: - if asset: - exchange = self.exchanges[asset.exchange] - return exchange.get_open_orders(asset) + def _get_open_orders(self, asset=None): + if asset: + exchange = self.exchanges[asset.exchange] + return exchange.get_open_orders(asset) - else: - open_orders = [] - for exchange_name in self.exchanges: - exchange = self.exchanges[exchange_name] - exchange_orders = exchange.get_open_orders() - open_orders.append(exchange_orders) + else: + open_orders = [] + for exchange_name in self.exchanges: + exchange = self.exchanges[exchange_name] + exchange_orders = exchange.get_open_orders() + open_orders.append(exchange_orders) - return open_orders - except ExchangeRequestError as e: - log.warn( - 'open orders attempt {}: {}'.format(attempt_index, e) - ) - if attempt_index < self.retry_get_open_orders: - sleep(self.retry_delay) - return self._get_open_orders(asset, attempt_index + 1) - else: - raise ExchangePortfolioDataError( - data_type='open-orders', - attempts=attempt_index, - error=e - ) + return open_orders @error_keywords(sid='Keyword argument `sid` is no longer supported for ' 'get_open_orders. Use `asset` instead.') @@ -810,7 +792,16 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): If an asset is passed then this will return a list of the open orders for this asset. """ - return self._get_open_orders(asset) + return retry( + action=self._get_open_orders, + attempts=self.attempts['get_open_orders_attempts'], + sleeptime=self.attempts['retry_sleeptime'], + retry_exceptions=(ExchangeRequestError,), + cleanup=lambda e: log.warn( + 'fetching open orders again: {}'.format(e) + ), + args=(asset,) + ) @api_method def get_order(self, order_id, exchange_name): diff --git a/catalyst/exchange/exchange_blotter.py b/catalyst/exchange/exchange_blotter.py index 19073b05..06494788 100644 --- a/catalyst/exchange/exchange_blotter.py +++ b/catalyst/exchange/exchange_blotter.py @@ -1,13 +1,10 @@ -from time import sleep - import pandas as pd from catalyst.assets._assets import TradingPair from logbook import Logger from redo import retry from catalyst.constants import LOG_LEVEL -from catalyst.exchange.exchange_errors import ExchangeRequestError, \ - ExchangePortfolioDataError, ExchangeTransactionError +from catalyst.exchange.exchange_errors import ExchangeRequestError from catalyst.finance.blotter import Blotter from catalyst.finance.commission import CommissionModel from catalyst.finance.order import ORDER_STATUS @@ -134,6 +131,7 @@ class TradingPairFixedSlippage(SlippageModel): class ExchangeBlotter(Blotter): def __init__(self, *args, **kwargs): self.simulate_orders = kwargs.pop('simulate_orders', False) + self.attempts = kwargs.pop('attempts', False) self.exchanges = kwargs.pop('exchanges', None) if not self.exchanges: @@ -153,32 +151,11 @@ class ExchangeBlotter(Blotter): TradingPair: TradingPairFeeSchedule() } - self.retry_delay = 5 - self.retry_check_open_orders = 5 - self.retry_order = 5 - - def exchange_order(self, asset, amount, style=None, attempt_index=0): - try: - exchange = self.exchanges[asset.exchange] - return exchange.order( - asset, amount, style - ) - except ExchangeRequestError as e: - log.warn( - 'order attempt {}: {}'.format(attempt_index, e) - ) - if attempt_index < self.retry_order: - sleep(self.retry_delay) - - return self.exchange_order( - asset, amount, style, attempt_index + 1 - ) - else: - raise ExchangeTransactionError( - transaction_type='order', - attempts=attempt_index, - error=e - ) + def exchange_order(self, asset, amount, style=None): + exchange = self.exchanges[asset.exchange] + return exchange.order( + asset, amount, style + ) @expect_types(asset=TradingPair) def order(self, asset, amount, style, order_id=None): @@ -193,8 +170,15 @@ class ExchangeBlotter(Blotter): ) else: - order = self.exchange_order( - asset, amount, style + order = retry( + action=self.get_exchange_transactions, + attempts=self.attempts['order_attempts'], + sleeptime=self.attempts['retry_sleeptime'], + retry_exceptions=(ExchangeRequestError,), + cleanup=lambda e: log.warn( + 'ordering again: {}'.format(e) + ), + args=(asset, amount, style), ) self.open_orders[order.asset].append(order) @@ -283,10 +267,10 @@ class ExchangeBlotter(Blotter): else: return retry( action=self.get_exchange_transactions, - attempts=self.retry_check_open_orders, - sleeptime=self.retry_delay, - retry_exceptions=ExchangeRequestError, + attempts=self.attempts['get_transactions_attempts'], + sleeptime=self.retry_sleeptime, + retry_exceptions=(ExchangeRequestError,), cleanup=lambda e: log.warn( 'fetching exchange transactions again: {}'.format(e) - ) + ), )