Source code for tradeexecutor.visual.benchmark

"""Compare portfolio performance against other strategies."""
import datetime
from typing import Optional, List, Union, Collection

import plotly.graph_objects as go
import pandas as pd

from tradeexecutor.state.statistics import PortfolioStatistics
from tradeexecutor.state.visualisation import Plot
from tradeexecutor.visual.technical_indicator import visualise_technical_indicator

[docs]def visualise_portfolio_equity_curve( name: str, portfolio_statistics: List[PortfolioStatistics], colour="#008800") -> go.Scatter: """Draw portfolio performance.""" plot = [] for s in portfolio_statistics: plot.append({ "timestamp": pd.Timestamp(s.calculated_at), "value": s.total_equity, }) df = pd.DataFrame(plot) df.set_index("timestamp", inplace=True) return go.Scatter( x=df.index, y=df["value"], mode="lines", name=name, line=dict(color=colour), )
[docs]def visualise_all_cash( start_at: pd.Timestamp, end_at: pd.Timestamp, all_cash: float, colour="#000088") -> go.Scatter: """Draw portfolio performance.""" plot = [] plot.append({ "timestamp": start_at, "value": all_cash, }) plot.append({ "timestamp": end_at, "value": all_cash, }) df = pd.DataFrame(plot) df.set_index("timestamp", inplace=True) return go.Scatter( x=df.index, y=df["value"], mode="lines", name="Hold cash", line=dict(color=colour), )
[docs]def visualise_buy_and_hold( buy_and_hold_asset_name: str, price_series: pd.Series, all_cash: float, colour="#880000") -> go.Scatter: """Draw portfolio performance.""" # Whatever we bought at the start initial_inventory = all_cash / float(price_series.iloc[0]) series = price_series * initial_inventory return go.Scatter( x=series.index, y=series, mode="lines", name=f"Buy and hold {buy_and_hold_asset_name}", # line=dict(color=colour), )
[docs]def visualise_benchmark( name: Optional[str] = None, portfolio_statistics: Optional[List[PortfolioStatistics]] = None, all_cash: Optional[float] = None, buy_and_hold_asset_name: Optional[str] = None, buy_and_hold_price_series: Optional[pd.Series] = None, benchmark_indexes: pd.DataFrame = None, additional_indicators: Collection[Plot] = None, height=1200, start_at: Optional[Union[pd.Timestamp, datetime.datetime]] = None, end_at: Optional[Union[pd.Timestamp, datetime.datetime]] = None, ) -> go.Figure: """Visualise strategy performance against benchmarks. - Live or backtested strategies - Benchmark against buy and hold of various assets - Benchmark against hold all cash Example: .. code-block:: python from tradeexecutor.visual.benchmark import visualise_benchmark TRADING_PAIRS = [ (ChainId.avalanche, "trader-joe", "WAVAX", "USDC"), # Avax (ChainId.polygon, "quickswap", "WMATIC", "USDC"), # Matic (ChainId.ethereum, "uniswap-v2", "WETH", "USDC"), # Eth (ChainId.ethereum, "uniswap-v2", "WBTC", "USDC"), # Btc ] # Benchmark against all of our assets benchmarks = pd.DataFrame() for pair_description in TRADING_PAIRS: token_symbol = pair_description[2] pair = universe.get_pair_by_human_description(pair_description) benchmarks[token_symbol] = universe.universe.candles.get_candles_by_pair(pair.internal_id)["close"] fig = visualise_benchmark( "Bollinger bands example strategy", portfolio_statistics=state.stats.portfolio, all_cash=state.portfolio.get_initial_deposit(), benchmark_indexes=benchmarks, start_at=START_AT, end_at=END_AT, height=800 ) :param portfolio_statistics: Portfolio performance record. :param all_cash: Set a linear line of just holding X amount :param buy_and_hold_asset_name: Visualise holding all_cash amount in the asset, bought at the start. This is basically price * all_cash. .. note :: This is a legacy argument. Use `benchmark_indexes` instead. :param buy_and_hold_price_series: Visualise holding all_cash amount in the asset, bought at the start. This is basically price * all_cash. .. note :: This is a legacy argument. Use `benchmark_indexes` instead. :param benchmark_indexes: List of other asset price series displayed on the timeline besides equity curve. DataFrame containing multiple series. Asset name is the series name. :param height: Chart height in pixels :param start_at: When the backtest started :param end_at: When the backtest ended :param additional_indicators: Additional technical indicators drawn on this chart. List of indicator names. The indicators must be plotted earlier using `state.visualisation.plot_indicator()`. **Note**: Currently not very useful due to Y axis scale """ fig = go.Figure() assert portfolio_statistics 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) first_portfolio_timestamp = pd.Timestamp(portfolio_statistics[0].calculated_at) last_portfolio_timestamp = pd.Timestamp(portfolio_statistics[-1].calculated_at) if not start_at or start_at < first_portfolio_timestamp: start_at = first_portfolio_timestamp if not end_at or end_at > last_portfolio_timestamp: end_at = last_portfolio_timestamp scatter = visualise_portfolio_equity_curve(name, portfolio_statistics) fig.add_trace(scatter) if all_cash: scatter = visualise_all_cash(start_at, end_at, all_cash) fig.add_trace(scatter) if benchmark_indexes is None: benchmark_indexes = pd.DataFrame() # Backwards compatible arguments if buy_and_hold_price_series is not None: benchmark_indexes[buy_and_hold_asset_name] = buy_and_hold_price_series # Plot all benchmark series for benchmark_name in benchmark_indexes: buy_and_hold_price_series = benchmark_indexes[benchmark_name] # Clip to the backtest time frame buy_and_hold_price_series = buy_and_hold_price_series[start_at:end_at] scatter = visualise_buy_and_hold(benchmark_name, buy_and_hold_price_series, all_cash) fig.add_trace(scatter) if additional_indicators: for plot in additional_indicators: scatter = visualise_technical_indicator(plot, start_at, end_at) fig.add_trace(scatter) if name: fig.update_layout(title=f"{name}", height=height) else: fig.update_layout(title=f"Portfolio value", height=height) fig.update_yaxes(title="Value $", showgrid=False) fig.update_xaxes(rangeslider={"visible": False}) # Move legend to the bottom so we have more space for # time axis in narrow notebook views # fig.update_layout(legend=dict( orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1 )) return fig