From 36881b03e281024eb4f1c452c20467fb12b2782c Mon Sep 17 00:00:00 2001 From: fredfortier Date: Thu, 7 Sep 2017 23:54:11 -0400 Subject: [PATCH] Working on multi-exchange implementation (not fully tested) --- catalyst/exchange/algorithm_exchange.py | 134 ++++++++++++++++----- catalyst/exchange/asset_finder_exchange.py | 17 +-- catalyst/exchange/data_portal_exchange.py | 78 ++++++++++-- catalyst/exchange/exchange_errors.py | 7 ++ catalyst/utils/run_algo.py | 77 +++++++----- 5 files changed, 226 insertions(+), 87 deletions(-) diff --git a/catalyst/exchange/algorithm_exchange.py b/catalyst/exchange/algorithm_exchange.py index e7bced7e..5321f971 100644 --- a/catalyst/exchange/algorithm_exchange.py +++ b/catalyst/exchange/algorithm_exchange.py @@ -23,6 +23,7 @@ import numpy as np import logbook import pandas as pd +from catalyst.utils.preprocess import preprocess import catalyst.protocol as zp from catalyst.algorithm import TradingAlgorithm @@ -45,7 +46,7 @@ from catalyst.gens.tradesimulation import AlgorithmSimulator from catalyst.utils.api_support import ( api_method, disallowed_in_before_trading_start) -from catalyst.utils.input_validation import error_keywords +from catalyst.utils.input_validation import error_keywords, ensure_upper_case log = logbook.Logger("ExchangeTradingAlgorithm") @@ -57,7 +58,7 @@ class ExchangeAlgorithmExecutor(AlgorithmSimulator): class ExchangeTradingAlgorithm(TradingAlgorithm): def __init__(self, *args, **kwargs): - self.exchange = kwargs.pop('exchange', None) + self.exchanges = kwargs.pop('exchanges', None) self.algo_namespace = kwargs.pop('algo_namespace', None) self.live_graph = kwargs.pop('live_graph', None) @@ -83,6 +84,7 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): self.stats_minutes = 5 super(self.__class__, self).__init__(*args, **kwargs) + # TODO: fix precision before re-enabling # self._create_minute_writer() signal.signal(signal.SIGINT, self.signal_handler) @@ -97,6 +99,7 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): writer = BcolzMinuteBarWriter.open( root, self.sim_params.end_session) else: + # TODO: need to be able to write more precise numbers writer = BcolzMinuteBarWriter( rootdir=root, calendar=self.trading_calendar, @@ -163,13 +166,11 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): if self.live_graph: self._clock = LiveGraphClock( self.sim_params.sessions, - time_skew=self.exchange.time_skew, context=self ) else: self._clock = SimpleClock( self.sim_params.sessions, - time_skew=self.exchange.time_skew ) return self._clock @@ -202,27 +203,31 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): portfolio directly. :return: """ - return self.exchange.portfolio + # TODO: build cumulative portfolio + return self.perf_tracker.get_portfolio(False) def updated_account(self): - return self.exchange.account + return self.perf_tracker.get_account(False) def _synchronize_portfolio(self, attempt_index=0): try: - self.exchange.synchronize_portfolio() + for exchange_name in self.exchanges: + exchange = self.exchanges[exchange_name] - # Applying the updated last_sales_price to the positions - # in the performance tracker. This seems a bit redundant - # but it will make sense when we have multiple exchange portfolios - # feeding into the same performance tracker. - tracker = self.perf_tracker.todays_performance.position_tracker - for asset in self.exchange.portfolio.positions: - position = self.exchange.portfolio.positions[asset] - tracker.update_position( - asset=asset, - last_sale_date=position.last_sale_date, - last_sale_price=position.last_sale_price - ) + exchange.synchronize_portfolio() + + # Applying the updated last_sales_price to the positions + # in the performance tracker. This seems a bit redundant + # but it will make sense when we have multiple exchange portfolios + # feeding into the same performance tracker. + tracker = self.perf_tracker.todays_performance.position_tracker + for asset in exchange.portfolio.positions: + position = exchange.portfolio.positions[asset] + tracker.update_position( + asset=asset, + last_sale_date=position.last_sale_date, + last_sale_price=position.last_sale_price + ) except ExchangeRequestError as e: log.warn( 'update portfolio attempt {}: {}'.format(attempt_index, e) @@ -239,7 +244,14 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): def _check_open_orders(self, attempt_index=0): try: - return self.exchange.check_open_orders() + orders = list() + for exchange_name in self.exchanges: + exchange = self.exchanges[exchange_name] + exchange_orders = exchange.check_open_orders() + + orders += exchange_orders + + return orders except ExchangeRequestError as e: log.warn( 'check open orders attempt {}: {}'.format(attempt_index, e) @@ -429,11 +441,13 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): log.warn('unable to save minute perfs to disk: {}'.format(e)) try: - save_algo_object( - algo_name=self.algo_namespace, - key='portfolio_{}'.format(self.exchange.name), - obj=self.exchange.portfolio - ) + for exchange_name in self.exchanges: + exchange = self.exchanges[exchange_name] + save_algo_object( + algo_name=self.algo_namespace, + key='portfolio_{}'.format(exchange_name), + obj=exchange.portfolio + ) except Exception as e: log.warn('unable to save portfolio to disk: {}'.format(e)) @@ -445,9 +459,10 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): style=None, attempt_index=0): try: - return self.exchange.order(asset, amount, limit_price, - stop_price, - style) + exchange = self.exchanges[asset.exchange] + return exchange.order(asset, amount, limit_price, + stop_price, + style) except ExchangeRequestError as e: log.warn( 'order attempt {}: {}'.format(attempt_index, e) @@ -500,7 +515,18 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): def _get_open_orders(self, asset=None, attempt_index=0): try: - return self.exchange.get_open_orders(asset) + 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) + + return open_orders except ExchangeRequestError as e: log.warn( 'open orders attempt {}: {}'.format(attempt_index, e) @@ -522,12 +548,54 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): return self._get_open_orders(asset) @api_method - def get_order(self, order_id): - return self.exchange.get_order(order_id) + def get_order(self, order_id, exchange_name): + exchange = self.exchanges[exchange_name] + return exchange.get_order(order_id) @api_method - def cancel_order(self, order_param): + def cancel_order(self, order_param, exchange_name): + exchange = self.exchanges[exchange_name] + order_id = order_param if isinstance(order_param, zp.Order): order_id = order_param.id - self.exchange.cancel_order(order_id) + + exchange.cancel_order(order_id) + + @api_method + @preprocess(symbol_str=ensure_upper_case) + def symbol(self, symbol_str, exchange_name=None): + """Lookup an Equity by its ticker symbol. + + Parameters + ---------- + symbol_str : str + The ticker symbol for the equity to lookup. + + Returns + ------- + equity : Equity + The equity that held the ticker symbol on the current + symbol lookup date. + + Raises + ------ + SymbolNotFound + Raised when the symbols was not held on the current lookup date. + + See Also + -------- + :func:`catalyst.api.set_symbol_lookup_date` + """ + # If the user has not set the symbol lookup date, + # use the end_session as the date for sybmol->sid resolution. + _lookup_date = self._symbol_lookup_date \ + if self._symbol_lookup_date is not None \ + else self.sim_params.end_session + + exchange = self.exchanges[exchange_name] + return self.asset_finder.lookup_symbol( + symbol_str, + as_of_date=_lookup_date, + exchange=exchange + ) diff --git a/catalyst/exchange/asset_finder_exchange.py b/catalyst/exchange/asset_finder_exchange.py index 2239f8a8..b95d1101 100644 --- a/catalyst/exchange/asset_finder_exchange.py +++ b/catalyst/exchange/asset_finder_exchange.py @@ -4,8 +4,7 @@ log = Logger('AssetFinderExchange') class AssetFinderExchange(object): - def __init__(self, exchange): - self.exchange = exchange + def __init__(self): self._asset_cache = {} @property @@ -47,7 +46,8 @@ class AssetFinderExchange(object): log.info('fetching asset: {}'.format(sid)) return list() - def lookup_symbol(self, symbol, as_of_date, fuzzy=False): + def lookup_symbol(self, symbol, as_of_date, exchange, + fuzzy=False): """Lookup an asset by symbol. Parameters @@ -81,11 +81,12 @@ class AssetFinderExchange(object): there are multiple candidates for the given ``symbol`` on the ``as_of_date``. """ - log.debug('looking up symbol: {}'.format(symbol)) + log.debug('looking up symbol: {} {}'.format(symbol, exchange.name)) - if symbol in self._asset_cache: - return self._asset_cache[symbol] + key = ','.join([exchange.name, symbol]) + if key in self._asset_cache: + return self._asset_cache[key] else: - asset = self.exchange.get_asset(symbol) - self._asset_cache[symbol] = asset + asset = exchange.get_asset(symbol) + self._asset_cache[key] = asset return asset diff --git a/catalyst/exchange/data_portal_exchange.py b/catalyst/exchange/data_portal_exchange.py index 77a7cb76..b1d91f56 100644 --- a/catalyst/exchange/data_portal_exchange.py +++ b/catalyst/exchange/data_portal_exchange.py @@ -12,7 +12,8 @@ # limitations under the License. from time import sleep - +import pandas as pd +from catalyst.assets._assets import TradingPair from logbook import Logger from catalyst.data.data_portal import DataPortal @@ -25,8 +26,8 @@ log = Logger('DataPortalExchange') class DataPortalExchange(DataPortal): - def __init__(self, exchange, *args, **kwargs): - self.exchange = exchange + def __init__(self, exchanges, *args, **kwargs): + self.exchanges = exchanges # TODO: put somewhere accessible by each algo self.retry_get_history_window = 5 @@ -45,14 +46,43 @@ class DataPortalExchange(DataPortal): ffill=True, attempt_index=0): try: - return self.exchange.get_history_window( - assets, - end_dt, - bar_count, - frequency, - field, - data_frequency, - ffill) + exchange_assets = dict() + for asset in assets: + if asset.exchange not in exchange_assets: + exchange_assets[asset.exchange] = list() + + exchange_assets[asset.exchange].append(asset) + + if len(exchange_assets) > 1: + df_list = [] + for exchange_name in exchange_assets: + exchange = self.exchanges[exchange_name] + assets = exchange_assets[exchange_name] + + df = exchange.get_history_window( + assets, + end_dt, + bar_count, + frequency, + field, + data_frequency, + ffill) + + df_list.append(df) + + return pd.concat(df_list) + + else: + exchange = self.exchanges[exchange_assets.keys()[0]] + return exchange.get_history_window( + assets, + end_dt, + bar_count, + frequency, + field, + data_frequency, + ffill) + except ExchangeRequestError as e: log.warn( 'get history attempt {}: {}'.format(attempt_index, e) @@ -93,8 +123,30 @@ class DataPortalExchange(DataPortal): def _get_spot_value(self, assets, field, dt, data_frequency, attempt_index=0): try: - return self.exchange.get_spot_value(assets, field, dt, - data_frequency) + if isinstance(assets, TradingPair): + exchange = self.exchanges[assets.exchange] + return exchange.get_spot_value( + assets, field, dt, data_frequency) + + else: + exchange_assets = dict() + for asset in assets: + if asset.exchange not in exchange_assets: + exchange_assets[asset.exchange] = list() + + exchange_assets[asset.exchange].append(asset) + + spot_values = [] + for exchange_name in exchange_assets: + exchange = self.exchanges[exchange_name] + assets = exchange_assets[exchange_name] + exchange_spot_values = exchange.get_spot_value( + assets, field, dt, data_frequency) + + spot_values += exchange_spot_values + + return spot_values + except ExchangeRequestError as e: log.warn( 'get spot value attempt {}: {}'.format(attempt_index, e) diff --git a/catalyst/exchange/exchange_errors.py b/catalyst/exchange/exchange_errors.py index 0cfa35b3..276be89c 100644 --- a/catalyst/exchange/exchange_errors.py +++ b/catalyst/exchange/exchange_errors.py @@ -34,6 +34,13 @@ class ExchangeTransactionError(ZiplineError): ).strip() +class ExchangeNotFoundError(ZiplineError): + msg = ( + 'Exchange {exchange_name} not found. Please specify exchanges ' + 'supported by Catalyst and verify spelling for accuracy.' + ).strip() + + class ExchangeAuthNotFound(ZiplineError): msg = ( 'Please create an auth.json file containing the api token and key for ' diff --git a/catalyst/utils/run_algo.py b/catalyst/utils/run_algo.py index fc9f6354..793c8919 100644 --- a/catalyst/utils/run_algo.py +++ b/catalyst/utils/run_algo.py @@ -45,7 +45,7 @@ from catalyst.exchange.exchange_portfolio import ExchangePortfolio from catalyst.exchange.exchange_errors import ( ExchangeRequestError, ExchangeRequestErrorTooManyAttempts, - BaseCurrencyNotFoundError) + BaseCurrencyNotFoundError, ExchangeNotFoundError) from catalyst.exchange.exchange_utils import get_exchange_auth, \ get_algo_object from logbook import Logger @@ -150,37 +150,42 @@ def _run(handle_data, if live and exchange is not None: exchange_name = exchange + start = pd.Timestamp.utcnow() end = start + timedelta(minutes=1439) + exchange_list = [x.strip().lower() for x in exchange.split(',')] - portfolio = get_algo_object( - algo_name=algo_namespace, - key='portfolio_{}'.format(exchange_name), - environ=environ - ) - if portfolio is None: - portfolio = ExchangePortfolio( - start_date=pd.Timestamp.utcnow() + exchanges = dict() + for exchange_name in exchange_list: + + portfolio = get_algo_object( + algo_name=algo_namespace, + key='portfolio_{}'.format(exchange_name), + environ=environ ) - exchange_auth = get_exchange_auth(exchange_name) - if exchange_name == 'bitfinex': - exchange = Bitfinex( - key=exchange_auth['key'], - secret=exchange_auth['secret'], - base_currency=base_currency, - portfolio=portfolio - ) - elif exchange_name == 'bittrex': - exchange = Bittrex( - key=exchange_auth['key'], - secret=exchange_auth['secret'], - base_currency=base_currency, - portfolio=portfolio - ) - else: - raise NotImplementedError( - 'exchange not supported: %s' % exchange_name) + if portfolio is None: + portfolio = ExchangePortfolio( + start_date=pd.Timestamp.utcnow() + ) + + exchange_auth = get_exchange_auth(exchange_name) + if exchange_name == 'bitfinex': + exchanges[exchange_name] = Bitfinex( + key=exchange_auth['key'], + secret=exchange_auth['secret'], + base_currency=base_currency, + portfolio=portfolio + ) + elif exchange_name == 'bittrex': + exchanges[exchange_name] = Bittrex( + key=exchange_auth['key'], + secret=exchange_auth['secret'], + base_currency=base_currency, + portfolio=portfolio + ) + else: + raise ExchangeNotFoundError(exchange_name=exchange_name) open_calendar = get_calendar('OPEN') sim_params = create_simulation_parameters( @@ -197,23 +202,24 @@ def _run(handle_data, exchange_tz='UTC', asset_db_path=None ) - env.asset_finder = AssetFinderExchange(exchange) + env.asset_finder = AssetFinderExchange() data = DataPortalExchange( - exchange=exchange, + exchanges=exchanges, asset_finder=env.asset_finder, trading_calendar=open_calendar, first_trading_day=pd.to_datetime('today', utc=True) ) choose_loader = None - def fetch_capital_base(attempt_index=0): + def fetch_capital_base(exchange, attempt_index=0): """ Fetch the base currency amount required to bootstrap the algorithm against the exchange. The algorithm cannot continue without this value. + :param exchange: the targeted exchange :param attempt_index: :return capital_base: the amount of base currency available for trading @@ -240,10 +246,15 @@ def _run(handle_data, exchange=exchange_name ) + capital_base = 0 + for exchange_name in exchanges: + exchange = exchanges[exchange_name] + capital_base += fetch_capital_base(exchange) + sim_params = create_simulation_parameters( start=start, end=end, - capital_base=fetch_capital_base(), + capital_base=capital_base, emission_rate='minute', data_frequency='minute' ) @@ -339,9 +350,9 @@ def _run(handle_data, choose_loader = None TradingAlgorithmClass = ( - partial(ExchangeTradingAlgorithm, exchange=exchange, + partial(ExchangeTradingAlgorithm, exchanges=exchanges, algo_namespace=algo_namespace, live_graph=live_graph) - if live and exchange else TradingAlgorithm) + if live and exchanges else TradingAlgorithm) perf = TradingAlgorithmClass( namespace=namespace,