DOC: For issue #151, significantly improved the way in which we are processing order for exchanges supporting "fetch-my-trades"

catalyst/exchange/ccxt/ccxt_exchange.py

@@ -6,6 +6,11 @@ from collections import defaultdict
 import ccxt
 import pandas as pd
 import six
+from ccxt import InvalidOrder, NetworkError, \
+    ExchangeError
+from logbook import Logger
+from six import string_types
+
 from catalyst.algorithm import MarketOrder
 from catalyst.assets._assets import TradingPair
 from catalyst.constants import LOG_LEVEL
@@ -19,10 +24,7 @@ from catalyst.exchange.utils.exchange_utils import mixin_market_params, \
     from_ms_timestamp, get_epoch, get_exchange_folder, get_catalyst_symbol, \
     get_exchange_auth
 from catalyst.finance.order import Order, ORDER_STATUS
-from ccxt import InvalidOrder, NetworkError, \
-    ExchangeError
-from logbook import Logger
-from six import string_types
+from catalyst.finance.transaction import Transaction
 
 log = Logger('CCXT', level=LOG_LEVEL)
 
@@ -759,13 +761,110 @@ class CCXT(Exchange):
 
         orders = []
         for order_status in result:
-            order, executed_price = self._create_order(order_status)
+            order, _ = self._create_order(order_status)
             if asset is None or asset == order.sid:
                 orders.append(order)
 
         return orders
 
-    def get_order(self, order_id, asset_or_symbol=None):
+    def _get_executed_order_fallback(self, order):
+        """
+        Fallback method for exchanges which do not play nice with
+        fetch-my-trades. Apparently, about 60% of exchanges will return
+        the correct executed values with this method. Others will support
+        fetch-my-trades.
+
+        Parameters
+        ----------
+        order: Order
+
+        Returns
+        -------
+        float
+
+        """
+        exc_order, price = self.get_order(
+            order.id, order.asset, return_price=True
+        )
+        order.status = exc_order.status
+
+        order.commission = exc_order.commission
+        if order.amount != exc_order.amount:
+            log.warn(
+                'executed order amount {} differs '
+                'from original'.format(
+                    exc_order.amount, order.amount
+                )
+            )
+            order.amount = exc_order.amount
+
+        if order.status == ORDER_STATUS.FILLED:
+            transaction = Transaction(
+                asset=order.asset,
+                amount=order.amount,
+                dt=pd.Timestamp.utcnow(),
+                price=price,
+                order_id=order.id,
+                commission=order.commission
+            )
+        return [transaction]
+
+    def process_order(self, order):
+        if not self.api.hasFetchMyTrades:
+            return self._get_executed_order_fallback(order)
+
+        try:
+            all_trades = self.get_trades(order.asset)
+        except ExchangeRequestError as e:
+            log.warn(
+                'unable to fetch account trades, trying an alternate '
+                'method to find executed order {} / {}: {}'.format(
+                    order.id, order.asset.symbol, e
+                )
+            )
+            return self._get_executed_order_fallback(order)
+
+        transactions = []
+        trades = [t for t in all_trades if t['order'] == order.id]
+        if not trades:
+            log.debug(
+                'order {} / {} not found in trades'.format(
+                    order.id, order.asset.symbol
+                )
+            )
+            return transactions
+
+        trades.sort(key=lambda t: t['timestamp'], reverse=False)
+        order.filled = 0
+        order.commission = 0
+        for trade in trades:
+            # status property will update automatically
+            filled = trade['amount'] * order.direction
+            order.filled += filled
+
+            commission = 0
+            if 'fee' in trade and 'cost' in trade['fee']:
+                commission = trade['fee']['cost']
+                order.commission += commission
+
+            order.check_triggers(
+                price=trade['price'],
+                dt=pd.to_datetime(trade['timestamp'], unit='ms', utc=True),
+            )
+            transaction = Transaction(
+                asset=order.asset,
+                amount=filled,
+                dt=pd.Timestamp.utcnow(),
+                price=trade['price'],
+                order_id=order.id,
+                commission=commission
+            )
+            transactions.append(transaction)
+
+        order.broker_order_id = ', '.join([t['id'] for t in trades])
+        return transactions
+
+    def get_order(self, order_id, asset_or_symbol=None, return_price=False):
         if asset_or_symbol is None:
             log.debug(
                 'order not found in memory, the request might fail '
@@ -777,6 +876,12 @@ class CCXT(Exchange):
             order_status = self.api.fetch_order(id=order_id, symbol=symbol)
             order, executed_price = self._create_order(order_status)
 
+            if return_price:
+                return order, executed_price
+
+            else:
+                return order
+
         except (ExchangeError, NetworkError) as e:
             log.warn(
                 'unable to fetch order {} / {}: {}'.format(
@@ -785,8 +890,6 @@ class CCXT(Exchange):
             )
             raise ExchangeRequestError(error=e)
 
-        return order, executed_price
-
     def cancel_order(self, order_param, asset_or_symbol=None):
         order_id = order_param.id \
             if isinstance(order_param, Order) else order_param
@@ -893,3 +996,22 @@ class CCXT(Exchange):
                 ))
 
         return result
