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 asyncronous task that updates graphs


class LiveExectutionVisualizer():
    """
    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
        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):
        """
        
        """
        print("running visualizer")
        self.setup_figures()
        print("finished setup")

        while len(self.live_executor.cumulative_data) < 2:
            print("visualiztion waiting for data")
            await asyncio.sleep(self.update_interval)

        while self.stop is False:
            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.product_id)
            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 id, o in self.live_executor.strategy.broker.orders.items() if o.side=="BUY" and o.status=="FILLED" and not o.type == "STOP"]
            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 id, o in self.live_executor.strategy.broker.orders.items() if o.side=="SELL" and o.status=="FILLED" and not o.type == "STOP"]
            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)
            await asyncio.sleep(self.update_interval)