Added the initial version of a live graph

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 == '-':

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(

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)
 

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)

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

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
     )