From d3e33c44bfd77c9649fd34347de59f1a53d65558 Mon Sep 17 00:00:00 2001 From: fredfortier Date: Wed, 11 Oct 2017 00:13:22 -0400 Subject: [PATCH] Reading data from bundles first and other fixes --- catalyst/__main__.py | 4 +- .../buy_low_sell_high_neo_with_interface.py | 38 ++--- catalyst/exchange/bitfinex/bitfinex.py | 1 - catalyst/exchange/bundle_utils.py | 46 ++++-- catalyst/exchange/data_portal_exchange.py | 4 +- catalyst/exchange/exchange.py | 90 +++++++++-- catalyst/exchange/exchange_bundle.py | 144 ++++++++++++------ tests/exchange/test_bundle.py | 5 +- 8 files changed, 236 insertions(+), 96 deletions(-) diff --git a/catalyst/__main__.py b/catalyst/__main__.py index fce7fa55..60758487 100644 --- a/catalyst/__main__.py +++ b/catalyst/__main__.py @@ -9,6 +9,7 @@ from six import text_type from catalyst.data import bundles as bundles_module from catalyst.exchange.exchange_bundle import ExchangeBundle +from catalyst.exchange.init_utils import get_exchange from catalyst.utils.cli import Date, Timestamp from catalyst.utils.run_algo import _run, load_extensions @@ -492,7 +493,8 @@ def ingest_exchange(exchange_name, data_frequency, start, end, """ Ingest data for the given exchange. """ - exchange_bundle = ExchangeBundle(exchange_name) + exchange=get_exchange(exchange_name) + exchange_bundle = ExchangeBundle(exchange) click.echo('ingesting exchange bundle {}'.format(exchange_name)) exchange_bundle.ingest( 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 2e216cb5..8cb1a5a6 100644 --- a/catalyst/examples/buy_low_sell_high_neo_with_interface.py +++ b/catalyst/examples/buy_low_sell_high_neo_with_interface.py @@ -36,8 +36,8 @@ def _handle_data(context, data): prices = data.history( context.asset, fields='price', - bar_count=30, - frequency='30m' + bar_count=50, + frequency='1m' ) rsi = talib.RSI(prices.values, timeperiod=14)[-1] log.info('got rsi: {}'.format(rsi)) @@ -148,27 +148,27 @@ def analyze(context, stats): pass -# run_algorithm( -# initialize=initialize, -# handle_data=handle_data, -# analyze=analyze, -# exchange_name='bitfinex', -# live=True, -# algo_namespace=algo_namespace, -# base_currency='btc', -# live_graph=False -# ) - -# Backtest run_algorithm( - 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, handle_data=handle_data, analyze=analyze, exchange_name='bitfinex', + live=True, algo_namespace=algo_namespace, - base_currency='btc' + base_currency='btc', + live_graph=False ) + +# Backtest +# run_algorithm( +# 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, +# handle_data=handle_data, +# analyze=analyze, +# exchange_name='bitfinex', +# algo_namespace=algo_namespace, +# base_currency='btc' +# ) diff --git a/catalyst/exchange/bitfinex/bitfinex.py b/catalyst/exchange/bitfinex/bitfinex.py index d4d368a9..38785c5f 100644 --- a/catalyst/exchange/bitfinex/bitfinex.py +++ b/catalyst/exchange/bitfinex/bitfinex.py @@ -14,7 +14,6 @@ import six from catalyst.assets._assets import TradingPair from logbook import Logger -# from websocket import create_connection from catalyst.exchange.exchange import Exchange from catalyst.exchange.exchange_errors import ( ExchangeRequestError, diff --git a/catalyst/exchange/bundle_utils.py b/catalyst/exchange/bundle_utils.py index 5078f52d..ade4f860 100644 --- a/catalyst/exchange/bundle_utils.py +++ b/catalyst/exchange/bundle_utils.py @@ -1,4 +1,5 @@ -import datetime, requests +import requests +from datetime import timedelta, datetime import os from logging import Logger import pandas as pd @@ -16,11 +17,11 @@ API_URL = 'http://data.enigma.co/api/v1' def get_date_from_ms(ms): - return datetime.datetime.fromtimestamp(ms / 1000.0) + return datetime.fromtimestamp(ms / 1000.0) def get_seconds_from_date(date): - epoch = datetime.datetime.utcfromtimestamp(0) + epoch = datetime.utcfromtimestamp(0) epoch = epoch.replace(tzinfo=pytz.UTC) return int((date - epoch).total_seconds()) @@ -107,14 +108,30 @@ def get_history(exchange_name, data_frequency, symbol, start=None, end=None): return data -def get_ffill_candles(candles, start_dt, end_dt, data_frequency, +def get_delta(periods, data_frequency): + return timedelta(minutes=periods) \ + if data_frequency == 'minute' else timedelta(days=periods) + + +def get_start_dt(end_dt, bar_count, data_frequency): + periods = bar_count - 1 + if periods > 1: + delta = get_delta(periods, data_frequency) + start_dt = end_dt - delta + else: + start_dt = end_dt + + return start_dt + + +def get_ffill_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 start_dt: + :param bar_count: :param end_dt: :param data_frequency: :param previous_candle: @@ -123,6 +140,8 @@ def get_ffill_candles(candles, start_dt, end_dt, data_frequency, """ all_dates = [] all_candles = [] + + start_dt = get_start_dt(end_dt, bar_count, data_frequency) date = start_dt while date <= end_dt: @@ -130,18 +149,15 @@ def get_ffill_candles(candles, start_dt, end_dt, data_frequency, candle for candle in candles if candle['last_traded'] == date ), previous_candle) - if candle is not None: - all_dates.append(date) - all_candles.append(candle) + if candle is None: + candle = candles[0] - previous_candle = candle + all_dates.append(date) + all_candles.append(candle) - if data_frequency == 'minute': - date += datetime.timedelta(minutes=1) - elif data_frequency == 'daily': - date += datetime.timedelta(days=1) - else: - raise ValueError('invalid data frequency') + previous_candle = candle + + date += get_delta(1, data_frequency) return all_dates, all_candles diff --git a/catalyst/exchange/data_portal_exchange.py b/catalyst/exchange/data_portal_exchange.py index 6d124b9f..e47ca435 100644 --- a/catalyst/exchange/data_portal_exchange.py +++ b/catalyst/exchange/data_portal_exchange.py @@ -260,8 +260,8 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): self.minute_history_loaders = dict() for exchange_name in self.exchanges: - self.exchange_bundles[exchange_name] = \ - ExchangeBundle(exchange_name) + exchange = self.exchanges[exchange_name] + self.exchange_bundles[exchange_name] = ExchangeBundle(exchange) def _get_first_trading_day(self, assets): first_date = None diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index 6bf6ee05..96969692 100644 --- a/catalyst/exchange/exchange.py +++ b/catalyst/exchange/exchange.py @@ -1,5 +1,6 @@ import abc import random +import re from abc import ABCMeta, abstractmethod, abstractproperty from datetime import timedelta from time import sleep @@ -11,8 +12,12 @@ from logbook import Logger from catalyst.data.data_portal import BASE_FIELDS from catalyst.exchange import bundle_utils +from catalyst.exchange.bundle_utils import get_ffill_candles, get_start_dt, \ + get_delta +from catalyst.exchange.exchange_bundle import ExchangeBundle from catalyst.exchange.exchange_errors import MismatchingBaseCurrencies, \ - InvalidOrderStyle, BaseCurrencyNotFoundError, SymbolNotFoundOnExchange + InvalidOrderStyle, BaseCurrencyNotFoundError, SymbolNotFoundOnExchange, \ + InvalidHistoryFrequencyError from catalyst.exchange.exchange_execution import ExchangeStopLimitOrder, \ ExchangeLimitOrder, ExchangeStopOrder from catalyst.exchange.exchange_portfolio import ExchangePortfolio @@ -456,7 +461,7 @@ class Exchange: The last trading date of the last bar. :return: """ - start = end - timedelta(minutes=bar_count) + start = get_start_dt(end, bar_count, data_frequency) exchange_start = None catalyst_end = None @@ -475,8 +480,8 @@ class Exchange: exchange_end = end else: - exchange_start = start exchange_end = end + exchange_start = start data = [] if catalyst_end is not None: @@ -495,7 +500,7 @@ class Exchange: data_frequency=data_frequency, assets=[asset], bar_count=bar_count, - start_dt=exchange_start, + start_dt=exchange_start if bar_count > 1 else None, end_dt=exchange_end ) data += candles[asset] @@ -545,25 +550,86 @@ class Exchange: A dataframe containing the requested data. """ - # TODO: try to read from bundle first - candles = self.get_history( + bundle = ExchangeBundle(self) + + freq_match = re.match(r'([0-9].*)(m|M|d|D)', frequency, re.M | re.I) + if freq_match: + candle_size = int(freq_match.group(1)) + unit = freq_match.group(2) + else: + raise InvalidHistoryFrequencyError(frequency) + + if unit.lower() == 'd': + data_frequency = 'daily' + + elif unit.lower() == 'm': + data_frequency = 'minute' + + else: + raise InvalidHistoryFrequencyError(frequency) + + adj_bar_count = candle_size * bar_count + start_dt = get_start_dt(end_dt, adj_bar_count, data_frequency) + + missing_assets = bundle.filter_existing_assets( assets=assets, + start_dt=start_dt, end_dt=end_dt, - bar_count=bar_count, data_frequency=data_frequency ) + if len(missing_assets) > 0: + writer = bundle.get_writer(start_dt, end_dt, data_frequency) + + bundle.ingest_chunk( + bar_count=adj_bar_count, + end_dt=end_dt, + data_frequency=data_frequency, + assets=missing_assets, + writer=writer + ) + + reader = bundle.get_reader(data_frequency) + values = reader.load_raw_arrays( + fields=[field], + start_dt=start_dt, + end_dt=end_dt, + sids=[asset.sid for asset in assets], + )[0] + series = dict() - for asset in assets: - asset_candles = candles[asset] + for asset_index, asset in enumerate(assets): + all_dates = [] + asset_values = [] - values = map(lambda candle: candle[field], asset_candles) - dates = map(lambda candle: candle['last_traded'], asset_candles) + date = start_dt + for value in values: + all_dates.append(date) + asset_values.append(value[asset_index]) - value_series = pd.Series(values, index=dates) + date += get_delta(1, data_frequency) + + value_series = pd.Series(asset_values, index=all_dates) series[asset] = value_series df = pd.DataFrame(series) + + if candle_size > 1: + if field == 'open': + agg = 'first' + elif field == 'high': + agg = 'max' + elif field == 'low': + agg = 'min' + elif field == 'close': + agg = 'last' + elif field == 'volume': + agg = 'sum' + else: + raise ValueError('invalid field') + + df = df.resample('{}T'.format(candle_size)).agg(agg) + return df def synchronize_portfolio(self): diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index 0e6e30af..7504f7c0 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/catalyst/exchange/exchange_bundle.py @@ -4,15 +4,15 @@ from datetime import timedelta import numpy as np import pandas as pd from logbook import Logger +from pandas import DatetimeIndex from catalyst import get_calendar from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \ - BcolzMinuteBarWriter, BcolzMinuteBarReader + BcolzMinuteBarWriter, BcolzMinuteBarReader, BcolzMinuteBarMetadata from catalyst.data.us_equity_pricing import BcolzDailyBarWriter, \ BcolzDailyBarReader -from catalyst.exchange.bundle_utils import get_ffill_candles +from catalyst.exchange.bundle_utils import get_ffill_candles, get_start_dt from catalyst.exchange.exchange_utils import get_exchange_folder -from catalyst.exchange.init_utils import get_exchange from catalyst.utils.cli import maybe_show_progress from catalyst.utils.paths import ensure_directory @@ -26,8 +26,8 @@ log = Logger('exchange_bundle') class ExchangeBundle: - def __init__(self, exchange_name, ): - self.exchange = get_exchange(exchange_name) + def __init__(self, exchange): + self.exchange = exchange self.minutes_per_day = 1440 self.default_ohlc_ratio = 1000000 self._writers = dict() @@ -72,7 +72,8 @@ class ExchangeBundle: :return: BcolzMinuteBarReader or BcolzDailyBarReader """ - if data_frequency in self._readers: + if data_frequency in self._readers \ + and self._readers[data_frequency] is not None: return self._readers[data_frequency] root = get_exchange_folder(self.exchange.name) @@ -81,6 +82,7 @@ class ExchangeBundle: frequency=data_frequency ) + self._readers[data_frequency] = None if data_frequency == 'minute': try: self._readers[data_frequency] = BcolzMinuteBarReader(input_dir) @@ -99,13 +101,16 @@ class ExchangeBundle: return self._readers[data_frequency] - def get_writer(self, data_frequency, start, end): + def update_metadata(self, writer, start_dt, end_dt): + pass + + def get_writer(self, start_dt, end_dt, data_frequency): """ Get a data writer object, either a new object or from cache :return: BcolzMinuteBarWriter or BcolzDailyBarWriter """ - key = (data_frequency, start, end) + key = data_frequency if key in self._writers: return self._writers[key] @@ -120,27 +125,59 @@ class ExchangeBundle: if data_frequency == 'minute': if len(os.listdir(output_dir)) > 0: + + metadata = BcolzMinuteBarMetadata.read(output_dir) + + write_metadata = False + if start_dt < metadata.start_session: + write_metadata = True + start_session = start_dt.floor('1d') + else: + start_session = metadata.start_session + + if end_dt > metadata.end_session: + write_metadata = True + + # TODO: workaround, improve the calendar logic? + if end_dt == start_dt: + end_dt += timedelta(days=1) + + end_session = end_dt.floor('1d') + else: + end_session = metadata.end_session + self._writers[key] = \ - BcolzMinuteBarWriter.open(output_dir, end) + BcolzMinuteBarWriter( + output_dir, + metadata.calendar, + start_session, + end_session, + metadata.minutes_per_day, + metadata.default_ohlc_ratio, + metadata.ohlc_ratios_per_sid, + write_metadata=write_metadata + ) else: self._writers[key] = BcolzMinuteBarWriter( rootdir=output_dir, calendar=open_calendar, minutes_per_day=self.minutes_per_day, - start_session=start, - end_session=end, + start_session=start_dt, + end_session=end_dt, write_metadata=True, default_ohlc_ratio=self.default_ohlc_ratio ) + elif data_frequency == 'daily': if len(os.listdir(output_dir)) > 0: - self._writers[key] = BcolzDailyBarWriter.open(output_dir, end) + self._writers[key] = \ + BcolzDailyBarWriter.open(output_dir, end_dt) else: - end_session = end.floor('1d') + end_session = end_dt.floor('1d') self._writers[key] = BcolzDailyBarWriter( filename=output_dir, calendar=open_calendar, - start_session=start, + start_session=start_dt, end_session=end_session ) else: @@ -150,7 +187,7 @@ class ExchangeBundle: return self._writers[key] - def filter_existing_assets(self, assets, start, end, data_frequency): + def filter_existing_assets(self, assets, start_dt, end_dt, data_frequency): """ For each asset, get the close on the start and end dates of the chunk. If the data exists, the chunk ingestion is complete. @@ -158,9 +195,9 @@ class ExchangeBundle: :param assets: list[TradingPair] The assets is scope. - :param start: + :param start_dt: The chunk start date. - :param end: + :param end_dt: The chunk end date. :return: list[TradingPair] The assets missing from the bundle @@ -171,13 +208,15 @@ class ExchangeBundle: has_data = True if has_data and reader is not None: try: - start_close = reader.get_value(asset.sid, start, 'close') + start_close = \ + reader.get_value(asset.sid, start_dt, 'close') if np.isnan(start_close): has_data = False else: - end_close = reader.get_value(asset.sid, end, 'close') + end_close = reader.get_value(asset.sid, end_dt, + 'close') if np.isnan(end_close): has_data = False @@ -193,8 +232,8 @@ class ExchangeBundle: return missing_assets - def ingest_chunk(self, chunk, previous_candle, data_frequency, assets, - writer): + def ingest_chunk(self, bar_count, end_dt, data_frequency, assets, + writer, previous_candle=dict()): """ Retrieve the specified OHLCV chunk and write it to the bundle @@ -205,18 +244,17 @@ class ExchangeBundle: :param writer: :return: """ - chunk_end = chunk['end'] - chunk_start = chunk_end - timedelta(minutes=chunk['bar_count']) chunk_assets = [] for asset in assets: - if asset.start_date <= chunk_end: + if asset.start_date <= end_dt: chunk_assets.append(asset) + start_dt = get_start_dt(end_dt, bar_count, data_frequency) missing_assets = self.filter_existing_assets( assets=chunk_assets, - start=chunk_start, - end=chunk_end, + start_dt=start_dt, + end_dt=end_dt, data_frequency=data_frequency ) @@ -226,8 +264,8 @@ class ExchangeBundle: candles = self.exchange.get_history( assets=missing_assets, - end_dt=chunk_end, - bar_count=chunk['bar_count'], + end_dt=end_dt, + bar_count=bar_count, data_frequency=data_frequency ) @@ -240,7 +278,7 @@ class ExchangeBundle: 'no data: {symbols} on {exchange}, date {end}'.format( symbols=missing_assets, exchange=self.exchange.name, - end=chunk_end + end=end_dt ) ) continue @@ -250,14 +288,18 @@ class ExchangeBundle: all_dates, all_candles = get_ffill_candles( candles=asset_candles, - start_dt=chunk_start, - end_dt=chunk_end, + bar_count=bar_count, + end_dt=end_dt, data_frequency=data_frequency, previous_candle=previous ) previous_candle[asset] = all_candles[-1] - df = pd.DataFrame(all_candles, index=all_dates) + df = pd.DataFrame( + data=all_candles, + index=all_dates, + columns=['open', 'high', 'low', 'close', 'volume'] + ) if not df.empty: df.sort_index(inplace=True) @@ -268,24 +310,37 @@ class ExchangeBundle: try: log.debug( - 'writing {num_candles} candles from {start} to {end}'.format( + 'writing {num_candles} candles for {bar_count} bars' + 'ending {end}'.format( num_candles=num_candles, - start=chunk_start, - end=chunk_end + bar_count=bar_count, + end=end_dt ) ) - for pair in data: - log.debug('data for sid {}\n{}\n{}'.format( - pair[0], pair[1].head(2), pair[1].tail(2))) - writer.write( data=data, show_progress=False, invalid_data_behavior='raise' ) except BcolzMinuteOverlappingData as e: - log.warn('chunk already exists {}: {}'.format(chunk, e)) + log.warn('chunk already exists: {}'.format(e)) + except Exception as e: + log.warn('error when writing data: {}, trying again'.format(e)) + + # This is workaround, there is an issue with empty + # session_label when using a newly created writer + del self._writers[data_frequency] + + # TODO: these are the dates of the chunk, not the job + writer = self.get_writer(start_dt, end_dt, data_frequency) + writer.write( + data=data, + show_progress=False, + invalid_data_behavior='raise' + ) + + return data def ingest(self, data_frequency, include_symbols=None, exclude_symbols=None, start=None, end=None, @@ -327,7 +382,7 @@ class ExchangeBundle: else: raise ValueError('frequency not supported') - writer = self.get_writer(data_frequency, start, end) + writer = self.get_writer(start, end, data_frequency) if delta_periods > self.exchange.num_candles_limit: bar_count = self.exchange.num_candles_limit @@ -359,9 +414,10 @@ class ExchangeBundle: previous_candle = dict() for chunk in it: self.ingest_chunk( - chunk=chunk, - previous_candle=previous_candle, + bar_count=chunk['bar_count'], + end_dt=chunk['end'], data_frequency=data_frequency, assets=assets, - writer=writer + writer=writer, + previous_candle=previous_candle, ) diff --git a/tests/exchange/test_bundle.py b/tests/exchange/test_bundle.py index f0199425..643326a9 100644 --- a/tests/exchange/test_bundle.py +++ b/tests/exchange/test_bundle.py @@ -3,6 +3,7 @@ from logging import Logger import pandas as pd from catalyst.exchange.exchange_bundle import ExchangeBundle +from catalyst.exchange.init_utils import get_exchange log = Logger('test_exchange_bundle') @@ -14,7 +15,7 @@ class ExchangeBundleTestCase: start = pd.to_datetime('2017-09-01', utc=True) end = pd.Timestamp.utcnow() - exchange_bundle = ExchangeBundle(exchange_name) + exchange_bundle = ExchangeBundle(get_exchange(exchange_name)) log.info('ingesting exchange bundle {}'.format(exchange_name)) exchange_bundle.ingest( @@ -33,7 +34,7 @@ class ExchangeBundleTestCase: start = pd.to_datetime('2017-09-01', utc=True) end = pd.Timestamp.utcnow() - exchange_bundle = ExchangeBundle(exchange_name) + exchange_bundle = ExchangeBundle(get_exchange(exchange_name)) log.info('ingesting exchange bundle {}'.format(exchange_name)) exchange_bundle.ingest(