Plot Live Execution

Python source code: ../../../src/boatwright/Visualization/live_execution_visualization2.py

from ..LiveExecution import LiveExecution
import matplotlib.pyplot as plt
import matplotlib.animation as animation
import matplotlib.dates as mdates
import time
import asyncio

# This version defines an asynchronous task that updates graphs

class LiveExecutionVisualizer():
    """
    matplotlib plotting of live execution of a trading strategy. 
    matplotlib 'FuncAnimation' figures runs a function every 'interval' milliseconds, which is utilized to update figures with the latest live executor data
    """

    def __init__(self, live_executor: LiveExecution, update_interval: int = 1, figsize=(7.5, 4), debug: bool = False):
        self.live_executor = live_executor
        self.update_interval = update_interval
        self.figsize = figsize
        self.debug = debug
        self.stop = False
        self.data_event = asyncio.Event()
        self.live_executor.register_listener(self)
        print("initializing live_execution_visualization")
    
    def setup_figures(self):
        print("setting up figures")
        
        self.figs = []
        self.axs = []

        plt.ion()
        
        # AUM and buy/hold reference
        fig, ax = plt.subplots(figsize=self.figsize)
        self.figs.append(fig)
        self.axs.append(ax)
        
        # ASSET fraction
        fig = plt.figure(figsize=self.figsize)
        ax = fig.add_subplot(sharex=self.axs[0])
        self.figs.append(fig)
        self.axs.append(ax)

        # CLOSE
        fig = plt.figure(figsize=self.figsize)
        ax = fig.add_subplot(sharex=self.axs[0])
        self.figs.append(fig)
        self.axs.append(ax)

        # plot any additional signals as specified by the strategy
        for key, info in self.live_executor.strategy.plot_info().items():
            while len(self.axs) - 1 < info["ax_n"]:
                fig = plt.figure(figsize=self.figsize)
                ax = fig.add_subplot(sharex=self.axs[0])
                self.figs.append(fig)
                self.axs.append(ax)
        
        plt.pause(0.01)
        plt.show()
    
    async def run(self):
        """
        Run the visualizer
        """
        print("running visualizer")
        self.setup_figures()
        print("finished setup")

        loop = asyncio.get_event_loop()
        self.data_event._loop = loop

        while not self.live_executor.stop:
            await self.data_event.wait()
            self.data_event.clear()
            if self.debug: print("LiveExecutionVisualizer received new data")
            if len(self.live_executor.cumulative_data) < 2: 
                # skip if there's not enough data to plot
                if self.debug: print("not enough data to plot yet...")
                continue 
            self.update_figures()

    def update_figures(self):
        dates = self.live_executor.cumulative_data["datetime"]

        # AUM ------------------------------------------------------
        self.axs[0].clear()
        self.axs[0].plot(dates, self.live_executor.cumulative_data["aum"], label="aum", color="blue")
        buy_and_hold = (self.live_executor.strategy.broker.starting_aum / self.live_executor.cumulative_data["close"].iloc[0]) * self.live_executor.cumulative_data["close"]
        self.axs[0].plot(dates, buy_and_hold, label="buy & hold", color="black")
        self.axs[0].legend(loc='upper right')
        self.axs[0].set_ylabel(self.live_executor.strategy.broker.quote_symbol)

        # ASSET ----------------------------------------------------
        self.axs[1].clear()
        quote_fraction = self.live_executor.cumulative_data["quote_amount"] / self.live_executor.cumulative_data["aum"]
        base_fraction = self.live_executor.cumulative_data["base_amount"] * self.live_executor.cumulative_data["close"] / self.live_executor.cumulative_data["aum"]
        self.axs[1].plot(dates, quote_fraction, color="black", label=self.live_executor.strategy.broker.quote_symbol)
        self.axs[1].plot(dates, base_fraction, color="blue", label=self.live_executor.strategy.symbol)
        self.axs[1].legend(loc='upper right')
        self.axs[1].set_ylabel("fraction")

        # CLOSE ----------------------------------------------------
        self.axs[2].clear()
        self.axs[2].plot(dates, self.live_executor.cumulative_data["close"], label=self.live_executor.strategy.symbol, color="black")
        self.axs[2].set_ylabel(f"close price [{self.live_executor.strategy.broker.quote_symbol}]")

        # buy / sell orders:    
        buy_orders = [o for o in self.live_executor.strategy.broker.orders["FILLED"].values() if o.side == "BUY" and (o.type == "MARKET" or o.type == "LIMIT")]
        buy_dates = [order.time_closed for order in buy_orders]
        buy_prices = [order.info["avg_price"] for order in buy_orders]
        self.axs[2].scatter(buy_dates, buy_prices, color="lime", marker="^", label="buy", zorder=5, alpha=0.75)

        sell_orders = [o for o in self.live_executor.strategy.broker.orders["FILLED"].values() if o.side == "SELL" and (o.type == "MARKET" or o.type == "LIMIT")]
        sell_dates = [order.time_closed for order in sell_orders]
        sell_prices = [order.info["avg_price"] for order in sell_orders]
        self.axs[2].scatter(sell_dates, sell_prices, color="red", marker="v", label="sell", zorder=5, alpha=0.75)
        self.axs[2].legend(loc='upper right')

        # plot any additional signals as specified by the strategy
        for key, info in self.live_executor.strategy.plot_info().items():
            ax = self.axs[info["ax_n"]]
            if info["ax_n"] > 2: ax.clear() 
            plot_kwargs = info
            plot_kwargs.pop("ax_n")
            if not "label" in plot_kwargs.keys(): plot_kwargs["label"] = key
            ax.plot(dates, self.live_executor.cumulative_data[key], **plot_kwargs)
            ax.legend(loc='upper right')

        for a in self.axs:
            a.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d, %H:%M'))
            for label in a.get_xticklabels(which='major'):
                label.set(rotation=20, horizontalalignment='right')
        
        for f in self.figs:
            f.tight_layout()
            f.canvas.draw()

        plt.pause(0.01)