BLD: retry refactoring for issue #121, more testing required

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 = []

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
 

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))
 

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):

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)
-                )
+                ),
             )