From 1e8b0c36a1be2dfa38b81b92c7150b676b7977bc Mon Sep 17 00:00:00 2001 From: fredfortier Date: Mon, 20 Nov 2017 20:00:48 -0500 Subject: [PATCH] BUG: fixed #74, a problematic scenario when retrieving the history of multiple assets. --- catalyst/exchange/exchange_bundle.py | 101 ++++++++++----------- catalyst/support/issue_74.py | 127 +++++++++++++++++++++++++++ 2 files changed, 173 insertions(+), 55 deletions(-) create mode 100644 catalyst/support/issue_74.py diff --git a/catalyst/exchange/exchange_bundle.py b/catalyst/exchange/exchange_bundle.py index 6c36fa9e..d6e1d558 100644 --- a/catalyst/exchange/exchange_bundle.py +++ b/catalyst/exchange/exchange_bundle.py @@ -826,6 +826,9 @@ class ExchangeBundle: delta = get_delta(trailing_bar_count, data_frequency) end_dt += delta + # This is an attempt to resolve some caching with the reader + # when auto-ingesting data. + # TODO: needs more work reader = self.get_reader(data_frequency) if reset_reader: del self._readers[reader._rootdir] @@ -844,11 +847,11 @@ class ExchangeBundle: end_dt=end_dt ) + series = dict() for asset in assets: asset_start_dt, asset_end_dt = self.get_adj_dates( start_dt, end_dt, assets, data_frequency ) - in_bundle = range_in_bundle( asset, asset_start_dt, asset_end_dt, reader ) @@ -864,75 +867,63 @@ class ExchangeBundle: end_dt=asset_end_dt ) - series = dict() - try: + periods = self.get_calendar_periods_range( + asset_start_dt, asset_end_dt, data_frequency + ) + # This does not behave well when requesting multiple assets + # when the start or end date of one asset is outside of the range + # looking at the logic in load_raw_arrays(), we are not achieving + # any performance gain by requesting multiple sids at once. It's + # looping through the sids and making separate requests anyway. arrays = reader.load_raw_arrays( - sids=[asset.sid for asset in assets], + sids=[asset.sid], fields=[field], start_dt=start_dt, end_dt=end_dt ) + field_values = arrays[0][:, 0] - except Exception: - symbols = [asset.symbol.encode('utf-8') for asset in assets] - raise PricingDataNotLoadedError( - field=field, - first_trading_day=min([asset.start_date for asset in assets]), - exchange=self.exchange.name, - symbols=symbols, - symbol_list=','.join(symbols), - data_frequency=data_frequency, - start_dt=start_dt, - end_dt=end_dt - ) - - periods = self.get_calendar_periods_range( - start_dt, end_dt, data_frequency - ) - - for asset_index, asset in enumerate(assets): - asset_values = arrays[asset_index] - - value_series = pd.Series(asset_values.flatten(), index=periods) + value_series = pd.Series(field_values, index=periods) series[asset] = value_series return series - def clean(self, data_frequency): - """ - Removing the bundle data from the catalyst folder. - Parameters - ---------- - data_frequency: str +def clean(self, data_frequency): + """ + Removing the bundle data from the catalyst folder. - """ - log.debug('cleaning exchange {}, frequency {}'.format( - self.exchange.name, data_frequency - )) - root = get_exchange_folder(self.exchange.name) + Parameters + ---------- + data_frequency: str - symbols = os.path.join(root, 'symbols.json') - if os.path.isfile(symbols): - os.remove(symbols) + """ + log.debug('cleaning exchange {}, frequency {}'.format( + self.exchange.name, data_frequency + )) + root = get_exchange_folder(self.exchange.name) - temp_bundles = os.path.join(root, 'temp_bundles') + symbols = os.path.join(root, 'symbols.json') + if os.path.isfile(symbols): + os.remove(symbols) - 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)) + temp_bundles = os.path.join(root, 'temp_bundles') - frequencies = ['daily', 'minute'] if data_frequency is None \ - else [data_frequency] + 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)) - for frequency in frequencies: - label = '{}_bundle'.format(frequency) - frequency_bundle = os.path.join(root, label) + frequencies = ['daily', 'minute'] if data_frequency is None \ + else [data_frequency] - 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)) + 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)) diff --git a/catalyst/support/issue_74.py b/catalyst/support/issue_74.py new file mode 100644 index 00000000..ad6d6fee --- /dev/null +++ b/catalyst/support/issue_74.py @@ -0,0 +1,127 @@ +from __future__ import division +import os +import pytz +import numpy as np +import pandas as pd +from scipy.optimize import minimize +import matplotlib.pyplot as plt +from datetime import datetime + +from catalyst.api import record, symbol, symbols, order_target_percent +from catalyst.utils.run_algo import run_algorithm + +np.set_printoptions(threshold='nan', suppress=True) + + +def initialize(context): + # Portfolio assets list + context.assets = symbols('btc_usdt', 'eth_usdt', 'ltc_usdt', 'dash_usdt', + 'xmr_usdt') + context.nassets = len(context.assets) + # Set the time window that will be used to compute expected return + # and asset correlations + context.window = 180 + # Set the number of days between each portfolio rebalancing + context.rebalance_period = 30 + context.i = 0 + + +def handle_data(context, data): + # Only rebalance at the beggining of the algorithm execution and + # every multiple of the rebalance period + if context.i == 0 or context.i % context.rebalance_period == 0: + n = context.window + prices = data.history(context.assets, fields='price', + bar_count=n + 1, frequency='daily') + pr = np.asmatrix(prices) + t_prices = prices.iloc[1:n + 1] + t_val = t_prices.values + tminus_prices = prices.iloc[0:n] + tminus_val = tminus_prices.values + # Compute daily returns (r) + r = np.asmatrix(t_val / tminus_val - 1) + # Compute the expected returns of each asset with the average + # daily return for the selected time window + m = np.asmatrix(np.mean(r, axis=0)) + # ### + stds = np.std(r, axis=0) + # Compute excess returns matrix (xr) + xr = r - m + # Matrix algebra to get variance-covariance matrix + cov_m = np.dot(np.transpose(xr), xr) / n + # Compute asset correlation matrix (informative only) + corr_m = cov_m / np.dot(np.transpose(stds), stds) + + # Define portfolio optimization parameters + n_portfolios = 50000 + results_array = np.zeros((3 + context.nassets, n_portfolios)) + for p in xrange(n_portfolios): + weights = np.random.random(context.nassets) + weights /= np.sum(weights) + w = np.asmatrix(weights) + p_r = np.sum(np.dot(w, np.transpose(m))) * 365 + p_std = np.sqrt( + np.dot(np.dot(w, cov_m), np.transpose(w))) * np.sqrt(365) + + # store results in results array + results_array[0, p] = p_r + results_array[1, p] = p_std + # store Sharpe Ratio (return / volatility) - risk free rate element + # excluded for simplicity + results_array[2, p] = results_array[0, p] / results_array[1, p] + i = 0 + for iw in weights: + results_array[3 + i, p] = weights[i] + i += 1 + + # convert results array to Pandas DataFrame + results_frame = pd.DataFrame(np.transpose(results_array), + columns=['r', 'stdev', + 'sharpe'] + context.assets) + # locate position of portfolio with highest Sharpe Ratio + max_sharpe_port = results_frame.iloc[results_frame['sharpe'].idxmax()] + # locate positon of portfolio with minimum standard deviation + min_vol_port = results_frame.iloc[results_frame['stdev'].idxmin()] + + # order optimal weights for each asset + for asset in context.assets: + if data.can_trade(asset): + order_target_percent(asset, max_sharpe_port[asset]) + + # create scatter plot coloured by Sharpe Ratio + plt.scatter(results_frame.stdev, results_frame.r, + c=results_frame.sharpe, cmap='RdYlGn') + plt.xlabel('Volatility') + plt.ylabel('Returns') + plt.colorbar() + # plot red star to highlight position of portfolio with highest Sharpe Ratio + plt.scatter(max_sharpe_port[1], max_sharpe_port[0], marker='o', + color='b', s=200) + # plot green star to highlight position of minimum variance portfolio + plt.show() + print(max_sharpe_port) + record(pr=pr, r=r, m=m, stds=stds, max_sharpe_port=max_sharpe_port, + corr_m=corr_m) + context.i += 1 + + +def analyze(context=None, results=None): + # Form DataFrame with selected data + data = results[['pr', 'r', 'm', 'stds', 'max_sharpe_port', 'corr_m', + 'portfolio_value']] + + # Save results in CSV file + filename = os.path.splitext(os.path.basename(__file__))[0] + data.to_csv(filename + '.csv') + + +# Bitcoin data is available from 2015-3-2. Dates vary for other tokens. +start = datetime(2017, 1, 1, 0, 0, 0, 0, pytz.utc) +end = datetime(2017, 8, 16, 0, 0, 0, 0, pytz.utc) +results = run_algorithm(initialize=initialize, + handle_data=handle_data, + analyze=analyze, + start=start, + end=end, + exchange_name='poloniex', + capital_base=100000, )