diff --git a/catalyst/__main__.py b/catalyst/__main__.py index 8b7f001b..c1872d3f 100644 --- a/catalyst/__main__.py +++ b/catalyst/__main__.py @@ -9,6 +9,7 @@ from six import text_type from catalyst.data import bundles as bundles_module from catalyst.exchange.exchange_bundle import ExchangeBundle +from catalyst.exchange.exchange_utils import delete_algo_folder from catalyst.exchange.factory import get_exchange from catalyst.utils.cli import Date, Timestamp from catalyst.utils.run_algo import _run, load_extensions @@ -490,8 +491,19 @@ def live(ctx, default=True, help='Print progress information to the terminal.' ) +@click.option( + '--verbose/--no-verbose`', + default=False, + help='Show a progress indicator for every currency pair.' +) +@click.option( + '--validate/--no-validate`', + default=False, + help='Report potential anomalies found in data bundles.' +) def ingest_exchange(exchange_name, data_frequency, start, end, - include_symbols, exclude_symbols, show_progress): + include_symbols, exclude_symbols, show_progress, verbose, + validate): """ Ingest data for the given exchange. """ @@ -509,10 +521,26 @@ def ingest_exchange(exchange_name, data_frequency, start, end, exclude_symbols=exclude_symbols, start=start, end=end, - show_progress=show_progress + show_progress=show_progress, + show_breakdown=verbose, + show_report=validate ) +@main.command(name='clean-algo') +@click.option( + '-n', + '--algo-namespace', + help='The label of the algorithm to for which to clean the state.' +) +@click.pass_context +def clean_algo(ctx, algo_namespace): + click.echo( + 'Deleting the state folder of algo: {}...'.format(algo_namespace) + ) + delete_algo_folder(algo_namespace) + + @main.command(name='clean-exchange') @click.option( '-x', diff --git a/catalyst/constants.py b/catalyst/constants.py index c5f8d4ef..cde29914 100644 --- a/catalyst/constants.py +++ b/catalyst/constants.py @@ -2,4 +2,8 @@ import logbook -LOG_LEVEL = logbook.INFO \ No newline at end of file +LOG_LEVEL = logbook.INFO + +DATE_TIME_FORMAT = '%Y-%m-%d %H:%M' + +AUTO_INGEST = False \ No newline at end of file diff --git a/catalyst/curate/poloniex.py b/catalyst/curate/poloniex.py index c81ddf04..572cd777 100644 --- a/catalyst/curate/poloniex.py +++ b/catalyst/curate/poloniex.py @@ -6,9 +6,8 @@ from catalyst.exchange.exchange_utils import get_exchange_symbols_filename DT_START = int(time.mktime(datetime(2010, 1, 1, 0, 0).timetuple())) -DT_END = int(time.time()) -CSV_OUT_FOLDER = '/var/tmp/catalyst/data/poloniex/' -CSV_OUT_FOLDER = '/Volumes/enigma/data/poloniex/' +DT_END = pd.to_datetime('today').value // 10 ** 9 +CSV_OUT_FOLDER = os.environ.get('CSV_OUT_FOLDER', '/efs/exchanges/poloniex/') CONN_RETRIES = 2 logbook.StderrHandler().push_application() @@ -27,13 +26,15 @@ class PoloniexCurator(object): try: os.makedirs(CSV_OUT_FOLDER) except Exception as e: - log.error('Failed to create data folder: %s' % CSV_OUT_FOLDER) + log.error('Failed to create data folder: {}'.format( + CSV_OUT_FOLDER)) log.exception(e) - ''' - Retrieves and returns all currency pairs from the exchange - ''' + def get_currency_pairs(self): + ''' + Retrieves and returns all currency pairs from the exchange + ''' url = self._api_path + 'command=returnTicker' try: @@ -49,89 +50,136 @@ class PoloniexCurator(object): self.currency_pairs.append(ticker) self.currency_pairs.sort() - log.debug('Currency pairs retrieved successfully: %d' % (len(self.currency_pairs))) + log.debug('Currency pairs retrieved successfully: {}'.format( + len(self.currency_pairs) + )) + - ''' - Helper function that reads tradeID and date fields from CSV readline - ''' def _retrieve_tradeID_date(self, row): + ''' + Helper function that reads tradeID and date fields from CSV readline + ''' tId = int(row.split(',')[0]) - d = pd.to_datetime( row.split(',')[1], infer_datetime_format=True).value // 10 ** 9 + d = pd.to_datetime(row.split(',')[1], + infer_datetime_format=True).value // 10 ** 9 return tId, d - ''' - Retrieves TradeHistory from exchange for a given currencyPair between start and end dates. - If no start date is provided, uses a system-wide one (beginning of time for cryptotrading) - If no end date is provided, 'now' is used + + def retrieve_trade_history(self, currencyPair, start=DT_START, + end=DT_END, temp=None): + ''' + Retrieves TradeHistory from exchange for a given currencyPair + between start and end dates. If no start date is provided, uses + a system-wide one (beginning of time for cryptotrading). + If no end date is provided, 'now' is used. + Stores results in CSV file on disk. - This function is called recursively to work around the limitations imposed by the provider API. - ''' - def retrieve_trade_history(self, currencyPair, start=DT_START, end=DT_END, temp=None): + + This function is called recursively to work around the + limitations imposed by the provider API. + ''' csv_fn = CSV_OUT_FOLDER + 'crypto_trades-' + currencyPair + '.csv' ''' - Check what data we already have on disk, reading first and last lines from file. - Data is stored on file from NEWEST to OLDEST. + Check what data we already have on disk, reading first and last + lines from file. Data is stored on file from NEWEST to OLDEST. ''' try: with open(csv_fn, 'ab+') as f: f.seek(0, os.SEEK_END) - if(f.tell() > 2): # First check file is not zero size - f.seek(0) # Go to the beginning to read first line - last_tradeID, end_file = self._retrieve_tradeID_date(f.readline()) - f.seek(-2, os.SEEK_END) # Jump to the second last byte. - while f.read(1) != b"\n": # Until EOL is found... - f.seek(-2, os.SEEK_CUR) # ...jump back the read byte plus one more. + if(f.tell() > 2): # Check file size is not 0 + f.seek(0) # Go to start to read + last_tradeID, end_file = self._retrieve_tradeID_date(f.readline()) + f.seek(-2, os.SEEK_END) # Jump to the 2nd last byte + while f.read(1) != b"\n": # Until EOL is found... + f.seek(-2, os.SEEK_CUR) # ...jump back the read byte plus one more. first_tradeID, start_file = self._retrieve_tradeID_date(f.readline()) - if( first_tradeID == 1 and end_file + 3600 > DT_END ): + if( end_file + 3600 * 6 > DT_END and ( first_tradeID == 1 + or (currencyPair == 'BTC_HUC' and first_tradeID == 2) + or (currencyPair == 'BTC_RIC' and first_tradeID == 2) + or (currencyPair == 'BTC_XCP' and first_tradeID == 2) + or (currencyPair == 'BTC_NAV' and first_tradeID == 4569) + or (currencyPair == 'BTC_POT' and first_tradeID == 23511) ) ): return except Exception as e: - log.error('Error opening file: %s' % csv_fn) + log.error('Error opening file: {}'.format(csv_fn)) log.exception(e) ''' - Poloniex API limits querying TradeHistory to intervals smaller than 1 month, - so we make sure that start date is never more than 1 month apart from end date + Poloniex API limits querying TradeHistory to intervals smaller + than 1 month, so we make sure that start date is never more than + 1 month apart from end date ''' - if( end - start > 2419200 ): # 60 s/min * 60 min/hr * 24 hr/day * 28 days + if( end - start > 2419200 ): # 60s/min * 60min/hr * 24hr/day * 28days newstart = end - 2419200 else: newstart = start - log.debug(currencyPair+': Retrieving from '+str(newstart)+' to '+str(end) +'\t ' - + time.ctime(newstart) + ' - '+ time.ctime(end)) + log.debug('{}: Retrieving from {} to {}\t {} - {}'.format( + currencyPair, str(newstart), str(end), + time.ctime(newstart), time.ctime(end))) - url = self._api_path + 'command=returnTradeHistory¤cyPair=' + currencyPair + '&start=' + str(newstart) + '&end=' + str(end) + url = '{path}command=returnTradeHistory¤cyPair={pair}' \ + '&start={start}&end={end}'.format( + path = self._api_path, + pair = currencyPair, + start = str(newstart), + end = str(end) + ) + print url - try: - response = requests.get(url) - except Exception as e: - log.error('Failed to retrieve trade history data for %s' % currencyPair) - log.exception(e) + attempts = 0 + success = 0 + while attempts < CONN_RETRIES: + try: + response = requests.get(url) + except Exception as e: + log.error('Failed to retrieve trade history data for {}'.format( + currencyPair + )) + log.exception(e) + attempts += 1 + else: + try: + if isinstance(response.json(), dict) and response.json()['error']: + log.error('Failed to to retrieve trade history data ' + 'for {}: {}'.format( + currencyPair, + response.json()['error'] + )) + attempts += 1 + except Exception as e: + log.exception(e) + attempts += 1 + else: + success = 1 + break + + if not success: return None - else: - if isinstance(response.json(), dict) and response.json()['error']: - log.error('Failed to to retrieve trade history data for %s: %s' % (currencyPair,response.json()['error'])) - exit(1) + ''' - If we get to transactionId == 1, and we already have that on disk, - we got to the end of TradeHistory for this coin. + If we get to transactionId == 1, and we already have that on + disk, we got to the end of TradeHistory for this coin. ''' - if('first_tradeID' in locals() and response.json()[-1]['tradeID'] == first_tradeID): + if('first_tradeID' in locals() + and response.json()[-1]['tradeID'] == first_tradeID): return ''' There are primarily two scenarios: - a) There is newer data available that we need to add at the beginning - of the file. We'll retrieve all what we need until we get to what - we already have, writing it to a temporary file; and we will write - that at the beginning of our existing file. - b) We are going back in time, appending at the end of our existing - TradeHistory until the first transaction for this currencyPair + a) There is newer data available that we need to add at + the beginning of the file. We'll retrieve all what we + need until we get to what we already have, writing it + to a temporary file; and we will write that at the + beginning of our existing file. + b) We are going back in time, appending at the end of + our existing TradeHistory until the first transaction + for this currencyPair ''' try: if( 'end_file' in locals() and end_file + 3600 < end): @@ -151,8 +199,10 @@ class PoloniexCurator(object): item['globalTradeID'] ]) if( response.json()[-1]['tradeID'] > last_tradeID ): - end = pd.to_datetime( response.json()[-1]['date'], infer_datetime_format=True).value // 10 ** 9 - self.retrieve_trade_history(currencyPair, start, end, temp=temp) + end = pd.to_datetime( response.json()[-1]['date'], + infer_datetime_format=True).value // 10 ** 9 + self.retrieve_trade_history(currencyPair, start, + end, temp=temp) else: with open(csv_fn,'rb+') as f: shutil.copyfileobj(f,temp) @@ -165,7 +215,8 @@ class PoloniexCurator(object): with open(csv_fn, 'ab') as csvfile: csvwriter = csv.writer(csvfile) for item in response.json(): - if( 'first_tradeID' in locals() and item['tradeID'] >= first_tradeID ): + if( 'first_tradeID' in locals() + and item['tradeID'] >= first_tradeID ): continue csvwriter.writerow([ item['tradeID'], @@ -176,84 +227,112 @@ class PoloniexCurator(object): item['total'], item['globalTradeID'] ]) - end = pd.to_datetime( response.json()[-1]['date'], infer_datetime_format=True).value // 10 ** 9 + end = pd.to_datetime(response.json()[-1]['date'], + infer_datetime_format=True).value // 10 ** 9 except Exception as e: - log.error('Error opening %s' % csv_fn) + log.error('Error opening {}'.format(csv_fn)) log.exception(e) ''' - If we got here, we aren't done yet. Call recursively with 'end' times - that go sequentially back in time. + If we got here, we aren't done yet. Call recursively with + 'end' times that go sequentially back in time. ''' self.retrieve_trade_history(currencyPair, start, end) - ''' + + def generate_ohlcv(self, df): + ''' Generates OHLCV dataframe from a dataframe containing all TradeHistory by resampling with 1-minute period - ''' - def generate_ohlcv(self, df): - df.set_index('date', inplace=True) # Index by date - vol = df['total'].to_frame('volume') # Will deal with vol separately, as ohlc() messes it up - df.drop('total', axis=1, inplace=True) # Drop volume data from dataframe - ohlc = df.resample('T').ohlc() # Resample OHLC in 1min bins - ohlc.columns = ohlc.columns.map(lambda t: t[1]) # Raname columns by dropping 'rate' - closes = ohlc['close'].fillna(method='pad') # Pad forward missing 'close' - ohlc = ohlc.apply(lambda x: x.fillna(closes)) # Fill N/A with last close - vol = vol.resample('T').sum().fillna(0) # Add volumes by bin - ohlcv = pd.concat([ohlc,vol], axis=1) # Concatenate OHLC + Volume + ''' + df.set_index('date', inplace=True) # Index by date + vol = df['total'].to_frame('volume') # set Vol aside + df.drop('total', axis=1, inplace=True) # Drop volume data + ohlc = df.resample('T').ohlc() # Resample OHLC 1min + ohlc.columns = ohlc.columns.map(lambda t: t[1]) # Raname columns by dropping 'rate' + closes = ohlc['close'].fillna(method='pad') # Pad fwd missing 'close' + ohlc = ohlc.apply(lambda x: x.fillna(closes)) # Fill N/A with last close + vol = vol.resample('T').sum().fillna(0) # Add volumes by bin + ohlcv = pd.concat([ohlc,vol], axis=1) # Concatenate OHLC + Vol return ohlcv - ''' + + def write_ohlcv_file(self, currencyPair): + ''' Generates OHLCV data file with 1minute bars from TradeHistory on disk - ''' - def write_ohlcv_file(self, currencyPair): + ''' csv_trades = CSV_OUT_FOLDER + 'crypto_trades-' + currencyPair + '.csv' csv_1min = CSV_OUT_FOLDER + 'crypto_1min-' + currencyPair + '.csv' - #if( os.path.isfile(csv_1min) ): - # log.debug(currencyPair+': 1min data already present. Delete the file if you want to rebuild it.') - #else: - df = pd.read_csv(csv_trades, names=['tradeID','date','type','rate','amount','total','globalTradeID'], - dtype = {'tradeID': int, 'date': str, 'type': str, 'rate': float, 'amount': float, 'total': float, 'globalTradeID': int } ) - df.drop(['tradeID','type','amount','globalTradeID'], axis=1, inplace=True) - df['date'] = pd.to_datetime(df['date'], infer_datetime_format=True) - ohlcv = self.generate_ohlcv(df) - try: - with open(csv_1min, 'w') as csvfile: - csvwriter = csv.writer(csvfile) - for item in ohlcv.itertuples(): - if item.Index == 0: - continue - csvwriter.writerow([ - item.Index.value // 10 ** 9, - item.open, - item.high, - item.low, - item.close, - item.volume, - ]) - except Exception as e: - log.error('Error opening %s' % csv_fn) - log.exception(e) - log.debug(currencyPair+': Generated 1min OHLCV data.') + if( os.path.getmtime(csv_1min) > time.time() - 7200 ): + log.debug(currencyPair+': 1min data file already up to date. ' + 'Delete the file if you want to rebuild it.') + else: + df = pd.read_csv(csv_trades, + names=['tradeID', + 'date', + 'type', + 'rate', + 'amount', + 'total', + 'globalTradeID'], + dtype = {'tradeID': int, + 'date': str, + 'type': str, + 'rate': float, + 'amount': float, + 'total': float, + 'globalTradeID': int } + ) + df.drop(['tradeID','type','amount','globalTradeID'], + axis=1, inplace=True) + df['date'] = pd.to_datetime(df['date'], infer_datetime_format=True) + ohlcv = self.generate_ohlcv(df) + try: + with open(csv_1min, 'w') as csvfile: + csvwriter = csv.writer(csvfile) + for item in ohlcv.itertuples(): + if item.Index == 0: + continue + csvwriter.writerow([ + item.Index.value // 10 ** 9, + item.open, + item.high, + item.low, + item.close, + item.volume, + ]) + except Exception as e: + log.error('Error opening {}'.format(csv_fn)) + log.exception(e) + log.debug('{}: Generated 1min OHLCV data.'.format(currencyPair)) + - ''' - Returns a data frame for a given currencyPair from data on disk - ''' def onemin_to_dataframe(self, currencyPair, start, end): + ''' + Returns a data frame for a given currencyPair from data on disk + ''' csv_fn = CSV_OUT_FOLDER + 'crypto_1min-' + currencyPair + '.csv' - df = pd.read_csv(csv_fn, names=['date', 'open', 'high', 'low', 'close', 'volume']) + df = pd.read_csv(csv_fn, names=['date', + 'open', + 'high', + 'low', + 'close', + 'volume'] + ) df['date'] = pd.to_datetime(df['date'],unit='s') df.set_index('date', inplace=True) return df[start : end] - ''' - Generates a symbols.json file with corresponding start_date for each currencyPair - ''' + def generate_symbols_json(self, filename=None): + ''' + Generates a symbols.json file with corresponding start_date + for each currencyPair + ''' symbol_map = {} if(filename is None): @@ -262,14 +341,16 @@ class PoloniexCurator(object): with open(filename, 'w') as symbols: for currencyPair in self.currency_pairs: start = None - csv_fn = CSV_OUT_FOLDER + 'crypto_trades-' + currencyPair + '.csv' + csv_fn = '{}crypto_trades-{}.csv'.format( + CSV_OUT_FOLDER, currencyPair) with open(csv_fn, 'r') as f: f.seek(0, os.SEEK_END) - if(f.tell() > 2): # First check file is not zero size - f.seek(-2, os.SEEK_END) # Jump to the second last byte. - while f.read(1) != b"\n": # Until EOL is found... - f.seek(-2, os.SEEK_CUR) # ...jump back the read byte plus one more. - start = pd.to_datetime( f.readline().split(',')[1], infer_datetime_format=True) + if(f.tell() > 2): # Check file size is not 0 + f.seek(-2, os.SEEK_END) # Jump to 2nd last byte + while f.read(1) != b"\n": # Until EOL is found... + f.seek(-2, os.SEEK_CUR) # ...jump back the read byte plus one more. + start = pd.to_datetime( f.readline().split(',')[1], + infer_datetime_format=True) if(start is None): start = time.gmtime() @@ -279,7 +360,8 @@ class PoloniexCurator(object): symbol = symbol, start_date = start.strftime("%Y-%m-%d") ) - json.dump(symbol_map, symbols, sort_keys=True, indent=2, separators=(',',':')) + json.dump(symbol_map, symbols, sort_keys=True, indent=2, + separators=(',',':')) if __name__ == '__main__': @@ -289,6 +371,6 @@ if __name__ == '__main__': for currencyPair in pc.currency_pairs: pc.retrieve_trade_history(currencyPair) + log.debug('{} up to date.'.format(currencyPair)) pc.write_ohlcv_file(currencyPair) - \ No newline at end of file diff --git a/catalyst/examples/mean_reversion.py b/catalyst/examples/mean_reversion.py new file mode 100644 index 00000000..2e891593 --- /dev/null +++ b/catalyst/examples/mean_reversion.py @@ -0,0 +1,283 @@ +# 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. +from datetime import timedelta + +import pandas as pd +import talib +# To run an algorithm in Catalyst, you need two functions: initialize and +# handle_data. +from logbook import Logger +from talib.common import MA_Type + +from catalyst import run_algorithm +from catalyst.api import symbol, record, order_target_percent, \ + get_open_orders +# 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.exchange.stats_utils import extract_transactions, trend_direction + +algo_namespace = 'momentum' +log = Logger(algo_namespace) + + +def initialize(context): + # This initialize function sets any data or variables that you'll use in + # your algorithm. For instance, you'll want to define the trading pair (or + # trading pairs) you want to backtest. You'll also want to define any + # parameters or values you're going to use. + + # In our example, we're looking at Ether in USD Tether. + context.eth_btc = symbol('etc_usdt') + context.base_price = None + context.current_day = None + context.trigger = None + + +def handle_data(context, data): + # This handle_data function is where the real work is done. Our data is + # minute-level tick data, and each minute is called a frame. This function + # runs on each frame of the data. + + # We flag the first period of each day. + # Since cryptocurrencies trade 24/7 the `before_trading_starts` handle + # would only execute once. This method works with minute and daily + # frequencies. + today = data.current_dt.floor('1D') + if today != context.current_day: + context.traded_today = False + context.current_day = today + + # We're computing the volume-weighted-average-price of the security + # defined above, in the context.eth_btc 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.eth_btc, + fields='close', + bar_count=50, + frequency='15T' + ) + + # Ta-lib calculates various technical indicator based on price and + # volume arrays. + + # In this example, we are comp + rsi = talib.RSI(prices.values, timeperiod=14) + upper, middle, lower = talib.BBANDS( + prices.values, + timeperiod=20, + nbdevup=2, + nbdevdn=2, + matype=MA_Type.EMA + ) + + # 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.eth_btc, fields=['close', 'volume']) + price = current['close'] + + # If base_price is not set, we use the current value. This is the + # price at the first bar which we reference to calculate price_change. + if context.base_price is None: + context.base_price = price + + price_change = (price - context.base_price) / context.base_price + cash = context.portfolio.cash + + # Now that we've collected all current data for this frame, we use + # the record() method to save it. This data will be available as + # a parameter of the analyze() function for further analysis. + record( + price=price, + volume=current['volume'], + upper_band=upper[-1], + lower_band=lower[-1], + price_change=price_change, + rsi=rsi[-1], + cash=cash + ) + + # We are trying to avoid over-trading by limiting our trades to + # one per day. + if context.traded_today: + return + + # 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.eth_btc) + if len(orders) > 0: + return + + # Exit if we cannot trade + if not data.can_trade(context.eth_btc): + 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.eth_btc].amount + + # In this example, we're using a trigger instead of buying directly after + # a signal. Since this is mean reversion, our signals go against the + # momentum. Using a trigger allow us to spot the opportunity but trade + # only when a trade reversal begins. + if context.trigger is not None: + # The tread_direction() method determines the trend based on the last + # two bars of the series. + direction = trend_direction(rsi) + if context.trigger[1] == 'buy' and direction == 'up': + log.info( + '{}: buying - price: {}, rsi: {}, bband: {}'.format( + data.current_dt, price, rsi[-1], lower[-1] + ) + ) + order_target_percent(context.eth_btc, 1) + context.traded_today = True + context.trigger = None + + elif context.trigger[1] == 'sell' and direction == 'down': + log.info( + '{}: selling - price: {}, rsi: {}, bband: {}'.format( + data.current_dt, price, rsi[-1], upper[-1] + ) + ) + order_target_percent(context.eth_btc, 0) + context.traded_today = True + context.trigger = None + + # If we found a signal but no trade reversal within two hours, we + # reset the trigger. + elif context.trigger[0] + timedelta(hours=2) < data.current_dt: + context.trigger = None + + else: + # Determining the entry and exit signals based on RSI and SMA + if rsi[-1] <= 30 and pos_amount == 0: + context.trigger = (data.current_dt, 'buy') + + elif rsi[-1] >= 80 and pos_amount > 0: + context.trigger = (data.current_dt, 'sell') + + +def analyze(context=None, perf=None): + import matplotlib.pyplot as plt + + # The base currency of the algo exchange + base_currency = context.exchanges.values()[0].base_currency.upper() + + # Plot the portfolio value over time. + ax1 = plt.subplot(611) + perf.loc[:, 'portfolio_value'].plot(ax=ax1) + ax1.set_ylabel('Portfolio Value ({})'.format(base_currency)) + + # Plot the price increase or decrease over time. + ax2 = plt.subplot(612, sharex=ax1) + perf.loc[:, 'price'].plot(ax=ax2, label='Price') + perf.loc[:, 'upper_band'].plot(ax=ax2, label='Upper') + perf.loc[:, 'lower_band'].plot(ax=ax2, label='Lower') + + ax2.set_ylabel('{asset} ({base})'.format( + asset=context.eth_btc.symbol, base=base_currency + )) + + transaction_df = extract_transactions(perf) + if not transaction_df.empty: + buy_df = transaction_df[transaction_df['amount'] > 0] + sell_df = transaction_df[transaction_df['amount'] < 0] + ax2.scatter( + buy_df.index.to_pydatetime(), + perf.loc[buy_df.index, 'price'], + marker='^', + s=100, + c='green', + label='' + ) + ax2.scatter( + sell_df.index.to_pydatetime(), + perf.loc[sell_df.index, 'price'], + marker='v', + s=100, + c='red', + label='' + ) + + ax4 = plt.subplot(613, sharex=ax1) + perf.loc[:, 'cash'].plot( + ax=ax4, label='Base Currency ({})'.format(base_currency) + ) + ax4.set_ylabel('Cash ({})'.format(base_currency)) + + perf['algorithm'] = perf.loc[:, 'algorithm_period_return'] + + ax5 = plt.subplot(614, sharex=ax1) + perf.loc[:, ['algorithm', 'price_change']].plot(ax=ax5) + ax5.set_ylabel('Percent Change') + + ax6 = plt.subplot(615, sharex=ax1) + perf.loc[:, 'rsi'].plot(ax=ax6, label='RSI') + ax6.axhline(70, color='darkgoldenrod') + ax6.axhline(30, color='darkgoldenrod') + + if not transaction_df.empty: + ax6.scatter( + buy_df.index.to_pydatetime(), + perf.loc[buy_df.index, 'rsi'], + marker='^', + s=100, + c='green', + label='' + ) + ax6.scatter( + sell_df.index.to_pydatetime(), + perf.loc[sell_df.index, 'rsi'], + marker='v', + s=100, + c='red', + label='' + ) + plt.legend(loc=3) + + # Show the plot. + plt.gcf().set_size_inches(18, 8) + plt.show() + pass + + +if __name__ == '__main__': + # The execution mode: backtest or live + MODE = 'backtest' + + if MODE == 'backtest': + run_algorithm( + capital_base=1, + data_frequency='minute', + initialize=initialize, + handle_data=handle_data, + analyze=analyze, + exchange_name='poloniex', + algo_namespace=algo_namespace, + base_currency='usdt', + start=pd.to_datetime('2017-7-1', utc=True), + # end=pd.to_datetime('2017-9-30', utc=True), + end=pd.to_datetime('2017-10-31', utc=True), + ) + + elif MODE == 'live': + run_algorithm( + initialize=initialize, + handle_data=handle_data, + analyze=analyze, + exchange_name='poloniex', + live=True, + algo_namespace=algo_namespace, + base_currency='usdt', + live_graph=True + ) diff --git a/catalyst/examples/mean_reversion_simple.py b/catalyst/examples/mean_reversion_simple.py new file mode 100644 index 00000000..7780e3ac --- /dev/null +++ b/catalyst/examples/mean_reversion_simple.py @@ -0,0 +1,248 @@ +# 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. +from datetime import timedelta + +import pandas as pd +import talib +# To run an algorithm in Catalyst, you need two functions: initialize and +# handle_data. +from logbook import Logger +from talib.common import MA_Type + +from catalyst import run_algorithm +from catalyst.api import symbol, record, order_target_percent, \ + get_open_orders +# 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.exchange.stats_utils import extract_transactions, trend_direction + +algo_namespace = 'momentum' +log = Logger(algo_namespace) + + +def initialize(context): + # This initialize function sets any data or variables that you'll use in + # your algorithm. For instance, you'll want to define the trading pair (or + # trading pairs) you want to backtest. You'll also want to define any + # parameters or values you're going to use. + + # In our example, we're looking at Ether in USD Tether. + context.eth_btc = symbol('etc_usdt') + context.base_price = None + context.current_day = None + + +def handle_data(context, data): + # This handle_data function is where the real work is done. Our data is + # minute-level tick data, and each minute is called a frame. This function + # runs on each frame of the data. + + # We flag the first period of each day. + # Since cryptocurrencies trade 24/7 the `before_trading_starts` handle + # would only execute once. This method works with minute and daily + # frequencies. + today = data.current_dt.floor('1D') + if today != context.current_day: + context.traded_today = False + context.current_day = today + + # We're computing the volume-weighted-average-price of the security + # defined above, in the context.eth_btc 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.eth_btc, + fields='close', + bar_count=50, + frequency='15T' + ) + + # Ta-lib calculates various technical indicator based on price and + # volume arrays. + + # In this example, we are comp + rsi = talib.RSI(prices.values, timeperiod=14) + + # 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.eth_btc, fields=['close', 'volume']) + price = current['close'] + + # If base_price is not set, we use the current value. This is the + # price at the first bar which we reference to calculate price_change. + if context.base_price is None: + context.base_price = price + + price_change = (price - context.base_price) / context.base_price + cash = context.portfolio.cash + + # Now that we've collected all current data for this frame, we use + # the record() method to save it. This data will be available as + # a parameter of the analyze() function for further analysis. + record( + price=price, + volume=current['volume'], + price_change=price_change, + rsi=rsi[-1], + cash=cash + ) + + # We are trying to avoid over-trading by limiting our trades to + # one per day. + if context.traded_today: + return + + # 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.eth_btc) + if len(orders) > 0: + return + + # Exit if we cannot trade + if not data.can_trade(context.eth_btc): + 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.eth_btc].amount + + if rsi[-1] <= 30 and pos_amount == 0: + log.info( + '{}: buying - price: {}, rsi: {}'.format( + data.current_dt, price, rsi[-1] + ) + ) + order_target_percent(context.eth_btc, 1) + context.traded_today = True + + elif rsi[-1] >= 80 and pos_amount > 0: + log.info( + '{}: selling - price: {}, rsi: {}'.format( + data.current_dt, price, rsi[-1] + ) + ) + order_target_percent(context.eth_btc, 0) + context.traded_today = True + + +def analyze(context=None, perf=None): + import matplotlib.pyplot as plt + + # The base currency of the algo exchange + base_currency = context.exchanges.values()[0].base_currency.upper() + + # Plot the portfolio value over time. + ax1 = plt.subplot(611) + perf.loc[:, 'portfolio_value'].plot(ax=ax1) + ax1.set_ylabel('Portfolio Value ({})'.format(base_currency)) + + # Plot the price increase or decrease over time. + ax2 = plt.subplot(612, sharex=ax1) + perf.loc[:, 'price'].plot(ax=ax2, label='Price') + + ax2.set_ylabel('{asset} ({base})'.format( + asset=context.eth_btc.symbol, base=base_currency + )) + + transaction_df = extract_transactions(perf) + if not transaction_df.empty: + buy_df = transaction_df[transaction_df['amount'] > 0] + sell_df = transaction_df[transaction_df['amount'] < 0] + ax2.scatter( + buy_df.index.to_pydatetime(), + perf.loc[buy_df.index, 'price'], + marker='^', + s=100, + c='green', + label='' + ) + ax2.scatter( + sell_df.index.to_pydatetime(), + perf.loc[sell_df.index, 'price'], + marker='v', + s=100, + c='red', + label='' + ) + + ax4 = plt.subplot(613, sharex=ax1) + perf.loc[:, 'cash'].plot( + ax=ax4, label='Base Currency ({})'.format(base_currency) + ) + ax4.set_ylabel('Cash ({})'.format(base_currency)) + + perf['algorithm'] = perf.loc[:, 'algorithm_period_return'] + + ax5 = plt.subplot(614, sharex=ax1) + perf.loc[:, ['algorithm', 'price_change']].plot(ax=ax5) + ax5.set_ylabel('Percent Change') + + ax6 = plt.subplot(615, sharex=ax1) + perf.loc[:, 'rsi'].plot(ax=ax6, label='RSI') + ax6.axhline(70, color='darkgoldenrod') + ax6.axhline(30, color='darkgoldenrod') + + if not transaction_df.empty: + ax6.scatter( + buy_df.index.to_pydatetime(), + perf.loc[buy_df.index, 'rsi'], + marker='^', + s=100, + c='green', + label='' + ) + ax6.scatter( + sell_df.index.to_pydatetime(), + perf.loc[sell_df.index, 'rsi'], + marker='v', + s=100, + c='red', + label='' + ) + plt.legend(loc=3) + + # Show the plot. + plt.gcf().set_size_inches(18, 8) + plt.show() + pass + + +if __name__ == '__main__': + # The execution mode: backtest or live + MODE = 'backtest' + + if MODE == 'backtest': + # catalyst run -f catalyst/examples/mean_reversion_simple.py -x poloniex -s 2017-7-1 -e 2017-7-31 -c usdt -n mean-reversion --data-frequency minute --capital-base 10000 + run_algorithm( + capital_base=10000, + data_frequency='minute', + initialize=initialize, + handle_data=handle_data, + analyze=analyze, + exchange_name='poloniex', + algo_namespace=algo_namespace, + base_currency='usdt', + start=pd.to_datetime('2017-7-1', utc=True), + end=pd.to_datetime('2017-7-31', utc=True), + ) + + elif MODE == 'live': + run_algorithm( + initialize=initialize, + handle_data=handle_data, + analyze=analyze, + exchange_name='poloniex', + live=True, + algo_namespace=algo_namespace, + base_currency='usdt', + live_graph=True + ) diff --git a/catalyst/examples/rsi_profit_target.py b/catalyst/examples/rsi_profit_target.py new file mode 100644 index 00000000..a426c402 --- /dev/null +++ b/catalyst/examples/rsi_profit_target.py @@ -0,0 +1,276 @@ +from datetime import timedelta + +import pandas as pd +import numpy as np +import talib +from logbook import Logger + +from catalyst.api import ( + order, + symbol, + record, + get_open_orders, +) +from catalyst.exchange.stats_utils import crossover, crossunder +from catalyst.utils.run_algo import run_algorithm + +algo_namespace = 'rsi' +log = Logger(algo_namespace) + + +def initialize(context): + log.info('initializing algo') + context.asset = symbol('eth_btc') + context.base_price = None + + context.MAX_HOLDINGS = 0.2 + context.RSI_OVERSOLD = 30 + context.RSI_OVERSOLD_BBANDS = 45 + context.RSI_OVERBOUGHT_BBANDS = 55 + context.SLIPPAGE_ALLOWED = 0.03 + + context.TARGET = 0.15 + context.STOP_LOSS = 0.1 + context.STOP = 0.03 + context.position = None + + context.last_bar = None + + context.errors = [] + pass + + +def _handle_buy_sell_decision(context, data, signal, price): + orders = get_open_orders(context.asset) + if len(orders) > 0: + log.info('skipping bar until all open orders execute') + return + + positions = context.portfolio.positions + if context.position is None and context.asset in positions: + position = positions[context.asset] + context.position = dict( + cost_basis=position['cost_basis'], + amount=position['amount'], + stop=None + ) + + action = None + if context.position is not None: + cost_basis = context.position['cost_basis'] + amount = context.position['amount'] + log.info( + 'found {amount} positions with cost basis {cost_basis}'.format( + amount=amount, + cost_basis=cost_basis + ) + ) + stop = context.position['stop'] + + target = cost_basis * (1 + context.TARGET) + if price >= target: + context.position['cost_basis'] = price + context.position['stop'] = context.STOP + + stop_target = context.STOP_LOSS if stop is None else context.STOP + if price < cost_basis * (1 - stop_target): + log.info('executing stop loss') + order( + asset=context.asset, + amount=-amount, + limit_price=price * (1 - context.SLIPPAGE_ALLOWED), + ) + action = 0 + context.position = None + + else: + if signal == 'long': + log.info('opening position') + buy_amount = context.MAX_HOLDINGS / price + order( + asset=context.asset, + amount=buy_amount, + limit_price=price * (1 + context.SLIPPAGE_ALLOWED), + ) + context.position = dict( + cost_basis=price, + amount=buy_amount, + stop=None + ) + action = 0 + + +def _handle_data_rsi_only(context, data): + price = data.current(context.asset, 'close') + log.info('got price {price}'.format(price=price)) + + if price is np.nan: + log.warn('no pricing data') + return + + if context.base_price is None: + context.base_price = price + + try: + prices = data.history( + context.asset, + fields='price', + bar_count=17, + frequency='30T' + ) + except Exception as e: + log.warn('historical data not available: '.format(e)) + return + + rsi = talib.RSI(prices.values, timeperiod=16)[-1] + log.info('got rsi {}'.format(rsi)) + + signal = None + if rsi < context.RSI_OVERSOLD: + signal = 'long' + + # Making sure that the price is still current + price = data.current(context.asset, 'close') + cash = context.portfolio.cash + log.info( + 'base currency available: {cash}, cap: {cap}'.format( + cash=cash, + cap=context.MAX_HOLDINGS + ) + ) + volume = data.current(context.asset, 'volume') + price_change = (price - context.base_price) / context.base_price + record( + price=price, + price_change=price_change, + rsi=rsi, + volume=volume, + cash=cash, + starting_cash=context.portfolio.starting_cash, + leverage=context.account.leverage, + ) + + _handle_buy_sell_decision(context, data, signal, price) + + +def handle_data(context, data): + dt = data.current_dt + + if context.last_bar is None or ( + context.last_bar + timedelta(minutes=15)) <= dt: + context.last_bar = dt + else: + return + + log.info('BAR {}'.format(dt)) + try: + _handle_data_rsi_only(context, data) + except Exception as e: + log.warn('aborting the bar on error {}'.format(e)) + context.errors.append(e) + + if len(context.errors) > 0: + log.info('the errors:\n{}'.format(context.errors)) + + +def analyze(context=None, results=None): + import matplotlib.pyplot as plt + + base_currency = context.exchanges.values()[0].base_currency.upper() + # Plot the portfolio and asset data. + ax1 = plt.subplot(611) + results.loc[:, 'portfolio_value'].plot(ax=ax1) + ax1.set_ylabel('Portfolio Value ({})'.format(base_currency)) + + ax2 = plt.subplot(612, sharex=ax1) + results.loc[:, 'price'].plot(ax=ax2) + ax2.set_ylabel('{asset} ({base})'.format( + asset=context.asset.symbol, base=base_currency + )) + + trans = results.loc[[t != [] for t in results.transactions], :] + buys = trans.loc[[t[0]['amount'] > 0 for t in trans.transactions], :] + sells = trans.loc[[t[0]['amount'] < 0 for t in trans.transactions], :] + # buys = results.loc[results['action'] == 1, :] + # sells = results.loc[results['action'] == 0, :] + + ax2.plot( + buys.index, + results.loc[buys.index, 'price'], + '^', + markersize=10, + color='g', + ) + ax2.plot( + sells.index, + results.loc[sells.index, 'price'], + 'v', + markersize=10, + color='r', + ) + + ax3 = plt.subplot(613, sharex=ax1) + results.loc[:, ['alpha', 'beta']].plot(ax=ax3) + ax3.set_ylabel('Alpha / Beta ') + + ax4 = plt.subplot(614, sharex=ax1) + results.loc[:, ['starting_cash', 'cash']].plot(ax=ax4) + ax4.set_ylabel('Base Currency ({})'.format(base_currency)) + + results['algorithm'] = results.loc[:, 'algorithm_period_return'] + + ax5 = plt.subplot(615, sharex=ax1) + results.loc[:, ['algorithm', 'price_change']].plot(ax=ax5) + ax5.set_ylabel('Percent Change') + + ax6 = plt.subplot(616, sharex=ax1) + results.loc[:, 'rsi'].plot(ax=ax6) + ax6.set_ylabel('RSI') + + ax6.plot( + buys.index, + results.loc[buys.index, 'rsi'], + '^', + markersize=10, + color='g', + ) + ax6.plot( + sells.index, + results.loc[sells.index, 'rsi'], + 'v', + markersize=10, + color='r', + ) + + plt.legend(loc=3) + + # Show the plot. + plt.gcf().set_size_inches(18, 8) + plt.show() + pass + + +run_algorithm( + initialize=initialize, + handle_data=handle_data, + analyze=analyze, + exchange_name='bittrex', + live=True, + algo_namespace=algo_namespace, + base_currency='btc', + live_graph=False +) + +# Backtest +# run_algorithm( +# capital_base=0.5, +# data_frequency='minute', +# initialize=initialize, +# handle_data=handle_data, +# analyze=analyze, +# exchange_name='poloniex', +# algo_namespace=algo_namespace, +# base_currency='btc', +# start=pd.to_datetime('2017-9-1', utc=True), +# end=pd.to_datetime('2017-10-1', utc=True), +# ) diff --git a/catalyst/examples/simple_loop.py b/catalyst/examples/simple_loop.py index 282c304e..8e6ce22e 100644 --- a/catalyst/examples/simple_loop.py +++ b/catalyst/examples/simple_loop.py @@ -7,7 +7,7 @@ from catalyst.api import symbol def initialize(context): print('initializing') - context.asset = symbol('eth_btc') + context.asset = symbol('swift_btc') def handle_data(context, data): @@ -20,8 +20,8 @@ def handle_data(context, data): prices = data.history( context.asset, fields='price', - bar_count=16, - frequency='5T' + bar_count=15, + frequency='1D' ) rsi = talib.RSI(prices.values, timeperiod=14)[-1] print('got rsi: {}'.format(rsi)) @@ -31,13 +31,13 @@ def handle_data(context, data): run_algorithm( capital_base=250, - start=pd.to_datetime('2016-6-1', utc=True), - end=pd.to_datetime('2016-12-31', utc=True), + start=pd.to_datetime('2015-4-1', utc=True), + end=pd.to_datetime('2017-11-1', utc=True), data_frequency='daily', initialize=initialize, handle_data=handle_data, analyze=None, - exchange_name='bitfinex', + exchange_name='bittrex', algo_namespace='simple_loop', base_currency='btc' ) diff --git a/catalyst/exchange/bittrex/bittrex_api.py b/catalyst/exchange/bittrex/bittrex_api.py index bc31607d..ca5f9c4a 100644 --- a/catalyst/exchange/bittrex/bittrex_api.py +++ b/catalyst/exchange/bittrex/bittrex_api.py @@ -3,11 +3,12 @@ import json import time import hmac import hashlib - +import ssl # Workaround for backwards compatibility # https://stackoverflow.com/questions/3745771/urllib-request-in-python-2-7 from six.moves import urllib + urlopen = urllib.request.urlopen @@ -48,7 +49,8 @@ class Bittrex_api(object): headers = {} req = urllib.request.Request(url, headers=headers) - response = json.loads(urlopen(req).read()) + response = json.loads(urlopen( + req, context=ssl._create_unverified_context()).read()) if response["result"]: return response["result"] diff --git a/catalyst/exchange/bundle_utils.py b/catalyst/exchange/bundle_utils.py index 2210aa7b..9b0c8c49 100644 --- a/catalyst/exchange/bundle_utils.py +++ b/catalyst/exchange/bundle_utils.py @@ -149,7 +149,7 @@ def get_periods(start_dt, end_dt, freq): return len(get_periods_range(start_dt, end_dt, freq)) -def get_start_dt(end_dt, bar_count, data_frequency): +def get_start_dt(end_dt, bar_count, data_frequency, include_first=True): """ The start date based on specified end date and data frequency. @@ -168,6 +168,9 @@ def get_start_dt(end_dt, bar_count, data_frequency): if periods > 1: delta = get_delta(periods, data_frequency) start_dt = end_dt - delta + + if not include_first: + start_dt += get_delta(1, data_frequency) else: start_dt = end_dt diff --git a/catalyst/exchange/exchange_algorithm.py b/catalyst/exchange/exchange_algorithm.py index c05e559f..f05bee7e 100644 --- a/catalyst/exchange/exchange_algorithm.py +++ b/catalyst/exchange/exchange_algorithm.py @@ -10,7 +10,6 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -import os import pickle import signal import sys @@ -27,8 +26,6 @@ from catalyst.assets._assets import TradingPair import catalyst.protocol as zp from catalyst.algorithm import TradingAlgorithm from catalyst.constants import LOG_LEVEL -from catalyst.data.minute_bars import BcolzMinuteBarWriter, \ - BcolzMinuteBarReader from catalyst.errors import OrderInBeforeTradingStart from catalyst.exchange.exchange_blotter import ExchangeBlotter from catalyst.exchange.exchange_errors import ( @@ -38,8 +35,8 @@ from catalyst.exchange.exchange_errors import ( OrphanOrderError) from catalyst.exchange.exchange_execution import ExchangeStopLimitOrder, \ ExchangeLimitOrder, ExchangeStopOrder -from catalyst.exchange.exchange_utils import get_exchange_minute_writer_root, \ - save_algo_object, get_algo_object, get_algo_folder, get_algo_df, \ +from catalyst.exchange.exchange_utils import save_algo_object, get_algo_object, \ + get_algo_folder, get_algo_df, \ save_algo_df from catalyst.exchange.live_graph_clock import LiveGraphClock from catalyst.exchange.simple_clock import SimpleClock @@ -182,17 +179,19 @@ class ExchangeTradingAlgorithmBase(TradingAlgorithm): # we want the key to be absent, not just empty # Only include transactions for given dt - stats['transactions'] = dict() + stats['transactions'] = [] for date in period.processed_transactions: if start_dt <= date < end_dt: - stats['transactions'][date] = \ - period.processed_transactions[date] + transactions = period.processed_transactions[date] + for t in transactions: + stats['transactions'].append(t.to_dict()) - stats['orders'] = dict() + stats['orders'] = [] for date in period.orders_by_modified: if start_dt <= date < end_dt: - stats['orders'][date] = \ - period.orders_by_modified[date] + orders = period.orders_by_modified[date] + for order in orders: + stats['orders'].append(orders[order].to_dict()) return stats @@ -201,6 +200,7 @@ class ExchangeTradingAlgorithmBacktest(ExchangeTradingAlgorithmBase): def __init__(self, *args, **kwargs): super(ExchangeTradingAlgorithmBacktest, self).__init__(*args, **kwargs) + self.frame_stats = list() self.blotter = ExchangeBlotter( data_frequency=self.data_frequency, # Default to NeverCancel in catalyst @@ -245,6 +245,19 @@ class ExchangeTradingAlgorithmBacktest(ExchangeTradingAlgorithmBase): else: return MarketOrder() + def handle_data(self, data): + super(ExchangeTradingAlgorithmBacktest, self).handle_data(data) + + minute_stats = self.prepare_period_stats( + data.current_dt, data.current_dt + timedelta(minutes=1)) + self.frame_stats.append(minute_stats) + + def analyze(self, perf): + stats = pd.DataFrame(self.frame_stats) + stats.set_index('period_close', inplace=True, drop=False) + + super(ExchangeTradingAlgorithmBacktest, self).analyze(stats) + class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): def __init__(self, *args, **kwargs): @@ -273,34 +286,11 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): self.stats_minutes = 5 super(ExchangeTradingAlgorithmLive, self).__init__(*args, **kwargs) - # TODO: fix precision before re-enabling - # self._create_minute_writer() signal.signal(signal.SIGINT, self.signal_handler) log.info('initialized trading algorithm in live mode') - def _create_minute_writer(self): - root = get_exchange_minute_writer_root(self.exchange.name) - filename = os.path.join(root, 'metadata.json') - - if os.path.isfile(filename): - writer = BcolzMinuteBarWriter.open( - root, self.sim_params.end_session) - else: - # TODO: need to be able to write more precise numbers - writer = BcolzMinuteBarWriter( - rootdir=root, - calendar=self.trading_calendar, - minutes_per_day=1440, - start_session=self.sim_params.start_session, - end_session=self.sim_params.end_session, - write_metadata=True - ) - - self.exchange.minute_writer = writer - self.exchange.minute_reader = BcolzMinuteBarReader(root) - def signal_handler(self, signal, frame): """ Handles the keyboard interruption signal. diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index b5c66d7b..48d3294f 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/catalyst/exchange/exchange_bundle.py @@ -1,47 +1,24 @@ import os import shutil +from functools import partial from itertools import chain +from operator import is_not +import numpy as np import pandas as pd from catalyst.assets._assets import TradingPair +from datetime import datetime, timedelta from logbook import Logger -from pandas.tslib import Timestamp 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_delta, get_month_start_end, \ - get_year_start_end, get_df_from_arrays, get_start_dt, get_period_label -from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \ - BcolzExchangeBarWriter -from catalyst.exchange.exchange_errors import EmptyValuesInBundleError, \ - TempBundleNotFoundError, \ - NoDataAvailableOnExchange, \ - PricingDataNotLoadedError -from catalyst.exchange.exchange_utils import get_exchange_folder -from catalyst.utils.cli import maybe_show_progress -from catalyst.utils.paths import ensure_directory -import os -import shutil -from itertools import chain - -import pandas as pd -from catalyst.assets._assets import TradingPair -from logbook import Logger -from pandas.tslib import Timestamp -from pytz import UTC -from six import itervalues - -from catalyst import get_calendar -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_delta, get_month_start_end, \ + get_bcolz_chunk, get_month_start_end, \ get_year_start_end, get_df_from_arrays, get_start_dt, get_period_label from catalyst.exchange.exchange_bcolz import BcolzExchangeBarReader, \ BcolzExchangeBarWriter @@ -244,8 +221,91 @@ class ExchangeBundle: 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='strip'): + empty_rows_behavior='warn', duplicates_threshold=None): """ Ingest a DataFrame of OHLCV data for a given market. @@ -258,50 +318,16 @@ class ExchangeBundle: empty_rows_behavior: str """ + problems = [] if empty_rows_behavior is not 'ignore': - nan_rows = ohlcv_df[ohlcv_df.isnull().T.any().T].index + problems += self._spot_empty_periods( + ohlcv_df, asset, data_frequency, empty_rows_behavior + ) - if len(nan_rows) > 0: - dates = [] - previous_date = None - for row_date in nan_rows.values: - row_date = pd.to_datetime(row_date) - - if previous_date is None: - dates.append(row_date) - - else: - seq_date = previous_date + get_delta(1, data_frequency) - - if row_date > seq_date: - dates.append(previous_date) - dates.append(row_date) - - previous_date = row_date - - dates.append(pd.to_datetime(nan_rows.values[-1])) - - name = '{} from {} to {}'.format( - asset.symbol, ohlcv_df.index[0], ohlcv_df.index[-1] - ) - if empty_rows_behavior == 'warn': - log.warn( - '\n{name} with end minute {end_minute} has empty rows ' - 'in ranges: {dates}'.format( - name=name, - end_minute=asset.end_minute, - dates=dates - ) - ) - - elif empty_rows_behavior == 'raise': - raise EmptyValuesInBundleError( - name=name, - end_minute=asset.end_minute, - dates=dates - ) - else: - ohlcv_df.dropna(inplace=True) + # if duplicates_threshold is not None: + # problems += self._spot_duplicates( + # ohlcv_df, asset, data_frequency, duplicates_threshold + # ) data = [] if not ohlcv_df.empty: @@ -310,8 +336,11 @@ class ExchangeBundle: self._write(data, writer, data_frequency) + return problems + def ingest_ctable(self, asset, data_frequency, period, - writer, empty_rows_behavior='strip', cleanup=False): + writer, empty_rows_behavior='strip', + duplicates_threshold=100, cleanup=False): """ Merge a ctable bundle chunk into the main bundle for the exchange. @@ -327,8 +356,14 @@ class ExchangeBundle: cleanup: bool Remove the temp bundle directory after ingestion. - :return: + 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, @@ -375,12 +410,13 @@ class ExchangeBundle: start_dt, end_dt, data_frequency ) df = get_df_from_arrays(arrays, periods) - self.ingest_df( + problems += self.ingest_df( ohlcv_df=df, data_frequency=data_frequency, asset=asset, writer=writer, - empty_rows_behavior=empty_rows_behavior + empty_rows_behavior=empty_rows_behavior, + duplicates_threshold=duplicates_threshold ) if cleanup: @@ -390,7 +426,7 @@ class ExchangeBundle: ) shutil.rmtree(reader._rootdir) - return reader._rootdir + return filter(partial(is_not, None), problems) def get_adj_dates(self, start, end, assets, data_frequency): """ @@ -528,7 +564,8 @@ class ExchangeBundle: return chunks def ingest_assets(self, assets, data_frequency, start_dt=None, end_dt=None, - show_progress=False, asset_chunks=False): + show_progress=False, show_breakdown=False, + show_report=False): """ Determine if data is missing from the bundle and attempt to ingest it. @@ -539,7 +576,7 @@ class ExchangeBundle: start_dt: datetime end_dt: datetime show_progress: bool - asset_chunks: bool + show_breakdown: bool """ if start_dt is None: @@ -562,10 +599,11 @@ class ExchangeBundle: 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 asset_chunks: + if show_breakdown: for asset in chunks: with maybe_show_progress( chunks[asset], @@ -577,7 +615,7 @@ class ExchangeBundle: symbol=asset.symbol )) as it: for chunk in it: - self.ingest_ctable( + problems += self.ingest_ctable( asset=chunk['asset'], data_frequency=data_frequency, period=chunk['period'], @@ -601,7 +639,7 @@ class ExchangeBundle: frequency=data_frequency, )) as it: for chunk in it: - self.ingest_ctable( + problems += self.ingest_ctable( asset=chunk['asset'], data_frequency=data_frequency, period=chunk['period'], @@ -610,9 +648,14 @@ class ExchangeBundle: cleanup=True ) + if show_report and len(problems) > 0: + log.info('problems during ingestion:{}\n'.format( + '\n'.join(problems) + )) + def ingest(self, data_frequency, include_symbols=None, exclude_symbols=None, start=None, end=None, - show_progress=True, environ=os.environ): + show_progress=True, show_breakdown=True, show_report=True): """ Inject data based on specified parameters. @@ -631,7 +674,7 @@ class ExchangeBundle: for frequency in data_frequency.split(','): self.ingest_assets(assets, frequency, start, end, - show_progress, True) + show_progress, show_breakdown, show_report) def get_history_window_series_and_load(self, assets, @@ -658,7 +701,46 @@ class ExchangeBundle: Series """ - try: + if 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 + ) + 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, + 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 + ) + return series + + else: series = self.get_history_window_series( assets=assets, end_dt=end_dt, @@ -668,34 +750,6 @@ class ExchangeBundle: ) 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, - data_frequency=data_frequency, - show_progress=True, - asset_chunks=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=False - ) - return series - def get_spot_values(self, assets, field, @@ -707,12 +761,18 @@ class ExchangeBundle: The spot values for the gives assets, field and date. Reads from the exchange data bundle. - :param assets: - :param field: - :param dt: - :param data_frequency: - :param reset_reader: - :return: + Parameters + ---------- + assets: list[TradingPair] + field: str + dt: pd.Timestamp + data_frequency: str + reset_reader: + + Returns + ------- + float + """ values = [] try: @@ -739,7 +799,9 @@ class ExchangeBundle: exchange=self.exchange.name, symbols=symbols, symbol_list=','.join(symbols), - data_frequency=data_frequency + data_frequency=data_frequency, + start_dt=dt, + end_dt=dt ) def get_history_window_series(self, @@ -749,7 +811,7 @@ class ExchangeBundle: field, data_frequency, reset_reader=False): - start_dt = get_start_dt(end_dt, bar_count, data_frequency) + start_dt = get_start_dt(end_dt, bar_count, data_frequency, False) start_dt, end_dt = self.get_adj_dates( start_dt, end_dt, assets, data_frequency ) @@ -767,7 +829,9 @@ class ExchangeBundle: exchange=self.exchange.name, symbols=symbols, symbol_list=','.join(symbols), - data_frequency=data_frequency + data_frequency=data_frequency, + start_dt=start_dt, + end_dt=end_dt ) for asset in assets: @@ -785,7 +849,9 @@ class ExchangeBundle: exchange=self.exchange.name, symbols=asset.symbol, symbol_list=asset.symbol, - data_frequency=data_frequency + data_frequency=data_frequency, + start_dt=asset_start_dt, + end_dt=asset_end_dt ) series = dict() @@ -805,7 +871,9 @@ class ExchangeBundle: exchange=self.exchange.name, symbols=symbols, symbol_list=','.join(symbols), - data_frequency=data_frequency + data_frequency=data_frequency, + start_dt=start_dt, + end_dt=end_dt ) periods = self.get_calendar_periods_range( diff --git a/catalyst/exchange/exchange_data_portal.py b/catalyst/exchange/exchange_data_portal.py index a6cf4db7..43feaa9c 100644 --- a/catalyst/exchange/exchange_data_portal.py +++ b/catalyst/exchange/exchange_data_portal.py @@ -6,7 +6,7 @@ import pandas as pd from catalyst.assets._assets import TradingPair from logbook import Logger -from catalyst.constants import LOG_LEVEL +from catalyst.constants import LOG_LEVEL, AUTO_INGEST from catalyst.data.data_portal import DataPortal from catalyst.exchange.exchange_bundle import ExchangeBundle from catalyst.exchange.exchange_errors import ( @@ -378,24 +378,28 @@ class DataPortalExchangeBacktest(DataPortalExchangeBase): else: dt = dt.floor('1 min') - try: - return bundle.get_spot_values(assets, field, dt, data_frequency) - - except PricingDataNotLoadedError: - log.info( - 'pricing data for {symbol} not found on {dt}' - ', updating the bundles.'.format( - symbol=[asset.symbol for asset in assets], - dt=dt + if AUTO_INGEST: + try: + return bundle.get_spot_values( + assets, field, dt, data_frequency ) - ) - bundle.ingest_assets( - assets=assets, - start_dt=self._first_trading_day, - end_dt=self._last_available_session, - data_frequency=data_frequency, - show_progress=True - ) - return bundle.get_spot_values( - assets, field, dt, data_frequency, True - ) + except PricingDataNotLoadedError: + log.info( + 'pricing data for {symbol} not found on {dt}' + ', updating the bundles.'.format( + symbol=[asset.symbol for asset in assets], + dt=dt + ) + ) + bundle.ingest_assets( + assets=assets, + start_dt=self._first_trading_day, + end_dt=self._last_available_session, + data_frequency=data_frequency, + show_progress=True + ) + return bundle.get_spot_values( + assets, field, dt, data_frequency, True + ) + else: + return bundle.get_spot_values(assets, field, dt, data_frequency) diff --git a/catalyst/exchange/exchange_errors.py b/catalyst/exchange/exchange_errors.py index 00e53395..35f320d8 100644 --- a/catalyst/exchange/exchange_errors.py +++ b/catalyst/exchange/exchange_errors.py @@ -211,12 +211,11 @@ class PricingDataBeforeTradingError(ZiplineError): class PricingDataNotLoadedError(ZiplineError): - msg = ('Pricing data {field} for trading pairs {symbols} trading on ' - 'exchange {exchange} since {first_trading_day} is unavailable. ' - 'The bundle data is either out-of-date or has not been loaded yet. ' - 'Please ingest data using the command ' - '`catalyst ingest-exchange -x {exchange} -f {data_frequency} -i {symbol_list}`. ' - 'See catalyst documentation for details.').strip() + msg = ('Missing data for {exchange} {symbols} in date range ' + '[{start_dt} - {end_dt}]' + '\nPlease run: `catalyst ingest-exchange -x {exchange} -f ' + '{data_frequency} -i {symbol_list}`. See catalyst documentation ' + 'for details.').strip() class ApiCandlesError(ZiplineError): diff --git a/catalyst/exchange/exchange_utils.py b/catalyst/exchange/exchange_utils.py index b4f828fe..554825b4 100644 --- a/catalyst/exchange/exchange_utils.py +++ b/catalyst/exchange/exchange_utils.py @@ -2,6 +2,7 @@ import json import os import pickle import re +import shutil from datetime import date, datetime import pandas as pd @@ -158,6 +159,24 @@ def get_exchange_auth(exchange_name, environ=None): return data +def delete_algo_folder(algo_name, environ=None): + """ + Delete the folder containing the algo state. + + Parameters + ---------- + algo_name: str + environ: + + Returns + ------- + str + + """ + folder = get_algo_folder(algo_name, environ) + shutil.rmtree(folder) + + def get_algo_folder(algo_name, environ=None): """ The algorithm root folder of the algorithm. diff --git a/catalyst/exchange/poloniex/poloniex_api.py b/catalyst/exchange/poloniex/poloniex_api.py index 8bf6bb83..894ba08b 100644 --- a/catalyst/exchange/poloniex/poloniex_api.py +++ b/catalyst/exchange/poloniex/poloniex_api.py @@ -3,6 +3,7 @@ import json import time import hmac import hashlib +import ssl from six.moves import urllib @@ -104,9 +105,10 @@ class Poloniex_api(object): req = urllib.request.Request( url, data=post_data, - headers=headers + headers=headers, ) - return json.loads(urlopen(req).read()) + return json.loads( + urlopen(req, context=ssl._create_unverified_context()).read()) def returnticker(self): return self.query('returnTicker', {}) diff --git a/catalyst/exchange/stats_utils.py b/catalyst/exchange/stats_utils.py index b7bfda98..1290f71f 100644 --- a/catalyst/exchange/stats_utils.py +++ b/catalyst/exchange/stats_utils.py @@ -1,7 +1,19 @@ +import numbers + import numpy as np import pandas as pd +def trend_direction(series): + if series[-1] is np.nan or series[-1] is np.nan: + return None + + if series[-1] > series[-2]: + return 'up' + else: + return 'down' + + def crossover(source, target): """ The `x`-series is defined as having crossed over `y`-series if the value @@ -44,14 +56,56 @@ def crossunder(source, target): bool """ - if source[-1] is np.nan or source[-2] is np.nan \ - or target[-1] is np.nan or target[-2] is np.nan: - return False + if isinstance(target, numbers.Number): + if source[-1] is np.nan or source[-2] is np.nan \ + or target is np.nan: + return False - if source[-1] < target[-1] and source[-2] > target[-2]: - return True + if source[-1] < target <= source[-2]: + return True + else: + return False else: - return False + if source[-1] is np.nan or source[-2] is np.nan \ + or target[-1] is np.nan or target[-2] is np.nan: + return False + + if source[-1] < target[-1] and source[-2] >= target[-2]: + return True + else: + return False + + +def vwap(df): + """ + Volume-weighted average price (VWAP) is a ratio generally used by + institutional investors and mutual funds to make buys and sells so as not + to disturb the market prices with large orders. It is the average share + price of a stock weighted against its trading volume within a particular + time frame, generally one day. + + Read more: Volume Weighted Average Price - VWAP + https://www.investopedia.com/terms/v/vwap.asp#ixzz4xt922daE + + Parameters + ---------- + df: pd.DataFrame + + Returns + ------- + + """ + if 'close' not in df.columns or 'volume' not in df.columns: + raise ValueError('price data must include `volume` and `close`') + + vol_sum = np.nansum(df['volume'].values) + + try: + ret = np.nansum(df['close'].values * df['volume'].values) / vol_sum + except ZeroDivisionError: + ret = np.nan + + return ret def get_pretty_stats(stats_df, recorded_cols=None, num_rows=10): @@ -129,3 +183,28 @@ def df_to_string(df): pd.set_option('display.max_colwidth', 1000) return df.to_string() + + +def extract_transactions(perf): + """ + Compute indexes for buy and sell transactions + + Parameters + ---------- + perf: DataFrame + The algo performance DataFrame. + + Returns + ------- + DataFrame + A DataFrame of transactions. + + """ + trans_list = perf.transactions.values + all_trans = [t for sublist in trans_list for t in sublist] + all_trans.sort(key=lambda t: t['dt']) + + transactions = pd.DataFrame(all_trans) + if not transactions.empty: + transactions.set_index('dt', inplace=True, drop=True) + return transactions diff --git a/catalyst/finance/execution.py b/catalyst/finance/execution.py index fe7c1c89..8fec0af6 100644 --- a/catalyst/finance/execution.py +++ b/catalyst/finance/execution.py @@ -77,6 +77,7 @@ class LimitOrder(ExecutionStyle): Execution style representing an order to be executed at a price equal to or better than a specified limit price. """ + def __init__(self, limit_price, exchange=None): """ Store the given price. @@ -99,6 +100,7 @@ class StopOrder(ExecutionStyle): Execution style representing an order to be placed once the market price reaches a specified stop price. """ + def __init__(self, stop_price, exchange=None): """ Store the given price. @@ -121,6 +123,7 @@ class StopLimitOrder(ExecutionStyle): Execution style representing a limit order to be placed with a specified limit price once the market reaches a specified stop price. """ + def __init__(self, limit_price, stop_price, exchange=None): """ Store the given prices @@ -144,31 +147,20 @@ class StopLimitOrder(ExecutionStyle): def asymmetric_round_price_to_penny(price, prefer_round_down, diff=(0.0095 - .005)): """ - Asymmetric rounding function for adjusting prices to two places in a way - that "improves" the price. For limit prices, this means preferring to - round down on buys and preferring to round up on sells. For stop prices, - it means the reverse. + Modified the original function because we do not want to round + prices on crypto exchange. - If prefer_round_down == True: - When .05 below to .95 above a penny, use that penny. - If prefer_round_down == False: - When .95 below to .05 above a penny, use that penny. + Parameters + ---------- + price: float + + Returns + ------- + float - In math-speak: - If prefer_round_down: [.0095, X.0195) -> round to X.01. - If not prefer_round_down: (.0005, X.0105] -> round to X.01. """ - # Subtracting an epsilon from diff to enforce the open-ness of the upper - # bound on buys and the lower bound on sells. Using the actual system - # epsilon doesn't quite get there, so use a slightly less epsilon-ey value. - epsilon = float_info.epsilon * 10 - diff = diff - epsilon - - # relies on rounding half away from zero, unlike numpy's bankers' rounding - rounded = round(price - (diff if prefer_round_down else -diff), 2) - if zp_math.tolerant_equals(rounded, 0.0): - return 0.0 - return rounded + # TODO: consider overriding outside of the original function + return price def check_stoplimit_prices(price, label): diff --git a/catalyst/finance/risk/cumulative.py b/catalyst/finance/risk/cumulative.py index 2a1b70bb..324547f5 100644 --- a/catalyst/finance/risk/cumulative.py +++ b/catalyst/finance/risk/cumulative.py @@ -22,7 +22,7 @@ from pandas.tseries.tools import normalize_date from six import iteritems -from . risk import ( +from .risk import ( check_entry, choose_treasury ) @@ -37,15 +37,16 @@ from empyrical import ( sharpe_ratio, sortino_ratio, ) - +import warnings from catalyst.constants import LOG_LEVEL log = logbook.Logger('Risk Cumulative', level=LOG_LEVEL) - choose_treasury = functools.partial(choose_treasury, lambda *args: '10year', compound=False) +warnings.filterwarnings('error') + class RiskMetricsCumulative(object): """ @@ -191,9 +192,12 @@ class RiskMetricsCumulative(object): if len(self.benchmark_returns) == 1: self.benchmark_returns = np.append(0.0, self.benchmark_returns) - self.benchmark_cumulative_returns[dt_loc] = cum_returns( - self.benchmark_returns - )[-1] + try: + self.benchmark_cumulative_returns[dt_loc] = cum_returns( + self.benchmark_returns + )[-1] + except Exception as e: + log.debug('cumulative returns error: {}'.format(e)) benchmark_cumulative_returns_to_date = \ self.benchmark_cumulative_returns[:dt_loc + 1] @@ -268,10 +272,15 @@ algorithm_returns ({algo_count}) in range {start} : {end} on {dt}" self.downside_risk[dt_loc] = downside_risk( self.algorithm_returns ) - self.sortino[dt_loc] = sortino_ratio( - self.algorithm_returns, - _downside_risk=self.downside_risk[dt_loc] - ) + + try: + self.sortino[dt_loc] = sortino_ratio( + self.algorithm_returns, + _downside_risk=self.downside_risk[dt_loc] + ) + except Exception as e: + log.debug('sortino ratio error: {}'.format(e)) + self.information[dt_loc] = information_ratio( self.algorithm_returns, self.benchmark_returns, @@ -294,18 +303,18 @@ algorithm_returns ({algo_count}) in range {start} : {end} on {dt}" rval = { 'trading_days': self.num_trading_days, 'benchmark_volatility': - self.benchmark_volatility[dt_loc], + self.benchmark_volatility[dt_loc], 'algo_volatility': - self.algorithm_volatility[dt_loc], + self.algorithm_volatility[dt_loc], 'treasury_period_return': self.treasury_period_return, # Though the two following keys say period return, # they would be more accurately called the cumulative return. # However, the keys need to stay the same, for now, for backwards # compatibility with existing consumers. 'algorithm_period_return': - self.algorithm_cumulative_returns[dt_loc], + self.algorithm_cumulative_returns[dt_loc], 'benchmark_period_return': - self.benchmark_cumulative_returns[dt_loc], + self.benchmark_cumulative_returns[dt_loc], 'beta': self.beta[dt_loc], 'alpha': self.alpha[dt_loc], 'sharpe': self.sharpe[dt_loc], diff --git a/catalyst/support/issue_57.py b/catalyst/support/issue_57.py index a7bf0226..f7bfcd18 100644 --- a/catalyst/support/issue_57.py +++ b/catalyst/support/issue_57.py @@ -39,4 +39,8 @@ run_algorithm( exchange_name='bittrex', algo_namespace='issue_57', base_currency='btc' +<<<<<<< HEAD ) +======= +) +>>>>>>> develop diff --git a/docs/source/releases.rst b/docs/source/releases.rst index 01b215cb..6945a1b7 100644 --- a/docs/source/releases.rst +++ b/docs/source/releases.rst @@ -2,6 +2,25 @@ Release Notes ============= +Version 0.3.7 +^^^^^^^^^^^^^ +**Release Date**: 2017-11-14 + +Bug Fixes +~~~~~~~~~ + +- Fixed an SSL cert issue (:issue:`64`) +- Fixed cumulative stats warnings (:issue:`63`) +- Disabled auto-ingestion because of unresolved caching issues (:issue:`47`) +- Standardized live-trading stats (:issue:`61`) + +Build +~~~~~ + +- Added a mean-reversion sample algo +- Added minutely stats in the analyze() function (:issue:`62`) +- Added specificity to some error messages + Version 0.3.6 ^^^^^^^^^^^^^ **Release Date**: 2017-11-4 @@ -31,7 +50,8 @@ Bug Fixes - Fixed issue with sell orders in backtesting - Fixed data frequency issues with data.history() in backtesting - Fixed an issue with can_trade() -- Reduced the commission and slippage values to account for lower volume transactions +- Reduced the commission and slippage values to account for lower volume + transactions Build ~~~~~ @@ -42,12 +62,18 @@ Documentation ~~~~~~~~~~~~~ - Improved installation notes for Windows C++ compiler and Conda -- Addition of `Jupyter Notebook guide `_ -- Addition of `Live Trading page `_ -- Addition of `Videos page `_ -- Addition of `Resources page `_ -- Addition of `Development Guidelines `_ -- Addition of `Release Notes `_ +- Addition of + `Jupyter Notebook guide `_ +- Addition of + `Live Trading page `_ +- Addition of + `Videos page `_ +- Addition of + `Resources page `_ +- Addition of `Development Guidelines + `_ +- Addition of + `Release Notes `_ - Updated code docstrings @@ -97,9 +123,11 @@ Bug Fixes ~~~~~~~~~ - Fixed OS-dependent path issue in data bundle -- Changed handling of empty ``auth.json``, instead of throwing an error for missing file +- Changed handling of empty ``auth.json``, instead of throwing an error for + missing file - Updated ``etc/python2.7-environment.yml`` to work with Catalyst version 0.3 -- Updated ``catalyst/examples/buy_and_hodl.py`` and ``catalyst/examples/buy_low_sell_high.py`` to work with Catalyst version 0.3 +- Updated ``catalyst/examples/buy_and_hodl.py`` and + ``catalyst/examples/buy_low_sell_high.py`` to work with Catalyst version 0.3 Version 0.3 @@ -118,15 +146,19 @@ Version 0.2.dev5 ^^^^^^^^^^^^^^^^ **Release Date**: 2017-10-03 -- Fixes bug in data.history function that was formatting 'volume' data as integers, now they are returned as floats with up to 9 decimals of precision. Data bundles redone. +- Fixes bug in data.history function that was formatting 'volume' data as + integers, now they are returned as floats with up to 9 decimals of precision. + Data bundles redone. Version 0.2.dev4 ^^^^^^^^^^^^^^^^ **Release Date**: 2017-09-20 -- Fixes bug in the pricing resolution of 1-minute data, now set to 8 decimal places. Pricing resolution of daily data remains set to 9 decimal places. -- The current data bundle takes 340MB compressed for download, and 460MB uncompressed on disk for Catalyst to use. +- Fixes bug in the pricing resolution of 1-minute data, now set to 8 decimal + places. Pricing resolution of daily data remains set to 9 decimal places. +- The current data bundle takes 340MB compressed for download, and 460MB + uncompressed on disk for Catalyst to use. Version 0.2.dev3 ^^^^^^^^^^^^^^^^ @@ -135,9 +167,12 @@ Version 0.2.dev3 - 1-minute resolution OHLCV data bundle for backtesting from Poloniex exchange - Implementation of trading of fractional crypto assets (i.e. 0.01 BTC) -- Minimum trade size of a coin can be configured on a per-coin basis, defaults to 0.00000001 in backtesting (most exchanges set the minimum trade to larger amounts, which will impact live trading) +- Minimum trade size of a coin can be configured on a per-coin basis, defaults + to 0.00000001 in backtesting (most exchanges set the minimum trade to larger + amounts, which will impact live trading) - Increased pricing resolution from 3 to 9 decimal places -- The current data bundle takes 40MB compressed for download, and 99MB uncompressed on disk for Catalyst to use. +- The current data bundle takes 40MB compressed for download, and 99MB + uncompressed on disk for Catalyst to use. Version 0.2.dev2 ^^^^^^^^^^^^^^^^ @@ -155,19 +190,24 @@ Version 0.2.dev1 - Comprehensive trading functionality against exchanges Bitfinex and Bittrex. - Support for all trading pairs available on each exchange. - - Multiple algorithms can trade simultaneously against a single exchange using the same account. - - Each algorithm has a persisted state (i.e. algorithm can be stopped and restarted preserving the state without data loss) that tracks all open orders, executed transactions and portfolio positions. + - Multiple algorithms can trade simultaneously against a single exchange + using the same account. + - Each algorithm has a persisted state (i.e. algorithm can be stopped and + restarted preserving the state without data loss) that tracks all open + orders, executed transactions and portfolio positions. - Minute by minute portfolio performance metrics. - - Daily summary performance statistics compatible with pyfolio, a Python library for performance and risk analysis of financial portfolios + - Daily summary performance statistics compatible with pyfolio, a Python + library for performance and risk analysis of financial portfolios Version 0.1.dev9 ^^^^^^^^^^^^^^^^ **Release Date**: 2017-08-28 -- Retrieval of crypto benchmark from bundle, instead of hitting Poloniex exchange directly +- Retrieval of crypto benchmark from bundle, instead of hitting Poloniex + exchange directly - Change of bundle storage provider from Dropbox to AWS - Fix issue with 1/1000 scaling issue of prices in bundle diff --git a/tests/exchange/test_bundle.py b/tests/exchange/test_bundle.py index ba055dca..9492438e 100644 --- a/tests/exchange/test_bundle.py +++ b/tests/exchange/test_bundle.py @@ -42,17 +42,16 @@ class TestExchangeBundle: def test_ingest_minute(self): data_frequency = 'minute' - exchange_name = 'bitfinex' + exchange_name = 'poloniex' exchange = get_exchange(exchange_name) exchange_bundle = ExchangeBundle(exchange) assets = [ - exchange.get_asset('xmr_btc') + exchange.get_asset('eth_btc') ] - # start = pd.to_datetime('2017-09-01', utc=True) - start = pd.to_datetime('2016-01-01', utc=True) - end = pd.to_datetime('2017-9-30', utc=True) + start = pd.to_datetime('2016-03-01', utc=True) + end = pd.to_datetime('2017-11-1', utc=True) log.info('ingesting exchange bundle {}'.format(exchange_name)) exchange_bundle.ingest( @@ -122,8 +121,8 @@ class TestExchangeBundle: def test_ingest_daily(self): exchange_name = 'bitfinex' - data_frequency = 'daily' - include_symbols = 'btc_usd' + data_frequency = 'minute' + include_symbols = 'neo_btc' # exchange_name = 'poloniex' # data_frequency = 'daily' @@ -422,7 +421,8 @@ class TestExchangeBundle: data_frequency=data_frequency, asset=asset, writer=writer, - empty_rows_behavior='raise' + empty_rows_behavior='raise', + duplicates_behavior='raise' ) bundle_series = bundle.get_history_window_series( @@ -442,22 +442,26 @@ class TestExchangeBundle: data_frequency = 'minute' exchange = get_exchange(exchange_name) - asset = exchange.get_asset('neo_usd') + asset = exchange.get_asset('eth_btc') + start_dt = pd.to_datetime('2016-5-31', utc=True) + end_dt = pd.to_datetime('2016-6-1', utc=True) self._bundle_to_csv( asset=asset, exchange=exchange, data_frequency=data_frequency, filename='{}_{}_{}'.format( exchange_name, data_frequency, asset.symbol - ) + ), + start_dt=start_dt, + end_dt=end_dt ) def bundle_to_csv(self): - exchange_name = 'bitfinex' + exchange_name = 'poloniex' data_frequency = 'minute' - period = '2017-10' - symbol = 'neo_btc' + period = '2017-09' + symbol = 'eth_btc' exchange = get_exchange(exchange_name) asset = exchange.get_asset(symbol) @@ -478,12 +482,15 @@ class TestExchangeBundle: pass def _bundle_to_csv(self, asset, exchange, data_frequency, filename, - path=None): + path=None, start_dt=None, end_dt=None): bundle = ExchangeBundle(exchange) reader = bundle.get_reader(data_frequency, path=path) - start_dt = reader.first_trading_day - end_dt = reader.last_available_dt + if start_dt is None: + start_dt = reader.first_trading_day + + if end_dt is None: + end_dt = reader.last_available_dt if data_frequency == 'daily': end_dt = end_dt - pd.Timedelta(hours=23, minutes=59)