diff --git a/catalyst/examples/buy_and_hold_live.py b/catalyst/examples/buy_the_dip_live.py similarity index 67% rename from catalyst/examples/buy_and_hold_live.py rename to catalyst/examples/buy_the_dip_live.py index 430abc18..21623f40 100644 --- a/catalyst/examples/buy_and_hold_live.py +++ b/catalyst/examples/buy_the_dip_live.py @@ -7,8 +7,10 @@ from catalyst.api import ( record, get_open_orders, ) +from catalyst.exchange.exchange_errors import ExchangeRequestError -log = Logger('buy_and_hold_live') +algo_namespace = 'buy_the_dip_live' +log = Logger(algo_namespace) def initialize(context): @@ -18,9 +20,28 @@ def initialize(context): context.TARGET_POSITIONS = 100 context.BUY_INCREMENT = 1 + context.retry_check_open_orders = 2 + context.retry_update_portfolio = 2 + context.retry_order = 2 + def handle_data(context, data): log.info('handling bar {data}'.format(data=data)) + # ohlc = data.history(context.asset, + # fields='price', + # bar_count=20, + # frequency='1d' + # ) + # ohlc = data.history([context.asset, symbol('iot_usd')], + # fields='price', + # bar_count=20, + # frequency='1d' + # ) + ohlc = data.history([context.asset, symbol('iot_usd')], + fields=['price', 'volume'], + bar_count=20, + frequency='1d' + ) cash = context.portfolio.cash log.info('base currency available: {cash}'.format(cash=cash)) @@ -82,10 +103,13 @@ def handle_data(context, data): leverage=context.account.leverage, ) - context.perf_tracker.update_performance() - log.info('the performance:\n{}'.format( - context.perf_tracker.to_dict('minute') - )) + try: + context.perf_tracker.update_performance() + log.info('the performance:\n{}'.format( + context.perf_tracker.to_dict('minute') + )) + except Exception as e: + log.warn('unable to calculate performance: {}'.format(e)) pass @@ -101,5 +125,5 @@ run_algorithm( capital_base=100000, exchange_conn=exchange_conn, live=True, - algo_namespace='buy_and_hold_live' + algo_namespace=algo_namespace ) diff --git a/catalyst/exchange/algorithm_exchange.py b/catalyst/exchange/algorithm_exchange.py index 23bb109b..2307e55d 100644 --- a/catalyst/exchange/algorithm_exchange.py +++ b/catalyst/exchange/algorithm_exchange.py @@ -24,6 +24,9 @@ from catalyst.utils.api_support import ( disallowed_in_before_trading_start) from catalyst.utils.calendars.trading_calendar import days_at_time +from catalyst.exchange.exchange_errors import ( + ExchangeRequestError +) log = logbook.Logger("ExchangeTradingAlgorithm") @@ -38,6 +41,11 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): self.exchange = kwargs.pop('exchange', None) self.orders = {} + self.retry_check_open_orders = 5 + self.retry_update_portfolio = 5 + self.retry_order = 1 + self.retry_delay = 5 + super(self.__class__, self).__init__(*args, **kwargs) log.info('exchange trading algorithm successfully initialized') @@ -113,9 +121,32 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): def updated_account(self): return self.exchange.account + def _update_portfolio(self, attempt_index=0): + try: + self.exchange.update_portfolio() + except ExchangeRequestError as e: + log.warn( + 'update portfolio attempt {}: {}'.format(attempt_index, e) + ) + if attempt_index < self.retry_update_portfolio: + time.sleep(self.retry_delay) + self._update_portfolio(attempt_index + 1) + + def _check_open_orders(self, attempt_index=0): + try: + return self.exchange.check_open_orders() + except ExchangeRequestError as e: + log.warn( + 'check open orders attempt {}: {}'.format(attempt_index, e) + ) + if attempt_index < self.retry_check_open_orders: + time.sleep(self.retry_delay) + return self._check_open_orders(attempt_index + 1) + def handle_data(self, data): - self.exchange.update_portfolio() - transactions = self.exchange.check_open_orders() + self._update_portfolio() + + transactions = self._check_open_orders() for transaction in transactions: self.perf_tracker.process_transaction(transaction) @@ -127,6 +158,27 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): # every bar no matter if the algorithm places an order or not. self.validate_account_controls() + def _order(self, + asset, + amount, + limit_price=None, + stop_price=None, + style=None, + attempt_index=0): + try: + return self.exchange.order(asset, amount, limit_price, + stop_price, + style) + except ExchangeRequestError as e: + log.warn( + 'order attempt {}: {}'.format(attempt_index, e) + ) + if attempt_index < self.retry_order: + time.sleep(self.retry_delay) + return self._order( + asset, amount, limit_price, stop_price, style, + attempt_index + 1) + @api_method @disallowed_in_before_trading_start(OrderInBeforeTradingStart()) def order(self, @@ -136,11 +188,12 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): stop_price=None, style=None): amount, style = self._calculate_order(asset, amount, - limit_price, stop_price, style) + limit_price, stop_price, + style) - order_id = self.exchange.order(asset, amount, limit_price, stop_price, - style) + order_id = self._order(asset, amount, limit_price, stop_price, style) order = self.portfolio.open_orders[order_id] + self.perf_tracker.process_order(order) return order diff --git a/catalyst/exchange/bitfinex.py b/catalyst/exchange/bitfinex.py index e321a799..2c06b64c 100644 --- a/catalyst/exchange/bitfinex.py +++ b/catalyst/exchange/bitfinex.py @@ -7,19 +7,26 @@ import json import time import requests import pandas as pd -import collections +from datetime import timedelta, datetime from catalyst.protocol import Portfolio, Account # from websocket import create_connection from catalyst.exchange.exchange import Exchange from logbook import Logger +from catalyst.assets._assets import Asset from catalyst.finance.order import ORDER_STATUS from catalyst.exchange.exchange_order import ExchangeOrder from catalyst.finance.execution import (MarketOrder, LimitOrder, StopOrder, StopLimitOrder) -from catalyst.data.data_portal import BASE_FIELDS from catalyst.exchange.exchange_portfolio import ExchangePortfolio +from catalyst.errors import ( + IncompatibleHistoryFrequency, +) +from catalyst.exchange.exchange_errors import ( + ExchangeRequestError, + InvalidHistoryFrequencyError +) BITFINEX_URL = 'https://api.bitfinex.com' ASSETS = '{ "USDT_BTC": {"symbol":"btc_usd", "start_date": "2010-01-01"}, "ltcusd": {"symbol":"ltc_usd", "start_date": "2010-01-01"}, "ltcbtc": {"symbol":"ltc_btc", "start_date": "2010-01-01"}, "ethusd": {"symbol":"eth_usd", "start_date": "2010-01-01"}, "ethbtc": {"symbol":"eth_btc", "start_date": "2010-01-01"}, "etcbtc": {"symbol":"etc_btc", "start_date": "2010-01-01"}, "etcusd": {"symbol":"etc_usd", "start_date": "2010-01-01"}, "rrtusd": {"symbol":"rrt_usd", "start_date": "2010-01-01"}, "rrtbtc": {"symbol":"rrt_btc", "start_date": "2010-01-01"}, "zecusd": {"symbol":"zec_usd", "start_date": "2010-01-01"}, "zecbtc": {"symbol":"zec_btc", "start_date": "2010-01-01"}, "xmrusd": {"symbol":"xmr_usd", "start_date": "2010-01-01"}, "xmrbtc": {"symbol":"xmr_btc", "start_date": "2010-01-01"}, "dshusd": {"symbol":"dsh_usd", "start_date": "2010-01-01"}, "dshbtc": {"symbol":"dsh_btc", "start_date": "2010-01-01"}, "bccbtc": {"symbol":"bcc_btc", "start_date": "2010-01-01"}, "bcubtc": {"symbol":"bcu_btc", "start_date": "2010-01-01"}, "bccusd": {"symbol":"bcc_usd", "start_date": "2010-01-01"}, "bcuusd": {"symbol":"bcu_usd", "start_date": "2010-01-01"}, "xrpusd": {"symbol":"xrp_usd", "start_date": "2010-01-01"}, "xrpbtc": {"symbol":"xrp_btc", "start_date": "2010-01-01"}, "iotusd": {"symbol":"iot_usd", "start_date": "2010-01-01"}, "iotbtc": {"symbol":"iot_btc", "start_date": "2010-01-01"}, "ioteth": {"symbol":"iot_eth", "start_date": "2010-01-01"}, "eosusd": {"symbol":"eos_usd", "start_date": "2010-01-01"}, "eosbtc": {"symbol":"eos_btc", "start_date": "2010-01-01"}, "eoseth": {"symbol":"eos_eth", "start_date": "2010-01-01"} }' @@ -175,8 +182,8 @@ class Bitfinex(Exchange): response = self._request('balances', None) balances = response.json() if 'message' in balances: - raise ValueError( - 'unable to fetch balance %s' % balances['message'] + raise ExchangeRequestError( + error='unable to fetch balance {}'.format(balances['message']) ) base_position = None @@ -187,7 +194,7 @@ class Bitfinex(Exchange): if position is None: raise ValueError( - 'Base currency %s not found in portfolio' % self.base_currency + error='Base currency %s not found in portfolio' % self.base_currency ) portfolio = self.store.portfolio @@ -265,99 +272,81 @@ class Bitfinex(Exchange): def subscribe_to_market_data(self, symbol): pass - def get_spot_value(self, assets, field, dt=None, data_frequency='minute'): - """ - Public API method that returns a scalar value representing the value - of the desired asset's field at either the given dt. + def get_candles(self, data_frequency, assets, + end_dt=None, bar_count=None, limit=None): - Parameters - ---------- - assets : Asset, ContinuousFuture, or iterable of same. - The asset or assets whose data is desired. - field : {'open', 'high', 'low', 'close', 'volume', - 'price', 'last_traded'} - The desired field of the asset. - dt : pd.Timestamp - The timestamp for the desired value. - data_frequency : str - The frequency of the data to query; i.e. whether the data is - 'daily' or 'minute' bars - - Returns - ------- - value : float, int, or pd.Timestamp - The spot value of ``field`` for ``asset`` The return type is based - on the ``field`` requested. If the field is one of 'open', 'high', - 'low', 'close', or 'price', the value will be a float. If the - ``field`` is 'volume' the value will be a int. If the ``field`` is - 'last_traded' the value will be a Timestamp. - - Bitfinex timeframes - ------------------- - Available values: '1m', '5m', '15m', '30m', '1h', '3h', '6h', '12h', - '1D', '7D', '14D', '1M' - """ - if field not in BASE_FIELDS: - raise KeyError('Invalid column: ' + str(field)) - - if isinstance(assets, collections.Iterable): - values = list() - for asset in assets: - value = self.get_single_spot_value( - asset, field, data_frequency) - values.append(value) - - return values - else: - return self.get_single_spot_value( - assets, field, data_frequency) - - def get_single_spot_value(self, asset, field, data_frequency): - symbol = self._get_v2_symbol(asset) - log.debug( - 'fetching spot value {field} for symbol {symbol}'.format( - symbol=symbol, - field=field - ) - ) - - if data_frequency == 'minute': + # TODO: support all available frequencies + start_dt = None + if data_frequency == 'minute' or data_frequency == '1m': frequency = '1m' - elif data_frequency == 'daily': + if bar_count and end_dt: + start_dt = end_dt - timedelta(minutes=bar_count) + elif data_frequency == 'daily' or data_frequency == '1d': frequency = '1D' + if bar_count and end_dt: + start_dt = end_dt - timedelta(days=bar_count) else: - raise NotImplementedError( - 'Unsupported frequency %s' % data_frequency + raise InvalidHistoryFrequencyError( + frequency=data_frequency ) - response = requests.get( - '{url}/v2/candles/trade:{frequency}:{symbol}/last'.format( + # Making sure that assets are iterable + asset_list = [assets] if isinstance(assets, Asset) else assets + ohlc_list = dict() + for asset in asset_list: + symbol = self._get_v2_symbol(asset) + url = '{url}/v2/candles/trade:{frequency}:{symbol}'.format( url=self.url, frequency=frequency, symbol=symbol ) - ) - candles = response.json() - if 'message' in candles: - raise ValueError( - 'Unable to retrieve candles: %s' % candles['message'] - ) + if start_dt and end_dt: + is_list = True + url += '/hist?start={start}&end={end}'.format( + start=time.mktime(start_dt.timetuple()) * 1000, + end=time.mktime(end_dt.timetuple()) * 1000, + ) + else: + is_list = False + url += '/last' - ohlc = dict( - open=candles[1], - high=candles[3], - low=candles[4], - close=candles[2], - volume=candles[5], - price=candles[2], - last_traded=pd.Timestamp.utcfromtimestamp(candles[0] / 1000.0), - ) + response = requests.get(url) + candles = response.json() - if field not in ohlc: - raise KeyError('Invalid column: %s' % field) + if 'message' in candles: + raise ExchangeRequestError( + error='Unable to retrieve candles: {}'.format( + candles['message']) + ) + + def ohlc_from_candle(candle): + return dict( + open=candle[1], + high=candle[3], + low=candle[4], + close=candle[2], + volume=candle[5], + price=candle[2], + last_traded=pd.Timestamp.utcfromtimestamp( + candle[0] / 1000.0), + ) + + if is_list: + ohlc_bars = [] + for candle in candles: + ohlc = ohlc_from_candle(candle) + ohlc_bars.append(ohlc) + + ohlc_list[asset] = ohlc_bars + + else: + ohlc = ohlc_from_candle(candles) + ohlc_list[asset] = ohlc + + return ohlc_list[assets] \ + if isinstance(assets, Asset) else ohlc_list - return ohlc[field] def order(self, asset, amount, limit_price, stop_price, style): """Place an order. @@ -460,9 +449,9 @@ class Bitfinex(Exchange): response = self._request('order/new', req) exchange_order = response.json() if 'message' in exchange_order: - raise ValueError( - 'unable to create Bitfinex order %s' % exchange_order[ - 'message'] + raise ExchangeRequestError( + error='unable to create Bitfinex order {}'.format( + exchange_order['message']) ) order_id = exchange_order['id'] @@ -501,9 +490,9 @@ class Bitfinex(Exchange): response = self._request('orders', None) order_statuses = response.json() if 'message' in order_statuses: - raise ValueError( - 'Unable to retrieve open orders: %s' % order_statuses[ - 'message'] + raise ExchangeRequestError( + error='Unable to retrieve open orders: {}'.format( + order_statuses['message']) ) orders = list() @@ -533,8 +522,9 @@ class Bitfinex(Exchange): order_status = response.json() if 'message' in order_status: - raise ValueError( - 'Unable to retrieve order status: %s' % order_status['message'] + raise ExchangeRequestError( + error='Unable to retrieve order status: {}'.format( + order_status['message']) ) return self._create_order(order_status) @@ -552,8 +542,9 @@ class Bitfinex(Exchange): response = self._request('order/cancel', {'order_id': order_id}) status = response.json() if 'message' in status: - raise ValueError( - 'Unable to cancel order: %s %s' % (order_id, status['message']) + raise ExchangeRequestError( + error='Unable to cancel order: {} {}'.format( + order_id, status['message']) ) def tickers(self, assets): @@ -576,8 +567,9 @@ class Bitfinex(Exchange): tickers = request.json() if 'message' in tickers: - raise ValueError( - 'Unable to retrieve tickers: %s' % tickers['message'] + raise ExchangeRequestError( + error='Unable to retrieve tickers: {}'.format( + tickers['message']) ) formatted_tickers = [] diff --git a/catalyst/exchange/data_portal_exchange.py b/catalyst/exchange/data_portal_exchange.py index b61a234f..74efc073 100644 --- a/catalyst/exchange/data_portal_exchange.py +++ b/catalyst/exchange/data_portal_exchange.py @@ -33,7 +33,7 @@ class DataPortalExchange(DataPortal): field, data_frequency, ffill=True): - history_window = super(self.__class__, self).get_history_window( + return self.exchange.get_history_window( assets, end_dt, bar_count, @@ -42,12 +42,6 @@ class DataPortalExchange(DataPortal): data_frequency, ffill) - # The returned dataframe contains today's value as a NaN because - # end_dt points to the current wall clock. We drop today's - # value to be in sync with the simulation's behavior. - today = pd.Timestamp.utcnow() - return history_window[history_window.index.date != today] - def get_spot_value(self, assets, field, dt, data_frequency): return self.exchange.get_spot_value(assets, field, dt, data_frequency) @@ -55,4 +49,5 @@ class DataPortalExchange(DataPortal): perspective_dt, data_frequency, spot_value=None): + # TODO: does this pertain to cryptocurrencies? raise NotImplementedError("get_adjusted_value is not implemented yet!") diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index f0844db2..15057f7a 100644 --- a/catalyst/exchange/exchange.py +++ b/catalyst/exchange/exchange.py @@ -1,11 +1,13 @@ import abc import json from abc import ABCMeta, abstractmethod, abstractproperty +import collections import pandas as pd from catalyst.assets._assets import Asset from catalyst.finance.order import ORDER_STATUS from catalyst.finance.transaction import Transaction +from catalyst.data.data_portal import BASE_FIELDS from catalyst.errors import ( MultipleSymbolsFound, @@ -265,6 +267,141 @@ class Exchange: """ pass + @abstractmethod + def get_candles(self, data_frequency, assets, + end_dt=None, bar_count=None, limit=None): + """ + Retrieve OHLC candles + """ + pass + + def get_spot_value(self, assets, field, dt=None, data_frequency='minute'): + """ + Public API method that returns a scalar value representing the value + of the desired asset's field at either the given dt. + + Parameters + ---------- + assets : Asset, ContinuousFuture, or iterable of same. + The asset or assets whose data is desired. + field : {'open', 'high', 'low', 'close', 'volume', + 'price', 'last_traded'} + The desired field of the asset. + dt : pd.Timestamp + The timestamp for the desired value. + data_frequency : str + The frequency of the data to query; i.e. whether the data is + 'daily' or 'minute' bars + + Returns + ------- + value : float, int, or pd.Timestamp + The spot value of ``field`` for ``asset`` The return type is based + on the ``field`` requested. If the field is one of 'open', 'high', + 'low', 'close', or 'price', the value will be a float. If the + ``field`` is 'volume' the value will be a int. If the ``field`` is + 'last_traded' the value will be a Timestamp. + + Bitfinex timeframes + ------------------- + Available values: '1m', '5m', '15m', '30m', '1h', '3h', '6h', '12h', + '1D', '7D', '14D', '1M' + """ + if field not in BASE_FIELDS: + raise KeyError('Invalid column: ' + str(field)) + + if isinstance(assets, collections.Iterable): + values = list() + for asset in assets: + value = self.get_single_spot_value( + asset, field, data_frequency) + values.append(value) + + return values + else: + return self.get_single_spot_value( + assets, field, data_frequency) + + def get_single_spot_value(self, asset, field, data_frequency): + log.debug( + 'fetching spot value {field} for symbol {symbol}'.format( + symbol=asset.symbol, + field=field + ) + ) + + ohlc = self.get_candles(data_frequency, asset) + if field not in ohlc: + raise KeyError('Invalid column: %s' % field) + + return ohlc[field] + + def get_history_window(self, + assets, + end_dt, + bar_count, + frequency, + fields, + data_frequency, + ffill=True): + + """ + + :param assets: + :param end_dt: + :param bar_count: + :param frequency: + :param fields: + :param data_frequency: + :param ffill: + + :return df: + If a single security and a single field were passed into data.history, + a pandas Series is returned, indexed by date. + + If multiple securities and single field are passed in, the returned + pandas DataFrame is indexed by date, and has assets as columns. + + If a single security and multiple fields are passed in, the returned + pandas DataFrame is indexed by date, and has fields as columns. + + If multiple assets and multiple fields are passed in, the returned + pandas Panel is indexed by field, has date as the major axis, and + securities as the minor axis. + """ + + candles = self.get_candles( + data_frequency=frequency, + assets=assets, + bar_count=bar_count, + end_dt=end_dt + ) + + def get_single_field_series(candles): + return pd.Series( + map(lambda candle: candle[fields], candles), + index=map(lambda candle: candle['last_traded'], candles) + ) + + df = None + if len(assets) == 1: + if type(fields) is str: + asset = assets[0] + df = get_single_field_series(candles[asset]) + else: + raise NotImplementedError() + else: + if type(fields) is str: + series = [] + for asset in assets: + item = get_single_field_series(candles[asset]) + series.append(item) + df = pd.concat(series, axis=1) + else: + raise NotImplementedError() + + return df + @abc.abstractmethod def tickers(self, date, pairs): return diff --git a/catalyst/exchange/exchange_errors.py b/catalyst/exchange/exchange_errors.py new file mode 100644 index 00000000..cdb6734c --- /dev/null +++ b/catalyst/exchange/exchange_errors.py @@ -0,0 +1,13 @@ +from catalyst.errors import ZiplineError + + +class ExchangeRequestError(ZiplineError): + msg = ( + 'Request failed: {error}' + ).strip() + + +class InvalidHistoryFrequencyError(ZiplineError): + msg = ( + 'History frequency {frequency} not supported by the exchange.' + ).strip() diff --git a/catalyst/exchange/exchange_utils.py b/catalyst/exchange/exchange_utils.py new file mode 100644 index 00000000..c9ffee57 --- /dev/null +++ b/catalyst/exchange/exchange_utils.py @@ -0,0 +1,9 @@ +import os +from catalyst.utils.paths import data_root + + +def get_exchange_folder(environ=None): + if not environ: + environ = os.environ + + root = data_root(environ)