diff --git a/catalyst/__main__.py b/catalyst/__main__.py index d58b5c30..dd15e6f7 100644 --- a/catalyst/__main__.py +++ b/catalyst/__main__.py @@ -38,7 +38,7 @@ except NameError: '--default-extension/--no-default-extension', is_flag=True, default=True, - help="Don't load the default catalyst extension.py file in $ZIPLINE_HOME.", + help="Don't load the default catalyst extension.py file in $CATALYST_HOME.", ) @click.version_option() def main(extension, strict_extensions, default_extension): @@ -495,6 +495,10 @@ def ingest_exchange(exchange_name, data_frequency, start, end, """ Ingest data for the given exchange. """ + + if exchange_name is None: + ctx.fail("must specify an exchange name '-x'") + exchange = get_exchange(exchange_name) exchange_bundle = ExchangeBundle(exchange) diff --git a/catalyst/examples/simple_loop.py b/catalyst/examples/simple_loop.py index d9bf988e..52400485 100644 --- a/catalyst/examples/simple_loop.py +++ b/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' ) diff --git a/catalyst/exchange/bitfinex/bitfinex.py b/catalyst/exchange/bitfinex/bitfinex.py index 47e64cab..ea243a1f 100644 --- a/catalyst/exchange/bitfinex/bitfinex.py +++ b/catalyst/exchange/bitfinex/bitfinex.py @@ -23,7 +23,7 @@ from catalyst.exchange.exchange_errors import ( from catalyst.exchange.exchange_execution import ExchangeLimitOrder, \ ExchangeStopLimitOrder, ExchangeStopOrder from catalyst.exchange.exchange_utils import get_exchange_symbols_filename, \ - download_exchange_symbols + download_exchange_symbols, get_symbols_string from catalyst.finance.order import Order, ORDER_STATUS from catalyst.protocol import Account @@ -255,6 +255,16 @@ class Bitfinex(Exchange): '1m', '5m', '15m', '30m', '1h', '3h', '6h', '12h', '1D', '7D', '14D', '1M' """ + log.debug( + 'retrieving {bars} {freq} candles on {exchange} from ' + '{end_dt} for markets {symbols}, '.format( + bars=bar_count, + freq=data_frequency, + exchange=self.name, + end_dt=end_dt, + symbols=get_symbols_string(assets) + ) + ) freq_match = re.match(r'([0-9].*)(m|h|d)', data_frequency, re.M | re.I) if freq_match: diff --git a/catalyst/exchange/bittrex/bittrex.py b/catalyst/exchange/bittrex/bittrex.py index 1bca9422..1bbfc68b 100644 --- a/catalyst/exchange/bittrex/bittrex.py +++ b/catalyst/exchange/bittrex/bittrex.py @@ -1,6 +1,7 @@ import json import pandas as pd +import time from catalyst.assets._assets import TradingPair from logbook import Logger from six.moves import urllib @@ -13,10 +14,12 @@ from catalyst.exchange.exchange_errors import InvalidHistoryFrequencyError, \ ExchangeRequestError, InvalidOrderStyle, OrderNotFound, OrderCancelError, \ CreateOrderError from catalyst.exchange.exchange_utils import get_exchange_symbols_filename, \ - download_exchange_symbols + download_exchange_symbols, get_symbols_string from catalyst.finance.execution import LimitOrder, StopLimitOrder from catalyst.finance.order import Order, ORDER_STATUS +# TODO: consider using this: https://github.com/mondeja/bittrex_v2 + log = Logger('Bittrex', level=LOG_LEVEL) URL2 = 'https://bittrex.com/Api/v2.0' @@ -217,10 +220,27 @@ class Bittrex(Exchange): :param data_frequency: :param assets: :param bar_count: + :param start_dt + :param end_dt :return: """ - log.info('retrieving candles') + # TODO: this has no effect at the moment + if end_dt is None: + end_dt = pd.Timestamp.utcnow() + + log.debug( + 'retrieving {bars} {freq} candles on {exchange} from ' + '{end_dt} for markets {symbols}, '.format( + bars=bar_count, + freq=data_frequency, + exchange=self.name, + end_dt=end_dt, + symbols=get_symbols_string(assets) + ) + ) + + data_frequency = data_frequency.lower() if data_frequency == 'minute' or data_frequency == '1m': frequency = 'oneMin' elif data_frequency == '5m': @@ -229,7 +249,7 @@ class Bittrex(Exchange): frequency = 'thirtyMin' elif data_frequency == '1h': frequency = 'hour' - elif data_frequency == 'daily' or data_frequency == '1D': + elif data_frequency == 'daily' or data_frequency == '1d': frequency = 'day' else: raise InvalidHistoryFrequencyError( @@ -238,13 +258,14 @@ class Bittrex(Exchange): # Making sure that assets are iterable asset_list = [assets] if isinstance(assets, TradingPair) else assets - ohlc_map = dict() for asset in asset_list: + end = int(time.mktime(end_dt.timetuple())) url = '{url}/pub/market/GetTicks?marketName={symbol}' \ - '&tickInterval={frequency}&_=1499127220008'.format( + '&tickInterval={frequency}&_={end}'.format( url=URL2, symbol=self.get_symbol(asset), - frequency=frequency + frequency=frequency, + end=end ) try: @@ -272,6 +293,7 @@ class Bittrex(Exchange): return ohlc ordered_candles = list(reversed(candles)) + ohlc_map = dict() if bar_count is None: ohlc_map[asset] = ohlc_from_candle(ordered_candles[0]) else: diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index bb4018d5..31037882 100644 --- a/catalyst/exchange/exchange.py +++ b/catalyst/exchange/exchange.py @@ -373,7 +373,7 @@ class Exchange: return value def get_series_from_candles(self, candles, start_dt, end_dt, - field, previous_value=None): + data_frequency, field, previous_value=None): """ Get a series of field data for the specified candles. @@ -388,9 +388,12 @@ class Exchange: dates = [candle['last_traded'] for candle in candles] values = [candle[field] for candle in candles] - periods = pd.date_range(start_dt, end_dt) + periods = self.bundle.get_calendar_periods_range( + start_dt, end_dt, data_frequency + ) series = pd.Series(values, index=dates) + #TODO: ensure that this working as expected, if not use fillna series.reindex(periods, method='ffill', fill_value=previous_value) return series @@ -487,6 +490,7 @@ class Exchange: data_frequency=data_frequency, assets=asset, bar_count=trailing_bar_count, + start_dt=start_dt, end_dt=end_dt ) @@ -497,6 +501,7 @@ class Exchange: candles=candles, start_dt=trailing_dt, end_dt=end_dt, + data_frequency=data_frequency, field=field, previous_value=last_value ) @@ -784,13 +789,14 @@ class Exchange: pass @abc.abstractmethod - def get_orderbook(self, asset, order_type): + def get_orderbook(self, asset, order_type, limit): """ Retrieve the the orderbook for the given trading pair. :param asset: TradingPair :param order_type: str The type of orders: bid, ask or all + :param limit :return: """ diff --git a/catalyst/exchange/exchange_bcolz.py b/catalyst/exchange/exchange_bcolz.py index 209f2d91..326745ac 100644 --- a/catalyst/exchange/exchange_bcolz.py +++ b/catalyst/exchange/exchange_bcolz.py @@ -16,7 +16,7 @@ class BcolzExchangeBarWriter(BcolzMinuteBarWriter): end_session = end_session.floor('1d') minutes_per_day = 1440 if self._data_frequency == 'minute' else 1 - default_ohlc_ratio = kwargs.pop('default_ohlc_ratio', 1000000) + default_ohlc_ratio = kwargs.pop('default_ohlc_ratio', 100000000) calendar = get_calendar('OPEN') super(BcolzExchangeBarWriter, self) \ @@ -79,8 +79,9 @@ class BcolzExchangeBarReader(BcolzMinuteBarReader): if mask is None: mask = a != 0 + inverse_ratio = self._ohlc_ratio_inverse_for_sid(sid) out[:len(mask), i][mask] = ( - a[mask] * self._ohlc_ratio_inverse_for_sid(sid) + a[mask] * inverse_ratio ) if field in fields: diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index b86639aa..353c5fc9 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/catalyst/exchange/exchange_bundle.py @@ -11,7 +11,7 @@ from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \ BcolzMinuteBarMetadata from catalyst.exchange.bundle_utils import range_in_bundle, \ get_bcolz_chunk, get_delta, get_month_start_end, \ - get_year_start_end, get_periods_range, get_df_from_arrays, get_start_dt + get_year_start_end, get_df_from_arrays, get_start_dt from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \ BcolzExchangeBarWriter from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \ @@ -194,6 +194,71 @@ class ExchangeBundle: if data_frequency == 'minute' \ else self.calendar.sessions_in_range(start_dt, end_dt) + def ingest_df(self, ohlcv_df, data_frequency, asset, writer, + empty_rows_behavior='strip'): + """ + Ingest a DataFrame of OHLCV data for a given market. + + :param ohlcv_df: + :param data_frequency: + :param asset: + :param writer: + :param path: + :param empty_rows_behavior: + :return: + """ + if empty_rows_behavior is not 'ignore': + nan_rows = ohlcv_df[ohlcv_df.isnull().T.any().T].index + + if len(nan_rows) > 0: + dates = [] + previous_date = None + for row_date in nan_rows.values: + row_date = pd.to_datetime(row_date) + + if previous_date is None: + dates.append(row_date) + + else: + seq_date = previous_date + get_delta(1, data_frequency) + + if row_date > seq_date: + dates.append(previous_date) + dates.append(row_date) + + previous_date = row_date + + dates.append(pd.to_datetime(nan_rows.values[-1])) + + name = '{} from {} to {}'.format( + asset.symbol, ohlcv_df.index[0], ohlcv_df.index[-1] + ) + if empty_rows_behavior == 'warn': + log.warn( + '\n{name} with end minute {end_minute} has empty rows ' + 'in ranges: {dates}'.format( + name=name, + end_minute=asset.end_minute, + dates=dates + ) + ) + + elif empty_rows_behavior == 'raise': + raise EmptyValuesInBundleError( + name=name, + end_minute=asset.end_minute, + dates=dates + ) + else: + ohlcv_df.dropna(inplace=True) + + data = [] + if not ohlcv_df.empty: + ohlcv_df.sort_index(inplace=True) + data.append((asset.sid, ohlcv_df)) + + self._write(data, writer, data_frequency) + def ingest_ctable(self, asset, data_frequency, period, start_dt, end_dt, writer, empty_rows_behavior='strip', cleanup=False): """ @@ -242,62 +307,19 @@ class ExchangeBundle: 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 - - if len(nan_rows) > 0: - dates = [] - previous_date = None - for row_date in nan_rows.values: - row_date = pd.to_datetime(row_date) - - if previous_date is None: - dates.append(row_date) - - else: - seq_date = previous_date + get_delta(1, data_frequency) - - if row_date > seq_date: - dates.append(previous_date) - dates.append(row_date) - - previous_date = row_date - - dates.append(pd.to_datetime(nan_rows.values[-1])) - - name = path.split('/')[-1] - if empty_rows_behavior == 'warn': - log.warn( - '\n{name} with end minute {end_minute} has empty rows ' - 'in ranges: {dates}'.format( - name=name, - end_minute=asset.end_minute, - dates=dates - ) - ) - - elif empty_rows_behavior == 'raise': - raise EmptyValuesInBundleError( - name=name, - end_minute=asset.end_minute, - dates=dates - ) - else: - df.dropna(inplace=True) - - data = [] - if not df.empty: - df.sort_index(inplace=True) - data.append((asset.sid, df)) - - self._write(data, writer, data_frequency) + self.ingest_df( + ohlcv_df=df, + data_frequency=data_frequency, + asset=asset, + writer=writer, + empty_rows_behavior=empty_rows_behavior + ) if cleanup: - log.debug('removing bundle folder following ' - 'ingestion: {}'.format(path)) + log.debug( + 'removing bundle folder following ingestion: {}'.format(path) + ) shutil.rmtree(path) return path @@ -315,15 +337,22 @@ class ExchangeBundle: earliest_trade = None last_entry = None for asset in assets: - if (earliest_trade is None or earliest_trade > asset.start_date) \ - and asset.start_date >= self.calendar.first_session: - earliest_trade = asset.start_date + if earliest_trade is None or earliest_trade > asset.start_date: + if asset.start_date >= self.calendar.first_session: + earliest_trade = asset.start_date + + else: + earliest_trade = self.calendar.first_session end_asset = asset.end_minute if data_frequency == 'minute' else \ asset.end_daily - if end_asset is not None and \ - (last_entry is None or end_asset > last_entry): - last_entry = end_asset + if end_asset is not None: + if last_entry is None or end_asset > last_entry: + last_entry = end_asset + + else: + end = None + last_entry = None if start is None or \ (earliest_trade is not None and earliest_trade > start): @@ -363,8 +392,9 @@ class ExchangeBundle: start_dt, end_dt, [asset], data_frequency ) - except NoDataAvailableOnExchange: + except NoDataAvailableOnExchange as e: # If not, we continue to the next asset + log.debug('skipping {}: {}'.format(asset.symbol, e)) continue # This is either the first trading day of the asset or the @@ -391,11 +421,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 +439,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 ) @@ -427,11 +447,11 @@ class ExchangeBundle: period_start = first_trading_dt _, asset_end_year = get_year_start_end( - asset.end_minute + asset.end_daily ) if asset_end_year == period_end \ - and period_end > asset.end_minute: - period_end = asset.end_minute + and period_end > asset.end_daily: + period_end = asset.end_daily else: raise InvalidHistoryFrequencyError( diff --git a/catalyst/exchange/exchange_utils.py b/catalyst/exchange/exchange_utils.py index 2e3982f0..77e0fc56 100644 --- a/catalyst/exchange/exchange_utils.py +++ b/catalyst/exchange/exchange_utils.py @@ -1,6 +1,8 @@ import json import os import pickle + +from catalyst.assets._assets import TradingPair from six.moves.urllib import request from datetime import date, datetime @@ -57,6 +59,11 @@ def get_exchange_symbols(exchange_name, environ=None): ) +def get_symbols_string(assets): + array = [assets] if isinstance(assets, TradingPair) else assets + return ', '.join([asset.symbol for asset in array]) + + def get_exchange_auth(exchange_name, environ=None): exchange_folder = get_exchange_folder(exchange_name, environ) filename = os.path.join(exchange_folder, 'auth.json') diff --git a/catalyst/exchange/poloniex/poloniex.py b/catalyst/exchange/poloniex/poloniex.py index 76d51332..468463b0 100644 --- a/catalyst/exchange/poloniex/poloniex.py +++ b/catalyst/exchange/poloniex/poloniex.py @@ -22,7 +22,7 @@ from catalyst.exchange.exchange_errors import ( from catalyst.exchange.exchange_execution import ExchangeLimitOrder, \ ExchangeStopLimitOrder from catalyst.exchange.exchange_utils import get_exchange_symbols_filename, \ - download_exchange_symbols + download_exchange_symbols, get_symbols_string from catalyst.exchange.poloniex.poloniex_api import Poloniex_api from catalyst.finance.order import Order, ORDER_STATUS from catalyst.finance.transaction import Transaction @@ -186,22 +186,35 @@ 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' + log.debug( + 'retrieving {bars} {freq} candles on {exchange} from ' + '{end_dt} for markets {symbols}, '.format( + bars=bar_count, + freq=data_frequency, + exchange=self.name, + end_dt=end_dt, + symbols=get_symbols_string(assets) + ) + ) + + if data_frequency == '5m': frequency = 300 - elif (data_frequency == '15m'): + elif data_frequency == '15m': frequency = 900 - elif (data_frequency == '30m'): + elif data_frequency == '30m': frequency = 1800 - elif (data_frequency == '2h'): + elif data_frequency == '2h': frequency = 7200 - elif (data_frequency == '4h'): + elif data_frequency == '4h': frequency = 14400 - elif (data_frequency == '1D' or data_frequency == 'daily'): + elif data_frequency == '1D' or data_frequency == 'daily': frequency = 86400 else: + # Poloniex does not offer 1m data candles + # It is likely to error out there frequently raise InvalidHistoryFrequencyError( frequency=data_frequency ) @@ -212,8 +225,8 @@ class Poloniex(Exchange): for asset in asset_list: - end = int(time.time()) - if (bar_count is None): + end = int(time.mktime(end_dt.timetuple())) + if bar_count is None: start = end - 2 * frequency else: start = end - bar_count * frequency diff --git a/catalyst/exchange/stats_utils.py b/catalyst/exchange/stats_utils.py index 2cf65b43..602694d6 100644 --- a/catalyst/exchange/stats_utils.py +++ b/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() diff --git a/tests/exchange/test_bittrex.py b/tests/exchange/test_bittrex.py index bbf52fa1..779c2a08 100644 --- a/tests/exchange/test_bittrex.py +++ b/tests/exchange/test_bittrex.py @@ -1,3 +1,4 @@ +import pandas as pd from catalyst.exchange.bittrex.bittrex import Bittrex from catalyst.finance.order import Order from base import BaseExchangeTestCase @@ -7,15 +8,15 @@ from catalyst.exchange.exchange_utils import get_exchange_auth log = Logger('test_bittrex') -class TestBittrexTestCase(BaseExchangeTestCase): +class TestBittrex(BaseExchangeTestCase): @classmethod def setup(self): - print ('creating bittrex object') auth = get_exchange_auth('bittrex') self.exchange = Bittrex( key=auth['key'], secret=auth['secret'], - base_currency='btc' + base_currency=None, + portfolio=None ) def test_order(self): @@ -52,15 +53,18 @@ class TestBittrexTestCase(BaseExchangeTestCase): log.info('retrieving candles') ohlcv_neo = self.exchange.get_candles( data_frequency='5m', - assets=self.exchange.get_asset('neo_btc') + assets=self.exchange.get_asset('neo_btc'), + bar_count=20, + end_dt=pd.to_datetime('2017-10-20', utc=True) ) ohlcv_neo_ubq = self.exchange.get_candles( - data_frequency='5m', + data_frequency='1d', assets=[ self.exchange.get_asset('neo_btc'), self.exchange.get_asset('ubq_btc') ], - bar_count=14 + bar_count=14, + end_dt=pd.to_datetime('2017-10-20', utc=True) ) pass diff --git a/tests/exchange/test_bundle.py b/tests/exchange/test_bundle.py index bd79c16b..9c39e4e4 100644 --- a/tests/exchange/test_bundle.py +++ b/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') @@ -44,11 +45,11 @@ class TestExchangeBundle: exchange = get_exchange(exchange_name) exchange_bundle = ExchangeBundle(exchange) assets = [ - exchange.get_asset('neo_eth') + exchange.get_asset('iot_btc') ] # start = pd.to_datetime('2017-09-01', utc=True) - start = pd.to_datetime('2017-9-15', utc=True) + start = pd.to_datetime('2017-9-01', utc=True) end = pd.to_datetime('2017-9-30', utc=True) log.info('ingesting exchange bundle {}'.format(exchange_name)) @@ -315,3 +316,113 @@ class TestExchangeBundle: hashlib.sha256(symbol.encode('utf-8')).hexdigest(), 16 ) % 10 ** 6 pass + + def test_validate_data(self): + exchange_name = 'bitfinex' + data_frequency = 'minute' + + exchange = get_exchange(exchange_name) + exchange_bundle = ExchangeBundle(exchange) + assets = [exchange.get_asset('iot_btc')] + + end_dt = pd.to_datetime('2017-9-2 1:00', utc=True) + bar_count = 60 + + 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 + ) + exchange_series = exchange.get_series_from_candles( + candles=candles[asset], + start_dt=start_dt, + end_dt=end_dt, + data_frequency=data_frequency, + 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 + + def test_ingest_candles(self): + exchange_name = 'bitfinex' + data_frequency = 'minute' + + exchange = get_exchange(exchange_name) + bundle = ExchangeBundle(exchange) + assets = [exchange.get_asset('iot_btc')] + + end_dt = pd.to_datetime('2017-10-20', utc=True) + bar_count = 100 + + start_dt = get_start_dt(end_dt, bar_count, data_frequency) + candles = exchange.get_candles( + assets=assets, + start_dt=start_dt, + end_dt=end_dt, + bar_count=bar_count, + data_frequency=data_frequency + ) + + writer = bundle.get_writer(start_dt, end_dt, data_frequency) + for asset in assets: + dates = [candle['last_traded'] for candle in candles[asset]] + + values = dict() + for field in ['open', 'high', 'low', 'close', 'volume']: + values[field] = [candle[field] for candle in candles[asset]] + + periods = bundle.get_calendar_periods_range( + start_dt, end_dt, data_frequency + ) + df = pd.DataFrame(values, index=dates) + df = df.loc[periods].fillna(method='ffill') + + # TODO: why do I get an extra bar? + bundle.ingest_df( + ohlcv_df=df, + data_frequency=data_frequency, + asset=asset, + writer=writer, + empty_rows_behavior='raise' + ) + + bundle_series = bundle.get_history_window_series( + assets=assets, + end_dt=end_dt, + bar_count=bar_count, + field='close', + data_frequency=data_frequency, + reset_reader=True + ) + df = pd.DataFrame(bundle_series) + print('\n' + df_to_string(df)) + pass