diff --git a/catalyst/examples/simple_loop.py b/catalyst/examples/simple_loop.py index 7081178f..5b7fcd26 100644 --- a/catalyst/examples/simple_loop.py +++ b/catalyst/examples/simple_loop.py @@ -27,15 +27,25 @@ def handle_data(context, data): pass +# run_algorithm( +# capital_base=250, +# start=pd.to_datetime('2017-08-01', utc=True), +# end=pd.to_datetime('2017-9-30', utc=True), +# data_frequency='daily', +# initialize=initialize, +# handle_data=handle_data, +# analyze=None, +# exchange_name='poloniex', +# algo_namespace='simple_loop', +# base_currency='eth' +# ) run_algorithm( - capital_base=250, - start=pd.to_datetime('2015-2-19', utc=True), - end=pd.to_datetime('2017-9-30', utc=True), - data_frequency='daily', initialize=initialize, handle_data=handle_data, analyze=None, exchange_name='poloniex', + live=True, algo_namespace='simple_loop', - base_currency='eth' + base_currency='eth', + live_graph=False ) diff --git a/catalyst/exchange/bundle_utils.py b/catalyst/exchange/bundle_utils.py index 0aa05661..d4ace25d 100644 --- a/catalyst/exchange/bundle_utils.py +++ b/catalyst/exchange/bundle_utils.py @@ -135,11 +135,11 @@ def get_adj_dates(start, end, assets, data_frequency): end = last_entry if end is None: - raise NoDataAvailableOnExchange( + raise NoDataAvailableOnExchange( exchange=asset.exchange.title(), symbol=[asset.symbol.encode('utf-8')], data_frequency=data_frequency, - ) + ) if end is None or start >= end: raise PricingDataBeforeTradingError( @@ -184,8 +184,21 @@ def get_year_start_end(dt): return year_start, year_end -def get_ffill_candles(candles, bar_count, end_dt, data_frequency, - previous_candle=None): +def get_df_from_arrays(arrays, periods): + ohlcv = dict() + for index, field in enumerate( + ['open', 'high', 'low', 'close', 'volume']): + ohlcv[field] = arrays[index].flatten() + + df = pd.DataFrame( + data=ohlcv, + index=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. diff --git a/catalyst/exchange/data_portal_exchange.py b/catalyst/exchange/data_portal_exchange.py index ce8ab3ab..6d0d3fc6 100644 --- a/catalyst/exchange/data_portal_exchange.py +++ b/catalyst/exchange/data_portal_exchange.py @@ -12,6 +12,7 @@ # limitations under the License. import abc +from datetime import timedelta from time import sleep import pandas as pd @@ -19,6 +20,7 @@ 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.exchange_bundle import ExchangeBundle from catalyst.exchange.exchange_errors import ( ExchangeRequestError, @@ -293,6 +295,18 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): 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: diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index d21cace4..083ad1a0 100644 --- a/catalyst/exchange/exchange.py +++ b/catalyst/exchange/exchange.py @@ -1,5 +1,4 @@ import abc -import random import re from abc import ABCMeta, abstractmethod, abstractproperty from datetime import timedelta @@ -12,7 +11,8 @@ from logbook import Logger from catalyst.data.data_portal import BASE_FIELDS from catalyst.exchange.bundle_utils import get_start_dt, \ - get_delta, get_trailing_candles_dt, get_periods, get_adj_dates + get_delta, get_trailing_candles_dt, get_periods, get_adj_dates, \ + get_df_from_candles from catalyst.exchange.exchange_bundle import ExchangeBundle from catalyst.exchange.exchange_errors import MismatchingBaseCurrencies, \ InvalidOrderStyle, BaseCurrencyNotFoundError, SymbolNotFoundOnExchange, \ @@ -100,12 +100,6 @@ class Exchange: delta = now - cpt_date sleep_period = 60 - delta.total_seconds() - - # log.debug( - # 'max requests {} reached, sleeping for {} seconds'.format( - # self.max_requests_per_minute, - # sleep_period - # )) sleep(sleep_period) now = pd.Timestamp.utcnow() @@ -174,7 +168,8 @@ class Exchange: asset = self.assets[key] if not asset: - supported_symbols = [pair.symbol.encode('utf-8') for pair in self.assets.values()] + supported_symbols = [pair.symbol.encode('utf-8') for pair in + self.assets.values()] raise SymbolNotFoundOnExchange( symbol=symbol, exchange=self.name.title(), @@ -367,35 +362,76 @@ class Exchange: ) ) - # Don't use a timezone here - dt = pd.Timestamp.utcnow().floor('1 min') ohlc = self.get_candles(data_frequency, asset) if field not in ohlc: raise KeyError('Invalid column: %s' % field) - if self.minute_writer is not None: - df = pd.DataFrame( - [ohlc], - index=pd.DatetimeIndex([dt]), - columns=['open', 'high', 'low', 'close', 'volume'] - ) - - try: - # TODO: use victor's modified branch using int64 - self.minute_writer.write_sid( - sid=asset.sid, - df=df - ) - log.debug('wrote minute data: {}'.format(dt)) - except Exception as e: - log.warn( - 'unable to write minute data: {} {}'.format(dt, e)) - - value = ohlc[field] - log.debug('got spot value: {}'.format(value)) + value = ohlc[field] + log.debug('got spot value: {}'.format(value)) 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 retreive from bundle: {}'.format(e)) + + return series + + def get_series_from_candles(self, candles, start_dt, end_dt, + field, previous_value=None): + """ + Get a series of field data for the specified candles. + + :param candles: + :param start_dt: + :param end_dt: + :param field: + :param previous_value: + :return: + """ + + dates = [candle['last_traded'] for candle in candles] + values = [candle[field] for candle in candles] + + periods = pd.date_range(start_dt, end_dt) + series = pd.Series(values, index=dates) + + series.reindex(periods, method='ffill', fill_value=previous_value) + + return series + def get_history_window(self, assets, end_dt, @@ -448,11 +484,8 @@ class Exchange: raise InvalidHistoryFrequencyError(frequency) if unit.lower() == 'd': - if data_frequency != 'daily': - raise MismatchingFrequencyError( - frequency=frequency, - data_frequency=data_frequency - ) + if data_frequency == 'minute': + data_frequency = 'daily' elif unit.lower() == 'm': if data_frequency != 'minute': @@ -467,94 +500,70 @@ class Exchange: adj_bar_count = candle_size * bar_count start_dt = get_start_dt(end_dt, adj_bar_count, data_frequency) - start_dt, end_dt = get_adj_dates(start_dt, end_dt, assets, - data_frequency) + adj_start_dt, adj_end_dt = get_adj_dates( + start_dt, end_dt, assets, data_frequency + ) missing_assets = self.bundle.filter_existing_assets( assets=assets, - start_dt=start_dt, - end_dt=end_dt, + start_dt=adj_start_dt, + end_dt=adj_end_dt, data_frequency=data_frequency ) if missing_assets: self.bundle.ingest_assets( assets=assets, - start_dt=start_dt, - end_dt=end_dt, + start_dt=adj_start_dt, + end_dt=adj_end_dt, data_frequency=data_frequency ) - # We check again for data which may be too recent for the consolidated - # exchanges service - trailing_assets = self.bundle.filter_existing_assets( + series = self.get_series_from_bundle( assets=assets, - start_dt=start_dt, - end_dt=end_dt, - data_frequency=data_frequency + start_dt=adj_start_dt, + end_dt=adj_end_dt, + data_frequency=data_frequency, + field=field ) - if trailing_assets: - # Adding bars too recent to be contained in the consolidated - # exchanges bundles. We go directly against the exchange - # to retrieve the candles. - for asset in trailing_assets: - trailing_candles_dt = get_trailing_candles_dt( - asset=asset, - start_dt=start_dt, - end_dt=end_dt, - data_frequency=data_frequency + + for asset in assets: + if asset not in series or series[asset].index[-1] < end_dt: + # Adding bars too recent to be contained in the consolidated + # exchanges bundles. We go directly against the exchange + # to retrieve the candles. + + trailing_dt = \ + series[asset].index[-1] + get_delta(1, data_frequency) \ + if asset in series else start_dt + + trailing_bar_count = \ + get_periods(trailing_dt, end_dt, data_frequency) + + # The get_history method supports multiple asset + candles = self.get_candles( + data_frequency=data_frequency, + assets=asset, + bar_count=trailing_bar_count, + end_dt=end_dt ) - if trailing_candles_dt is not None: - trailing_bar_count = \ - get_periods(start_dt, end_dt, data_frequency) + last_value = series[asset].iloc(0) if asset in series \ + else np.nan - # The get_history method supports multiple asset - candles = self.get_candles( - data_frequency=data_frequency, - assets=[asset], - bar_count=trailing_bar_count, - end_dt=end_dt - ) + candle_series = self.get_series_from_candles( + candles=candles, + start_dt=trailing_dt, + end_dt=end_dt, + field=field, + previous_value=last_value + ) - # TODO: Do I need the previous_candle? - self.bundle.ingest_candles( - candles=candles, - bar_count=trailing_bar_count, - start_dt=start_dt, - end_dt=end_dt, - data_frequency=data_frequency - ) + if asset in series: + series[asset].append(candle_series) - reader = self.bundle.get_reader(data_frequency) - if reader is None: - raise BundleNotFoundError( - exchange=self.name.title(), - data_frequency=data_frequency - ) - - values = reader.load_raw_arrays( - sids=[asset.sid for asset in assets], - fields=[field], - start_dt=start_dt, - end_dt=end_dt - ) - - series = dict() - for asset_index, asset in enumerate(assets): - all_dates = [] - asset_values = [] - - # TODO: use numpy to avoid the loop - date = start_dt - for value in values[0]: - all_dates.append(date) - asset_values.append(value[asset_index]) - - date += get_delta(1, data_frequency) - - value_series = pd.Series(asset_values, index=all_dates) - series[asset] = value_series + else: + series[asset] = candle_series df = pd.DataFrame(series) diff --git a/catalyst/exchange/exchange_bcolz.py b/catalyst/exchange/exchange_bcolz.py index cfd323f6..523eae28 100644 --- a/catalyst/exchange/exchange_bcolz.py +++ b/catalyst/exchange/exchange_bcolz.py @@ -61,12 +61,13 @@ class BcolzExchangeBarReader(BcolzMinuteBarReader): num_days = len(periods) shape = num_days, len(sids) - if len(fields) == 1 and fields[0] == 'volume': - fields.insert(0, 'close') + all_fields = fields[:] + if len(all_fields) == 1 and all_fields[0] == 'volume': + all_fields.insert(0, 'close') mask = None data = [] - for field in fields: + for field in all_fields: if field != 'volume': out = np.full(shape, np.nan) else: @@ -83,6 +84,7 @@ class BcolzExchangeBarReader(BcolzMinuteBarReader): a[mask] * self._ohlc_ratio_inverse_for_sid(sid) ) - data.append(out) + if field in fields: + data.append(out) return data diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index db4bb613..9fc5b9c9 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/catalyst/exchange/exchange_bundle.py @@ -2,16 +2,15 @@ import os import shutil from datetime import timedelta -import numpy as np import pandas as pd from logbook import Logger, INFO from catalyst import get_calendar from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \ BcolzMinuteBarMetadata -from catalyst.exchange.bundle_utils import get_ffill_candles, range_in_bundle, \ +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_year_start_end, get_periods_range, get_df_from_arrays from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \ BcolzExchangeBarWriter from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \ @@ -192,79 +191,6 @@ class ExchangeBundle: invalid_data_behavior='raise' ) - def ingest_candles(self, candles, bar_count, start_dt, end_dt, - data_frequency, - previous_candle=dict()): - """ - Ingest candles obtained via the get_candles API of an exchange. - - Since exchange APIs generally only do not return candles when there - are no transactions in the period, we ffill values using the - previous candle to ensure that each period has a candle. - - :param bar_count: - :param end_dt: - :param data_frequency: - :param asset: - :param writer: - :param previous_candle - :return: - """ - - writer = self.get_writer(start_dt, end_dt, data_frequency) - - num_candles = 0 - data = [] - for asset in candles: - asset_candles = candles[asset] - if not asset_candles: - log.debug( - 'no data: {symbols} on {exchange}, date {end}'.format( - symbols=asset, - exchange=self.exchange.name, - end=end_dt - ) - ) - continue - - previous = previous_candle[asset] \ - if asset in previous_candle else None - - all_dates, all_candles = get_ffill_candles( - candles=asset_candles, - bar_count=bar_count, - end_dt=end_dt, - data_frequency=data_frequency, - previous_candle=previous - ) - previous_candle[asset] = all_candles[-1] - - df = pd.DataFrame( - data=all_candles, - index=all_dates, - columns=['open', 'high', 'low', 'close', 'volume'] - ) - - if not df.empty: - df.sort_index(inplace=True) - - sid = asset.sid - num_candles += len(df.values) - - data.append((sid, df)) - - log.debug( - 'writing {num_candles} candles for {bar_count} bars' - 'ending {end}'.format( - num_candles=num_candles, - bar_count=bar_count, - end=end_dt - ) - ) - self._write(data, writer, data_frequency) - - return data - def get_calendar_periods_range(self, start_dt, end_dt, data_frequency): return self.calendar.minutes_in_range(start_dt, end_dt) \ if data_frequency == 'minute' \ @@ -295,9 +221,6 @@ class ExchangeBundle: period=period ) - periods = self.get_calendar_periods_range( - start_dt, end_dt, data_frequency - ) reader = self.get_reader(data_frequency, path=path) if reader is None: raise TempBundleNotFoundError(path=path) @@ -312,16 +235,12 @@ class ExchangeBundle: if not arrays: return path - ohlcv = dict() - for index, field in enumerate( - ['open', 'high', 'low', 'close', 'volume']): - ohlcv[field] = arrays[index].flatten() - - df = pd.DataFrame( - data=ohlcv, - index=periods + periods = self.get_calendar_periods_range( + start_dt, end_dt, data_frequency ) + df = get_df_from_arrays(arrays, periods) + if empty_rows_behavior is not 'ignore': nan_rows = df[df.isnull().T.any().T].index @@ -369,7 +288,6 @@ class ExchangeBundle: if not df.empty: df.sort_index(inplace=True) data.append((asset.sid, df)) - self._write(data, writer, data_frequency) if cleanup: