From 6e3017010f223fbd735da05b4967b7ceb95eb28b Mon Sep 17 00:00:00 2001 From: fredfortier Date: Tue, 5 Sep 2017 00:51:38 -0400 Subject: [PATCH] Added the initial version of a live graph --- catalyst/__main__.py | 12 +- catalyst/exchange/algorithm_exchange.py | 96 +++++++++- catalyst/exchange/exchange_utils.py | 32 ++++ catalyst/exchange/live_graph_clock.py | 177 ++++++++++++++++++ .../{exchange_clock.py => simple_clock.py} | 2 +- catalyst/utils/run_algo.py | 14 +- 6 files changed, 317 insertions(+), 16 deletions(-) create mode 100644 catalyst/exchange/live_graph_clock.py rename catalyst/exchange/{exchange_clock.py => simple_clock.py} (98%) diff --git a/catalyst/__main__.py b/catalyst/__main__.py index 36593bab..7a9e6316 100644 --- a/catalyst/__main__.py +++ b/catalyst/__main__.py @@ -210,6 +210,12 @@ def ipython_only(option): help='The base currency used to calculate statistics ' '(e.g. usd, btc, eth).', ) +@click.option( + '--live-graph/--no-live-graph', + is_flag=True, + default=False, + help='Display live graph.', +) @click.pass_context def run(ctx, algofile, @@ -227,7 +233,8 @@ def run(ctx, live, exchange_name, algo_namespace, - base_currency): + base_currency, + live_graph): """Run a backtest for the given algorithm. """ @@ -283,7 +290,8 @@ def run(ctx, live=live, exchange=exchange_name, algo_namespace=algo_namespace, - base_currency=base_currency + base_currency=base_currency, + live_graph=live_graph ) if output == '-': diff --git a/catalyst/exchange/algorithm_exchange.py b/catalyst/exchange/algorithm_exchange.py index 80b1c5c5..e7bced7e 100644 --- a/catalyst/exchange/algorithm_exchange.py +++ b/catalyst/exchange/algorithm_exchange.py @@ -19,6 +19,7 @@ from time import sleep from os import listdir from os.path import isfile, join from collections import deque +import numpy as np import logbook import pandas as pd @@ -28,14 +29,16 @@ from catalyst.algorithm import TradingAlgorithm from catalyst.data.minute_bars import BcolzMinuteBarWriter, \ BcolzMinuteBarReader from catalyst.errors import OrderInBeforeTradingStart -from catalyst.exchange.exchange_clock import ExchangeClock +from catalyst.exchange.simple_clock import SimpleClock +from catalyst.exchange.live_graph_clock import LiveGraphClock from catalyst.exchange.exchange_errors import ( ExchangeRequestError, ExchangePortfolioDataError, ExchangeTransactionError ) from catalyst.exchange.exchange_utils import get_exchange_minute_writer_root, \ - save_algo_object, get_algo_object, get_algo_folder + save_algo_object, get_algo_object, get_algo_folder, get_algo_df, \ + save_algo_df from catalyst.exchange.stats_utils import get_pretty_stats from catalyst.finance.performance.period import calc_period_stats from catalyst.gens.tradesimulation import AlgorithmSimulator @@ -56,8 +59,19 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): def __init__(self, *args, **kwargs): self.exchange = kwargs.pop('exchange', None) self.algo_namespace = kwargs.pop('algo_namespace', None) - self.orders = {} + self.live_graph = kwargs.pop('live_graph', None) + + self._clock = None self.minute_stats = deque(maxlen=60) + + self.pnl_stats = get_algo_df(self.algo_namespace, 'pnl_stats') + + self.custom_signals_stats = \ + get_algo_df(self.algo_namespace, 'custom_signals_stats') + + self.exposure_stats = \ + get_algo_df(self.algo_namespace, 'exposure_stats') + self.is_running = True self.retry_check_open_orders = 5 @@ -122,6 +136,13 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): sys.exit(0) + @property + def clock(self): + if self._clock is None: + return self._create_clock() + else: + return self._clock + def _create_clock(self): # The calendar's execution times are the minutes over which we actually @@ -137,10 +158,21 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): # This method is taken from TradingAlgorithm. # The clock has been replaced to use RealtimeClock # TODO: should we apply a time skew? not sure to understand the utility. - return ExchangeClock( - self.sim_params.sessions, - time_skew=self.exchange.time_skew - ) + + log.debug('creating clock') + if self.live_graph: + self._clock = LiveGraphClock( + self.sim_params.sessions, + time_skew=self.exchange.time_skew, + context=self + ) + else: + self._clock = SimpleClock( + self.sim_params.sessions, + time_skew=self.exchange.time_skew + ) + + return self._clock def _create_generator(self, sim_params): if self.perf_tracker is None: @@ -156,7 +188,7 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): self, sim_params, self.data_portal, - self._create_clock(), + self.clock, self._create_benchmark_source(), self.restrictions, universe_func=self._calculate_universe @@ -222,6 +254,49 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): error=e ) + def add_pnl_stats(self, period_stats): + starting = period_stats['starting_cash'] + current = period_stats['portfolio_value'] + appreciation = (current / starting) - 1 + perc = (appreciation * 100) if current != 0 else 0 + + log.debug('adding pnl stats: {:6f}%'.format(perc)) + + df = pd.DataFrame( + data=[dict(performance=perc)], + index=[period_stats['period_close']] + ) + self.pnl_stats = pd.concat([self.pnl_stats, df]) + + save_algo_df(self.algo_namespace, 'pnl_stats', self.pnl_stats) + + def add_custom_signals_stats(self, period_stats): + log.debug('adding custom signals stats: {}'.format(self.recorded_vars)) + df = pd.DataFrame( + data=[self.recorded_vars], + index=[period_stats['period_close']], + ) + self.custom_signals_stats = pd.concat([self.custom_signals_stats, df]) + + save_algo_df(self.algo_namespace, 'custom_signals_stats', + self.custom_signals_stats) + + def add_exposure_stats(self, period_stats): + data = dict( + long_exposure=period_stats['long_exposure'], + base_currency=period_stats['ending_cash'] + ) + log.debug('adding exposure stats: {}'.format(data)) + + df = pd.DataFrame( + data=[data], + index=[period_stats['period_close']], + ) + self.exposure_stats = pd.concat([self.exposure_stats, df]) + + save_algo_df(self.algo_namespace, 'exposure_stats', + self.exposure_stats) + def prepare_period_stats(self, start_dt, end_dt): """ Creates a dictionary representing the state of the tracker. @@ -314,9 +389,14 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): minute_stats = self.prepare_period_stats( data.current_dt, data.current_dt + timedelta(minutes=1)) + # Saving the last hour in memory self.minute_stats.append(minute_stats) + self.add_pnl_stats(minute_stats) + self.add_custom_signals_stats(minute_stats) + self.add_exposure_stats(minute_stats) + print_df = pd.DataFrame(list(self.minute_stats)) log.debug( 'statistics for the last {stats_minutes} minutes:\n{stats}'.format( diff --git a/catalyst/exchange/exchange_utils.py b/catalyst/exchange/exchange_utils.py index 3b559726..cc91b470 100644 --- a/catalyst/exchange/exchange_utils.py +++ b/catalyst/exchange/exchange_utils.py @@ -3,6 +3,7 @@ import os import pickle import urllib from datetime import date, datetime +import pandas as pd from catalyst.exchange.exchange_errors import ExchangeAuthNotFound, \ ExchangeSymbolsNotFound @@ -116,6 +117,37 @@ def append_algo_object(algo_name, key, obj, environ=None): pickle.dump(obj, handle, protocol=pickle.HIGHEST_PROTOCOL) +def get_algo_df(algo_name, key, environ=None, rel_path=None): + folder = get_algo_folder(algo_name, environ) + + if rel_path is not None: + folder = os.path.join(folder, rel_path) + + filename = os.path.join(folder, key + '.csv') + + if os.path.isfile(filename): + try: + with open(filename, 'rb') as handle: + return pd.read_csv(handle, index_col=0, parse_dates=True) + except IOError: + return pd.DataFrame() + else: + return pd.DataFrame() + + +def save_algo_df(algo_name, key, df, environ=None, rel_path=None): + folder = get_algo_folder(algo_name, environ) + + if rel_path is not None: + folder = os.path.join(folder, rel_path) + ensure_directory(folder) + + filename = os.path.join(folder, key + '.csv') + + with open(filename, 'wb') as handle: + df.to_csv(handle) + + def get_exchange_minute_writer_root(exchange_name, environ=None): exchange_folder = get_exchange_folder(exchange_name, environ) diff --git a/catalyst/exchange/live_graph_clock.py b/catalyst/exchange/live_graph_clock.py new file mode 100644 index 00000000..589446ff --- /dev/null +++ b/catalyst/exchange/live_graph_clock.py @@ -0,0 +1,177 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# 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. + +from time import sleep + +import numpy as np +from matplotlib import pyplot as plt +from matplotlib import style +import pandas as pd +from catalyst.gens.sim_engine import ( + BAR, + SESSION_START, + MINUTE_END, + SESSION_END +) +from logbook import Logger +import matplotlib.dates as mdates + +log = Logger('LiveGraphClock') + +style.use('dark_background') + +days = mdates.DayLocator() +hours = mdates.HourLocator() +fmt = mdates.DateFormatter('%Y-%m-%d %H:%M') + + +class LiveGraphClock(object): + """Realtime clock for live trading. + + This class is a drop-in replacement for + :class:`zipline.gens.sim_engine.MinuteSimulationClock`. + + This is a stripped down version because crypto exchanges run around the clock. + + The :param:`time_skew` parameter represents the time difference between + the Broker and the live trading machine's clock. + """ + + def __init__(self, sessions, context, time_skew=pd.Timedelta('0s')): + + self.sessions = sessions + self.time_skew = time_skew + self._last_emit = None + self._before_trading_start_bar_yielded = True + self.context = context + + fig = plt.figure() + fig.canvas.set_window_title('Enigma Catalyst: {}'.format( + self.context.algo_namespace)) + + self.ax_pnl = fig.add_subplot(311) + self.draw_pnl() + + self.ax_custom_signals = fig.add_subplot(312, sharex=self.ax_pnl) + self.draw_custom_signals() + + self.ax_exposure = fig.add_subplot(313, sharex=self.ax_pnl) + self.draw_exposure() + + # rotates and right aligns the x labels, and moves the bottom of the + # axes up to make room for them + fig.autofmt_xdate() + fig.subplots_adjust(hspace=0.5) + + plt.tight_layout() + plt.ion() + plt.show() + + def format_ax(self, ax): + ax.xaxis.set_major_locator(hours) + ax.xaxis.set_major_formatter(fmt) + ax.xaxis.set_minor_locator(days) + ax.grid(True) + + def draw_pnl(self): + ax = self.ax_pnl + df = self.context.pnl_stats + + ax.clear() + ax.set_title('Performance') + ax.plot(df.index, df['performance'], '-', + color='green', + linewidth=1.0, + label='Performance' + ) + + ax.legend(loc='upper left', ncol=1, fontsize=10, numpoints=1) + + def perc(val): + return '{:2f}'.format(val) + + ax.format_ydata = perc + + self.format_ax(ax) + + def draw_custom_signals(self): + ax = self.ax_custom_signals + df = self.context.custom_signals_stats + + colors = ['blue', 'green', 'red', 'black', 'orange', 'yellow', 'pink'] + + ax.clear() + ax.set_title('Custom Signals') + for index, column in enumerate(df.columns.values.tolist()): + ax.plot(df.index, df[column], '-', + color=colors[index], + linewidth=1.0, + label=column + ) + ax.legend(loc='upper left', ncol=1, fontsize=10, numpoints=1) + self.format_ax(ax) + + def draw_exposure(self): + ax = self.ax_exposure + context = self.context + df = context.exposure_stats + + ax.clear() + ax.set_title('Exposure') + ax.plot(df.index, df['base_currency'], '-', + color='green', + linewidth=1.0, + label='Base Currency: {}'.format( + context.exchange.base_currency.upper() + ) + ) + + positions = context.exchange.portfolio.positions + symbols = [] + for position in positions: + symbols.append(position.symbol) + + ax.plot(df.index, df['long_exposure'], '-', + color='blue', + linewidth=1.0, + label='Long Exposure: {}'.format( + ', '.join(symbols).upper() + ) + ) + ax.legend(loc='upper left', ncol=1, fontsize=10, numpoints=1) + self.format_ax(ax) + + def __iter__(self): + yield pd.Timestamp.utcnow(), SESSION_START + + while True: # plot x^1, x^2, ..., x^4 + current_time = pd.Timestamp.utcnow() + current_minute = current_time.floor('1 min') + + if self._last_emit is None or current_minute > self._last_emit: + log.debug('emitting minutely bar: {}'.format(current_minute)) + + self._last_emit = current_minute + yield current_minute, BAR + + self.draw_pnl() + self.draw_custom_signals() + self.draw_exposure() + + plt.draw() + + else: + # I can't use the "animate" reactive approach here because + # I need to yield from the main loop. + # Workaround: https://stackoverflow.com/a/33050617/814633 + plt.pause(0.001) diff --git a/catalyst/exchange/exchange_clock.py b/catalyst/exchange/simple_clock.py similarity index 98% rename from catalyst/exchange/exchange_clock.py rename to catalyst/exchange/simple_clock.py index 4e180816..1f4d3ffa 100644 --- a/catalyst/exchange/exchange_clock.py +++ b/catalyst/exchange/simple_clock.py @@ -25,7 +25,7 @@ from logbook import Logger log = Logger('ExchangeClock') -class ExchangeClock(object): +class SimpleClock(object): """Realtime clock for live trading. This class is a drop-in replacement for diff --git a/catalyst/utils/run_algo.py b/catalyst/utils/run_algo.py index 42f4f349..fc9f6354 100644 --- a/catalyst/utils/run_algo.py +++ b/catalyst/utils/run_algo.py @@ -95,7 +95,8 @@ def _run(handle_data, live, exchange, algo_namespace, - base_currency): + base_currency, + live_graph): """Run a backtest for the given algorithm. This is shared between the cli and :func:`catalyst.run_algo`. @@ -277,7 +278,8 @@ def _run(handle_data, ) env = TradingEnvironment( - load=partial(load_crypto_market_data, bundle=b, bundle_data=bundle_data, environ=environ), + load=partial(load_crypto_market_data, bundle=b, + bundle_data=bundle_data, environ=environ), bm_symbol='USDT_BTC', trading_calendar=open_calendar, asset_db_path=connstr, @@ -338,7 +340,7 @@ def _run(handle_data, TradingAlgorithmClass = ( partial(ExchangeTradingAlgorithm, exchange=exchange, - algo_namespace=algo_namespace) + algo_namespace=algo_namespace, live_graph=live_graph) if live and exchange else TradingAlgorithm) perf = TradingAlgorithmClass( @@ -439,7 +441,8 @@ def run_algorithm(initialize, live=False, exchange_name=None, base_currency=None, - algo_namespace=None): + algo_namespace=None, + live_graph=False): """Run a trading algorithm. Parameters @@ -552,5 +555,6 @@ def run_algorithm(initialize, live=live, exchange=exchange_name, algo_namespace=algo_namespace, - base_currency=base_currency + base_currency=base_currency, + live_graph=live_graph )