+
+    def get_trades(self, asset, my_trades=True, start_dt=None, limit=None):
+        # TODO: is it possible to sort this? Limit is useless otherwise.
+        ccxt_symbol = self.get_symbol(asset)
+        try:
+            trades = self.api.fetch_my_trades(
+                symbol=ccxt_symbol,
+                since=start_dt,
+                limit=limit,
+            )
+        except (ExchangeError, NetworkError) as e:
+            log.warn(
+                'unable to fetch trades {} / {}: {}'.format(
+                    self.name, asset.symbol, e
+                )
+            )
+            raise ExchangeRequestError(error=e)
+
+        return trades

catalyst/exchange/exchange.py

@@ -899,6 +899,22 @@ class Exchange:
         """
         pass
 
+    @abstractmethod
+    def process_order(self, order):
+        """
+        Similar to get_order but looks only for executed orders.
+
+        Parameters
+        ----------
+        order: Order
+
+        Returns
+        -------
+        float
+            Avg execution price
+
+        """
+
     @abstractmethod
     def cancel_order(self, order_param, symbol_or_asset=None):
         """Cancel an open order.
@@ -979,7 +995,7 @@ class Exchange:
     @abc.abstractmethod
     def get_orderbook(self, asset, order_type, limit):
         """
-        Retrieve the the orderbook for the given trading pair.
+        Retrieve the orderbook for the given trading pair.
 
         Parameters
         ----------
@@ -993,3 +1009,20 @@ class Exchange:
         list[dict[str, float]
         """
         pass
+
+    @abc.abstractmethod
+    def get_trades(self, asset, my_trades, start_dt, limit):
+        """
+        Retrieve a list of trades.
+
+        Parameters
+        ----------
+        my_trades: bool
+            List only my trades.
+        start_dt
+        limit
+
+        Returns
+        -------
+
+        """

catalyst/exchange/exchange_blotter.py

@@ -1,4 +1,8 @@
+import numpy as np
 import pandas as pd
+from logbook import Logger
+from redo import retry
+
 from catalyst.assets._assets import TradingPair
 from catalyst.constants import LOG_LEVEL
 from catalyst.exchange.exchange_errors import ExchangeRequestError
@@ -8,8 +12,6 @@ from catalyst.finance.order import ORDER_STATUS
 from catalyst.finance.slippage import SlippageModel
 from catalyst.finance.transaction import create_transaction, Transaction
 from catalyst.utils.input_validation import expect_types
-from logbook import Logger
-from redo import retry
 
 log = Logger('exchange_blotter', level=LOG_LEVEL)
 
@@ -93,7 +95,6 @@ class TradingPairFixedSlippage(SlippageModel):
 
     def simulate(self, data, asset, orders_for_asset):
         self._volume_for_bar = 0
-
         price = data.current(asset, 'close')
 
         dt = data.current_dt
@@ -103,18 +104,20 @@ class TradingPairFixedSlippage(SlippageModel):
 
             order.check_triggers(price, dt)
             if not order.triggered:
