diff --git a/catalyst/exchange/ccxt/ccxt_exchange.py b/catalyst/exchange/ccxt/ccxt_exchange.py index bade7221..5ba2dc06 100644 --- a/catalyst/exchange/ccxt/ccxt_exchange.py +++ b/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 diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index 75f4ec3c..a0f3a4bf 100644 --- a/catalyst/exchange/exchange.py +++ b/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 + ------- + + """ diff --git a/catalyst/exchange/exchange_blotter.py b/catalyst/exchange/exchange_blotter.py index d638e4bd..2051ba43 100644 --- a/catalyst/exchange/exchange_blotter.py +++ b/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 diff --git a/tests/exchange/test_ccxt.py b/tests/exchange/test_ccxt.py index 7ef939b1..04112675 100644 --- a/tests/exchange/test_ccxt.py +++ b/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() diff --git a/tests/exchange/test_suites/test_suite_exchange.py b/tests/exchange/test_suites/test_suite_exchange.py index 3eb3b38b..29baf739 100644 --- a/tests/exchange/test_suites/test_suite_exchange.py +++ b/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]