BUG: fixed issue #71 with the last candle of a resampled set

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'
 

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]

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)

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

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
+    )

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)

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

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
 

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