BLD: improved portfolio synchronization in live trading to handle situations where positions in the exchange are less than tracked by the algo

catalyst/exchange/ccxt/ccxt_exchange.py

@@ -68,6 +68,7 @@ class CCXT(Exchange):
 
         self.num_candles_limit = 2000
         self.max_requests_per_minute = 60
+        self.low_balance_threshold = 0.1
         self.request_cpt = dict()
 
         self.bundle = ExchangeBundle(self.name)

catalyst/exchange/exchange.py

@@ -16,7 +16,8 @@ from catalyst.exchange.exchange_bundle import ExchangeBundle
 from catalyst.exchange.exchange_errors import MismatchingBaseCurrencies, \
     BaseCurrencyNotFoundError, SymbolNotFoundOnExchange, \
     PricingDataNotLoadedError, \
-    NoDataAvailableOnExchange, NoValueForField, LastCandleTooEarlyError
+    NoDataAvailableOnExchange, NoValueForField, LastCandleTooEarlyError, \
+    TickerNotFoundError, BalanceNotFoundError, BalanceTooLowError
 from catalyst.exchange.exchange_utils import get_exchange_symbols, \
     get_frequency, resample_history_df, has_bundle
 from catalyst.utils.deprecate import deprecated
@@ -40,6 +41,8 @@ class Exchange:
         self.request_cpt = None
         self.bundle = ExchangeBundle(self.name)
 
+        self.low_balance_threshold = None
+
     @abstractproperty
     def account(self):
         pass
@@ -651,16 +654,56 @@ class Exchange:
 
         return df
 
-    def calculate_totals(self, check_cash=False, positions=None):
+    def _check_low_balance(self, currency, balances, amount):
+        free = balances[currency]['free'] \
+            if currency in balances else None
+
+        if free is None or free == 0:
+            raise BalanceNotFoundError(
+                currency=currency,
+                exchange=self.name,
+                balances=balances,
+            )
+
+        if free < amount:
+            limit = amount * (1 - self.low_balance_threshold)
+            if free < limit:
+                raise BalanceTooLowError(
+                    currency=currency,
+                    exchange=self.name,
+                    free=free,
+                    amount=amount,
+                )
+
+            log.debug(
+                'detected lower balance for {} on {}: {} < {}, '
+                'updating position amount'.format(
+                    currency, self.name, free, amount
+                )
+            )
+            return free, True
+
+        else:
+            return free, False
+
+    def sync_positions(self, positions, check_balances=False):
         """
         Update the portfolio cash and position balances based on the
         latest ticker prices.
 
+        Parameters
+        ----------
+        positions:
+            The positions to synchronize.
+
+        check_balances:
+            Check balances amounts against the exchange.
+
         """
         log.debug('synchronizing portfolio with exchange {}'.format(self.name))
 
         cash = None
-        if check_cash:
+        if check_balances:
             balances = self.get_balances()
 
             cash = balances[self.base_currency]['free'] \
@@ -669,26 +712,51 @@ class Exchange:
             if cash is None:
                 raise BaseCurrencyNotFoundError(
                     base_currency=self.base_currency,
-                    exchange=self.name
+                    exchange=self.name,
+                    balances=balances,
                 )
             log.debug('found base currency balance: {}'.format(cash))
 
         positions_value = 0.0
-        if positions:
+        if positions is not None:
             assets = set([position.asset for position in positions])
             tickers = self.tickers(assets)
-            log.debug('got tickers for positions: {}'.format(tickers))
 
-            for asset in tickers:
+            for position in positions:
+                asset = position.asset
+                if asset not in tickers:
+                    raise TickerNotFoundError(
+                        symbol=asset.symbol,
+                        exchange=self.name,
+                    )
+
                 ticker = tickers[asset]
-                positions = [p for p in positions if p.asset == asset]
+                log.debug(
+                    'updating {} position with ticker: {}'.format(
+                        asset.symbol, ticker
+                    )
+                )
+                position.last_sale_price = ticker['last_price']
+                position.last_sale_date = ticker['last_traded']
 
