Source code for tradeexecutor.visual.single_pair

"""Tools to visualise live trading/backtest outcome for strategies trading only one pair."""
import datetime
import logging

from typing import Optional, Union, List, Collection

import plotly.graph_objects as go
import pandas as pd
from plotly.graph_objs.layout import Annotation

from tradeexecutor.state.portfolio import Portfolio
from tradeexecutor.state.position import TradingPosition
from tradeexecutor.state.state import State
from tradeexecutor.state.trade import TradeExecution
from tradeexecutor.state.visualisation import Plot
from tradeexecutor.strategy.trade_pricing import format_fees_dollars
from tradeexecutor.state.visualisation import PlotKind

from tradeexecutor.state.types import PairInternalId
from tradeexecutor.visual.technical_indicator import overlay_all_technical_indicators

from tradingstrategy.candle import GroupedCandleUniverse
from tradingstrategy.charting.candle_chart import visualise_ohlcv, make_candle_labels, VolumeBarMode

logger = logging.getLogger(__name__)


[docs]def export_trade_for_dataframe(p: Portfolio, t: TradeExecution) -> dict: """Export data for a Pandas dataframe presentation. - Decimal roundings are based on rule of thumb and may need to be tuned """ position = p.get_position_by_id(t.position_id) price_prefix = f"{t.pair.base.token_symbol} / USD" label = [] if t.is_failed(): label += [f"Failed trade"] type = "failed" else: if t.is_sell(): if t.is_stop_loss(): label += [f"Stop loss {t.pair.base.token_symbol}", "", f"Trigger was at {position.stop_loss:.4f} {price_prefix}"] type = "stop-loss" else: label += [f"Sell {t.pair.base.token_symbol}"] type = "sell" else: if t.is_take_profit(): type = "take-profit" label += [f"Take profit {t.pair.base.token_symbol}", "", "Trigger was at {position.take_profit:.4f} {price_prefix}"] else: type = "buy" label += [f"Buy {t.pair.base.token_symbol}"] label += [ "", f"Executed at: {t.executed_at}", f"Value: {t.get_value():.4f} USD", f"Quantity: {abs(t.get_position_quantity()):.6f} {t.pair.base.token_symbol}", "", ] label += [ f"Mid-price: {t.planned_mid_price:.4f} {price_prefix}" if t.planned_mid_price else "", f"Executed at price: {t.executed_price:.4f} {price_prefix}", f"Estimated execution price: {t.planned_price:.4f} {price_prefix}", "", ] if t.lp_fees_estimated is not None: if t.executed_price and t.planned_mid_price: realised_fees = abs(1 - t.planned_mid_price / t.executed_price) label += [ f"Fees paid: {format_fees_dollars(t.get_fees_paid())}", f"Fees planned: {format_fees_dollars(t.lp_fees_estimated)}", f"Fees: {realised_fees:.4f} %" ] else: label += [ f"Fees paid: {format_fees_dollars(t.get_fees_paid())}", f"Fees planned: {format_fees_dollars(t.lp_fees_estimated)}", ] # See Plotly Scatter usage https://stackoverflow.com/a/61349739/315168 return { "timestamp": t.executed_at, "success": t.is_success(), "type": type, "label": "<br>".join(label), "price": t.planned_mid_price if t.planned_mid_price else t.planned_price, }
[docs]def export_trades_as_dataframe( portfolio: Portfolio, pair_id: PairInternalId, start: Optional[pd.Timestamp] = None, end: Optional[pd.Timestamp] = None, ) -> pd.DataFrame: """Convert executed trades to a dataframe, so it is easier to work with them in Plotly. :param start_at: Crop range :param end_at: Crop range """ if start: assert isinstance(start, pd.Timestamp) if end: assert isinstance(end, pd.Timestamp) assert start data = [] for t in portfolio.get_all_trades(): if t.pair.internal_id != pair_id: continue # Crop if start or end: if not t.started_at: # Hotfix to some invalid data? logger.warning("Trade lacks start date: %s", t) continue if t.started_at < start or t.started_at > end: continue data.append(export_trade_for_dataframe(portfolio, t)) return pd.DataFrame(data)
[docs]def visualise_trades( fig: go.Figure, candles: pd.DataFrame, trades_df: pd.DataFrame, ): """Plot individual trades over the candlestick chart.""" # If we have used stop loss, do different categories advanced_trade_types = ("stop-loss", "take-profit") advanced_trades = len(trades_df.loc[trades_df["type"].isin(advanced_trade_types)]) > 0 if advanced_trades: buys_df = trades_df.loc[trades_df["type"] == "buy"] sells_df = trades_df.loc[trades_df["type"] == "sell"] stop_loss_df = trades_df.loc[trades_df["type"] == "stop-loss"] take_profit_df = trades_df.loc[trades_df["type"] == "take-profit"] else: buys_df = trades_df.loc[trades_df["type"] == "buy"] sells_df = trades_df.loc[trades_df["type"] == "sell"] stop_loss_df = None take_profit_df = None # Buys fig.add_trace( go.Scatter( name="Buy", mode="markers", x=buys_df["timestamp"], y=buys_df["price"], text=buys_df["label"], marker={"color": "#aaaaff", "symbol": 'triangle-right', "size": 12, "line": {"width": 1, "color": "#3333aa"}}, hoverinfo="text", ), secondary_y=False, ) # Sells fig.add_trace( go.Scatter( name="Sell", mode="markers", x=sells_df["timestamp"], y=sells_df["price"], text=sells_df["label"], marker={"color": "#aaaaff", "symbol": 'triangle-left', "size": 12, "line": {"width": 1, "color": "#3333aa"}}, hoverinfo="text", ), secondary_y=False, ) if stop_loss_df is not None: fig.add_trace( go.Scatter( name="Stop loss", mode="markers", x=stop_loss_df["timestamp"], y=stop_loss_df["price"], text=stop_loss_df["label"], marker={"symbol": 'triangle-left', "size": 12, "line": {"width": 1, "color": "black"}}, hoverinfo="text", ), secondary_y=False, ) if take_profit_df is not None: fig.add_trace( go.Scatter( name="Take profit", mode="markers", x=take_profit_df["timestamp"], y=take_profit_df["price"], text=take_profit_df["label"], marker={"symbol": 'triangle-left', "size": 12, "line": {"width": 1, "color": "black"}}, hoverinfo="text", ), secondary_y=False, ) return fig
[docs]def get_position_hover_text(p: TradingPosition) -> str: """Get position hover text for Plotly.""" # First draw a position as a re first_trade = p.get_first_trade() last_trade = p.get_last_trade() duration = last_trade.executed_at - first_trade.executed_at started_at = first_trade.started_at.strftime("%Y-%m-%d, %H:%M:%S UTC") ended_at = last_trade.executed_at.strftime("%Y-%m-%d, %H:%M:%S UTC") entry_diff = (first_trade.executed_price - first_trade.planned_price) / first_trade.planned_price entry_dur = (first_trade.executed_at - first_trade.started_at) exit_diff = (last_trade.executed_price - last_trade.planned_price) / last_trade.planned_price exit_dur = (last_trade.executed_at - last_trade.started_at) text = [] text += [ f"Position #{p.position_id}", "" ] # Add remarks if p.is_open(): text += [ "Position currently open", "" ] elif p.is_stop_loss(): text += [ f"Stop loss triggered at: {p.stop_loss:.2f} USD", "" ] else: pass if p.is_closed(): text += [ f"Profit: {p.get_realised_profit_usd():.2f} USD", f"Profit: {p.get_total_profit_percent() * 100:.4f} %", "" ] text += [ f"Entry price: {first_trade.planned_mid_price:.2f} USD (mid price)", f"Entry price: {first_trade.planned_price:.2f} USD (expected)", f"Entry price: {first_trade.executed_price:.2f} USD (executed)", f"Entry slippage: {entry_diff * 100:.4f} %", f"Entry duration: {entry_dur}", "" ] if p.is_closed(): text += [ f"Exit price: {last_trade.planned_price:.2f} USD (expected)", f"Exit price: {last_trade.executed_price:.2f} USD (executed)", f"Exit slippage: {exit_diff * 100:.4f} %", f"Exit duration: {exit_dur}", ] if p.has_buys() or p.has_sells(): if p.has_buys(): text += [ f"Avg buy price: {p.get_average_buy():.2f} USD", ] if p.has_sells(): text += [ f"Avg sell price: {p.get_average_sell():.2f} USD", ] text += [""] if p.is_closed(): text += [ f"Duration: {duration}", f"Started: {started_at} (first trade started)", f"Ended: {ended_at} (last trade executed at)", "" ] else: text += [ f"Started: {started_at} (first trade started)", "" ] return "<br>".join(text)
[docs]def visualise_positions_with_duration_and_slippage( fig: go.Figure, candles: pd.DataFrame, positions: Collection[TradingPosition]): """Visualise trades as coloured area over time. Add arrow indicators to point start and end duration, and slippage. """ # TODO: Figure out how to add a Y coordinate # for Scatter in Plotly paper space max_price = max(candles["high"]) # https://stackoverflow.com/a/58128982/315168 annotations: List[Annotation] = [] buys = { "x": [], "y": [], "text": [], } sells = { "x": [], "y": [], "text": [], } for position in positions: # First draw a position as a re first_trade = position.get_first_trade() last_trade = position.get_last_trade() left_x = pd.Timestamp(first_trade.started_at) right_x = pd.Timestamp(last_trade.executed_at) if position.is_profitable(): colour = "LightGreen" else: colour = "LightPink" # https://plotly.com/python/shapes/ fig.add_vrect( x0=left_x, x1=right_x, xref="x", fillcolor=colour, opacity=0.5, layer="below", line_width=0, ) position_text = get_position_hover_text(position) # Add tooltips as the dot market at the top left corner # of the position fig.add_trace( go.Scatter( x=[left_x + (right_x - left_x) / 2], y=[max_price], hovertext=position_text, hoverinfo="text", showlegend=False, mode='markers', marker={"color": colour, "size": 12} )) # Visualise trades as lines # TODO: Plotly arrow drawing broken for small arrows t: TradeExecution for t in position.trades.values(): colour = "black" fig.add_shape( type="line", x0=t.started_at, x1=t.executed_at, xref="x", y0=t.planned_price, y1=t.executed_price, yref="y", line={ "color": colour, "width": 1, } ) if t.is_buy(): trade_markers = buys else: trade_markers = sells trade_markers["x"].append(t.executed_at) trade_markers["y"].append(t.executed_price) trade_markers["text"].append(str(t)) # Plotly does not render arrows if they are # too small. # # ann = { # "showarrow": True, # "ax": t.started_at, # "axref": "x", # "x": t.executed_at, # "xref": "x", # "ay":t.planned_price, # "ayref": "y", # "y" :t.executed_price, # "yref": "y", # "arrowwidth": 2, # "arrowhead": 5, # "arrowcolor": colour, # } # # annotations.append(ann) # dict( # x= x_end, # y= y_end, # xref="x", yref="y", # text="", # showarrow=True, # axref = "x", ayref='y', # ax= x_start, # ay= y_start, # arrowhead = 3, # arrowwidth=1.5, # arrowcolor='rgb(255,51,0)',) # ) # Add "arrowheads" to trade lines fig.add_trace( go.Scatter( x=buys["x"], y=buys["y"], text=buys["text"], showlegend=False, mode='markers', marker={"symbol": "arrow-right", "color": "black", "size": 12, "line": {"width": 0}}, ) ) fig.add_trace( go.Scatter( x=sells["x"], y=sells["y"], text=sells["text"], showlegend=False, mode='markers', marker={"symbol": "arrow-left", "color": "black", "size": 12, "line": {"width": 0}}, ) ) # TODO: Currently does not work # https://stackoverflow.com/questions/58095322/draw-multiple-arrows-using-plotly-python if annotations: print(annotations) fig.update_layout(annotations=annotations) return fig
[docs]def visualise_single_pair( state: Optional[State], candle_universe: GroupedCandleUniverse | pd.DataFrame, start_at: Optional[Union[pd.Timestamp, datetime.datetime]] = None, end_at: Optional[Union[pd.Timestamp, datetime.datetime]] = None, pair_id: Optional[PairInternalId] = None, height=800, axes=True, technical_indicators=True, title: Union[str, bool] = True, theme="plotly_white", volume_bar_mode=VolumeBarMode.overlay, vertical_spacing = 0.05, subplot_font_size = 11, relative_sizing: list[float] = None, volume_axis_name: str = "Volume USD", ) -> go.Figure: """Visualise single-pair trade execution. :param state: The recorded state of the strategy execution. You must give either `state` or `positions`. :param pair_id: The visualised pair in the case the strategy contains trades for multiple pairs. If the strategy contains trades only for one pair this is not needed. :param candle_universe: Price candles we used for the strategy :param height: Chart height in pixels :param start_at: When the backtest started or when we crop the content :param end_at: When the backtest ended or when we crop the content :param axes: Draw axes labels :param technical_indicators: Extract technical indicators from the state and overlay them on the price action. Only makes sense if the indicators were drawn against the price action of this pair. :param title: Draw the chart title. Set to string to give your own name. Set `True` to use the state name as a title. TODO: True is a legacy option and will be removed. :param theme: Plotly colour scheme to use :param volume_bar_mode: How to draw the volume bars :param vertical_spacing: Vertical spacing between the subplots. Default is 0.05. :param subplot_font_size: Font size of the subplot titles. Default is 11. :param relative_sizing: Optional relative sizes of each plot. Starts with first main candle plot, then the volume plot if it is detached, then the other detached technical indicators. e.g. [1, 0.2, 0.3, 0.3] would mean the second plot is 20% the size of the first, and the third and fourth plots are 30% the size of the first. Remember to account for whether the volume subplot is detached or not. If it is detached, it should take up the second element in the list. :param volume_axis_name: Name of the volume axis. Default is "Volume USD". """ logger.info("Visualising %s", state) start_at, end_at = _get_start_and_end(start_at, end_at) if isinstance(candle_universe, GroupedCandleUniverse): if not pair_id: assert candle_universe.get_pair_count() == 1, "visualise_single_pair() can be only used for a trading universe with a single pair, please pass pair_id" pair_id = next(iter(candle_universe.get_pair_ids())) candles = candle_universe.get_candles_by_pair(pair_id) else: # Raw dataframe candles = candle_universe # Get all positions for the trading pair we want to visualise if pair_id: positions = _get_all_positions(state, pair_id) else: positions = [] if len(positions) > 0: first_trade = positions[0].get_first_trade() else: first_trade = None if first_trade: pair_name = _get_pair_name_from_first_trade(first_trade) pair = first_trade.pair base_token = pair.base.token_symbol quote_token = pair.quote.token_symbol else: pair_name = None base_token = None quote_token = None if not start_at: # No trades made, use the first candle timestamp start_at = candle_universe.get_timestamp_range()[0] if not end_at: end_at = candle_universe.get_timestamp_range()[1] logger.info(f"Visualising single pair strategy for range {start_at} - {end_at}") # Candles define our diagram X axis # Crop it to the trading range candles = candles.loc[candles["timestamp"].between(start_at, end_at)] candle_start_ts = candles["timestamp"].min() candle_end_ts = candles["timestamp"].max() logger.info(f"Candles are {candle_start_ts} = {candle_end_ts}") trades_df = export_trades_as_dataframe( state.portfolio, pair_id, start_at, end_at, ) labels = make_candle_labels( candles, base_token_name=base_token, quote_token_name=quote_token, ) fig = _get_figure_grid_with_indicators( state=state, start_at=start_at, end_at=end_at, height=height, axes=axes, technical_indicators=technical_indicators, title=title, theme=theme, volume_bar_mode=volume_bar_mode, vertical_spacing=vertical_spacing, subplot_font_size=subplot_font_size, relative_sizing=relative_sizing, candles=candles, pair_name=pair_name, labels=labels, volume_axis_name=volume_axis_name ) # Add trade markers if any trades have been made if len(trades_df) > 0: visualise_trades(fig, candles, trades_df) return fig
[docs]def visualise_single_pair_positions_with_duration_and_slippage( state: State, candles: pd.DataFrame, pair_id: Optional[PairInternalId] = None, start_at: Optional[Union[pd.Timestamp, datetime.datetime]] = None, end_at: Optional[Union[pd.Timestamp, datetime.datetime]] = None, height=800, axes=True, title: Union[bool, str] = True, theme="plotly_white", technical_indicators=True, vertical_spacing = 0.05, relative_sizing: list[float] = None, subplot_font_size: int = 11, ) -> go.Figure: """Visualise performance of a live trading strategy. Unlike :py:func:`visualise_single_pair` attempt to visualise - position duration, as a colored area - more position tooltip text - trade duration (started at - executed) - slippage :param state: The recorded state of the strategy execution. Either live or backtest. :param candle_universe: Price candles we used for the strategy :param pair_id: The visualised pair in the case the strategy contains trades for multiple pairs. If the strategy contains trades only for one pair this is not needed. :param height: Chart height in pixels :param start_at: When the backtest started or when we crop the content :param end_at: When the backtest ended or when we crop the content :param axes: Draw axes labels :param title: Draw the chart title. Set to string to give your own name. Set `True` to use the state name as a title. TODO: True is a legacy option and will be removed. :param technical_indicators: Extract technical indicators from the state and overlay them on the price action. Only makes sense if the indicators were drawn against the price action of this pair. :param theme: Plotly colour scheme to use :param vertical_spacing: Vertical spacing between subplots :param relative_sizing: Optional relative sizes of each plot. Starts with first main candle plot. In this function, there is no volume plot (neither overlayed, hidden, or detached), so the first plot is the candle plot, and the rest are the technical indicator plots. e.g. [1, 0.2, 0.3, 0.3] would mean the second plot is 20% the size of the first, and the third and fourth plots are 30% the size of the first. :param subplot_font_size: Font size of the subplot titles """ logger.info("Visualising %s", state) start_at, end_at = _get_start_and_end(start_at, end_at) try: first_trade = next(iter(state.portfolio.get_all_trades())) except StopIteration: first_trade = None if first_trade: pair_name = _get_pair_name_from_first_trade(first_trade) else: pair_name = None candle_start_ts = candles.iloc[0]["timestamp"] if not start_at: # No trades made, use the first candle timestamp start_at = candle_start_ts candle_end_ts = candles.iloc[-1]["timestamp"] if not end_at: end_at = candle_end_ts logger.info(f"Visualising single pair strategy for range {start_at} - {end_at}") # Candles define our diagram X axis # Crop it to the trading range candles = candles.loc[candles["timestamp"].between(start_at, end_at)] if not pair_id: pair_id = int(candles.iloc[0]["pair_id"]) logger.info(f"Candles are {candle_start_ts} - {candle_end_ts}") positions = _get_all_positions(state, pair_id) logging.info("State has %d positions for pair id %d", len(positions), pair_id) # hide volume bar volume_bar_mode = VolumeBarMode.hidden fig = _get_figure_grid_with_indicators( state=state, start_at=start_at, end_at=end_at, height=height, axes=axes, technical_indicators=technical_indicators, title=title, theme=theme, volume_bar_mode=volume_bar_mode, vertical_spacing=vertical_spacing, subplot_font_size=subplot_font_size, relative_sizing=relative_sizing, candles=candles, pair_name=pair_name, labels=None, ) # Add trade markers if any trades have been made visualise_positions_with_duration_and_slippage(fig, candles, positions) return fig
def _get_figure_grid_with_indicators( *, state: State, start_at: pd.Timestamp | None, end_at: pd.Timestamp | None, height: int, axes: bool, technical_indicators: bool, title: str | bool, theme: str, volume_bar_mode: VolumeBarMode, vertical_spacing: float, subplot_font_size: int, relative_sizing: list[float], candles: pd.DataFrame, pair_name: str | None, labels: pd.Series, volume_axis_name: str = "Volume USD", ): """Gets figure grid with indicators overlayered already. Main price plot is not yet added""" title_text, axes_text, volume_text = _get_all_text(state.name, axes, title, pair_name, volume_axis_name) plots = state.visualisation.plots.values() num_detached_indicators, subplot_names = _get_num_detached_and_names(plots, volume_bar_mode, volume_axis_name) # visualise candles and volume and create empty grid space for technical indicators fig = visualise_ohlcv( candles, height=height, theme=theme, chart_name=title_text, y_axis_name=axes_text, volume_axis_name=volume_text, labels=labels, volume_bar_mode=volume_bar_mode, num_detached_indicators=num_detached_indicators, vertical_spacing=vertical_spacing, relative_sizing=relative_sizing, subplot_names=subplot_names, subplot_font_size=subplot_font_size, ) # Draw EMAs etc. if technical_indicators: overlay_all_technical_indicators( fig, state.visualisation, start_at, end_at, volume_bar_mode, ) return fig def _get_all_text( state_name: str, axes: bool, title: str | None, pair_name: str | None, volume_axis_name: str, ): title_text = _get_title(state_name, title) axes_text, volume_text = _get_axes_and_volume_text(axes, pair_name, volume_axis_name) return title_text,axes_text,volume_text def _get_num_detached_and_names( plots: list[Plot], volume_bar_mode: VolumeBarMode, volume_axis_name: str ): """Get num_detached_indicators and subplot_names""" num_detached_indicators = _get_num_detached_indicators(plots, volume_bar_mode) subplot_names = _get_subplot_names(plots, volume_bar_mode, volume_axis_name) return num_detached_indicators,subplot_names def _get_title(name: str, title: str): if title is True: return name elif type(title) == str: return title else: return None def _get_num_detached_indicators(plots: list[Plot], volume_bar_mode: VolumeBarMode): """Get the number of detached technical indicators""" num_detached_indicators = sum( plot.kind == PlotKind.technical_indicator_detached for plot in plots ) if volume_bar_mode in {VolumeBarMode.hidden, VolumeBarMode.overlay}: pass elif volume_bar_mode == VolumeBarMode.separate: num_detached_indicators += 1 else: raise ValueError(f"Unknown volume bar mode {VolumeBarMode}") return num_detached_indicators def _get_subplot_names(plots: list[Plot], volume_bar_mode: VolumeBarMode, volume_axis_name: str = "Volume USD"): """Get subplot names for detached technical indicators. Overlaid names are appended to the detached plot name.""" if volume_bar_mode in {VolumeBarMode.hidden, VolumeBarMode.overlay}: subplot_names = [] detached_without_overlay_count = 0 else: subplot_names = [volume_axis_name] detached_without_overlay_count = 1 # for allowing multiple overlays on detached plots # list of detached plot names that already have overlays already_overlaid_names = [] for plot in plots: # get subplot names for detached technical indicators without any overlay if (plot.kind == PlotKind.technical_indicator_detached) and (plot.name not in [plot.detached_overlay_name for plot in plots if plot.kind == PlotKind.technical_indicator_overlay_on_detached]): subplot_names.append(plot.name) detached_without_overlay_count += 1 # get subplot names for detached technical indicators with overlay if plot.kind == PlotKind.technical_indicator_overlay_on_detached: # check that detached plot exists detached_plots = [plot.name for plot in plots if plot.kind == PlotKind.technical_indicator_detached] assert plot.detached_overlay_name in detached_plots, f"Overlay name {plot.detached_overlay_name} not in available detached plots {detached_plots}" # check if another overlay exists if plot.detached_overlay_name in already_overlaid_names: # add to existing overlay subplot_names[detached_without_overlay_count + already_overlaid_names.index(plot.detached_overlay_name)] += f"<br> + {plot.name}" else: # add to list subplot_names.append(plot.detached_overlay_name + f"<br> + {plot.name}") already_overlaid_names.append(plot.detached_overlay_name) # Insert blank name for main candle chart subplot_names.insert(0, None) return subplot_names def _get_start_and_end( start_at: pd.Timestamp | datetime.datetime | None, end_at: pd.Timestamp | datetime.datetime | None ): """Get and validate start and end timestamps""" if isinstance(start_at, datetime.datetime): start_at = pd.Timestamp(start_at) if isinstance(end_at, datetime.datetime): end_at = pd.Timestamp(end_at) if start_at is not None: assert isinstance(start_at, pd.Timestamp) if end_at is not None: assert isinstance(end_at, pd.Timestamp) return start_at,end_at def _get_all_positions(state: State, pair_id): """Get all positions for a given pair""" assert type(pair_id) == int positions = [p for p in state.portfolio.get_all_positions() if p.pair.internal_id == pair_id] return positions def _get_axes_and_volume_text(axes: bool, pair_name: str | None, volume_axis_name: str = "Volume USD"): """Get axes and volume text""" if axes: axes_text = pair_name volume_text = volume_axis_name else: axes_text = None volume_text = None return axes_text,volume_text def _get_pair_name_from_first_trade(first_trade: TradeExecution): return f"{first_trade.pair.base.token_symbol} - {first_trade.pair.quote.token_symbol}"