Add calculation of strategy volatility for plot_portfolio.py. Implement test functions for volatility calculation.

Author: niklasniedermeier

src/backtest_bay/plot/plot_portfolio.py

@@ -1,3 +1,4 @@
+import numpy as np
 import pandas as pd
 import plotly.graph_objects as go
 from plotly.subplots import make_subplots
@@ -7,27 +8,40 @@
 pd.options.plotting.backend = "plotly"
 
 
-def plot_portfolio(portfolio, title, tac):
+def plot_portfolio(portfolio, title, tac, cash):
     """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"])
+    annualized_volatility = _calculate_annualized_volatility(portfolio["assets"])
+
+    # Benchmark: Buy and Hold Strategy
+    portfolio["buy_and_hold"] = _buy_and_hold_strategy(cash, portfolio["Close"])
+    buy_and_hold_return = _calculate_annualized_return(portfolio["buy_and_hold"])
+    buy_and_hold_volatility = _calculate_annualized_volatility(
+        portfolio["buy_and_hold"]
+    )
 
+    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,
+        row_heights=[0.60, 0.40],
+        vertical_spacing=0.15,
         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
+        portfolio_return,
+        annualized_return,
+        annualized_volatility,
+        trades,
+        buy_and_hold_return,
+        buy_and_hold_volatility,
+        tac,
     )
     fig.add_trace(metrics_table, row=2, col=1)
 
@@ -51,7 +65,13 @@ def _create_portfolio_traces(portfolio):
 
 
 def _create_metrics_table(
-    portfolio_return, annualized_return, trades, buy_and_hold_return, tac
+    portfolio_return,
+    annualized_return,
+    annualized_volatility,
+    trades,
+    buy_and_hold_return,
+    buy_and_hold_volatility,
+    tac,
 ):
     """Create a Plotly table for the portfolio metrics."""
     table = go.Table(
@@ -63,18 +83,22 @@ def _create_metrics_table(
         cells={
             "values": [
                 [
-                    "Total Return",
-                    "Annualized Return",
+                    "Total Strategy Return",
+                    "Annualized Strategy Return",
+                    "Annualized Strategy Volatility",
                     "Trades",
                     "Assumed TAC",
-                    "Benchmark: Annualized Buy and Hold Return",
+                    "Annualized Buy and Hold Return",
+                    "Annualized Buy and Hold Volatility",
                 ],
                 [
                     f"{portfolio_return:.2f}%",
                     f"{annualized_return:.2f}%",
+                    f"{annualized_volatility:.2f}%",
                     trades,
                     f"{tac * 100:.2f}%",
                     f"{buy_and_hold_return:.2f}%",
+                    f"{buy_and_hold_volatility:.2f}%",
                 ],
             ],
             "align": "left",
@@ -131,3 +155,29 @@ def _calculate_trades(shares):
     """Calculate the number of trades by counting changes in the shares held."""
     trades = shares.diff().fillna(0).ne(0).sum()
     return trades
+
+
+def _calculate_annualized_volatility(stock):
+    if len(stock) <= 1:
+        return 0
+
+    daily_log_returns = np.log(stock / stock.shift(1)).dropna()
+    daily_volatility = daily_log_returns.std()
+    years = _calculate_years(stock.index)
+    days_per_year = 365 / years
+
+    if years == 0:
+        return 0
+
+    annualized_volatility = daily_volatility * np.sqrt(days_per_year) * 100
+    return annualized_volatility
+
+
+def _buy_and_hold_strategy(initial_cash, prices):
+    first_price = prices.iloc[0]
+
+    if first_price == 0:
+        return 0
+
+    shares = np.floor(initial_cash / first_price)
+    return shares * prices

src/backtest_bay/plot/task_plot.py

@@ -1,7 +1,7 @@
 import pandas as pd
 import pytask
 
-from backtest_bay.config import BLD, PARAMS, SRC, TAC
+from backtest_bay.config import BLD, INITIAL_CASH, PARAMS, SRC, TAC
 from backtest_bay.plot.plot_portfolio import plot_portfolio
 from backtest_bay.plot.plot_signals import plot_signals
 
@@ -35,5 +35,5 @@ def task_plot(
         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 = plot_portfolio(portfolio, id_backtest, TAC, INITIAL_CASH)
         fig.write_html(produces.get("plot_portfolio"))

tests/plot/test_plot_portfolio.py

@@ -3,7 +3,9 @@
 import pytest
 
 from backtest_bay.plot.plot_portfolio import (
+    _buy_and_hold_strategy,
     _calculate_annualized_return,
+    _calculate_annualized_volatility,
     _calculate_portfolio_return,
     _calculate_trades,
     _calculate_years,
@@ -22,7 +24,7 @@
     ],
 )
 def test_calculate_portfolio_return_valid_calculation(stock, expected_return):
-    """Check if calculated return equals expected value."""
+    """Test if _calculate_portfolio_return returns the correct return value."""
     result = _calculate_portfolio_return(stock)
     assert result == expected_return
 
@@ -39,6 +41,7 @@ def test_calculate_portfolio_return_valid_calculation(stock, expected_return):
     ],
 )
 def test_calculate_years(index, expected):
+    """Test if _calculate_years correctly computes the number of years."""
     result = _calculate_years(index)
     assert np.isclose(result, expected, atol=1e-2)
 
@@ -69,6 +72,7 @@ def test_calculate_years(index, expected):
     ],
 )
 def test_calculate_annualized_return(stock, expected):
+    """Test if _calculate_annualized_return correctly calculates annualized returns."""
     result = _calculate_annualized_return(stock)
     assert np.isclose(result, expected, atol=1e-1)
 
@@ -83,5 +87,42 @@ def test_calculate_annualized_return(stock, expected):
     ],
 )
 def test_calculate_trades(shares, expected):
+    """Test if _calculate_trades correctly counts the number of trades."""
     result = _calculate_trades(shares)
     assert result == expected
+
+
+# Tests for _calculate_annualized_volatility
+@pytest.mark.parametrize(
+    ("stock_prices", "expected_volatility"),
+    [
+        ([100, 100, 100, 100, 100], 0),
+        ([100, 101, 102, 103, 104], pytest.approx(2.26, rel=1e-2)),
+        ([100], 0),
+    ],
+)
+def test_calculate_annualized_volatility(stock_prices, expected_volatility):
+    stock = pd.Series(
+        stock_prices, index=pd.date_range(start="2022-01-01", periods=len(stock_prices))
+    )
+    result = _calculate_annualized_volatility(stock)
+    assert result == expected_volatility
+
+
+# Tests for _buy_and_hold_strategy
+@pytest.mark.parametrize(
+    ("initial_cash", "prices", "expected"),
+    [
+        # Standard case with positive prices
+        (100, pd.Series([10, 12, 15, 18]), pd.Series([100.0, 120.0, 150.0, 180.0])),
+        #  Not enough cash to buy even one share
+        (5, pd.Series([10, 12, 15, 18]), pd.Series([0.0, 0.0, 0.0, 0.0])),
+        # Cash exactly enough to buy one share
+        (10, pd.Series([10, 12, 15, 18]), pd.Series([10.0, 12.0, 15.0, 18.0])),
+    ],
+)
+def test_buy_and_hold_strategy(initial_cash, prices, expected):
+    """Test if _calculate_annualized_volatility correctly calculates annualized
+    volatility."""
+    result = _buy_and_hold_strategy(initial_cash, prices)
+    pd.testing.assert_series_equal(result, expected)