-                for position in positions:
-                    position.last_sale_price = ticker['last_price']
-                    position.last_sale_date = ticker['last_traded']
+                positions_value += \
+                    position.amount * position.last_sale_price
 
-                    positions_value += \
-                        position.amount * position.last_sale_price
+                if check_balances:
+                    free, is_lower = self._check_low_balance(
+                        currency=asset.base_currency,
+                        balances=balances,
+                        amount=position.amount,
+                    )
+
+                    if is_lower:
+                        log.debug(
+                            'detected lower balance for {} on {}: {} < {}, '
+                            'updating position amount'.format(
+                                asset.symbol, self.name, free, position.amount
+                            )
+                        )
+                        position.amount = free
 
         return cash, positions_value
 

catalyst/exchange/exchange_algorithm.py

@@ -18,6 +18,7 @@ from os import listdir
 from os.path import isfile, join
 from time import sleep
 
+import copy
 import logbook
 import pandas as pd
 
@@ -28,7 +29,7 @@ from catalyst.exchange.exchange_blotter import ExchangeBlotter
 from catalyst.exchange.exchange_errors import (
     ExchangeRequestError,
     ExchangePortfolioDataError,
-    OrderTypeNotSupported, )
+    OrderTypeNotSupported, CashTooLowError)
 from catalyst.exchange.exchange_execution import ExchangeLimitOrder
 from catalyst.exchange.exchange_utils import (
     save_algo_object,
@@ -495,6 +496,8 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase):
             The total value of all tracked positions.
 
         """
+        check_balances = (not self.simulate_orders)
+        base_currency = None
         tracker = self.perf_tracker.position_tracker
         total_cash = 0.0
         total_positions_value = 0.0
@@ -508,33 +511,42 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase):
                 assets = exchange_assets[exchange_name] \
                     if exchange_name in exchange_assets else []
 
-                exchange_positions = \
+                exchange_positions = copy.deepcopy(
                     [positions[asset] for asset in assets]
-
-                check_cash = (not self.simulate_orders)
+                )
 
                 exchange = self.exchanges[exchange_name]  # Type: Exchange
-                cash, positions_value = exchange.calculate_totals(
-                    positions=exchange_positions,
-                    check_cash=check_cash,
-                )
-                total_positions_value += positions_value
 
+                if base_currency is None:
+                    base_currency = exchange.base_currency
+
+                cash, positions_value = exchange.sync_positions(
+                    positions=exchange_positions,
+                    check_balances=check_balances,
+                )
                 if cash is not None:
                     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
+                        last_sale_price=position.last_sale_price,
                     )
 
-            if cash is None:
+            if not check_balances:
                 total_cash = self.portfolio.cash
 
             elif total_cash < self.portfolio.cash:
-                raise ValueError('Cash on exchanges is lower than the algo.')
+                raise CashTooLowError(
+                    currency=self.exchanges[0].base_currency,
+                    free=total_cash,
+                    cash=self.portfolio.cash,
+                )
 
             return total_cash, total_positions_value
 

catalyst/exchange/exchange_errors.py

@@ -33,22 +33,22 @@ class ExchangeRequestErrorTooManyAttempts(ZiplineError):
 
 class ExchangeBarDataError(ZiplineError):
     msg = (
-            'Unable to retrieve bar data: {data_type}, ' +
-            'giving up after {attempts} attempts: {error}'
+        'Unable to retrieve bar data: {data_type}, ' +
+        'giving up after {attempts} attempts: {error}'
     ).strip()
 
 
 class ExchangePortfolioDataError(ZiplineError):
     msg = (
-            'Unable to retrieve portfolio data: {data_type}, ' +
-            'giving up after {attempts} attempts: {error}'
+        'Unable to retrieve portfolio data: {data_type}, ' +
+        'giving up after {attempts} attempts: {error}'
     ).strip()
 
 
 class ExchangeTransactionError(ZiplineError):
     msg = (
-            'Unable to execute transaction: {transaction_type}, ' +
-            'giving up after {attempts} attempts: {error}'
+        'Unable to execute transaction: {transaction_type}, ' +
+        'giving up after {attempts} attempts: {error}'
     ).strip()
 
 
@@ -168,8 +168,8 @@ class SidHashError(ZiplineError):
 
 class BaseCurrencyNotFoundError(ZiplineError):
     msg = (
-        'Algorithm base currency {base_currency} not found in exchange '
-        '{exchange}.'
+        'Algorithm base currency {base_currency} not found in account '
+        'balances on {exchange}: {balances}'
     ).strip()
 
 
@@ -232,16 +232,20 @@ class PricingDataValueError(ZiplineError):
 
 
 class DataCorruptionError(ZiplineError):
-    msg = ('Unable to validate data for {exchange} {symbols} in date range '
-           '[{start_dt} - {end_dt}]. The data is either corrupted or '
-           'unavailable. Please try deleting this bundle:'
-           '\n`catalyst clean-exchange -x {exchange}\n'
-           'Then, ingest the data again. Please contact the Catalyst team if '
-           'the issue persists.').strip()
+    msg = (
+        'Unable to validate data for {exchange} {symbols} in date range '
+        '[{start_dt} - {end_dt}]. The data is either corrupted or '
+        'unavailable. Please try deleting this bundle:'
+        '\n`catalyst clean-exchange -x {exchange}\n'
+        'Then, ingest the data again. Please contact the Catalyst team if '
+        'the issue persists.'
+    ).strip()
 
 
 class ApiCandlesError(ZiplineError):
-    msg = ('Unable to fetch candles from the remote API: {error}.').strip()
+    msg = (
+        'Unable to fetch candles from the remote API: {error}.'
+    ).strip()
 
 
 class NoDataAvailableOnExchange(ZiplineError):
@@ -254,13 +258,16 @@ class NoDataAvailableOnExchange(ZiplineError):
 
 
 class NoValueForField(ZiplineError):
-    msg = ('Value not found for field: {field}.').strip()
+    msg = (
+        'Value not found for field: {field}.'
+    ).strip()
 
 
 class OrderTypeNotSupported(ZiplineError):
     msg = (
-        'Order type `{order_type}` not currencly supported by Catalyst. '
-        'Please use `limit` or `market` orders only.').strip()
+        'Order type `{order_type}` not currency supported by Catalyst. '
+        'Please use `limit` or `market` orders only.'
+    ).strip()
 
 
 class NotEnoughCapitalError(ZiplineError):
@@ -268,11 +275,43 @@ class NotEnoughCapitalError(ZiplineError):
         'Not enough capital on exchange {exchange} for trading. Each '
         'exchange should contain at least as much {base_currency} '
         'as the specified `capital_base`. The current balance {balance} is '
-        'lower than the `capital_base`: {capital_base}').strip()
+        'lower than the `capital_base`: {capital_base}'
+    ).strip()
 
 
 class LastCandleTooEarlyError(ZiplineError):
     msg = (
         'The trade date of the last candle {last_traded} is before the '
         'specified end date minus one candle {end_dt}. Please verify how '
-        '{exchange} calculates the start date of OHLCV candles.').strip()
+        '{exchange} calculates the start date of OHLCV candles.'
+    ).strip()
+
+
+class TickerNotFoundError(ZiplineError):
+    msg = (
+        'Unable to fetch ticker for {symbol} on {exchange}.'
+    ).strip()
+
+
+class BalanceNotFoundError(ZiplineError):
+    msg = (
+        '{currency} not found in account balance on {exchange}: {balances}.'
+    ).strip()
+
+
+class BalanceTooLowError(ZiplineError):
+    msg = (
+        'Balance for {currency} on {exchange} too low: {free} < {amount}. '
+        'Positions have likely been sold outside of this algorithm. Please '
+        'add positions to hold a free amount greater than {amount}, or clean '
+        'the state of this algo and restart.'
+    ).strip()
+
+
+class CashTooLowError(ZiplineError):
+    msg = (
+        'Total {currency} amount on exchanges is lower than the cash reserved '
+        'for this algo: {free} < {cash}. While trades can be made on the '
+        'exchange accounts outside of the algo, they must not compromise '
+        'the required amount of free {currency}.'
+    ).strip()