Add task_plot.py for plotting the backtesting results. Plots are included for visualizing signals and portfolio over time. Adjust task_backtesting.py to implement plotting.

Author: niklasniedermeier

src/backtest_bay/backtest/task_backtest.py

@@ -15,23 +15,45 @@
     id_backtest = (
         f"{row.stock}_{row.start_date}_{row.end_date}_" f"{row.interval}_{row.strategy}"
     )
-    data_path = (
+    stock_data_path = (
         BLD / "data" / f"{row.stock}_{row.start_date}_{row.end_date}_{row.interval}.pkl"
     )
     produces = BLD / "backtest" / f"{id_backtest}.pkl"
     strategy = row.strategy
 
     @pytask.task(id=id_backtest)
     def task_backtest(
-        scripts=scripts, data_path=data_path, produces=produces, strategy=strategy
+        scripts=scripts,
+        stock_data_path=stock_data_path,
+        produces=produces,
+        strategy=strategy,
     ):
-        data = pd.read_pickle(data_path)
-        signals = generate_signals(data=data, method=strategy)
-        portfolio = backtest_signals(
-            data=data,
+        stock_data = pd.read_pickle(stock_data_path)
+        signals = generate_signals(data=stock_data, method=strategy)
+        backtested_portfolio = backtest_signals(
+            data=stock_data,
             signals=signals,
             initial_cash=INITIAL_CASH,
             tac=TAC,
             trade_pct=TRADE_PCT,
         )
-        portfolio.to_pickle(produces)
+
+        merged_portfolio = _merge_stock_data_with_portfolio(
+            stock_data, backtested_portfolio
+        )
+
+        merged_portfolio.to_pickle(produces)
+
+
+def _merge_stock_data_with_portfolio(data, portfolio):
+    """Merge data with portfolio using the index.
+
+    Args:
+        data (pd.DataFrame): DataFrame with downloaded data.
+        portfolio (pd.DataFrame): DataFrame to be merged with data using the index.
+
+    Returns:
+        pd.DataFrame: Merged DataFrame.
+    """
+    data.columns = data.columns.droplevel(1)
+    return data.merge(portfolio, how="left", left_index=True, right_index=True)

src/backtest_bay/config.py

@@ -17,8 +17,8 @@
 INTERVALS = ["1d"]
 STRATEGIES = ["bollinger", "macd", "roc", "rsi"]
 
-INITIAL_CASH = 1000
-TAC = 0.05
+INITIAL_CASH = 1000000
+TAC = 0.005
 TRADE_PCT = 0.05
 
 # Define PARAMS using input data

src/backtest_bay/plot/plot_portfolio.py

@@ -0,0 +1,125 @@
+import pandas as pd
+import plotly.graph_objects as go
+from plotly.subplots import make_subplots
+
+
+def plot_portfolio(portfolio, title, tac):
+    """Main function to plot the portfolio performance and metrics."""
+    portfolio_return = _calculate_portfolio_return(portfolio["assets"])
+    annualized_return = _calculate_annualized_return(portfolio["assets"])
+    buy_and_hold_return = _calculate_annualized_return(portfolio["Close"])
+    trades = _calculate_trades(portfolio["shares"])
+
+    fig = make_subplots(
+        rows=2,
+        cols=1,
+        shared_xaxes=True,
+        row_heights=[0.70, 0.30],
+        vertical_spacing=0.2,
+        specs=[[{"type": "xy"}], [{"type": "domain"}]],
+    )
+
+    for trace in _create_portfolio_traces(portfolio):
+        fig.add_trace(trace, row=1, col=1)
+
+    metrics_table = _create_metrics_table(
+        portfolio_return, annualized_return, trades, buy_and_hold_return, tac
+    )
+    fig.add_trace(metrics_table, row=2, col=1)
+
+    fig.update_layout(_create_plot_layout(title))
+
+    return fig
+
+
+def _calculate_portfolio_return(stock):
+    """Calculate the total return of the stock."""
+    initial_value = stock.iloc[0]
+    final_value = stock.iloc[-1]
+    total_return = (final_value / initial_value) - 1
+    portfolio_return = total_return * 100
+    return portfolio_return
+
+
+def _calculate_annualized_return(stock):
+    """Calculate the annualized return for a stock."""
+    initial_value = stock.iloc[0]
+    final_value = stock.iloc[-1]
+    total_return = (final_value / initial_value) - 1
+
+    years = _calculate_years(stock)
+    annualized_return = ((1 + total_return) ** (1 / years) - 1) * 100
+
+    return annualized_return
+
+
+def _calculate_years(stock):
+    """Calculate the number of years between the first and last date of a stock."""
+    total_seconds = (stock.index[-1] - stock.index[0]).total_seconds()
+    years = total_seconds / pd.Timedelta(days=365).total_seconds()
+    return years
+
+
+def _calculate_trades(stock):
+    """Calculate the number of trades by counting changes in the shares held."""
+    trades = stock.diff().ne(0).sum()
+    return trades
+
+
+def _create_portfolio_traces(portfolio):
+    """Create Plotly traces for cash and assets over time."""
+    traces = [
+        go.Scatter(x=portfolio.index, y=portfolio["cash"], mode="lines", name="Cash"),
+        go.Scatter(
+            x=portfolio.index,
+            y=portfolio["assets"],
+            mode="lines",
+            name="Assets (Cash + Holdings)",
+        ),
+    ]
+    return traces
+
+
+def _create_metrics_table(
+    portfolio_return, annualized_return, trades, buy_and_hold_return, tac
+):
+    """Create a Plotly table for the portfolio metrics."""
+    table = go.Table(
+        header={
+            "values": ["Metric", "Value"],
+            "fill_color": "lightgrey",
+            "align": "center",
+        },
+        cells={
+            "values": [
+                [
+                    "Total Return",
+                    "Annualized Return",
+                    "Trades",
+                    "Assumed TAC",
+                    "Benchmark: Annualized Buy and Hold Return",
+                ],
+                [
+                    f"{portfolio_return:.2f}%",
+                    f"{annualized_return:.2f}%",
+                    trades,
+                    f"{tac * 100:.2f}%",
+                    f"{buy_and_hold_return:.2f}%",
+                ],
+            ],
+            "align": "left",
+        },
+    )
+    return table
+
+
+def _create_plot_layout(title):
+    """Create the layout configuration for the Plotly figure."""
+    layout = {
+        "title": title,
+        "xaxis_title": "Date",
+        "yaxis_title": "Value",
+        "legend_title": "Portfolio Components",
+        "template": "plotly",
+    }
+    return layout

src/backtest_bay/plot/plot_signals.py

@@ -0,0 +1,75 @@
+import plotly.graph_objects as go
+from plotly.subplots import make_subplots
+
+
+def plot_signals(df, title):
+    """Create Plotly figure to plot trading signals."""
+    df = _map_signals_for_plotting(df)
+
+    fig = make_subplots(
+        rows=2, cols=1, shared_xaxes=True, row_heights=[0.7, 0.3], vertical_spacing=0.1
+    )
+
+    fig.add_trace(_create_candlestick_trace(df), row=1, col=1)
+
+    buy_trace, sell_trace = _create_signal_traces(df)
+    fig.add_trace(buy_trace, row=2, col=1)
+    fig.add_trace(sell_trace, row=2, col=1)
+
+    fig.update_layout(
+        title=title,
+        xaxis_title="Date",
+        yaxis_title="Stock",
+        xaxis_rangeslider_visible=False,
+        legend_title="Legend",
+    )
+
+    fig.update_yaxes(
+        title_text="Signal",
+        row=2,
+        col=1,
+        tickvals=[-1, 0, 1],
+        ticktext=["Sell", "Hold", "Buy"],
+    )
+
+    return fig
+
+
+def _map_signals_for_plotting(df):
+    """Map signals to -1, 0, 1 for plotting."""
+    signal_mapping = {0: 0, 1: -1, 2: 1}
+    df["signal_plot"] = df["signal"].map(signal_mapping)
+    return df
+
+
+def _create_candlestick_trace(df):
+    """Create candlestick trace for stock data."""
+    return go.Candlestick(
+        x=df.index,
+        open=df["Open"],
+        high=df["High"],
+        low=df["Low"],
+        close=df["Close"],
+        name="Stock",
+    )
+
+
+def _create_signal_traces(df):
+    """Create signal bar traces for buy and sell signals."""
+    buy_signals = df["signal_plot"] > 0
+    buy_trace = go.Bar(
+        x=df.index[buy_signals],
+        y=df.loc[buy_signals, "signal_plot"],
+        marker_color="green",
+        name="Buy Signal",
+    )
+
+    sell_signals = df["signal_plot"] < 0
+    sell_trace = go.Bar(
+        x=df.index[sell_signals],
+        y=df.loc[sell_signals, "signal_plot"],
+        marker_color="red",
+        name="Sell Signal",
+    )
+
+    return buy_trace, sell_trace

src/backtest_bay/plot/task_plot.py

@@ -0,0 +1,39 @@
+import pandas as pd
+import pytask
+
+from backtest_bay.config import BLD, PARAMS, SRC, TAC
+from backtest_bay.plot.plot_portfolio import plot_portfolio
+from backtest_bay.plot.plot_signals import plot_signals
+
+scripts = [
+    SRC / "config.py",
+    SRC / "plot" / "plot_signals.py",
+    SRC / "plot" / "plot_portfolio.py",
+]
+
+for row in PARAMS.itertuples(index=False):
+    id_backtest = (
+        f"{row.stock}_{row.start_date}_{row.end_date}_{row.interval}_{row.strategy}"
+    )
+    backtest_path = BLD / "backtest" / f"{id_backtest}.pkl"
+    plot_path = f"{row.stock}_{row.start_date}_{row.end_date}_{row.interval}"
+    produces = {
+        "plot_signals": BLD / "plot" / plot_path / f"plot_signals_{row.strategy}.html",
+        "plot_portfolio": BLD
+        / "plot"
+        / plot_path
+        / f"plot_portfolio_{row.strategy}.html",
+    }
+
+    @pytask.task(id=id_backtest)
+    def task_plot(
+        scripts=scripts,
+        backtest_path=backtest_path,
+        produces=produces,
+        id_backtest=id_backtest,
+    ):
+        portfolio = pd.read_pickle(backtest_path)
+        fig = plot_signals(portfolio, id_backtest)
+        fig.write_html(produces.get("plot_signals"))
+        fig = plot_portfolio(portfolio, id_backtest, TAC)
+        fig.write_html(produces.get("plot_portfolio"))