From bc65c10fc6269c218ed11ee28906026204cac195 Mon Sep 17 00:00:00 2001 From: fredfortier Date: Fri, 22 Sep 2017 23:17:38 -0400 Subject: [PATCH] Implemented and tested the history() method in backtest mode. --- .../buy_low_sell_high_neo_with_interface.py | 39 +++--- catalyst/exchange/data_portal_exchange.py | 115 ++++++++++++++---- catalyst/exchange/exchange_blotter.py | 59 ++++----- catalyst/exchange/exchange_errors.py | 14 +++ tests/exchange/test_data_portal.py | 25 +++- 5 files changed, 167 insertions(+), 85 deletions(-) diff --git a/catalyst/examples/buy_low_sell_high_neo_with_interface.py b/catalyst/examples/buy_low_sell_high_neo_with_interface.py index 4fc81028..0e215aba 100644 --- a/catalyst/examples/buy_low_sell_high_neo_with_interface.py +++ b/catalyst/examples/buy_low_sell_high_neo_with_interface.py @@ -20,7 +20,7 @@ def initialize(context): log.info('initializing algo') context.asset = symbol('neo_btc', 'bitfinex') - context.TARGET_POSITIONS = 50 + context.TARGET_POSITIONS = 50000 context.PROFIT_TARGET = 0.1 context.SLIPPAGE_ALLOWED = 0.02 @@ -33,25 +33,24 @@ def initialize(context): def _handle_data(context, data): - # prices = data.history( - # context.asset, - # fields='price', - # bar_count=20, - # frequency='30m' - # ) - # rsi = talib.RSI(prices.values, timeperiod=14)[-1] - # log.info('got rsi: {}'.format(rsi)) + prices = data.history( + context.asset, + fields='price', + bar_count=30, + frequency='30m' + ) + rsi = talib.RSI(prices.values, timeperiod=14)[-1] + log.info('got rsi: {}'.format(rsi)) # Buying more when RSI is low, this should lower our cost basis - # if rsi <= 30: - # buy_increment = 1 - # elif rsi <= 40: - # buy_increment = 0.5 - # elif rsi <= 70: - # buy_increment = 0.1 - # else: - # buy_increment = None - buy_increment = 0.1 + if rsi <= 30: + buy_increment = 1 + elif rsi <= 40: + buy_increment = 0.5 + elif rsi <= 70: + buy_increment = 0.1 + else: + buy_increment = None cash = context.portfolio.cash log.info('base currency available: {cash}'.format(cash=cash)) @@ -159,8 +158,8 @@ def analyze(context, stats): # live_graph=True # ) run_algorithm( - capital_base=10000, - start=pd.to_datetime('2017-09-10', utc=True), + capital_base=250, + start=pd.to_datetime('2017-09-08', utc=True), end=pd.to_datetime('2017-09-15', utc=True), data_frequency='minute', initialize=initialize, diff --git a/catalyst/exchange/data_portal_exchange.py b/catalyst/exchange/data_portal_exchange.py index e7c35a1b..3dac7678 100644 --- a/catalyst/exchange/data_portal_exchange.py +++ b/catalyst/exchange/data_portal_exchange.py @@ -28,7 +28,8 @@ from catalyst.data.us_equity_pricing import BcolzDailyBarReader from catalyst.exchange.exchange_errors import ( ExchangeRequestError, ExchangeBarDataError, - BundleNotFoundError) + BundleNotFoundError, PricingDataBeforeTradingError, + PricingDataNotLoadedError) from catalyst.utils.paths import data_path log = Logger('DataPortalExchange') @@ -85,7 +86,8 @@ class DataPortalExchangeBase(DataPortal): else: exchange = self.exchanges[exchange_assets.keys()[0]] - return exchange.get_history_window( + return self.get_exchange_history_window( + exchange, assets, end_dt, bar_count, @@ -123,6 +125,10 @@ class DataPortalExchangeBase(DataPortal): field, data_frequency=None, ffill=True): + + if field == 'price': + field = 'close' + return self._get_history_window(assets, end_dt, bar_count, @@ -251,12 +257,14 @@ class DataPortalExchangeLive(DataPortalExchangeBase): class DataPortalExchangeBacktest(DataPortalExchangeBase): def __init__(self, *args, **kwargs): - super(DataPortalExchangeBacktest, self).__init__(*args, **kwargs) self.daily_bar_readers = dict() self.minute_bar_readers = dict() self.five_minute_bar_readers = dict() + + self.history_loaders = dict() + self.minute_history_loaders = dict() for exchange_name in self.exchanges: name = 'exchange_{}'.format(exchange_name) time_folder = \ @@ -289,6 +297,13 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): except IOError: self.minute_bar_readers[exchange_name] = None + def _get_first_trading_day(self, assets): + first_date = None + for asset in assets: + if first_date is None or asset.start_date > first_date: + first_date = asset.start_date + return first_date + @staticmethod def find_most_recent_time(bundle_name): try: @@ -320,30 +335,76 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): field, data_frequency, ffill=True): - # TODO: implement in the bundle - df = exchange.get_history_window( - assets, - end_dt, - bar_count, - frequency, - field, - data_frequency, - ffill) + if data_frequency == 'minute' or data_frequency == '1m': + reader = self.minute_bar_readers[exchange.name] + dts = self.trading_calendar.minutes_window( + end_dt, -bar_count + ) + + self.ensure_after_first_day(dts[0], assets) + + elif data_frequency == '5-minute' or data_frequency == '5m': + reader = self.five_minute_bar_readers[exchange.name] + elif data_frequency == 'daily' or data_frequency == '1d': + reader = self.daily_bar_readers[exchange.name] + + session = self.trading_calendar.minute_to_session_label(end_dt) + dts = self._get_days_for_window(session, bar_count) + + self.ensure_after_first_day(dts[0], assets) + else: + raise ValueError('Unsupported frequency') + + try: + values = reader.load_raw_arrays( + [field], + dts[0], + dts[-1], + assets, + )[0] + except Exception: + raise PricingDataNotLoadedError( + field=field, + first_trading_day=self._get_first_trading_day(assets), + exchange=exchange.name, + symbols=[asset.symbol for asset in assets], + ) + + series = dict() + for index, asset in enumerate(assets): + asset_values = [] + for value in values: + asset_values.append(value[index]) + + value_series = pd.Series(asset_values, index=dts) + series[asset] = value_series + + df = pd.DataFrame(series) return df + def ensure_after_first_day(self, dt, assets): + first_trading_day = self._get_first_trading_day(assets) + if dt < first_trading_day: + raise PricingDataBeforeTradingError( + first_trading_day=first_trading_day, + exchange=assets[0].exchange, + symbols=[asset.symbol for asset in assets], + ) + def get_exchange_spot_value(self, exchange, assets, field, dt, data_frequency): - - if data_frequency == 'minute': + if data_frequency == 'minute' or data_frequency == '1m': reader = self.minute_bar_readers[exchange.name] - elif data_frequency == '5-minute': + elif data_frequency == '5-minute' or data_frequency == '5m': reader = self.five_minute_bar_readers[exchange.name] - elif data_frequency == 'daily': + elif data_frequency == 'daily' or data_frequency == '1d': reader = self.daily_bar_readers[exchange.name] else: raise ValueError('Unsupported frequency') if isinstance(assets, TradingPair): + self.ensure_after_first_day(dt, [assets]) + try: value = reader.get_value( sid=assets.sid, @@ -351,10 +412,16 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): field=field ) return value - except Exception as e: - log.warn('minute data not found: {}'.format(e)) - return None + except Exception: + raise PricingDataNotLoadedError( + field=field, + first_trading_day=self._get_first_trading_day([assets]), + exchange=exchange.name, + symbols=assets.symbol, + ) else: + self.ensure_after_first_day(dt, assets) + values = [] for asset in assets: try: @@ -364,8 +431,12 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): field=field ) values.append(value) - except Exception as e: - log.warn('minute data not found: {}'.format(e)) - values.append(None) + except Exception: + raise PricingDataNotLoadedError( + field=field, + first_trading_day=self._get_first_trading_day(assets), + exchange=exchange.name, + symbols=[asset.symbol for asset in assets], + ) return values diff --git a/catalyst/exchange/exchange_blotter.py b/catalyst/exchange/exchange_blotter.py index 58994030..c4d451a1 100644 --- a/catalyst/exchange/exchange_blotter.py +++ b/catalyst/exchange/exchange_blotter.py @@ -1,18 +1,16 @@ +from catalyst.assets._assets import TradingPair from logbook import Logger from catalyst.finance.blotter import Blotter -from catalyst.finance.commission import PerShare, CommissionModel -from catalyst.finance.slippage import VolumeShareSlippage, SlippageModel, \ - LiquidityExceeded -from catalyst.assets._assets import TradingPair - -# It seems like we need to accept greate slippage risk in cryptos -# Orders won't often close at Equity levels. -# TODO: consider adjusting dynamically based on trading pair +from catalyst.finance.commission import CommissionModel +from catalyst.finance.slippage import SlippageModel from catalyst.finance.transaction import Transaction log = Logger('exchange_blotter') +# It seems like we need to accept greater slippage risk in cryptos +# Orders won't often close at Equity levels. +# TODO: consider adjusting dynamically based on trading pair DEFAULT_SLIPPAGE_SPREAD = 0.02 DEFAULT_MAKER_FEE = 0.001 DEFAULT_TAKER_FEE = 0.002 @@ -37,8 +35,7 @@ class TradingPairFeeSchedule(CommissionModel): def __repr__(self): return ( '{class_name}(maker_fee={maker_fee}, ' - 'taker_fee={taker_fee})' - .format( + 'taker_fee={taker_fee})'.format( class_name=self.__class__.__name__, maker_fee=self.maker_fee, taker_fee=self.taker_fee, @@ -83,44 +80,32 @@ class TradingPairFixedSlippage(SlippageModel): def simulate(self, data, asset, orders_for_asset): self._volume_for_bar = 0 - volume = data.current(asset, "volume") - if volume == 0: - return + price = data.current(asset, 'close') - # can use the close price, since we verified there's volume in this - # bar. - price = data.current(asset, "close") dt = data.current_dt - for order in orders_for_asset: if order.open_amount == 0: continue order.check_triggers(price, dt) if not order.triggered: + log.debug('order has not reached the trigger at current ' + 'price {}'.format(price)) continue - transaction = None - try: - execution_price, execution_volume = \ - self.process_order(data, order) + execution_price, execution_volume = self.process_order(data, order) - if execution_price is not None: - transaction = Transaction( - asset=order.asset, - amount=abs(execution_volume), - dt=data.current_dt, - price=execution_price, - order_id=order.id - ) + transaction = Transaction( + asset=order.asset, + amount=abs(execution_volume), + dt=dt, + price=execution_price, + order_id=order.id + ) - except LiquidityExceeded: - break - - if transaction: - 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') @@ -130,11 +115,11 @@ class TradingPairFixedSlippage(SlippageModel): adj_price = price * (1 + self.spread) else: # Sell order - adj_price = price & (1 - self.spread) + adj_price = price * (1 - self.spread) log.debug('added slippage to price: {} => {}'.format(price, adj_price)) - return (adj_price, order.amount) + return adj_price, order.amount class ExchangeBlotter(Blotter): diff --git a/catalyst/exchange/exchange_errors.py b/catalyst/exchange/exchange_errors.py index d823cd87..a6bfcbab 100644 --- a/catalyst/exchange/exchange_errors.py +++ b/catalyst/exchange/exchange_errors.py @@ -147,3 +147,17 @@ class BundleNotFoundError(ZiplineError): 'Please ingest data using the command ' '`catalyst ingest -b exchange_{exchange}`. ' 'See catalyst documentation for details.').strip() + + +class PricingDataBeforeTradingError(ZiplineError): + msg = ('Pricing data for trading pairs {symbols} on exchange {exchange} ' + 'starts on {first_trading_day}.').strip() + + +class PricingDataNotLoadedError(ZiplineError): + msg = ('Pricing data {field} for trading pairs {symbols} trading on ' + 'exchange {exchange} since {first_trading_day} is unavailable. ' + 'The bundle data is either out-of-date or has not been loaded yet.' + 'Please ingest data using the command ' + '`catalyst ingest -b exchange_{exchange}`. ' + 'See catalyst documentation for details.').strip() diff --git a/tests/exchange/test_data_portal.py b/tests/exchange/test_data_portal.py index 25d123cc..4e954acf 100644 --- a/tests/exchange/test_data_portal.py +++ b/tests/exchange/test_data_portal.py @@ -50,11 +50,10 @@ class ExchangeDataPortalTestCase: exchanges=dict(bitfinex=self.bitfinex), asset_finder=asset_finder, trading_calendar=open_calendar, - first_trading_day=pd.to_datetime('2017-09-10', utc=True) + first_trading_day=None # will set dynamically based on assets ) def test_get_history_window_live(self): - asset_finder = self.data_portal_live.asset_finder assets = [ @@ -82,8 +81,24 @@ class ExchangeDataPortalTestCase: assets, 'price', now, '1m') pass - def test_get_spot_value_backtest(self): + def test_get_history_window_backtest(self): + asset_finder = self.data_portal_live.asset_finder + assets = [ + asset_finder.lookup_symbol('neo_btc', self.bitfinex), + ] + + date = pd.to_datetime('2017-09-10', utc=True) + data = self.data_portal_backtest.get_history_window( + assets, + date, + 10, + '1m', + 'close', + 'minute') + pass + + def test_get_spot_value_backtest(self): asset_finder = self.data_portal_backtest.asset_finder assets = [ @@ -92,8 +107,6 @@ class ExchangeDataPortalTestCase: date = pd.to_datetime('2017-09-10', utc=True) value = self.data_portal_backtest.get_spot_value( - assets, 'close', date, 'minute') + assets[0], 'close', date, 'minute') pass - def test_get_history_window_backtest(self): - pass