diff --git a/catalyst/exchange/exchange_algorithm.py b/catalyst/exchange/exchange_algorithm.py index aeff9e4e..5c9f8771 100644 --- a/catalyst/exchange/exchange_algorithm.py +++ b/catalyst/exchange/exchange_algorithm.py @@ -16,7 +16,7 @@ import signal import sys from datetime import timedelta from os import listdir -from os.path import isfile, join +from os.path import isfile, join, exists import catalyst.protocol as zp import logbook @@ -36,9 +36,11 @@ from catalyst.exchange.utils.exchange_utils import ( get_algo_folder, get_algo_df, save_algo_df, + clear_frame_stats_directory, + remove_old_files, group_assets_by_exchange, ) -from catalyst.exchange.utils.stats_utils import get_pretty_stats, stats_to_s3, \ - stats_to_algo_folder +from catalyst.exchange.utils.stats_utils import \ + get_pretty_stats, stats_to_s3, stats_to_algo_folder from catalyst.finance.execution import MarketOrder from catalyst.finance.performance import PerformanceTracker from catalyst.finance.performance.period import calc_period_stats @@ -67,8 +69,8 @@ class ExchangeTradingAlgorithmBase(TradingAlgorithm): self.current_day = None - if self.simulate_orders is None \ - and self.sim_params.arena == 'backtest': + if self.simulate_orders is None and \ + self.sim_params.arena == 'backtest': self.simulate_orders = True # Operations with retry features @@ -118,7 +120,7 @@ class ExchangeTradingAlgorithmBase(TradingAlgorithm): # be in-line with CXXT and many exchanges. We'll consider # adding more order types in the future. if not isinstance(style, ExchangeLimitOrder) or \ - not isinstance(style, MarketOrder): + not isinstance(style, MarketOrder): raise OrderTypeNotSupported( order_type=style.__class__.__name__ ) @@ -368,6 +370,11 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): self._clock = None self.frame_stats = list() + # erase the frame_stats folder to avoid overloading the disk + error = clear_frame_stats_directory(self.algo_namespace) + if error: + log.warning(error) + self.pnl_stats = get_algo_df(self.algo_namespace, 'pnl_stats') self.custom_signals_stats = \ @@ -392,6 +399,19 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): "Exit should be handled by the user.") def interrupt_algorithm(self): + """ + + when algorithm comes to an end this function is called. + extracts the stats and calls analyze. + after finishing, it exits the run. + + Parameters + ---------- + + Returns + ------- + + """ self.is_running = False if self._analyze is None: @@ -401,21 +421,31 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): log.info('Exiting the algorithm. Calling `analyze()` ' 'before exiting the algorithm.') + # add the last day stats which is not saved in the directory + current_stats = pd.DataFrame(self.frame_stats) + current_stats.set_index('period_close', drop=False, inplace=True) + + # get the location of the directory algo_folder = get_algo_folder(self.algo_namespace) - folder = join(algo_folder, 'daily_performance') - files = [f for f in listdir(folder) if isfile(join(folder, f))] + folder = join(algo_folder, 'frame_stats') - daily_perf_list = [] - for item in files: - filename = join(folder, item) + if exists(folder): + files = [f for f in listdir(folder) if isfile(join(folder, f))] - with open(filename, 'rb') as handle: - perf_period = pickle.load(handle) - perf_period_dict = perf_period.to_dict() - daily_perf_list.append(perf_period_dict) + period_stats_list = [] + for item in files: + filename = join(folder, item) - stats = pd.DataFrame(daily_perf_list) - stats.set_index('period_close', drop=False, inplace=True) + with open(filename, 'rb') as handle: + perf_period = pickle.load(handle) + period_stats_list.extend(perf_period) + + stats = pd.DataFrame(period_stats_list) + stats.set_index('period_close', drop=False, inplace=True) + + stats = pd.concat([stats, current_stats]) + else: + stats = current_stats self.analyze(stats) @@ -709,6 +739,37 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): self.algo_namespace, 'exposure_stats', self.exposure_stats ) + def nullify_frame_stats(self, now): + """ + + Save all period_stats to local directory + erase old files from the folder and nullify + self.frame_stats + + Parameters + ---------- + now: Timestamp + + Returns + ------- + + """ + save_algo_object( + algo_name=self.algo_namespace, + key=now.floor('1D').strftime('%Y-%m-%d'), + obj=self.frame_stats, + rel_path='frame_stats' + ) + error = remove_old_files( + algo_name=self.algo_namespace, + today=now, + rel_path='frame_stats' + ) + if error: + log.warning(error) + + self.frame_stats = list() + def handle_data(self, data): """ Wrapper around the handle_data method of each algo. @@ -728,7 +789,7 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): # Resetting the frame stats every day to minimize memory footprint today = data.current_dt.floor('1D') if self.current_day is not None and today > self.current_day: - self.frame_stats = list() + self.nullify_frame_stats(now=data.current_dt) self.performance_needs_update = False orders = list(self.perf_tracker.todays_performance.orders_by_id.keys()) @@ -808,6 +869,8 @@ class ExchangeTradingAlgorithmLive(ExchangeTradingAlgorithmBase): # Saving the last hour in memory self.frame_stats.append(frame_stats) + # creating and saving the pnl_stats into the local + # directory self.add_pnl_stats(frame_stats) if self.recorded_vars: self.add_custom_signals_stats(frame_stats) diff --git a/catalyst/exchange/utils/exchange_utils.py b/catalyst/exchange/utils/exchange_utils.py index 3f8d6121..fb24f1c8 100644 --- a/catalyst/exchange/utils/exchange_utils.py +++ b/catalyst/exchange/utils/exchange_utils.py @@ -130,7 +130,7 @@ def get_exchange_symbols(exchange_name, is_local=False, environ=None): filename)).days > 1): try: download_exchange_symbols(exchange_name, environ) - except Exception as e: + except Exception: pass if os.path.isfile(filename): @@ -273,6 +273,7 @@ def get_algo_object(algo_name, key, environ=None, rel_path=None, how='pickle'): key: str environ: rel_path: str + how: str Returns ------- @@ -316,6 +317,7 @@ def save_algo_object(algo_name, key, obj, environ=None, rel_path=None, obj: Object environ: rel_path: str + how: str """ folder = get_algo_folder(algo_name, environ) @@ -392,6 +394,67 @@ def save_algo_df(algo_name, key, df, environ=None, rel_path=None): df.to_csv(handle, encoding='UTF_8') +def clear_frame_stats_directory(algo_name): + """ + remove the outdated directory + to avoid overloading the disk + + Parameters + ---------- + algo_name: str + + Returns + ------- + error: str + + """ + error = None + algo_folder = get_algo_folder(algo_name) + folder = os.path.join(algo_folder, 'frame_stats') + if os.path.exists(folder): + try: + shutil.rmtree(folder) + except OSError: + error = 'unable to remove {}, the analyze ' \ + 'data will be inconsistent'.format(folder) + return error + + +def remove_old_files(algo_name, today, rel_path): + """ + remove old files from a directory + to avoid overloading the disk + + Parameters + ---------- + algo_name: str + today: Timestamp + rel_path: str + + Returns + ------- + error: str + + """ + error = None + algo_folder = get_algo_folder(algo_name) + folder = os.path.join(algo_folder, rel_path) + + # run on all files in the folder + for f in os.listdir(folder): + creation_unix = os.path.getctime(f) + creation_time = pd.to_datetime(creation_unix, unit='s', ) + + # if the file is older than 30 days erase it + if today - pd.DateOffset(30) > creation_time: + try: + os.unlink(f) + except OSError: + error = 'unable to erase files in {}'.format(folder) + + return error + + def get_exchange_minute_writer_root(exchange_name, environ=None): """ The minute writer folder for the exchange. @@ -575,8 +638,9 @@ def mixin_market_params(exchange_name, params, market): params['maker'] = 0.001 params['taker'] = 0.002 - elif 'maker' in market and 'taker' in market \ - and market['maker'] is not None and market['taker'] is not None: + elif 'maker' in market and 'taker' in market and \ + market['maker'] is not None and market['taker'] is not None: + params['maker'] = market['maker'] params['taker'] = market['taker']