Poloniex get_candles fix and created a unit test to validate data.

catalyst/examples/simple_loop.py

@@ -7,7 +7,7 @@ from catalyst.api import symbol
 
 def initialize(context):
     print('initializing')
-    context.asset = symbol('btc_usd')
+    context.asset = symbol('burst_btc')
 
 
 def handle_data(context, data):
@@ -20,7 +20,7 @@ def handle_data(context, data):
         context.asset,
         fields='price',
         bar_count=15,
-        frequency='1m'
+        frequency='1d'
     )
     rsi = talib.RSI(prices.values, timeperiod=14)[-1]
     print('got rsi: {}'.format(rsi))
@@ -35,7 +35,7 @@ run_algorithm(
     initialize=initialize,
     handle_data=handle_data,
     analyze=None,
-    exchange_name='bitfinex',
+    exchange_name='poloniex',
     algo_namespace='simple_loop',
     base_currency='btc'
 )

catalyst/exchange/exchange_bundle.py

@@ -391,11 +391,6 @@ class ExchangeBundle:
                     if data_frequency == 'minute':
                         period_start, period_end = get_month_start_end(dt)
 
-                        # TODO: redundant gate, we are already filtering dates
-                        if first_trading_dt > period_start:
-                            dt += timedelta(days=1)
-                            continue
-
                         asset_start_month, _ = get_month_start_end(
                             first_trading_dt
                         )
@@ -414,11 +409,6 @@ class ExchangeBundle:
                     elif data_frequency == 'daily':
                         period_start, period_end = get_year_start_end(dt)
 
-                        # TODO: redundant gate, we are already filtering dates
-                        if first_trading_dt > period_start:
-                            dt += timedelta(days=1)
-                            continue
-
                         asset_start_year, _ = get_year_start_end(
                             first_trading_dt
                         )

catalyst/exchange/poloniex/poloniex.py

@@ -186,7 +186,8 @@ class Poloniex(Exchange):
         '5m', '15m', '30m', '2h', '4h', '1D'
         """
 
-        # TODO: implement end_dt and start_dt filters
+        if end_dt is None:
+            end_dt = pd.Timestamp.utcnow()
 
         if (
                         data_frequency == '5m' or data_frequency == 'minute'):  # TODO: Polo does not have '1m'
@@ -212,8 +213,8 @@ class Poloniex(Exchange):
 
         for asset in asset_list:
 
-            end = int(time.time())
+            end = int(time.mktime(end_dt.timetuple()))
-            if (bar_count is None):
+            if bar_count is None:
                 start = end - 2 * frequency
             else:
                 start = end - bar_count * frequency

catalyst/exchange/stats_utils.py

@@ -49,3 +49,12 @@ def get_pretty_stats(stats_df, recorded_cols=None, num_rows=10):
         columns=columns,
         formatters=formatters
     )
+
+
+def df_to_string(df):
+    pd.set_option('display.expand_frame_repr', False)
+    pd.set_option('precision', 8)
+    pd.set_option('display.width', 1000)
+    pd.set_option('display.max_colwidth', 1000)
+
+    return df.to_string()

tests/exchange/test_bundle.py

@@ -5,13 +5,14 @@ import pandas as pd
 
 from catalyst import get_calendar
 from catalyst.exchange.bundle_utils import get_bcolz_chunk, \
-    get_periods_range
+    get_periods_range, get_start_dt
 from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \
     BcolzExchangeBarWriter
 from catalyst.exchange.exchange_bundle import ExchangeBundle, \
     BUNDLE_NAME_TEMPLATE
 from catalyst.exchange.exchange_utils import get_exchange_folder
 from catalyst.exchange.init_utils import get_exchange
+from catalyst.exchange.stats_utils import df_to_string
 from catalyst.utils.paths import ensure_directory
 
 log = getLogger('test_exchange_bundle')
@@ -39,12 +40,12 @@ class TestExchangeBundle:
 
     def test_ingest_minute(self):
         data_frequency = 'minute'
-        exchange_name = 'bitfinex'
+        exchange_name = 'poloniex'
 
         exchange = get_exchange(exchange_name)
         exchange_bundle = ExchangeBundle(exchange)
         assets = [
-            exchange.get_asset('neo_eth')
+            exchange.get_asset('burst_btc')
         ]
 
         # start = pd.to_datetime('2017-09-01', utc=True)
@@ -315,3 +316,59 @@ class TestExchangeBundle:
             hashlib.sha256(symbol.encode('utf-8')).hexdigest(), 16
         ) % 10 ** 6
         pass
+
+    def test_validate_data(self):
+        exchange_name = 'poloniex'
+        data_frequency = 'minute'
+
+        exchange = get_exchange(exchange_name)
+        exchange_bundle = ExchangeBundle(exchange)
+        assets = [exchange.get_asset('neos_btc')]
+
+        end_dt = pd.to_datetime('2017-10-20', utc=True)
+        bar_count = 100
+
+        bundle_series = exchange_bundle.get_history_window_series(
+            assets=assets,
+            end_dt=end_dt,
+            bar_count=bar_count * 5,
+            field='close',
+            data_frequency='minute',
+        )
+        candles = exchange.get_candles(
+            assets=assets,
+            end_dt=end_dt,
+            bar_count=bar_count,
+            data_frequency='minute'
+        )
+        start_dt = get_start_dt(end_dt, bar_count, data_frequency)
+
+        frames = []
+        for asset in assets:
+            bundle_df = pd.DataFrame(
+                data=dict(bundle_price=bundle_series[asset]),
+                index=bundle_series[asset].index
+            )
+            bundle_df = bundle_df.resample('5T').last()
+
+            exchange_series = exchange.get_series_from_candles(
+                candles=candles[asset],
+                start_dt=start_dt,
+                end_dt=end_dt,
+                field='close'
+            )
+            exchange_df = pd.DataFrame(
+                data=dict(exchange_price=exchange_series),
+                index=exchange_series.index
+            )
+
+            df = exchange_df.join(bundle_df, how='left')
+            df['last_traded'] = df.index
+            df['asset'] = asset.symbol
+            df.set_index(['asset', 'last_traded'], inplace=True)
+
+            frames.append(df)
+
+        df = pd.concat(frames)
+        print('\n' + df_to_string(df))
+        pass