From 946d24bd7a355f09e0371b716e41e6ce76295b63 Mon Sep 17 00:00:00 2001 From: fredfortier Date: Thu, 19 Oct 2017 23:23:37 -0400 Subject: [PATCH] Refactoring related to auto-ingestion --- catalyst/data/loader.py | 22 ++- catalyst/data/minute_bars.py | 12 +- catalyst/examples/buy_low_sell_high_live.py | 36 ++-- .../buy_low_sell_high_neo_with_interface.py | 2 - catalyst/exchange/bundle_utils.py | 148 +---------------- catalyst/exchange/data_portal_exchange.py | 153 ++++++----------- catalyst/exchange/exchange.py | 82 +--------- catalyst/exchange/exchange_bcolz.py | 4 +- catalyst/exchange/exchange_bundle.py | 154 +++++++++++++++++- .../utils/calendars/exchange_calendar_open.py | 2 +- catalyst/utils/run_algo.py | 12 +- tests/exchange/test_bundle.py | 10 +- 12 files changed, 279 insertions(+), 358 deletions(-) diff --git a/catalyst/data/loader.py b/catalyst/data/loader.py index 7211938e..ce293e9b 100644 --- a/catalyst/data/loader.py +++ b/catalyst/data/loader.py @@ -95,7 +95,8 @@ def has_data_for_dates(series_or_df, first_date, last_date): def load_crypto_market_data(trading_day=None, trading_days=None, bm_symbol=None, bundle=None, bundle_data=None, - environ=None, exchange=None): + environ=None, exchange=None, start_dt=None, + end_dt=None): if trading_day is None: trading_day = get_calendar('OPEN').trading_day @@ -104,8 +105,11 @@ def load_crypto_market_data(trading_day=None, trading_days=None, # if trading_days is None: # trading_days = get_calendar('OPEN').schedule - first_date = get_calendar('OPEN').first_trading_session - now = pd.Timestamp.utcnow() + if start_dt is None: + start_dt = get_calendar('OPEN').first_trading_session + + if end_dt is None: + end_dt = pd.Timestamp.utcnow() # We expect to have benchmark and treasury data that's current up until # **two** full trading days prior to the most recently completed trading @@ -131,7 +135,7 @@ def load_crypto_market_data(trading_day=None, trading_days=None, else: last_date = trading_days[trading_days.get_loc(now, method='ffill') - 2] ''' - last_date = trading_days[trading_days.get_loc(now, method='ffill') - 1] + last_date = trading_days[trading_days.get_loc(end_dt, method='ffill') - 1] if exchange is None: # This is exceptional, since placing the import at the module scope @@ -146,14 +150,14 @@ def load_crypto_market_data(trading_day=None, trading_days=None, br = exchange.get_history_window( assets=[benchmark_asset], end_dt=last_date, - bar_count=pd.Timedelta(last_date - first_date).days, + bar_count=pd.Timedelta(last_date - start_dt).days, frequency='1d', field='close', data_frequency='daily') br.columns = ['close'] br = br.pct_change(1).iloc[1:] - br.loc[first_date]=0 - br=br.sort_index() + br.loc[start_dt] = 0 + br = br.sort_index() # Override first_date for treasury data since we have it for many more years # and is independent of crypto data @@ -162,10 +166,10 @@ def load_crypto_market_data(trading_day=None, trading_days=None, bm_symbol, first_date_treasury, last_date, - now, + end_dt, environ, ) - benchmark_returns = br[br.index.slice_indexer(first_date, last_date)] + benchmark_returns = br[br.index.slice_indexer(start_dt, last_date)] treasury_curves = tc[ tc.index.slice_indexer(first_date_treasury, last_date)] return benchmark_returns, treasury_curves diff --git a/catalyst/data/minute_bars.py b/catalyst/data/minute_bars.py index bd4d5b39..83dafa1c 100644 --- a/catalyst/data/minute_bars.py +++ b/catalyst/data/minute_bars.py @@ -44,7 +44,6 @@ from catalyst.utils.calendars import get_calendar from catalyst.utils.cli import maybe_show_progress from catalyst.utils.memoize import lazyval - logger = logbook.Logger('MinuteBars') US_EQUITIES_MINUTES_PER_DAY = 390 @@ -1125,7 +1124,7 @@ class BcolzMinuteBarReader(MinuteBarReader): else: return np.nan - #if field != 'volume': + # if field != 'volume': value *= self._ohlc_ratio_inverse_for_sid(sid) return value @@ -1206,7 +1205,7 @@ class BcolzMinuteBarReader(MinuteBarReader): minute_dt.value / NANOS_IN_MINUTE, self._minutes_per_day, False, - ) + ) def load_raw_arrays(self, fields, start_dt, end_dt, sids): """ @@ -1262,10 +1261,10 @@ class BcolzMinuteBarReader(MinuteBarReader): where = values != 0 # first slice down to len(where) because we might not have # written data for all the minutes requested - #if field != 'volume': + # if field != 'volume': out[:len(where), i][where] = ( values[where] * self._ohlc_ratio_inverse_for_sid(sid)) - #else: + # else: # out[:len(where), i][where] = values[where] results.append(out) @@ -1353,9 +1352,10 @@ class H5MinuteBarUpdateReader(MinuteBarUpdateReader): path : str The path of the HDF5 file from which to source data. """ + def __init__(self, path): self._panel = pd.read_hdf(path) def read(self, dts, sids): panel = self._panel[sids, dts, :] - return panel.iteritems() \ No newline at end of file + return panel.iteritems() diff --git a/catalyst/examples/buy_low_sell_high_live.py b/catalyst/examples/buy_low_sell_high_live.py index a6b1ea98..6742896f 100644 --- a/catalyst/examples/buy_low_sell_high_live.py +++ b/catalyst/examples/buy_low_sell_high_live.py @@ -1,6 +1,7 @@ import talib from logbook import Logger +import pandas as pd from catalyst.api import ( order, order_target_percent, @@ -17,10 +18,10 @@ log = Logger('buy low sell high') def initialize(context): log.info('initializing algo') - context.ASSET_NAME = 'XRP_BTC' + context.ASSET_NAME = 'etc_btc' context.asset = symbol(context.ASSET_NAME) - context.TARGET_POSITIONS = 300 + context.TARGET_POSITIONS = 3 context.PROFIT_TARGET = 0.1 context.SLIPPAGE_ALLOWED = 0.02 @@ -33,6 +34,9 @@ def initialize(context): def _handle_data(context, data): + price = data.current(context.asset, 'price') + log.info('got price {price}'.format(price=price)) + prices = data.history( context.asset, fields='price', @@ -44,20 +48,17 @@ def _handle_data(context, data): # Buying more when RSI is low, this should lower our cost basis if rsi <= 30: - buy_increment = 50 + buy_increment = 1 elif rsi <= 40: - buy_increment = 20 - # elif rsi <= 70: - # buy_increment = 5 + buy_increment = 0.5 + elif rsi <= 70: + buy_increment = 0.2 else: buy_increment = None cash = context.portfolio.cash log.info('base currency available: {cash}'.format(cash=cash)) - price = data.current(context.asset, 'price') - log.info('got price {price}'.format(price=price)) - record( price=price, rsi=rsi, @@ -146,11 +147,22 @@ def analyze(context, stats): run_algorithm( + capital_base=1, initialize=initialize, handle_data=handle_data, analyze=analyze, exchange_name='bitfinex', - live=True, - algo_namespace=algo_namespace, - base_currency='btc' + start=pd.to_datetime('2017-5-01', utc=True), + end=pd.to_datetime('2017-10-01', utc=True), + base_currency='btc', + data_frequency='daily' ) +# run_algorithm( +# initialize=initialize, +# handle_data=handle_data, +# analyze=analyze, +# exchange_name='poloniex', +# live=True, +# algo_namespace=algo_namespace, +# base_currency='btc' +# ) 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 e02b50de..a7630ef2 100644 --- a/catalyst/examples/buy_low_sell_high_neo_with_interface.py +++ b/catalyst/examples/buy_low_sell_high_neo_with_interface.py @@ -163,8 +163,6 @@ def analyze(context, stats): # Backtest run_algorithm( capital_base=250, - start=pd.to_datetime('2017-10-01', utc=True), - end=pd.to_datetime('2017-10-15', utc=True), data_frequency='minute', initialize=initialize, handle_data=handle_data, diff --git a/catalyst/exchange/bundle_utils.py b/catalyst/exchange/bundle_utils.py index 536baa28..fe622b52 100644 --- a/catalyst/exchange/bundle_utils.py +++ b/catalyst/exchange/bundle_utils.py @@ -1,18 +1,15 @@ import calendar -import tarfile - -import requests -from datetime import timedelta, datetime, date import os -import pandas as pd -import numpy as np +import tarfile +from datetime import timedelta, datetime, date +import numpy as np +import pandas as pd import pytz from catalyst.data.bundles import from_bundle_ingest_dirname from catalyst.data.bundles.core import download_without_progress -from catalyst.exchange.exchange_errors import ApiCandlesError, \ - PricingDataBeforeTradingError, NoDataAvailableOnExchange +from catalyst.exchange.exchange_errors import NoDataAvailableOnExchange from catalyst.exchange.exchange_utils import get_exchange_bundles_folder from catalyst.utils.deprecate import deprecated from catalyst.utils.paths import data_path @@ -189,60 +186,6 @@ def get_df_from_arrays(arrays, periods): return df -def get_df_from_candles(candles, bar_count, end_dt, data_frequency, - previous_candle=None): - """ - Create candles for each period of the specified range, forward-filling - missing candles with the previous value. - - :param candles: - :param bar_count: - :param end_dt: - :param data_frequency: - :param previous_candle: - - :return: - """ - all_dates = [] - all_candles = [] - - start_dt = get_start_dt(end_dt, bar_count, data_frequency) - date = start_dt - - # TODO: this works well with a small number of candles, consider using numpy as needed - while date <= end_dt: - candle = next(( - candle for candle in candles if candle['last_traded'] == date - ), previous_candle) - - if candle is None: - candle = candles[0] - - all_dates.append(date) - all_candles.append(candle) - - previous_candle = candle - - date += get_delta(1, data_frequency) - - return all_dates, all_candles - - -def get_trailing_candles_dt(asset, start_dt, end_dt, data_frequency): - missing_start = None - - if asset.end_minute is not None and start_dt < asset.end_minute: - if asset.end_minute < end_dt: - delta = get_delta(1, data_frequency) - - missing_start = asset.end_minute + delta - - else: - missing_start = start_dt - - return missing_start - - def range_in_bundle(asset, start_dt, end_dt, reader): """ Evaluate whether price data of an asset is included has been ingested in @@ -278,6 +221,7 @@ def range_in_bundle(asset, start_dt, end_dt, reader): return has_data +@deprecated def find_most_recent_time(bundle_name): """ Find most recent "time folder" for a given bundle. @@ -308,83 +252,3 @@ def find_most_recent_time(bundle_name): else: return None - -@deprecated -def get_history(exchange_name, data_frequency, symbol, start=None, end=None): - """ - History API provides OHLCV data for any of the supported exchanges up to yesterday. - - :param exchange_name: string - Required: The name identifier of the exchange (e.g. bitfinex, bittrex, poloniex). - :param data_frequency: string - Required: The bar frequency (minute or daily) - :param symbol: string - Required: The trading pair symbol, using Catalyst naming convention - :param start: datetime - Optional: The start date. - :param end: datetime - Optional: The end date. - - :return ohlcv: list[dict[string, float]] - Each row contains the following dictionary for the resulting bars: - 'ts' : int, the timestamp in seconds - 'open' : float - 'high' : float - 'low' : float - 'close' : float - 'volume' : float - - Notes - ===== - Using seconds for the start and end dates for ease of use in the - function query parameters. - - Sometimes, one minute goes by without completing a trade of the given - trading pair on the given exchange. To minimize the payload size, we - don't return identical sequential bars. Post-processing code will - forward fill missing bars outside of this function. - """ - - start_seconds = get_seconds_from_date(start) if start else None - end_seconds = get_seconds_from_date(end) if end else None - - if exchange_name not in EXCHANGE_NAMES: - raise ValueError( - 'get_history function only supports the following exchanges: {}'.format( - list(EXCHANGE_NAMES))) - - if data_frequency != 'daily' and data_frequency != 'minute': - raise ValueError( - 'get_history currently only supports daily and minute data.' - ) - - url = '{api_url}/candles?exchange={exchange}&market={symbol}&freq={data_frequency}'.format( - api_url=API_URL, - exchange=exchange_name, - symbol=symbol, - data_frequency=data_frequency, - ) - - if start_seconds: - url += '&start={}'.format(start_seconds) - - if end_seconds: - url += '&end={}'.format(end_seconds) - - try: - response = requests.get(url) - except Exception as e: - raise ValueError(e) - - data = response.json() - - if 'error' in data: - raise ApiCandlesError(error=data['error']) - - for candle in data: - last_traded = pd.Timestamp.utcfromtimestamp(candle['ts']) - last_traded = last_traded.replace(tzinfo=pytz.UTC) - - candle['last_traded'] = last_traded - - return data diff --git a/catalyst/exchange/data_portal_exchange.py b/catalyst/exchange/data_portal_exchange.py index 6d0d3fc6..468fb8a1 100644 --- a/catalyst/exchange/data_portal_exchange.py +++ b/catalyst/exchange/data_portal_exchange.py @@ -12,15 +12,15 @@ # limitations under the License. import abc -from datetime import timedelta from time import sleep +import numpy as np import pandas as pd from catalyst.assets._assets import TradingPair from logbook import Logger from catalyst.data.data_portal import DataPortal -from catalyst.errors import HistoryWindowStartsBeforeData +from catalyst.exchange.bundle_utils import get_start_dt from catalyst.exchange.exchange_bundle import ExchangeBundle from catalyst.exchange.exchange_errors import ( ExchangeRequestError, @@ -153,6 +153,10 @@ class DataPortalExchangeBase(DataPortal): exchange = self.exchanges[assets.exchange] spot_values = self.get_exchange_spot_value( exchange, [assets], field, dt, data_frequency) + + if not spot_values: + return np.nan + return spot_values[0] else: @@ -282,109 +286,60 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): field, data_frequency, ffill=True): + """ + Fetching price history window from the exchange bundle. + + Using a try... except approach to minimize reads most of the time, + when the data exists. + + :param exchange: + :param assets: + :param end_dt: + :param bar_count: + :param frequency: + :param field: + :param data_frequency: + :param ffill: + :return: + """ + bundle = self.exchange_bundles[exchange.name] - - if data_frequency == 'minute': - dts = self.trading_calendar.minutes_window( - end_dt, -bar_count - ) - - self.ensure_after_first_day(dts[0], assets) - - elif data_frequency == 'daily': - session = self.trading_calendar.minute_to_session_label(end_dt) - dts = self._get_days_for_window(session, bar_count) - - if len(dts) == 0: - symbols = [asset.symbol for asset in assets] - raise PricingDataNotLoadedError( - field=field, - symbols=symbols, - exchange=exchange.name, - first_trading_day= \ - min([asset.start_date for asset in assets]), - data_frequency=data_frequency, - symbol_list=','.join(symbols) - ) - - self.ensure_after_first_day(dts[0], assets) - - else: - raise InvalidHistoryFrequencyError(frequency=data_frequency) - - reader = bundle.get_reader(data_frequency) - if reader is None: - raise BundleNotFoundError( - exchange=exchange.name.title(), - data_frequency=data_frequency - ) - - try: - values = reader.load_raw_arrays( - sids=[asset.sid for asset in assets], - fields=[field], - start_dt=dts[0], - end_dt=dts[-1] - )[0] - - except Exception: - first_trading_day = self._get_first_trading_day(assets) - symbols = [asset.symbol.encode('utf-8') for asset in assets] - - symbol_list = ','.join(symbols) - raise PricingDataNotLoadedError( - field=field, - first_trading_day=first_trading_day, - exchange=exchange.name.title(), - symbols=symbols, - symbol_list=symbol_list, - data_frequency=data_frequency - ) - - series = dict() - for index, asset in enumerate(assets): - asset_values = values[:, index] - - value_series = pd.Series(asset_values, index=dts) - series[asset] = value_series - + series = bundle.get_history_window_series_and_load( + assets=assets, + end_dt=end_dt, + bar_count=bar_count, + field=field, + data_frequency=data_frequency + ) return pd.DataFrame(series) - 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.title(), - symbols=[asset.symbol.encode('utf-8') for asset in assets], - dt=dt, - ) - def get_exchange_spot_value(self, exchange, assets, field, dt, data_frequency): bundle = self.exchange_bundles[exchange.name] - reader = bundle.get_reader(data_frequency) - self.ensure_after_first_day(dt, assets) + if data_frequency == 'daily': + dt = dt.floor('1D') + else: + dt = dt.floor('1 min') - values = [] - for asset in assets: - try: - value = reader.get_value( - sid=asset.sid, - dt=dt, - field=field + try: + return bundle.get_spot_values(assets, field, dt, data_frequency) + + except PricingDataNotLoadedError: + log.info( + 'pricing data for {symbol} not found on {dt}' + ', updating the bundles.'.format( + symbol=[asset.symbol for asset in assets], + dt=dt ) - values.append(value) - except Exception: - raise PricingDataNotLoadedError( - field=field, - first_trading_day=self._get_first_trading_day(assets), - exchange=exchange.name.title(), - symbols=[asset.symbol.encode('utf-8') for asset in assets], - symbol_list=''.join( - [asset.symbol.encode('utf-8') for asset in assets]), - data_frequency=data_frequency - ) - - return values + ) + bundle.ingest_assets( + assets=assets, + start_dt=self._first_trading_day, + end_dt=self._last_available_session, + data_frequency=data_frequency, + show_progress=True + ) + return bundle.get_spot_values( + assets, field, dt, data_frequency, True + ) diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index c5884afe..f8c902a8 100644 --- a/catalyst/exchange/exchange.py +++ b/catalyst/exchange/exchange.py @@ -16,7 +16,7 @@ from catalyst.exchange.exchange_bundle import ExchangeBundle from catalyst.exchange.exchange_errors import MismatchingBaseCurrencies, \ InvalidOrderStyle, BaseCurrencyNotFoundError, SymbolNotFoundOnExchange, \ InvalidHistoryFrequencyError, MismatchingFrequencyError, \ - BundleNotFoundError, NoDataAvailableOnExchange + BundleNotFoundError, NoDataAvailableOnExchange, PricingDataNotLoadedError from catalyst.exchange.exchange_execution import ExchangeStopLimitOrder, \ ExchangeLimitOrder, ExchangeStopOrder from catalyst.exchange.exchange_portfolio import ExchangePortfolio @@ -370,44 +370,6 @@ class Exchange: return value - def get_series_from_bundle(self, assets, start_dt, end_dt, data_frequency, - field): - """ - - :return: - """ - reader = self.bundle.get_reader(data_frequency) - - if reader is None: - raise BundleNotFoundError( - exchange=self.name.title(), - data_frequency=data_frequency - ) - - series = dict() - try: - arrays = reader.load_raw_arrays( - sids=[asset.sid for asset in assets], - fields=[field], - start_dt=start_dt, - end_dt=end_dt - ) - - periods = self.bundle.get_calendar_periods_range( - start_dt, end_dt, data_frequency - ) - - for asset_index, asset in enumerate(assets): - asset_values = arrays[asset_index] - - value_series = pd.Series(asset_values[0], index=periods) - series[asset] = value_series - - except Exception as e: - log.debug('unable to retrieve from bundle: {}'.format(e)) - - return series - def get_series_from_candles(self, candles, start_dt, end_dt, field, previous_value=None): """ @@ -487,11 +449,6 @@ class Exchange: data_frequency = 'daily' elif unit.lower() == 'm': - # if data_frequency != 'minute': - # raise MismatchingFrequencyError( - # frequency=frequency, - # data_frequency=data_frequency - # ) if data_frequency == 'daily': data_frequency = 'minute' @@ -499,42 +456,15 @@ class Exchange: raise InvalidHistoryFrequencyError(frequency) adj_bar_count = candle_size * bar_count - start_dt = get_start_dt(end_dt, adj_bar_count, data_frequency) - try: - adj_start_dt, adj_end_dt = get_adj_dates( - start_dt, end_dt, assets, data_frequency - ) - in_bundle = True - - except NoDataAvailableOnExchange: - in_bundle = False - - if in_bundle: - missing_assets = self.bundle.filter_existing_assets( + series = self.bundle.get_history_window_series_and_load( assets=assets, - start_dt=adj_start_dt, - end_dt=adj_end_dt, + end_dt=end_dt, + bar_count=adj_bar_count, + field=field, data_frequency=data_frequency ) - - if missing_assets: - self.bundle.ingest_assets( - assets=assets, - start_dt=adj_start_dt, - end_dt=adj_end_dt, - data_frequency=data_frequency - ) - - series = self.get_series_from_bundle( - assets=assets, - start_dt=adj_start_dt, - end_dt=adj_end_dt, - data_frequency=data_frequency, - field=field - ) - - else: + except PricingDataNotLoadedError: series = dict() for asset in assets: diff --git a/catalyst/exchange/exchange_bcolz.py b/catalyst/exchange/exchange_bcolz.py index 523eae28..0d19d04f 100644 --- a/catalyst/exchange/exchange_bcolz.py +++ b/catalyst/exchange/exchange_bcolz.py @@ -48,9 +48,9 @@ class BcolzExchangeBarReader(BcolzMinuteBarReader): # else: # return self._load_daily_raw_arrays(fields, start_dt, end_dt, sids) - return self._load_daily_raw_arrays(fields, start_dt, end_dt, sids) + return self._load_raw_arrays(fields, start_dt, end_dt, sids) - def _load_daily_raw_arrays(self, fields, start_dt, end_dt, sids): + def _load_raw_arrays(self, fields, start_dt, end_dt, sids): start_idx = self._find_position_of_minute(start_dt) end_idx = self._find_position_of_minute(end_dt) diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index b8f4276a..ad014b8d 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/catalyst/exchange/exchange_bundle.py @@ -10,12 +10,13 @@ from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \ BcolzMinuteBarMetadata from catalyst.exchange.bundle_utils import range_in_bundle, \ get_bcolz_chunk, get_delta, get_adj_dates, get_month_start_end, \ - get_year_start_end, get_periods_range, get_df_from_arrays + get_year_start_end, get_periods_range, get_df_from_arrays, get_start_dt from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \ BcolzExchangeBarWriter from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \ InvalidHistoryFrequencyError, PricingDataBeforeTradingError, \ - TempBundleNotFoundError, NoDataAvailableOnExchange + TempBundleNotFoundError, NoDataAvailableOnExchange, \ + PricingDataNotLoadedError from catalyst.exchange.exchange_utils import get_exchange_folder from catalyst.utils.cli import maybe_show_progress from catalyst.utils.paths import ensure_directory @@ -451,3 +452,152 @@ class ExchangeBundle: for frequency in data_frequency.split(','): self.ingest_assets(assets, start_dt, end_dt, frequency, show_progress) + + def get_history_window_series_and_load(self, + assets, + end_dt, + bar_count, + field, + data_frequency): + try: + series = self.get_history_window_series( + assets=assets, + end_dt=end_dt, + bar_count=bar_count, + field=field, + data_frequency=data_frequency + ) + return pd.DataFrame(series) + + except PricingDataNotLoadedError: + start_dt = get_start_dt(end_dt, bar_count, data_frequency) + log.info( + 'pricing data for {symbol} not found in range ' + '{start} to {end}, updating the bundles.'.format( + symbol=[asset.symbol for asset in assets], + start=start_dt, + end=end_dt + ) + ) + self.ingest_assets( + assets=assets, + start_dt=start_dt, + end_dt=end_dt, + data_frequency=data_frequency, + show_progress=True + ) + series = self.get_history_window_series( + assets=assets, + end_dt=end_dt, + bar_count=bar_count, + field=field, + data_frequency=data_frequency, + reset_reader=True + ) + return series + + def get_spot_values(self, assets, field, dt, data_frequency, + reset_reader=False): + values = [] + try: + reader = self.get_reader(data_frequency) + if reset_reader: + del self._readers[reader._rootdir] + reader = self.get_reader(data_frequency) + + for asset in assets: + value = reader.get_value( + sid=asset.sid, + dt=dt, + field=field + ) + values.append(value) + + return values + + except Exception: + symbols = [asset.symbol.encode('utf-8') for asset in assets] + raise PricingDataNotLoadedError( + field=field, + first_trading_day=min([asset.start_date for asset in assets]), + exchange=self.exchange.name, + symbols=symbols, + symbol_list=','.join(symbols), + data_frequency=data_frequency + ) + + def get_history_window_series(self, + assets, + end_dt, + bar_count, + field, + data_frequency, + reset_reader=False): + start_dt = get_start_dt(end_dt, bar_count, data_frequency) + start_dt, end_dt = \ + get_adj_dates(start_dt, end_dt, assets, data_frequency) + + reader = self.get_reader(data_frequency) + if reset_reader: + del self._readers[reader._rootdir] + reader = self.get_reader(data_frequency) + + if reader is None: + symbols = [asset.symbol.encode('utf-8') for asset in assets] + raise PricingDataNotLoadedError( + field=field, + first_trading_day=min([asset.start_date for asset in assets]), + exchange=self.exchange.name, + symbols=symbols, + symbol_list=','.join(symbols), + data_frequency=data_frequency + ) + + for asset in assets: + asset_start_dt, asset_end_dt = \ + get_adj_dates(start_dt, end_dt, assets, data_frequency) + + in_bundle = range_in_bundle( + asset, asset_start_dt, asset_end_dt, reader + ) + if not in_bundle: + raise PricingDataNotLoadedError( + field=field, + first_trading_day=asset.start_date, + exchange=self.exchange.name, + symbols=asset.symbol, + symbol_list=asset.symbol, + data_frequency=data_frequency + ) + + series = dict() + try: + arrays = reader.load_raw_arrays( + sids=[asset.sid for asset in assets], + fields=[field], + start_dt=start_dt, + end_dt=end_dt + ) + + except Exception: + symbols = [asset.symbol.encode('utf-8') for asset in assets] + raise PricingDataNotLoadedError( + field=field, + first_trading_day=min([asset.start_date for asset in assets]), + exchange=self.exchange.name, + symbols=symbols, + symbol_list=','.join(symbols), + data_frequency=data_frequency + ) + + periods = self.get_calendar_periods_range( + start_dt, end_dt, data_frequency + ) + + for asset_index, asset in enumerate(assets): + asset_values = arrays[asset_index] + + value_series = pd.Series(asset_values[0], index=periods) + series[asset] = value_series + + return series diff --git a/catalyst/utils/calendars/exchange_calendar_open.py b/catalyst/utils/calendars/exchange_calendar_open.py index 68c385f5..54be460f 100644 --- a/catalyst/utils/calendars/exchange_calendar_open.py +++ b/catalyst/utils/calendars/exchange_calendar_open.py @@ -31,4 +31,4 @@ class OpenExchangeCalendar(TradingCalendar): return DateOffset(days=1) def __init__(self, *args, **kwargs): - super(OpenExchangeCalendar, self).__init__(start=Timestamp('2015-02-19', tz='UTC'), **kwargs) + super(OpenExchangeCalendar, self).__init__(start=Timestamp('2015-3-1', tz='UTC'), **kwargs) diff --git a/catalyst/utils/run_algo.py b/catalyst/utils/run_algo.py index bf149642..3a318748 100644 --- a/catalyst/utils/run_algo.py +++ b/catalyst/utils/run_algo.py @@ -191,7 +191,12 @@ def _run(handle_data, open_calendar = get_calendar('OPEN') env = TradingEnvironment( - load=partial(load_crypto_market_data, environ=environ), + load=partial( + load_crypto_market_data, + environ=environ, + start_dt=start, + end_dt=end + ), environ=environ, exchange_tz='UTC', asset_db_path=None # We don't need an asset db, we have exchanges @@ -263,7 +268,7 @@ def _run(handle_data, ) # TODO: use the constructor instead - sim_params._arena = 'live' + # sim_params._arena = 'live' algorithm_class = partial( ExchangeTradingAlgorithmLive, @@ -284,7 +289,8 @@ def _run(handle_data, exchanges=exchanges, asset_finder=None, trading_calendar=open_calendar, - first_trading_day=None, + first_trading_day=start, + last_available_session=end ) sim_params = create_simulation_parameters( diff --git a/tests/exchange/test_bundle.py b/tests/exchange/test_bundle.py index f7a3901d..301f71bb 100644 --- a/tests/exchange/test_bundle.py +++ b/tests/exchange/test_bundle.py @@ -3,7 +3,8 @@ from logging import Logger import pandas as pd from catalyst import get_calendar -from catalyst.exchange.bundle_utils import get_bcolz_chunk +from catalyst.exchange.bundle_utils import get_bcolz_chunk, get_periods, \ + get_periods_range from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \ BcolzExchangeBarWriter from catalyst.exchange.exchange_bundle import ExchangeBundle, \ @@ -78,12 +79,13 @@ class ExchangeBundleTestCase: # data_frequency = 'daily' # include_symbols = 'neo_btc,bch_btc,eth_btc' - exchange_name = 'bitfinex' + exchange_name = 'poloniex' data_frequency = 'daily' - include_symbols = 'etc_btc' + include_symbols = 'btc_usdt' - start = pd.to_datetime('2016-11-01', utc=True) + start = pd.to_datetime('2016-1-1', utc=True) end = pd.to_datetime('2017-10-16', utc=True) + periods = get_periods_range(start, end, data_frequency) exchange = get_exchange(exchange_name) exchange_bundle = ExchangeBundle(exchange)