-                log.debug('order has not reached the trigger at current '
-                          'price {}'.format(price))
+                log.info(
+                    'order has not reached the trigger at current '
+                    'price {}'.format(price)
+                )
                 continue
 
             execution_price, execution_volume = self.process_order(data, order)
+            if execution_price is not None:
+                transaction = create_transaction(
+                    order, dt, execution_price, execution_volume
+                )
 
-            transaction = create_transaction(
-                order, dt, execution_price, execution_volume
-            )
-
-            self._volume_for_bar += abs(transaction.amount)
-            yield order, transaction
+                self._volume_for_bar += abs(transaction.amount)
+                yield order, transaction
 
     def process_order(self, data, order):
         price = data.current(order.asset, 'close')
@@ -205,34 +208,24 @@ class ExchangeBlotter(Blotter):
             for order in self.open_orders[asset]:
                 log.debug('found open order: {}'.format(order.id))
 
-                new_order, executed_price = exchange.get_order(order.id, asset)
-                log.debug(
-                    'got updated order {} {}'.format(
-                        new_order, executed_price
+                transactions = exchange.process_order(order)
+                if transactions:
+                    avg_price = np.average(
+                        a=[t.price for t in transactions],
+                        weights=[t.amount for t in transactions],
                     )
-                )
-                order.status = new_order.status
-
-                if order.status == ORDER_STATUS.FILLED:
-                    order.commission = new_order.commission
-                    if order.amount != new_order.amount:
-                        log.warn(
-                            'executed order amount {} differs '
-                            'from original'.format(
-                                new_order.amount, order.amount
-                            )
+                    ostatus = 'filled' if order.open_amount == 0 else 'partial'
+                    log.info(
+                        '{} order {} / {}: {}, avg price: {}'.format(
+                            ostatus,
+                            order.id,
+                            asset.symbol,
+                            order.filled,
+                            avg_price,
                         )
-                        order.amount = new_order.amount
-
-                    transaction = Transaction(
-                        asset=order.asset,
-                        amount=order.amount,
-                        dt=pd.Timestamp.utcnow(),
-                        price=executed_price,
-                        order_id=order.id,
-                        commission=order.commission
                     )
-                    yield order, transaction
+                    for transaction in transactions:
+                        yield order, transaction
 
                 elif order.status == ORDER_STATUS.CANCELLED:
                     yield order, None

tests/exchange/test_ccxt.py

@@ -1,7 +1,7 @@
 import pandas as pd
 from logbook import Logger
 
-from base import BaseExchangeTestCase
+from .base import BaseExchangeTestCase
 from catalyst.exchange.ccxt.ccxt_exchange import CCXT
 from catalyst.exchange.exchange_execution import ExchangeLimitOrder
 from catalyst.exchange.utils.exchange_utils import get_exchange_auth
@@ -76,6 +76,22 @@ class TestCCXT(BaseExchangeTestCase):
         assert len(tickers) == 1
         pass
 
+    def test_my_trades(self):
+        asset = self.exchange.get_asset('eng_eth')
+
+        trades = self.exchange.get_trades(asset)
+        assert trades
+        pass
+
+    def test_get_executed_order(self):
+        log.info('retrieving executed order')
+        asset = self.exchange.get_asset('eng_eth')
+
+        order = self.exchange.get_order('165784', asset)
+        transactions = self.exchange.process_order(order)
+        assert transactions
+        pass
+
     def test_get_balances(self):
         log.info('testing wallet balances')
         # balances = self.exchange.get_balances()

tests/exchange/test_suites/test_suite_exchange.py

@@ -184,13 +184,13 @@ class TestSuiteExchange(WithLogger, ZiplineTestCase):
                 )
                 sleep(1)
 
-                open_order, _ = exchange.get_order(order.id, asset)
+                open_order = exchange.get_order(order.id, asset)
                 self.assertEqual(0, open_order.status)
 
                 exchange.cancel_order(open_order, asset)
                 sleep(1)
 
-                canceled_order, _ = exchange.get_order(open_order.id, asset)
+                canceled_order = exchange.get_order(open_order.id, asset)
                 warnings = [record for record in log_catcher.records if
                             record.level == WARNING]