From 0af592a5f47f98cb7753203210f7ec72899ecb32 Mon Sep 17 00:00:00 2001 From: fredfortier Date: Mon, 20 Nov 2017 17:52:29 -0500 Subject: [PATCH] BUG: fixed issue #71 with the last candle of a resampled set --- catalyst/constants.py | 2 +- catalyst/exchange/exchange_bundle.py | 24 ++-- catalyst/exchange/exchange_data_portal.py | 2 + catalyst/exchange/exchange_utils.py | 4 +- catalyst/exchange/validator.py | 145 ++++++++++++++++++++++ tests/exchange/test_bundle.py | 6 +- tests/exchange/test_data_portal.py | 45 +------ tests/exchange/test_poloniex.py | 7 +- tests/exchange/test_utils.py | 40 ++++++ 9 files changed, 219 insertions(+), 56 deletions(-) create mode 100644 catalyst/exchange/validator.py diff --git a/catalyst/constants.py b/catalyst/constants.py index cde29914..bbdae343 100644 --- a/catalyst/constants.py +++ b/catalyst/constants.py @@ -2,7 +2,7 @@ import logbook -LOG_LEVEL = logbook.INFO +LOG_LEVEL = logbook.DEBUG DATE_TIME_FORMAT = '%Y-%m-%d %H:%M' diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index 48d3294f..6c36fa9e 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/catalyst/exchange/exchange_bundle.py @@ -1,5 +1,6 @@ import os import shutil +from datetime import datetime, timedelta from functools import partial from itertools import chain from operator import is_not @@ -7,7 +8,6 @@ from operator import is_not import numpy as np import pandas as pd from catalyst.assets._assets import TradingPair -from datetime import datetime, timedelta from logbook import Logger from pytz import UTC from six import itervalues @@ -19,7 +19,8 @@ from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \ BcolzMinuteBarMetadata from catalyst.exchange.bundle_utils import range_in_bundle, \ get_bcolz_chunk, get_month_start_end, \ - get_year_start_end, get_df_from_arrays, get_start_dt, get_period_label + get_year_start_end, get_df_from_arrays, get_start_dt, get_period_label, \ + get_delta from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \ BcolzExchangeBarWriter from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \ @@ -682,7 +683,8 @@ class ExchangeBundle: bar_count, field, data_frequency, - algo_end_dt=None + algo_end_dt=None, + trailing_bar_count=None ): """ Retrieve price data history, ingest missing data. @@ -708,7 +710,8 @@ class ExchangeBundle: end_dt=end_dt, bar_count=bar_count, field=field, - data_frequency=data_frequency + data_frequency=data_frequency, + trailing_bar_count=trailing_bar_count ) return pd.DataFrame(series) @@ -725,7 +728,7 @@ class ExchangeBundle: self.ingest_assets( assets=assets, start_dt=start_dt, - end_dt=algo_end_dt, + end_dt=algo_end_dt, # TODO: apply trailing bars data_frequency=data_frequency, show_progress=True, show_breakdown=True @@ -736,7 +739,8 @@ class ExchangeBundle: bar_count=bar_count, field=field, data_frequency=data_frequency, - reset_reader=True + reset_reader=True, + trailing_bar_count=trailing_bar_count ) return series @@ -746,7 +750,8 @@ class ExchangeBundle: end_dt=end_dt, bar_count=bar_count, field=field, - data_frequency=data_frequency + data_frequency=data_frequency, + trailing_bar_count=trailing_bar_count ) return pd.DataFrame(series) @@ -810,12 +815,17 @@ class ExchangeBundle: bar_count, field, data_frequency, + trailing_bar_count=None, reset_reader=False): start_dt = get_start_dt(end_dt, bar_count, data_frequency, False) start_dt, end_dt = self.get_adj_dates( start_dt, end_dt, assets, data_frequency ) + if trailing_bar_count: + delta = get_delta(trailing_bar_count, data_frequency) + end_dt += delta + reader = self.get_reader(data_frequency) if reset_reader: del self._readers[reader._rootdir] diff --git a/catalyst/exchange/exchange_data_portal.py b/catalyst/exchange/exchange_data_portal.py index 43feaa9c..c5d7ac85 100644 --- a/catalyst/exchange/exchange_data_portal.py +++ b/catalyst/exchange/exchange_data_portal.py @@ -332,6 +332,7 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): frequency, data_frequency ) adj_bar_count = candle_size * bar_count + trailing_bar_count = candle_size - 1 if data_frequency == 'minute' and adj_data_frequency == 'daily': end_dt = end_dt.floor('1D') @@ -343,6 +344,7 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): field=field, data_frequency=adj_data_frequency, algo_end_dt=self._last_available_session, + trailing_bar_count=trailing_bar_count ) df = resample_history_df(pd.DataFrame(series), freq, field) diff --git a/catalyst/exchange/exchange_utils.py b/catalyst/exchange/exchange_utils.py index 554825b4..43144a0d 100644 --- a/catalyst/exchange/exchange_utils.py +++ b/catalyst/exchange/exchange_utils.py @@ -487,6 +487,7 @@ def resample_history_df(df, freq, field): DataFrame """ + print(df.tail(30)) if field == 'open': agg = 'first' elif field == 'high': @@ -500,4 +501,5 @@ def resample_history_df(df, freq, field): else: raise ValueError('Invalid field.') - return df.resample(freq).agg(agg) + resampled_df = df.resample(freq).agg(agg) + return resampled_df diff --git a/catalyst/exchange/validator.py b/catalyst/exchange/validator.py new file mode 100644 index 00000000..2037fd59 --- /dev/null +++ b/catalyst/exchange/validator.py @@ -0,0 +1,145 @@ +import os +import tempfile + +import six +from catalyst.assets._assets import TradingPair, get_calendar +from logbook import Logger + +import pandas as pd +from pandas.util.testing import assert_frame_equal + +from catalyst.constants import LOG_LEVEL +from catalyst.exchange.asset_finder_exchange import AssetFinderExchange +from catalyst.exchange.bundle_utils import get_start_dt +from catalyst.exchange.exchange_data_portal import DataPortalExchangeBacktest +from catalyst.exchange.factory import get_exchange, get_exchanges +from catalyst.utils.paths import ensure_directory +from catalyst.exchange.exchange import Exchange + +log = Logger('Validator', level=LOG_LEVEL) + + +def output_df(df, assets, name=None): + """ + Outputs a price DataFrame to a temp folder. + + Parameters + ---------- + df: pd.DataFrame + assets + name + + Returns + ------- + + """ + if isinstance(assets, TradingPair): + exchange_folder = assets.exchange + asset_folder = assets.symbol + else: + exchange_folder = ','.join([asset.exchange for asset in assets]) + asset_folder = ','.join([asset.symbol for asset in assets]) + + folder = os.path.join( + tempfile.gettempdir(), 'catalyst', exchange_folder, asset_folder + ) + ensure_directory(folder) + + if name is None: + name = 'output' + + path = os.path.join(folder, '{}.csv'.format(name)) + df.to_csv(path) + + return path + + +class Validator(object): + def __init__(self, data_portal): + self.data_portal = data_portal + + def compare_bundle_with_exchange(self, exchange, assets, end_dt, bar_count, + sample_minutes): + """ + Creates DataFrames from the bundle and exchange for the specified + data set. + + Parameters + ---------- + exchange: Exchange + assets + end_dt + bar_count + sample_minutes + + Returns + ------- + + """ + freq = '{}T'.format(sample_minutes) + + log.info('creating data sample from bundle') + df1 = self.data_portal.get_history_window( + assets=assets, + end_dt=end_dt, + bar_count=bar_count, + frequency=freq, + field='close', + data_frequency='minute' + ) + path = output_df(df1, assets, '{}_resampled'.format(freq)) + log.info('saved resampled bundle candles: {}\n{}'.format( + path, df1.tail(10)) + ) + + log.info('creating data sample from exchange api') + candles = exchange.get_candles( + end_dt=end_dt, + freq='{}T'.format(sample_minutes), + assets=assets, + bar_count=bar_count + ) + + series = dict() + for asset in assets: + series[asset] = pd.Series( + data=[candle['close'] for candle in candles[asset]], + index=[candle['last_traded'] for candle in candles[asset]] + ) + + df2 = pd.DataFrame(series) + path = output_df(df2, assets, '{}_api'.format(freq)) + log.info('saved exchange api candles: {}\n{}'.format( + path, df2.tail(10)) + ) + + try: + assert_frame_equal(df1, df2) + return True + except: + log.warn('differences found in dataframes') + return False + + +if __name__ == '__main__': + exchanges = get_exchanges(['poloniex']) + exchange = six.next(six.itervalues(exchanges)) + assets = exchange.get_assets(symbols=['eth_btc']) + + open_calendar = get_calendar('OPEN') + asset_finder = AssetFinderExchange() + data_portal = DataPortalExchangeBacktest( + exchanges=exchanges, + asset_finder=asset_finder, + trading_calendar=open_calendar, + first_trading_day=None # will set dynamically based on assets + ) + validator = Validator(data_portal=data_portal) + + validator.compare_bundle_with_exchange( + exchange=exchange, + assets=assets, + end_dt=pd.to_datetime('2017-11-10 1:00', utc=True), + bar_count=200, + sample_minutes=30 + ) diff --git a/tests/exchange/test_bundle.py b/tests/exchange/test_bundle.py index 3966a1e8..cd5f2bde 100644 --- a/tests/exchange/test_bundle.py +++ b/tests/exchange/test_bundle.py @@ -438,7 +438,7 @@ class TestExchangeBundle: pass def main_bundle_to_csv(self): - exchange_name = 'bitfinex' + exchange_name = 'poloniex' data_frequency = 'minute' exchange = get_exchange(exchange_name) @@ -460,8 +460,8 @@ class TestExchangeBundle: def bundle_to_csv(self): exchange_name = 'poloniex' data_frequency = 'minute' - period = '2017-02' - symbol = 'lsk_eth' + period = '2017-01' + symbol = 'eth_btc' exchange = get_exchange(exchange_name) asset = exchange.get_asset(symbol) diff --git a/tests/exchange/test_data_portal.py b/tests/exchange/test_data_portal.py index efd2ffe6..7b2b4720 100644 --- a/tests/exchange/test_data_portal.py +++ b/tests/exchange/test_data_portal.py @@ -1,16 +1,13 @@ import pandas as pd -from catalyst.exchange.exchange_data_portal import DataPortalExchangeBacktest, \ - DataPortalExchangeLive from logbook import Logger -from test_utils import rnd_history_date_days, rnd_bar_count from catalyst import get_calendar from catalyst.exchange.asset_finder_exchange import AssetFinderExchange -from catalyst.exchange.bitfinex.bitfinex import Bitfinex -from catalyst.exchange.bittrex.bittrex import Bittrex -from catalyst.exchange.exchange_utils import get_exchange_auth, \ - get_common_assets +from catalyst.exchange.exchange_data_portal import DataPortalExchangeBacktest, \ + DataPortalExchangeLive +from catalyst.exchange.exchange_utils import get_common_assets from catalyst.exchange.factory import get_exchange, get_exchanges +from test_utils import rnd_history_date_days, rnd_bar_count, output_df log = Logger('test_bitfinex') @@ -115,38 +112,4 @@ class TestExchangeDataPortal: log.info('found history window: {}'.format(data)) def test_validate_resample(self): - symbol = ['eth_btc'] - exchange_name = 'poloniex' - exchange = get_exchange(exchange_name, base_currency=symbol) - - assets = exchange.get_assets(symbols=symbol) - - date = rnd_history_date_days( - max_days=10, - last_dt=pd.to_datetime('2017-11-1', utc=True) - ) - bar_count = rnd_bar_count(max_bars=10) - sample_minutes = 15 - sample_data = self.data_portal_backtest.get_history_window( - assets=assets, - end_dt=date, - bar_count=bar_count, - frequency='{}T'.format(sample_minutes), - field='close', - data_frequency='daily' - ) - minute_data = self.data_portal_backtest.get_history_window( - assets=assets, - end_dt=date, - bar_count=bar_count * sample_minutes, - frequency='1T', - field='close', - data_frequency='daily' - ) - resampled_minute_data = minute_data.resample( - '{}T'.format(sample_minutes)) - - print(sample_data.tail(10)) - print(resampled_minute_data.tail(10)) - print(minute_data.tail(10)) pass diff --git a/tests/exchange/test_poloniex.py b/tests/exchange/test_poloniex.py index 60ff1e65..d84d1010 100644 --- a/tests/exchange/test_poloniex.py +++ b/tests/exchange/test_poloniex.py @@ -54,8 +54,9 @@ class TestPoloniex(BaseExchangeTestCase): log.info('retrieving candles') assets = self.exchange.get_asset('eth_btc') ohlcv = self.exchange.get_candles( - end_dt=pd.to_datetime('2017-11-01', utc=True), - freq='30T', + # end_dt=pd.to_datetime('2017-11-01', utc=True), + end_dt=None, + freq='5T', assets=assets, bar_count=200 ) @@ -63,7 +64,7 @@ class TestPoloniex(BaseExchangeTestCase): df.set_index('last_traded', drop=True, inplace=True) log.info(df.tail(25)) - path = output_df(df, assets, 'candles') + path = output_df(df, assets, '5min_candles') log.info('saved candles: {}'.format(path)) pass diff --git a/tests/exchange/test_utils.py b/tests/exchange/test_utils.py index d7f1df87..ddf76215 100644 --- a/tests/exchange/test_utils.py +++ b/tests/exchange/test_utils.py @@ -1,7 +1,12 @@ +import os +import tempfile from datetime import timedelta from random import randint import pandas as pd +from catalyst.assets._assets import TradingPair + +from catalyst.utils.paths import ensure_directory def rnd_history_date_days(max_days=30, last_dt=None): @@ -24,3 +29,38 @@ def rnd_bar_count(max_bars=21): now = pd.Timestamp.utcnow() return randint(0, max_bars) + + +def output_df(df, assets, name=None): + """ + Outputs a price DataFrame to a temp folder. + + Parameters + ---------- + df: pd.DataFrame + assets + name + + Returns + ------- + + """ + if isinstance(assets, TradingPair): + exchange_folder = assets.exchange + asset_folder = assets.symbol + else: + exchange_folder = ','.join([asset.exchange for asset in assets]) + asset_folder = ','.join([asset.symbol for asset in assets]) + + folder = os.path.join( + tempfile.gettempdir(), 'catalyst', exchange_folder, asset_folder + ) + ensure_directory(folder) + + if name is None: + name = 'output' + + path = os.path.join(folder, '{}.csv'.format(name)) + df.to_csv(path) + + return path