BLD: refactoring to decrease reliance on the Exchange in preparation to support ad-hoc CSV bundles

catalyst/__main__.py

@@ -499,6 +499,13 @@ def live(ctx,
     help='A list of symbols to exclude from the ingestion '
          '(optional comma separated list)',
 )
+@click.option(
+    '--csv',
+    default=None,
+    help='The path of a CSV file containing the data. If specified, start, '
+         'end, include-symbols and exclude-symbols will be ignored. Instead,'
+         'all data in the file will be ingested.',
+)
 @click.option(
     '--show-progress/--no-show-progress',
     default=True,
@@ -515,8 +522,8 @@ def live(ctx,
     help='Report potential anomalies found in data bundles.'
 )
 def ingest_exchange(exchange_name, data_frequency, start, end,
-                    include_symbols, exclude_symbols, show_progress, verbose,
-                    validate):
+                    include_symbols, exclude_symbols, csv, show_progress,
+                    verbose, validate):
     """
     Ingest data for the given exchange.
     """
@@ -524,8 +531,7 @@ def ingest_exchange(exchange_name, data_frequency, start, end,
     if exchange_name is None:
         ctx.fail("must specify an exchange name '-x'")
 
-    exchange = get_exchange(exchange_name)
-    exchange_bundle = ExchangeBundle(exchange)
+    exchange_bundle = ExchangeBundle(exchange_name)
 
     click.echo('Ingesting exchange bundle {}...'.format(exchange_name))
     exchange_bundle.ingest(
@@ -536,7 +542,8 @@ def ingest_exchange(exchange_name, data_frequency, start, end,
         end=end,
         show_progress=show_progress,
         show_breakdown=verbose,
-        show_report=validate
+        show_report=validate,
+        csv=csv
     )
 
 
@@ -579,8 +586,7 @@ def clean_exchange(ctx, exchange_name, data_frequency):
     if exchange_name is None:
         ctx.fail("must specify an exchange name '-x'")
 
-    exchange = get_exchange(exchange_name)
-    exchange_bundle = ExchangeBundle(exchange)
+    exchange_bundle = ExchangeBundle(exchange_name)
 
     click.echo('Cleaning exchange bundle {}...'.format(exchange_name))
     exchange_bundle.clean(

catalyst/constants.py

@@ -2,7 +2,7 @@
 
 import logbook
 
-LOG_LEVEL = logbook.DEBUG
+LOG_LEVEL = logbook.INFO
 
 DATE_TIME_FORMAT = '%Y-%m-%d %H:%M'
 

catalyst/examples/mean_reversion_simple.py

@@ -1,6 +1,9 @@
 # For this example, we're going to write a simple momentum script.  When the 
 # stock goes up quickly, we're going to buy; when it goes down quickly, we're
 # going to sell.  Hopefully we'll ride the waves.
+import os
+import tempfile
+import time
 
 import pandas as pd
 import talib
@@ -9,14 +12,16 @@ from logbook import Logger
 from catalyst import run_algorithm
 from catalyst.api import symbol, record, order_target_percent, get_open_orders
 from catalyst.exchange.stats_utils import extract_transactions
-
 # We give a name to the algorithm which Catalyst will use to persist its state.
 # In this example, Catalyst will create the `.catalyst/data/live_algos`
 # directory. If we stop and start the algorithm, Catalyst will resume its
 # state using the files included in the folder.
+from catalyst.utils.paths import ensure_directory
+
 NAMESPACE = 'mean_reversion_simple'
 log = Logger(NAMESPACE)
 
+
 # To run an algorithm in Catalyst, you need two functions: initialize and
 # handle_data.
 
@@ -27,10 +32,16 @@ def initialize(context):
     # parameters or values you're going to use.
 
     # In our example, we're looking at Ether in USD Tether.
-    context.neo_usd = symbol('neo_btc')
+    context.neo_eth = symbol('neo_eth')
     context.base_price = None
     context.current_day = None
 
+    context.RSI_OVERSOLD = 50
+    context.RSI_OVERBOUGHT = 80
+    context.CANDLE_SIZE = '5T'
+
+    context.start_time = time.time()
+
 
 def handle_data(context, data):
     # This handle_data function is where the real work is done.  Our data is
@@ -47,17 +58,17 @@ def handle_data(context, data):
         context.current_day = today
 
     # We're computing the volume-weighted-average-price of the security
-    # defined above, in the context.neo_usd variable.  For this example, we're 
+    # defined above, in the context.neo_eth variable.  For this example, we're 
     # using three bars on the 15 min bars.
 
     # The frequency attribute determine the bar size. We use this convention
     # for the frequency alias:
     # http://pandas.pydata.org/pandas-docs/stable/timeseries.html#offset-aliases
     prices = data.history(
-        context.neo_usd,
+        context.neo_eth,
         fields='close',
         bar_count=50,
-        frequency='30T'
+        frequency=context.CANDLE_SIZE
     )
 
     # Ta-lib calculates various technical indicator based on price and
@@ -69,7 +80,7 @@ def handle_data(context, data):
     # We need a variable for the current price of the security to compare to
     # the average. Since we are requesting two fields, data.current()
     # returns a DataFrame with
-    current = data.current(context.neo_usd, fields=['close', 'volume'])
+    current = data.current(context.neo_eth, fields=['close', 'volume'])
     price = current['close']
 
     # If base_price is not set, we use the current value. This is the
@@ -98,42 +109,51 @@ def handle_data(context, data):
 
     # Since we are using limit orders, some orders may not execute immediately
     # we wait until all orders are executed before considering more trades.
-    orders = get_open_orders(context.neo_usd)
+    orders = get_open_orders(context.neo_eth)
     if len(orders) > 0:
         return
 
     # Exit if we cannot trade
-    if not data.can_trade(context.neo_usd):
+    if not data.can_trade(context.neo_eth):
         return
 
     # Another powerful built-in feature of the Catalyst backtester is the
     # portfolio object.  The portfolio object tracks your positions, cash,
     # cost basis of specific holdings, and more.  In this line, we calculate
     # how long or short our position is at this minute.   
-    pos_amount = context.portfolio.positions[context.neo_usd].amount
+    pos_amount = context.portfolio.positions[context.neo_eth].amount
 
-    if rsi[-1] <= 30 and pos_amount == 0:
+    if rsi[-1] <= context.RSI_OVERSOLD and pos_amount == 0:
         log.info(
             '{}: buying - price: {}, rsi: {}'.format(
                 data.current_dt, price, rsi[-1]
             )
         )
-        order_target_percent(context.neo_usd, 1)
+        # Set a style for limit orders,
+        limit_price = price * 1.005
+        order_target_percent(
+            context.neo_eth, 1, limit_price=limit_price
+        )
         context.traded_today = True
 
-    elif rsi[-1] >= 80 and pos_amount > 0:
+    elif rsi[-1] >= context.RSI_OVERBOUGHT and pos_amount > 0:
         log.info(
             '{}: selling - price: {}, rsi: {}'.format(
                 data.current_dt, price, rsi[-1]
             )
         )
-        order_target_percent(context.neo_usd, 0)
+        limit_price = price * 0.995
+        order_target_percent(
+            context.neo_eth, 0, limit_price=limit_price
+        )
         context.traded_today = True
 
 
 def analyze(context=None, perf=None):
-    import matplotlib.pyplot as plt
+    end = time.time()
+    log.info('elapsed time: {}'.format(end - context.start_time))
 
+    import matplotlib.pyplot as plt
     # The base currency of the algo exchange
     base_currency = context.exchanges.values()[0].base_currency.upper()
 
@@ -147,7 +167,7 @@ def analyze(context=None, perf=None):
     perf.loc[:, 'price'].plot(ax=ax2, label='Price')
 
     ax2.set_ylabel('{asset} ({base})'.format(
-        asset=context.neo_usd.symbol, base=base_currency
+        asset=context.neo_eth.symbol, base=base_currency
     ))
 
     transaction_df = extract_transactions(perf)
@@ -156,7 +176,7 @@ def analyze(context=None, perf=None):
         sell_df = transaction_df[transaction_df['amount'] < 0]
         ax2.scatter(
             buy_df.index.to_pydatetime(),
-            perf.loc[buy_df.index, 'price'],
+            perf.loc[buy_df.index.floor('1 min'), 'price'],
             marker='^',
             s=100,
             c='green',
@@ -164,7 +184,7 @@ def analyze(context=None, perf=None):
         )
         ax2.scatter(
             sell_df.index.to_pydatetime(),
-            perf.loc[sell_df.index, 'price'],
+            perf.loc[sell_df.index.floor('1 min'), 'price'],
             marker='v',
             s=100,
             c='red',
@@ -191,7 +211,7 @@ def analyze(context=None, perf=None):
     if not transaction_df.empty:
         ax6.scatter(
             buy_df.index.to_pydatetime(),
-            perf.loc[buy_df.index, 'rsi'],
+            perf.loc[buy_df.index.floor('1 min'), 'rsi'],
             marker='^',
             s=100,
             c='green',
@@ -199,7 +219,7 @@ def analyze(context=None, perf=None):
         )
         ax6.scatter(
             sell_df.index.to_pydatetime(),
-            perf.loc[sell_df.index, 'rsi'],
+            perf.loc[sell_df.index.floor('1 min'), 'rsi'],
             marker='v',
             s=100,
             c='red',
@@ -218,6 +238,13 @@ if __name__ == '__main__':
     MODE = 'live'
 
     if MODE == 'backtest':
+        folder = os.path.join(
+            tempfile.gettempdir(), 'catalyst', NAMESPACE
+        )
+        ensure_directory(folder)
+
+        timestr = time.strftime('%Y%m%d-%H%M%S')
+        out = os.path.join(folder, '{}.p'.format(timestr))
         # catalyst run -f catalyst/examples/mean_reversion_simple.py -x poloniex -s 2017-10-1 -e 2017-11-10 -c usdt -n mean-reversion --data-frequency minute --capital-base 10000
         run_algorithm(
             capital_base=10000,
@@ -228,18 +255,21 @@ if __name__ == '__main__':
             exchange_name='bitfinex',
             algo_namespace=NAMESPACE,
             base_currency='usd',
-            start=pd.to_datetime('2017-10-1', utc=True),
+            start=pd.to_datetime('2017-10-01', utc=True),
             end=pd.to_datetime('2017-11-10', utc=True),
+            output=out
         )
+        log.info('saved perf stats: {}'.format(out))
 
     elif MODE == 'live':
         run_algorithm(
+            capital_base=0.5,
             initialize=initialize,
             handle_data=handle_data,
             analyze=analyze,
             exchange_name='bittrex',
             live=True,
             algo_namespace=NAMESPACE,
-            base_currency='btc',
+            base_currency='eth',
             live_graph=False
         )

catalyst/exchange/bitfinex/bitfinex.py

@@ -61,7 +61,7 @@ class Bitfinex(Exchange):
         self.max_requests_per_minute = 80
         self.request_cpt = dict()
 
-        self.bundle = ExchangeBundle(self)
+        self.bundle = ExchangeBundle(self.name)
 
     def _request(self, operation, data, version='v1'):
         payload_object = {

catalyst/exchange/bittrex/bittrex.py

@@ -46,7 +46,7 @@ class Bittrex(Exchange):
         self.assets = dict()
         self.load_assets()
 
-        self.bundle = ExchangeBundle(self)
+        self.bundle = ExchangeBundle(self.name)
 
     @property
     def account(self):

catalyst/exchange/bundle_utils.py

@@ -6,6 +6,7 @@ from datetime import timedelta, datetime, date
 import numpy as np
 import pandas as pd
 import pytz
+from catalyst.assets._assets import TradingPair
 
 from catalyst.data.bundles.core import download_without_progress
 from catalyst.exchange.exchange_utils import get_exchange_bundles_folder
@@ -13,7 +14,6 @@ from catalyst.exchange.exchange_utils import get_exchange_bundles_folder
 EXCHANGE_NAMES = ['bitfinex', 'bittrex', 'poloniex']
 API_URL = 'http://data.enigma.co/api/v1'
 
-
 def get_date_from_ms(ms):
     """
     The date from the number of miliseconds from the epoch.
@@ -317,3 +317,41 @@ def range_in_bundle(asset, start_dt, end_dt, reader):
             has_data = False
 
     return has_data
+
+
+def get_assets(exchange, include_symbols, exclude_symbols):
+    """
+    Get assets from an exchange, including or excluding the specified
+    symbols.
+
+    Parameters
+    ----------
+    exchange: Exchange
+    include_symbols: str
+    exclude_symbols: str
+
+    Returns
+    -------
+    list[TradingPair]
+
+    """
+    if include_symbols is not None:
+        include_symbols_list = include_symbols.split(',')
+
+        return exchange.get_assets(include_symbols_list)
+
+    else:
+        all_assets = exchange.get_assets()
+
+        if exclude_symbols is not None:
+            exclude_symbols_list = exclude_symbols.split(',')
+
+            assets = []
+            for asset in all_assets:
+                if asset.symbol not in exclude_symbols_list:
+                    assets.append(asset)
+
+            return assets
+
+        else:
+            return all_assets

catalyst/exchange/exchange.py

@@ -24,7 +24,6 @@ from catalyst.exchange.exchange_utils import get_exchange_symbols, \
     get_frequency, resample_history_df
 from catalyst.finance.order import ORDER_STATUS
 from catalyst.finance.transaction import Transaction
-from catalyst.utils.deprecate import deprecated
 
 log = Logger('Exchange', level=LOG_LEVEL)
 
@@ -43,7 +42,7 @@ class Exchange:
         self.num_candles_limit = None
         self.max_requests_per_minute = None
         self.request_cpt = None
-        self.bundle = ExchangeBundle(self)
+        self.bundle = ExchangeBundle(self.name)
 
     @property
     def positions(self):

catalyst/exchange/exchange_bundle.py

@@ -20,7 +20,7 @@ from catalyst.data.minute_bars import BcolzMinuteOverlappingData, \
 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_delta, get_assets
 from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \
     BcolzExchangeBarWriter
 from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \
@@ -41,23 +41,14 @@ def _cachpath(symbol, type_):
 
 
 class ExchangeBundle:
-    def __init__(self, exchange):
-        self.exchange = exchange
+    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')
-
-    def get_assets(self, include_symbols, exclude_symbols):
-        # TODO: filter exclude symbols assets
-        if include_symbols is not None:
-            include_symbols_list = include_symbols.split(',')
-
-            return self.exchange.get_assets(include_symbols_list)
-
-        else:
-            return self.exchange.get_assets()
+        self.exchange = None
 
     def get_reader(self, data_frequency, path=None):
         """
@@ -69,7 +60,7 @@ class ExchangeBundle:
 
         """
         if path is None:
-            root = get_exchange_folder(self.exchange.name)
+            root = get_exchange_folder(self.exchange_name)
             path = BUNDLE_NAME_TEMPLATE.format(
                 root=root,
                 frequency=data_frequency
@@ -100,7 +91,7 @@ class ExchangeBundle:
         BcolzMinuteBarWriter | BcolzDailyBarWriter
 
         """
-        root = get_exchange_folder(self.exchange.name)
+        root = get_exchange_folder(self.exchange_name)
         path = BUNDLE_NAME_TEMPLATE.format(
             root=root,
             frequency=data_frequency
@@ -158,9 +149,9 @@ class ExchangeBundle:
         ----------
         assets: list[TradingPair]
             The assets is scope.
-        start_dt: datetime
+        start_dt: pd.Timestamp
             The chunk start date.
-        end_dt: datetime
+        end_dt: pd.Timestamp
             The chunk end date.
         data_frequency: str
 
@@ -209,8 +200,8 @@ class ExchangeBundle:
 
         Parameters
         ----------
-        start_dt: datetime
-        end_dt: datetime
+        start_dt: pd.Timestamp
+        end_dt: pd.Timestamp
         data_frequency: str
 
         Returns
@@ -367,7 +358,7 @@ class ExchangeBundle:
 
         # Download and extract the bundle
         path = get_bcolz_chunk(
-            exchange_name=self.exchange.name,
+            exchange_name=self.exchange_name,
             symbol=asset.symbol,
             data_frequency=data_frequency,
             period=period
@@ -436,14 +427,14 @@ class ExchangeBundle:
 
         Parameters
         ----------
-        start: datetime
-        end: datetime
+        start: pd.Timestamp
+        end: pd.Timestamp
         assets: list[TradingPair]
         data_frequency: str
 
         Returns
         -------
-        datetime, datetime
+        pd.Timestamp, pd.Timestamp
         """
         earliest_trade = None
         last_entry = None
@@ -490,8 +481,8 @@ class ExchangeBundle:
         ----------
         assets: list[TradingPair]
         data_frequency: str
-        start_dt: datetime
-        end_dt: datetime
+        start_dt: pd.Timestamp
+        end_dt: pd.Timestamp
 
         Returns
         -------
@@ -574,8 +565,8 @@ class ExchangeBundle:
         ----------
         assets: list[TradingPair]
         data_frequency: str
-        start_dt: datetime
-        end_dt: datetime
+        start_dt: pd.Timestamp
+        end_dt: pd.Timestamp
         show_progress: bool
         show_breakdown: bool
 
@@ -611,7 +602,7 @@ class ExchangeBundle:
                         show_progress,
                         label='Ingesting {frequency} price data for '
                               '{symbol} on {exchange}'.format(
-                            exchange=self.exchange.name,
+                            exchange=self.exchange_name,
                             frequency=data_frequency,
                             symbol=asset.symbol
                         )) as it:
@@ -636,7 +627,7 @@ class ExchangeBundle:
                     show_progress,
                     label='Ingesting {frequency} price data on '
                           '{exchange}'.format(
-                        exchange=self.exchange.name,
+                        exchange=self.exchange_name,
                         frequency=data_frequency,
                     )) as it:
                 for chunk in it:
@@ -654,8 +645,41 @@ class ExchangeBundle:
                 '\n'.join(problems)
             ))
 
+    def ingest_csv(self, path, data_frequency):
+        """
+        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))
+        problems = []
+
+        df = pd.read_csv(
+            path,
+            names=['symbol', 'last_traded', 'open', 'high', 'close', 'volume'],
+            parse_dates=[1]
+        )
+        # 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
+        # )
+        return filter(partial(is_not, None), problems)
+
     def ingest(self, data_frequency, include_symbols=None,
-               exclude_symbols=None, start=None, end=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.
@@ -665,17 +689,34 @@ class ExchangeBundle:
         data_frequency: str
         include_symbols: str
         exclude_symbols: str
-        start: datetime
-        end: datetime
+        start: pd.Timestamp
+        end: pd.Timestamp
         show_progress: bool
         environ:
 
         """
-        assets = self.get_assets(include_symbols, exclude_symbols)
+        if csv is not None:
+            self.ingest_csv(csv, data_frequency)
 
-        for frequency in data_frequency.split(','):
-            self.ingest_assets(assets, frequency, start, end,
-                               show_progress, show_breakdown, show_report)
+        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,
@@ -693,11 +734,11 @@ class ExchangeBundle:
         Parameters
         ----------
         assets: list[TradingPair]
-        end_dt: datetime
+        end_dt: pd.Timestamp
         bar_count: int
         field: str
         data_frequency: str
-        algo_end_dt: datetime
+        algo_end_dt: pd.Timestamp
 
         Returns
         -------
@@ -802,7 +843,7 @@ class ExchangeBundle:
             raise PricingDataNotLoadedError(
                 field=field,
                 first_trading_day=min([asset.start_date for asset in assets]),
-                exchange=self.exchange.name,
+                exchange=self.exchange_name,
                 symbols=symbols,
                 symbol_list=','.join(symbols),
                 data_frequency=data_frequency,
@@ -840,7 +881,7 @@ class ExchangeBundle:
             raise PricingDataNotLoadedError(
                 field=field,
                 first_trading_day=min([asset.start_date for asset in assets]),
-                exchange=self.exchange.name,
+                exchange=self.exchange_name,
                 symbols=symbols,
                 symbol_list=','.join(symbols),
                 data_frequency=data_frequency,
@@ -860,7 +901,7 @@ class ExchangeBundle:
                 raise PricingDataNotLoadedError(
                     field=field,
                     first_trading_day=asset.start_date,
-                    exchange=self.exchange.name,
+                    exchange=self.exchange_name,
                     symbols=asset.symbol,
                     symbol_list=asset.symbol,
                     data_frequency=data_frequency,
@@ -884,7 +925,7 @@ class ExchangeBundle:
             )
             if len(arrays) == 0:
                 raise DataCorruptionError(
-                    exchange=self.exchange.name,
+                    exchange=self.exchange_name,
                     symbols=asset.symbol,
                     start_dt=asset_start_dt,
                     end_dt=asset_end_dt
@@ -897,42 +938,41 @@ class ExchangeBundle:
 
         return series
 
+    def clean(self, data_frequency):
+        """
+        Removing the bundle data from the catalyst folder.
 
-def clean(self, data_frequency):
-    """
-    Removing the bundle data from the catalyst folder.
+        Parameters
+        ----------
+        data_frequency: str
 
-    Parameters
-    ----------
-    data_frequency: str
+        """
+        log.debug('cleaning exchange {}, frequency {}'.format(
+            self.exchange_name, data_frequency
+        ))
+        root = get_exchange_folder(self.exchange_name)
 
-    """
-    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)
 
-    symbols = os.path.join(root, 'symbols.json')
-    if os.path.isfile(symbols):
-        os.remove(symbols)
+        temp_bundles = os.path.join(root, 'temp_bundles')
 
-    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))
 
-    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]
 
-    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)
 
-    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))
+            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))

catalyst/exchange/exchange_data_portal.py

@@ -21,7 +21,6 @@ log = Logger('DataPortalExchange', level=LOG_LEVEL)
 class DataPortalExchangeBase(DataPortal):
     def __init__(self, *args, **kwargs):
 
-        self.exchanges = kwargs.pop('exchanges', None)
         # TODO: put somewhere accessible by each algo
         self.retry_get_history_window = 5
         self.retry_get_spot_value = 5
@@ -49,11 +48,10 @@ class DataPortalExchangeBase(DataPortal):
             if len(exchange_assets) > 1:
                 df_list = []
                 for exchange_name in exchange_assets:
-                    exchange = self.exchanges[exchange_name]
                     assets = exchange_assets[exchange_name]
 
                     df_exchange = self.get_exchange_history_window(
-                        exchange,
+                        exchange_name,
                         assets,
                         end_dt,
                         bar_count,
@@ -68,9 +66,9 @@ class DataPortalExchangeBase(DataPortal):
                 return pd.concat(df_list)
 
             else:
-                exchange = self.exchanges[list(exchange_assets.keys())[0]]
+                exchange_name = list(exchange_assets.keys())[0]
                 return self.get_exchange_history_window(
-                    exchange,
+                    exchange_name,
                     assets,
                     end_dt,
                     bar_count,
@@ -122,7 +120,7 @@ class DataPortalExchangeBase(DataPortal):
 
     @abc.abstractmethod
     def get_exchange_history_window(self,
-                                    exchange,
+                                    exchange_name,
                                     assets,
                                     end_dt,
                                     bar_count,
@@ -136,9 +134,8 @@ class DataPortalExchangeBase(DataPortal):
                         attempt_index=0):
         try:
             if isinstance(assets, TradingPair):
-                exchange = self.exchanges[assets.exchange]
                 spot_values = self.get_exchange_spot_value(
-                    exchange, [assets], field, dt, data_frequency)
+                    assets.exchange, [assets], field, dt, data_frequency)
 
                 if not spot_values:
                     return np.nan
@@ -154,17 +151,16 @@ class DataPortalExchangeBase(DataPortal):
                     exchange_assets[asset.exchange].append(asset)
 
                 if len(list(exchange_assets.keys())) == 1:
-                    exchange = self.exchanges[list(exchange_assets.keys())[0]]
+                    exchange_name = list(exchange_assets.keys())[0]
                     return self.get_exchange_spot_value(
-                        exchange, assets, field, dt, data_frequency)
+                        exchange_name, assets, field, dt, data_frequency)
 
                 else:
                     spot_values = []
                     for exchange_name in exchange_assets:
-                        exchange = self.exchanges[exchange_name]
                         assets = exchange_assets[exchange_name]
                         exchange_spot_values = self.get_exchange_spot_value(
-                            exchange,
+                            exchange_name,
                             assets,
                             field,
                             dt,
@@ -199,7 +195,7 @@ class DataPortalExchangeBase(DataPortal):
         return self._get_spot_value(assets, field, dt, data_frequency)
 
     @abc.abstractmethod
-    def get_exchange_spot_value(self, exchange, assets, field, dt,
+    def get_exchange_spot_value(self, exchange_name, assets, field, dt,
                                 data_frequency):
         return
 
@@ -214,10 +210,11 @@ class DataPortalExchangeBase(DataPortal):
 
 class DataPortalExchangeLive(DataPortalExchangeBase):
     def __init__(self, *args, **kwargs):
+        self.exchanges = kwargs.pop('exchanges', None)
         super(DataPortalExchangeLive, self).__init__(*args, **kwargs)
 
     def get_exchange_history_window(self,
-                                    exchange,
+                                    exchange_name,
                                     assets,
                                     end_dt,
                                     bar_count,
@@ -230,7 +227,7 @@ class DataPortalExchangeLive(DataPortalExchangeBase):
 
         Parameters
         ----------
-        exchange: Exchange
+        exchange_name: Exchange
         assets: list[TradingPair]
         end_dt: datetime
         bar_count: int
@@ -244,6 +241,7 @@ class DataPortalExchangeLive(DataPortalExchangeBase):
         DataFrame
 
         """
+        exchange = self.exchanges[exchange_name]
         df = exchange.get_history_window(
             assets,
             end_dt,
@@ -254,14 +252,14 @@ class DataPortalExchangeLive(DataPortalExchangeBase):
             ffill)
         return df
 
-    def get_exchange_spot_value(self, exchange, assets, field, dt,
+    def get_exchange_spot_value(self, exchange_name, assets, field, dt,
                                 data_frequency):
         """
         A spot value for the exchange.
 
         Parameters
         ----------
-        exchange: Exchange
+        exchange_name: str
         assets: list[TradingPair]
         field: str
         dt: datetime
@@ -272,6 +270,7 @@ class DataPortalExchangeLive(DataPortalExchangeBase):
         float
 
         """
+        exchange = self.exchanges[exchange_name]
         exchange_spot_values = exchange.get_spot_value(
             assets, field, dt, data_frequency)
 
@@ -280,16 +279,16 @@ class DataPortalExchangeLive(DataPortalExchangeBase):
 
 class DataPortalExchangeBacktest(DataPortalExchangeBase):
     def __init__(self, *args, **kwargs):
+        self.exchange_names = kwargs.pop('exchange_names', None)
+
         super(DataPortalExchangeBacktest, self).__init__(*args, **kwargs)
 
         self.exchange_bundles = dict()
-
         self.history_loaders = dict()
         self.minute_history_loaders = dict()
 
-        for exchange_name in self.exchanges:
-            exchange = self.exchanges[exchange_name]
-            self.exchange_bundles[exchange_name] = ExchangeBundle(exchange)
+        for name in self.exchange_names:
+            self.exchange_bundles[name] = ExchangeBundle(name)
 
     def _get_first_trading_day(self, assets):
         first_date = None
@@ -299,7 +298,7 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
         return first_date
 
     def get_exchange_history_window(self,
-                                    exchange,
+                                    exchange_name,
                                     assets,
                                     end_dt,
                                     bar_count,
@@ -326,7 +325,7 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
         DataFrame
 
         """
-        bundle = self.exchange_bundles[exchange.name]  # type: ExchangeBundle
+        bundle = self.exchange_bundles[exchange_name]  # type: ExchangeBundle
 
         freq, candle_size, unit, adj_data_frequency = get_frequency(
             frequency, data_frequency
@@ -351,7 +350,7 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
         return df
 
     def get_exchange_spot_value(self,
-                                exchange,
+                                exchange_name,
                                 assets,
                                 field,
                                 dt,
@@ -363,7 +362,7 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
 
         Parameters
         ----------
-        exchange: Exchange
+        exchange_name: str
         assets: list[TradingPair]
         field: str
         dt: datetime
@@ -374,7 +373,7 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase):
         float
 
         """
-        bundle = self.exchange_bundles[exchange.name]
+        bundle = self.exchange_bundles[exchange_name]
         if data_frequency == 'daily':
             dt = dt.floor('1D')
         else:

catalyst/exchange/poloniex/poloniex.py

@@ -47,7 +47,7 @@ class Poloniex(Exchange):
         self.max_requests_per_minute = 60
         self.request_cpt = dict()
 
-        self.bundle = ExchangeBundle(self)
+        self.bundle = ExchangeBundle(self.name)
 
     def sanitize_curency_symbol(self, exchange_symbol):
         """

catalyst/exchange/validator.py

@@ -1,20 +1,17 @@
 import os
 import tempfile
 
+import pandas as pd
 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.exchange.factory import get_exchanges
 from catalyst.utils.paths import ensure_directory
-from catalyst.exchange.exchange import Exchange
 
 log = Logger('Validator', level=LOG_LEVEL)
 

catalyst/utils/run_algo.py

@@ -320,7 +320,7 @@ def _run(handle_data,
         # can handle this later.
 
         data = DataPortalExchangeBacktest(
-            exchanges=exchanges,
+            exchange_names=[exchange_name for exchange_name in exchanges],
             asset_finder=None,
             trading_calendar=open_calendar,
             first_trading_day=start,