catalyst/catalyst/exchange/exchange_bundle.py

import os
import shutil
from datetime import datetime, timedelta
from functools import partial
from itertools import chain
from operator import is_not

import numpy as np
import pandas as pd
import pytz
from catalyst.assets._assets import TradingPair
from logbook import Logger
from pytz import UTC
from six import itervalues

from catalyst import get_calendar
from catalyst.constants import DATE_TIME_FORMAT, AUTO_INGEST
from catalyst.constants import LOG_LEVEL
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_delta, get_assets
from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \
    BcolzExchangeBarWriter
from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \
    TempBundleNotFoundError, \
    NoDataAvailableOnExchange, \
    PricingDataNotLoadedError, DataCorruptionError, PricingDataValueError
from catalyst.exchange.exchange_utils import get_exchange_folder, \
    save_exchange_symbols, mixin_market_params
from catalyst.utils.cli import maybe_show_progress
from catalyst.utils.paths import ensure_directory

log = Logger('exchange_bundle', level=LOG_LEVEL)

BUNDLE_NAME_TEMPLATE = os.path.join('{root}', '{frequency}_bundle')


def _cachpath(symbol, type_):
    return '-'.join([symbol, type_])


class ExchangeBundle:
    def __init__(self, exchange_name):
        self.exchange_name = exchange_name
        self.minutes_per_day = 1440
        self.default_ohlc_ratio = 1000000
        self._writers = dict()
        self._readers = dict()
        self.calendar = get_calendar('OPEN')
        self.exchange = None

    def get_reader(self, data_frequency, path=None):
        """
        Get a data writer object, either a new object or from cache

        Returns
        -------
        BcolzMinuteBarReader | BcolzDailyBarReader

        """
        if path is None:
            root = get_exchange_folder(self.exchange_name)
            path = BUNDLE_NAME_TEMPLATE.format(
                root=root,
                frequency=data_frequency
            )

        if path in self._readers and self._readers[path] is not None:
            return self._readers[path]

        try:
            self._readers[path] = BcolzExchangeBarReader(
                rootdir=path,
                data_frequency=data_frequency
            )
        except IOError:
            self._readers[path] = None

        return self._readers[path]

    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

        Returns
        -------
        BcolzMinuteBarWriter | BcolzDailyBarWriter

        """
        root = get_exchange_folder(self.exchange_name)
        path = BUNDLE_NAME_TEMPLATE.format(
            root=root,
            frequency=data_frequency
        )

        if path in self._writers:
            return self._writers[path]

        ensure_directory(path)

        if len(os.listdir(path)) > 0:

            metadata = BcolzMinuteBarMetadata.read(path)

            write_metadata = False
            if start_dt < metadata.start_session:
                write_metadata = True
                start_session = start_dt
            else:
                start_session = metadata.start_session

            if end_dt > metadata.end_session:
                write_metadata = True

                end_session = end_dt
            else:
                end_session = metadata.end_session

            self._writers[path] = \
                BcolzExchangeBarWriter(
                    rootdir=path,
                    start_session=start_session,
                    end_session=end_session,
                    write_metadata=write_metadata,
                    data_frequency=data_frequency
                )
        else:
            self._writers[path] = BcolzExchangeBarWriter(
                rootdir=path,
                start_session=start_dt,
                end_session=end_dt,
                write_metadata=True,
                data_frequency=data_frequency
            )

        return self._writers[path]

    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.
            If any data is missing we ingest the data.

        Parameters
        ----------
        assets: list[TradingPair]
            The assets is scope.
        start_dt: pd.Timestamp
            The chunk start date.
        end_dt: pd.Timestamp
            The chunk end date.
        data_frequency: str

        Returns
        -------
        list[TradingPair]
            The assets missing from the bundle
        """
        reader = self.get_reader(data_frequency)
        missing_assets = []
        for asset in assets:
            has_data = range_in_bundle(asset, start_dt, end_dt, reader)

            if not has_data:
                missing_assets.append(asset)

        return missing_assets

    def _write(self, data, writer, data_frequency):
        try:
            writer.write(
                data=data,
                show_progress=False,
                invalid_data_behavior='raise'
            )
        except BcolzMinuteOverlappingData as e:
            log.debug('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[writer._rootdir]

            writer = self.get_writer(writer._start_session,
                                     writer._end_session, data_frequency)
            writer.write(
                data=data,
                show_progress=False,
                invalid_data_behavior='raise'
            )

    def get_calendar_periods_range(self, start_dt, end_dt, data_frequency):
        """
        Get a list of dates for the specified range.

        Parameters
        ----------
        start_dt: pd.Timestamp
        end_dt: pd.Timestamp
        data_frequency: str

        Returns
        -------
        list[datetime]

        """
        return self.calendar.minutes_in_range(start_dt, end_dt) \
            if data_frequency == 'minute' \
            else self.calendar.sessions_in_range(start_dt, end_dt)

    def _spot_empty_periods(self, ohlcv_df, asset, data_frequency,
                            empty_rows_behavior):
        problems = []

        nan_rows = ohlcv_df[ohlcv_df.isnull().T.any().T].index
        if len(nan_rows) > 0:
            dates = []
            for row_date in nan_rows.values:
                row_date = pd.to_datetime(row_date, utc=True)
                if row_date > asset.start_date:
                    dates.append(row_date)

            if len(dates) > 0:
                end_dt = asset.end_minute if data_frequency == 'minute' \
                    else asset.end_daily

                problem = '{name} ({start_dt} to {end_dt}) has empty ' \
                          'periods: {dates}'.format(
                    name=asset.symbol,
                    start_dt=asset.start_date.strftime(DATE_TIME_FORMAT),
                    end_dt=end_dt.strftime(DATE_TIME_FORMAT),
                    dates=[date.strftime(DATE_TIME_FORMAT) for date in dates]
                )
                if empty_rows_behavior == 'warn':
                    log.warn(problem)

                elif empty_rows_behavior == 'raise':
                    raise EmptyValuesInBundleError(
                        name=asset.symbol,
                        end_minute=end_dt,
                        dates=dates
                    )

                else:
                    ohlcv_df.dropna(inplace=True)

            else:
                problem = None

            problems.append(problem)

        return problems

    def _spot_duplicates(self, ohlcv_df, asset, data_frequency, threshold):
        # TODO: work in progress
        series = ohlcv_df.reset_index().groupby('close')['index'].apply(
            np.array
        )

        ref_delta = timedelta(minutes=1) if data_frequency == 'minute' \
            else timedelta(days=1)

        dups = series.loc[lambda values: [len(x) > 10 for x in values]]

        for index, dates in dups.iteritems():
            prev_date = None
            for date in dates:
                if prev_date is not None:
                    delta = (date - prev_date) / 1e9
                    if delta == ref_delta.seconds:
                        log.info('pex')

                prev_date = date

        problems = []
        for index, dates in dups.iteritems():
            end_dt = asset.end_minute if data_frequency == 'minute' \
                else asset.end_daily

            problem = '{name} ({start_dt} to {end_dt}) has {threshold} ' \
                      'identical close values on: {dates}'.format(
                name=asset.symbol,
                start_dt=asset.start_date.strftime(DATE_TIME_FORMAT),
                end_dt=end_dt.strftime(DATE_TIME_FORMAT),
                threshold=threshold,
                dates=[pd.to_datetime(date).strftime(DATE_TIME_FORMAT)
                       for date in dates]
            )

            problems.append(problem)

        return problems

    def ingest_df(self, ohlcv_df, data_frequency, asset, writer,
                  empty_rows_behavior='warn', duplicates_threshold=None):
        """
        Ingest a DataFrame of OHLCV data for a given market.

        Parameters
        ----------
        ohlcv_df: DataFrame
        data_frequency: str
        asset: TradingPair
        writer:
        empty_rows_behavior: str

        """
        problems = []
        if empty_rows_behavior is not 'ignore':
            problems += self._spot_empty_periods(
                ohlcv_df, asset, data_frequency, empty_rows_behavior
            )

        # if duplicates_threshold is not None:
        #     problems += self._spot_duplicates(
        #         ohlcv_df, asset, data_frequency, duplicates_threshold
        #     )

        data = []
        if not ohlcv_df.empty:
            ohlcv_df.sort_index(inplace=True)
            data.append((asset.sid, ohlcv_df))

        self._write(data, writer, data_frequency)

        return problems

    def ingest_ctable(self, asset, data_frequency, period,
                      writer, empty_rows_behavior='strip',
                      duplicates_threshold=100, cleanup=False):
        """
        Merge a ctable bundle chunk into the main bundle for the exchange.

        Parameters
        ----------
        asset: TradingPair
        data_frequency: str
        period: str
        writer:
        empty_rows_behavior: str
            Ensure that the bundle does not have any missing data.

        cleanup: bool
            Remove the temp bundle directory after ingestion.

        Returns
        -------
        list[str]
            A list of problems which occurred during ingestion.

        """
        problems = []

        # Download and extract the bundle
        path = get_bcolz_chunk(
            exchange_name=self.exchange_name,
            symbol=asset.symbol,
            data_frequency=data_frequency,
            period=period
        )

        reader = self.get_reader(data_frequency, path=path)
        if reader is None:
            try:
                log.warn('the reader is unable to use bundle: {}, '
                         'deleting it.'.format(path))
                shutil.rmtree(path)

            except Exception as e:
                log.warn('unable to remove temp bundle: {}'.format(e))

            raise TempBundleNotFoundError(path=path)

        start_dt = reader.first_trading_day
        end_dt = reader.last_available_dt

        if data_frequency == 'daily':
            end_dt = end_dt - pd.Timedelta(hours=23, minutes=59)

        arrays = None
        try:
            arrays = reader.load_raw_arrays(
                sids=[asset.sid],
                fields=['open', 'high', 'low', 'close', 'volume'],
                start_dt=start_dt,
                end_dt=end_dt
            )
        except Exception as e:
            log.warn('skipping ctable for {} from {} to {}: {}'.format(
                asset.symbol, start_dt, end_dt, e
            ))

        if not arrays:
            return reader._rootdir

        periods = self.get_calendar_periods_range(
            start_dt, end_dt, data_frequency
        )
        df = get_df_from_arrays(arrays, periods)
        problems += self.ingest_df(
            ohlcv_df=df,
            data_frequency=data_frequency,
            asset=asset,
            writer=writer,
            empty_rows_behavior=empty_rows_behavior,
            duplicates_threshold=duplicates_threshold
        )

        if cleanup:
            log.debug(
                'removing bundle folder following ingestion: {}'.format(
                    reader._rootdir)
            )
            shutil.rmtree(reader._rootdir)

        return filter(partial(is_not, None), problems)

    def get_adj_dates(self, start, end, assets, data_frequency):
        """
        Contains a date range to the trading availability of the specified
        markets.

        Parameters
        ----------
        start: pd.Timestamp
        end: pd.Timestamp
        assets: list[TradingPair]
        data_frequency: str

        Returns
        -------
        pd.Timestamp, pd.Timestamp
        """
        earliest_trade = None
        last_entry = None
        for asset in assets:
            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:
                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):
            start = earliest_trade

        if end is None or (last_entry is not None and end > last_entry):
            end = last_entry.replace(minute=59, hour=23) \
                if data_frequency == 'minute' else last_entry

        if end is None or start is None or start > end:
            raise NoDataAvailableOnExchange(
                exchange=[asset.exchange for asset in assets],
                symbol=[asset.symbol for asset in assets],
                data_frequency=data_frequency,
            )

        return start, end

    def prepare_chunks(self, assets, data_frequency, start_dt, end_dt):
        """
        Split a price data request into chunks corresponding to individual
        bundles.

        Parameters
        ----------
        assets: list[TradingPair]
        data_frequency: str
        start_dt: pd.Timestamp
        end_dt: pd.Timestamp

        Returns
        -------
        dict[TradingPair, list[dict(str, Object]]]

        """
        get_start_end = get_month_start_end \
            if data_frequency == 'minute' else get_year_start_end

        # Get a reader for the main bundle to verify if data exists
        reader = self.get_reader(data_frequency)

        chunks = dict()
        for asset in assets:
            try:
                # Checking if the the asset has price data in the specified
                # date range
                adj_start, adj_end = self.get_adj_dates(
                    start_dt, end_dt, [asset], data_frequency
                )

            except NoDataAvailableOnExchange as e:
                # If not, we continue to the next asset
                log.debug('skipping {}: {}'.format(asset.symbol, e))
                continue

            dates = pd.date_range(
                start=get_period_label(adj_start, data_frequency),
                end=get_period_label(adj_end, data_frequency),
                freq='MS' if data_frequency == 'minute' else 'AS',
                tz=UTC
            )

            # Adjusting the last date of the range to avoid
            # going over the asset's trading bounds
            dates.values[0] = adj_start
            dates.values[-1] = adj_end

            chunks[asset] = []
            for index, dt in enumerate(dates):
                period_start, period_end = get_start_end(
                    dt=dt,
                    first_day=dt if index == 0 else None,
                    last_day=dt if index == len(dates) - 1 else None
                )

                # Currencies don't always start trading at midnight.
                # Checking the last minute of the day instead.
                range_start = period_start.replace(hour=23, minute=59) \
                    if data_frequency == 'minute' else period_start

                # Checking if the data already exists in the bundle
                # for the date range of the chunk. If not, we create
                # a chunk for ingestion.
                has_data = range_in_bundle(
                    asset, range_start, period_end, reader
                )
                if not has_data:
                    period = get_period_label(dt, data_frequency)
                    chunk = dict(
                        asset=asset,
                        period=period,
                    )
                    chunks[asset].append(chunk)

            # We sort the chunks by end date to ingest most recent data first
            chunks[asset].sort(
                key=lambda chunk: pd.to_datetime(chunk['period'])
            )

        return chunks

    def ingest_assets(self, assets, data_frequency, start_dt=None, end_dt=None,
                      show_progress=False, show_breakdown=False,
                      show_report=False):
        """
        Determine if data is missing from the bundle and attempt to ingest it.

        Parameters
        ----------
        assets: list[TradingPair]
        data_frequency: str
        start_dt: pd.Timestamp
        end_dt: pd.Timestamp
        show_progress: bool
        show_breakdown: bool

        """
        if start_dt is None:
            start_dt = self.calendar.first_session

        if end_dt is None:
            end_dt = pd.Timestamp.utcnow()

        get_start_end = get_month_start_end \
            if data_frequency == 'minute' else get_year_start_end

        # Assign the first and last day of the period
        start_dt, _ = get_start_end(start_dt)
        _, end_dt = get_start_end(end_dt)

        chunks = self.prepare_chunks(
            assets=assets,
            data_frequency=data_frequency,
            start_dt=start_dt,
            end_dt=end_dt
        )

        problems = []
        # This is the common writer for the entire exchange bundle
        # we want to give an end_date far in time
        writer = self.get_writer(start_dt, end_dt, data_frequency)
        if show_breakdown:
            for asset in chunks:
                with maybe_show_progress(
                        chunks[asset],
                        show_progress,
                        label='Ingesting {frequency} price data for '
                              '{symbol} on {exchange}'.format(
                            exchange=self.exchange_name,
                            frequency=data_frequency,
                            symbol=asset.symbol
                        )) as it:
                    for chunk in it:
                        problems += self.ingest_ctable(
                            asset=chunk['asset'],
                            data_frequency=data_frequency,
                            period=chunk['period'],
                            writer=writer,
                            empty_rows_behavior='strip',
                            cleanup=True
                        )
        else:
            all_chunks = list(chain.from_iterable(itervalues(chunks)))

            # We sort the chunks by end date to ingest most recent data first
            all_chunks.sort(
                key=lambda chunk: pd.to_datetime(chunk['period'])
            )
            with maybe_show_progress(
                    all_chunks,
                    show_progress,
                    label='Ingesting {frequency} price data on '
                          '{exchange}'.format(
                        exchange=self.exchange_name,
                        frequency=data_frequency,
                    )) as it:
                for chunk in it:
                    problems += self.ingest_ctable(
                        asset=chunk['asset'],
                        data_frequency=data_frequency,
                        period=chunk['period'],
                        writer=writer,
                        empty_rows_behavior='strip',
                        cleanup=True
                    )

        if show_report and len(problems) > 0:
            log.info('problems during ingestion:{}\n'.format(
                '\n'.join(problems)
            ))

    def ingest_csv(self, path, data_frequency, empty_rows_behavior='strip',
                   duplicates_threshold=100):
        """
        Ingest price data from a CSV file.

        Parameters
        ----------
        path: str
        data_frequency: str

        Returns
        -------
        list[str]
            A list of potential problems detected during ingestion.

        """
        log.info('ingesting csv file: {}'.format(path))

        if self.exchange is None:
            # Avoid circular dependencies
            from catalyst.exchange.factory import get_exchange
            self.exchange = get_exchange(self.exchange_name)

        problems = []
        df = pd.read_csv(
            path,
            header=0,
            sep=',',
            dtype=dict(
                symbol=np.object_,
                last_traded=np.object_,
                open=np.float64,
                high=np.float64,
                close=np.float64,
                volume=np.float64
            ),
            parse_dates=['last_traded'],
            index_col=None
        )
        min_start_dt = None
        max_end_dt = None

        symbols = df['symbol'].unique()

        # Apply the timezone before creating an index for simplicity
        df['last_traded'] = df['last_traded'].dt.tz_localize(pytz.UTC)
        df.set_index(['symbol', 'last_traded'], drop=True, inplace=True)

        assets = dict()
        for symbol in symbols:
            start_dt = df.index.get_level_values(1).min()
            end_dt = df.index.get_level_values(1).max()
            end_dt_key = 'end_{}'.format(data_frequency)

            market = self.exchange.get_market(symbol)
            if market is None:
                raise ValueError('symbol not available in the exchange.')

            params = dict(
                exchange=self.exchange.name,
                data_source='local',
                exchange_symbol=market['id'],
            )
            mixin_market_params(self.exchange_name, params, market)

            asset_def = self.exchange.get_asset_def(market, True)
            if asset_def is not None:
                params['symbol'] = asset_def['symbol']

                params['start_date'] = asset_def['start_date'] \
                    if asset_def['start_date'] < start_dt else start_dt

                params['end_date'] = asset_def[end_dt_key] \
                    if asset_def[end_dt_key] > end_dt else end_dt

                params['end_daily'] = end_dt \
                    if data_frequency == 'daily' else asset_def['end_daily']

                params['end_minute'] = end_dt \
                    if data_frequency == 'minute' else asset_def['end_minute']

            else:
                params['symbol'] = self.exchange.get_catalyst_symbol(market)

                params['end_daily'] = end_dt \
                    if data_frequency == 'daily' else 'N/A'
                params['end_minute'] = end_dt \
                    if data_frequency == 'minute' else 'N/A'

            if min_start_dt is None or start_dt < min_start_dt:
                min_start_dt = start_dt

            if max_end_dt is None or end_dt > max_end_dt:
                max_end_dt = end_dt

            asset = TradingPair(**params)
            assets[market['id']] = asset

        save_exchange_symbols(self.exchange_name, assets, True)

        writer = self.get_writer(
            start_dt=min_start_dt.replace(hour=00, minute=00),
            end_dt=max_end_dt.replace(hour=23, minute=59),
            data_frequency=data_frequency
        )

        for symbol in assets:
            asset = assets[symbol]
            ohlcv_df = df.loc[
                (df.index.get_level_values(0) == symbol)
            ]  # type: pd.DataFrame
            ohlcv_df.index = ohlcv_df.index.droplevel(0)

            period_start = start_dt.replace(hour=00, minute=00)
            period_end = end_dt.replace(hour=23, minute=59)
            periods = self.get_calendar_periods_range(
                period_start, period_end, data_frequency
            )

            # We're not really resampling but ensuring that each frame
            # contains data
            ohlcv_df = ohlcv_df.reindex(periods, method='ffill')
            ohlcv_df['volume'] = ohlcv_df['volume'].fillna(0)

            problems += self.ingest_df(
                ohlcv_df=ohlcv_df,
                data_frequency=data_frequency,
                asset=asset,
                writer=writer,
                empty_rows_behavior=empty_rows_behavior,
                duplicates_threshold=duplicates_threshold
            )
        return filter(partial(is_not, None), problems)

    def ingest(self, data_frequency, include_symbols=None,
               exclude_symbols=None, start=None, end=None, csv=None,
               show_progress=True, show_breakdown=True, show_report=True):
        """
        Inject data based on specified parameters.

        Parameters
        ----------
        data_frequency: str
        include_symbols: str
        exclude_symbols: str
        start: pd.Timestamp
        end: pd.Timestamp
        show_progress: bool
        environ:

        """
        if csv is not None:
            self.ingest_csv(csv, data_frequency)

        else:
            if self.exchange is None:
                # Avoid circular dependencies
                from catalyst.exchange.factory import get_exchange
                self.exchange = get_exchange(self.exchange_name)

            assets = get_assets(
                self.exchange, include_symbols, exclude_symbols
            )
            for frequency in data_frequency.split(','):
                self.ingest_assets(
                    assets=assets,
                    data_frequency=frequency,
                    start_dt=start,
                    end_dt=end,
                    show_progress=show_progress,
                    show_breakdown=show_breakdown,
                    show_report=show_report
                )

    def get_history_window_series_and_load(self,
                                           assets,
                                           end_dt,
                                           bar_count,
                                           field,
                                           data_frequency,
                                           algo_end_dt=None,
                                           trailing_bar_count=None,
                                           force_auto_ingest=False
                                           ):
        """
        Retrieve price data history, ingest missing data.

        Parameters
        ----------
        assets: list[TradingPair]
        end_dt: pd.Timestamp
        bar_count: int
        field: str
        data_frequency: str
        algo_end_dt: pd.Timestamp

        Returns
        -------
        Series

        """
        if AUTO_INGEST or force_auto_ingest:
            try:
                series = self.get_history_window_series(
                    assets=assets,
                    end_dt=end_dt,
                    bar_count=bar_count,
                    field=field,
                    data_frequency=data_frequency,
                    trailing_bar_count=trailing_bar_count,
                )
                return pd.DataFrame(series)

            except PricingDataNotLoadedError:
                start_dt = get_start_dt(end_dt, bar_count, data_frequency)
                log.info(
                    'pricing data for {symbol} not found in range '
                    '{start} to {end}, updating the bundles.'.format(
                        symbol=[asset.symbol for asset in assets],
                        start=start_dt,
                        end=end_dt
                    )
                )
                self.ingest_assets(
                    assets=assets,
                    start_dt=start_dt,
                    end_dt=algo_end_dt,  # TODO: apply trailing bars
                    data_frequency=data_frequency,
                    show_progress=True,
                    show_breakdown=True
                )
                series = self.get_history_window_series(
                    assets=assets,
                    end_dt=end_dt,
                    bar_count=bar_count,
                    field=field,
                    data_frequency=data_frequency,
                    reset_reader=True,
                    trailing_bar_count=trailing_bar_count,
                )
                return series

        else:
            series = self.get_history_window_series(
                assets=assets,
                end_dt=end_dt,
                bar_count=bar_count,
                field=field,
                data_frequency=data_frequency,
                trailing_bar_count=trailing_bar_count,
            )
            return pd.DataFrame(series)

    def get_spot_values(self,
                        assets,
                        field,
                        dt,
                        data_frequency,
                        reset_reader=False
                        ):
        """
        The spot values for the gives assets, field and date. Reads from
        the exchange data bundle.

        Parameters
        ----------
        assets: list[TradingPair]
        field: str
        dt: pd.Timestamp
        data_frequency: str
        reset_reader:

        Returns
        -------
        float

        """
        values = []
        try:
            reader = self.get_reader(data_frequency)
            if reset_reader:
                del self._readers[reader._rootdir]
                reader = self.get_reader(data_frequency)

            for asset in assets:
                value = reader.get_value(
                    sid=asset.sid,
                    dt=dt,
                    field=field
                )
                values.append(value)

            return values

        except Exception:
            symbols = [asset.symbol for asset in assets]
            raise PricingDataNotLoadedError(
                field=field,
                first_trading_day=min([asset.start_date for asset in assets]),
                exchange=self.exchange_name,
                symbols=symbols,
                symbol_list=','.join(symbols),
                data_frequency=data_frequency,
                start_dt=dt,
                end_dt=dt
            )

    def get_history_window_series(self,
                                  assets,
                                  end_dt,
                                  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, _ = 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

        # This is an attempt to resolve some caching with the reader
        # when auto-ingesting data.
        # TODO: needs more work
        reader = self.get_reader(data_frequency)
        if reset_reader:
            del self._readers[reader._rootdir]
            reader = self.get_reader(data_frequency)

        if reader is None:
            symbols = [asset.symbol for asset in assets]
            raise PricingDataNotLoadedError(
                field=field,
                first_trading_day=min([asset.start_date for asset in assets]),
                exchange=self.exchange_name,
                symbols=symbols,
                symbol_list=','.join(symbols),
                data_frequency=data_frequency,
                start_dt=start_dt,
                end_dt=end_dt
            )

        series = dict()
        for asset in assets:
            asset_start_dt, _ = self.get_adj_dates(
                start_dt, end_dt, assets, data_frequency
            )
            in_bundle = range_in_bundle(
                asset, asset_start_dt, end_dt, reader
            )
            if not in_bundle:
                raise PricingDataNotLoadedError(
                    field=field,
                    first_trading_day=asset.start_date,
                    exchange=self.exchange_name,
                    symbols=asset.symbol,
                    symbol_list=asset.symbol,
                    data_frequency=data_frequency,
                    start_dt=asset_start_dt,
                    end_dt=end_dt
                )

            periods = self.get_calendar_periods_range(
                asset_start_dt, end_dt, data_frequency
            )
            # This does not behave well when requesting multiple assets
            # when the start or end date of one asset is outside of the range
            # looking at the logic in load_raw_arrays(), we are not achieving
            # any performance gain by requesting multiple sids at once. It's
            # looping through the sids and making separate requests anyway.
            arrays = reader.load_raw_arrays(
                sids=[asset.sid],
                fields=[field],
                start_dt=start_dt,
                end_dt=end_dt
            )
            if len(arrays) == 0:
                raise DataCorruptionError(
                    exchange=self.exchange_name,
                    symbols=asset.symbol,
                    start_dt=asset_start_dt,
                    end_dt=end_dt
                )

            field_values = arrays[0][:, 0]

            try:
                value_series = pd.Series(field_values, index=periods)
                series[asset] = value_series
            except ValueError as e:
                raise PricingDataValueError(
                    exchange=asset.exchange,
                    symbol=asset.symbol,
                    start_dt=asset_start_dt,
                    end_dt=end_dt,
                    error=e
                )

        return series

    def clean(self, data_frequency):
        """
        Removing the bundle data from the catalyst folder.

        Parameters
        ----------
        data_frequency: str

        """
        log.debug('cleaning exchange {}, frequency {}'.format(
            self.exchange_name, data_frequency
        ))
        root = get_exchange_folder(self.exchange_name)

        symbols = os.path.join(root, 'symbols.json')
        if os.path.isfile(symbols):
            os.remove(symbols)

        local_symbols = os.path.join(root, 'symbols_local.json')
        if os.path.isfile(local_symbols):
            os.remove(local_symbols)

        temp_bundles = os.path.join(root, 'temp_bundles')

        if os.path.isdir(temp_bundles):
            log.debug('removing folder and content: {}'.format(temp_bundles))
            shutil.rmtree(temp_bundles)
            log.debug('{} removed'.format(temp_bundles))

        frequencies = ['daily', 'minute'] if data_frequency is None \
            else [data_frequency]

        for frequency in frequencies:
            label = '{}_bundle'.format(frequency)
            frequency_bundle = os.path.join(root, label)

            if os.path.isdir(frequency_bundle):
                log.debug(
                    'removing folder and content: {}'.format(frequency_bundle)
                )
                shutil.rmtree(frequency_bundle)
                log.debug('{} removed'.format(frequency_bundle))