diff --git a/catalyst/__main__.py b/catalyst/__main__.py index 86e49ca2..ea20c957 100644 --- a/catalyst/__main__.py +++ b/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( diff --git a/catalyst/constants.py b/catalyst/constants.py index bbdae343..cde29914 100644 --- a/catalyst/constants.py +++ b/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' diff --git a/catalyst/examples/mean_reversion_simple.py b/catalyst/examples/mean_reversion_simple.py index 34d9c60c..6660c829 100644 --- a/catalyst/examples/mean_reversion_simple.py +++ b/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 ) diff --git a/catalyst/exchange/bitfinex/bitfinex.py b/catalyst/exchange/bitfinex/bitfinex.py index 4c878e04..66a72421 100644 --- a/catalyst/exchange/bitfinex/bitfinex.py +++ b/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 = { diff --git a/catalyst/exchange/bittrex/bittrex.py b/catalyst/exchange/bittrex/bittrex.py index d50d9bb8..fc248847 100644 --- a/catalyst/exchange/bittrex/bittrex.py +++ b/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): diff --git a/catalyst/exchange/bundle_utils.py b/catalyst/exchange/bundle_utils.py index 9b0c8c49..c010479a 100644 --- a/catalyst/exchange/bundle_utils.py +++ b/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 diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index bd9cca9d..d21bf273 100644 --- a/catalyst/exchange/exchange.py +++ b/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): diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index 23fa1737..da7908e2 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/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)) diff --git a/catalyst/exchange/exchange_data_portal.py b/catalyst/exchange/exchange_data_portal.py index c5d7ac85..4cfac2b7 100644 --- a/catalyst/exchange/exchange_data_portal.py +++ b/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: diff --git a/catalyst/exchange/poloniex/poloniex.py b/catalyst/exchange/poloniex/poloniex.py index 0d283d8b..ece0c659 100644 --- a/catalyst/exchange/poloniex/poloniex.py +++ b/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): """ diff --git a/catalyst/exchange/validator.py b/catalyst/exchange/validator.py index 2037fd59..5cd0b1e2 100644 --- a/catalyst/exchange/validator.py +++ b/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) diff --git a/catalyst/utils/run_algo.py b/catalyst/utils/run_algo.py index bf359ef4..af62f97d 100644 --- a/catalyst/utils/run_algo.py +++ b/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,