Add trading strategies macd, roc and rsi. Implement input validation for hyperparamters of the trading strategies. Add test cases for all implemented functions. Adjust config.py for additional trading strategies.

Author: niklasniedermeier

src/backtest_bay/analysis/generate_signals.py

@@ -1,4 +1,3 @@
-import numpy as np
 import pandas as pd
 
 
@@ -10,16 +9,22 @@ def generate_signals(data, method, **kwargs):
             (must include 'Close' column).
         method (str): The signal generation method. Currently supported:
              - 'bollinger_bands': Uses Bollinger Bands for signal generation.
-        **kwargs: Additional parameters specific to the chosen method
-            (e.g., window size, number of standard deviations).
+        **kwargs: Additional parameters specific to the chosen method.
 
     Returns:
         pd.Series: Trading signals (2: buy, 1: sell, 0: do nothing).
     """
-    if method == "flip":
-        signal = _flip_signals(prices=data["Close"].squeeze())
+    _validate_input_method(method)
+    closing_prices = data["Close"].squeeze()
+
     if method == "bollinger":
-        signal = _bollinger_signals(prices=data["Close"].squeeze(), **kwargs)
+        signal = _bollinger_signals(prices=closing_prices, **kwargs)
+    if method == "macd":
+        signal = _macd_signals(prices=closing_prices, **kwargs)
+    if method == "roc":
+        signal = _roc_signals(prices=closing_prices, **kwargs)
+    if method == "rsi":
+        signal = _rsi_signals(prices=closing_prices, **kwargs)
     return signal
 
 
@@ -37,36 +42,223 @@ def _bollinger_signals(prices, window=20, num_std_dev=2):
     Returns:
         pd.Series: Trading signals (2: buy, 1: sell, 0: do nothing).
     """
-    moving_avg = prices.rolling(window=window).mean().fillna(0)
+    _validate_input_bollinger_signals(window, num_std_dev)
+
+    moving_avg = prices.rolling(window=window).mean()
     std_dev = prices.rolling(window=window).std()
     upper_band = moving_avg + (num_std_dev * std_dev)
     lower_band = moving_avg - (num_std_dev * std_dev)
 
     signals = pd.Series(0, index=prices.index)
-    signals[prices < lower_band] = 2
-    signals[prices > upper_band] = 1
+    signals.loc[prices < lower_band] = 2
+    signals.loc[prices > upper_band] = 1
 
-    signals = _shift_signals_to_right(signals)
-    return pd.Series(signals, index=prices.index)
+    signals = signals.shift(periods=1, fill_value=0)
+    return signals
 
 
-def _flip_signals(prices):
-    """Generate trading signals based on price changes from the previous price.
+def _macd_signals(prices, short_window=12, long_window=26, signal_window=9):
+    """Generate trading signals based on the MACD indicator.
+
+    A buy signal (2) is generated when the MACD line crosses above the Signal Line.
+    A sell signal (1) is generated when the MACD line crosses below the Signal Line.
 
     Args:
-        prices (np.Series): Series of asset prices without index.
+        prices (pd.Series): Series of asset prices.
+        short_window (int): Window size for the short EMA (default: 12).
+        long_window (int): Window size for the long EMA (default: 26).
+        signal_window (int): Window size for the signal line EMA (default: 9).
 
     Returns:
-        np.ndarray: Trading signals (2: buy if previous price is lower,
-                    1: sell if previous price is higher,
-                    0: do nothing for the first price).
+        pd.Series: Trading signals (2: buy, 1: sell, 0: do nothing).
     """
-    price_diff = np.diff(prices, prepend=prices.iloc[0])
-    signals = np.zeros(len(prices), dtype=int)
-    signals[1:][price_diff[1:] > 0] = 1
-    signals[1:][price_diff[1:] < 0] = 2
+    _validate_input_macd_signals(short_window, long_window, signal_window)
+
+    short_ema = prices.ewm(span=short_window, adjust=False).mean()
+    long_ema = prices.ewm(span=long_window, adjust=False).mean()
+    macd_line = short_ema - long_ema
+    signal_line = macd_line.ewm(span=signal_window, adjust=False).mean()
+
+    signals = pd.Series(0, index=prices.index)
+    signals.loc[macd_line > signal_line] = 2
+    signals.loc[macd_line < signal_line] = 1
+
+    signals = signals.shift(periods=1, fill_value=0)
     return signals
 
 
-def _shift_signals_to_right(signals, shift=1):
-    return np.concatenate(([0] * shift, signals[:-shift]))
+def _roc_signals(prices, window=10):
+    """Generate trading signals based on the Rate of Change (ROC) indicator.
+
+    A buy signal (2) is generated when the ROC is positive,
+    and a sell signal (1) is generated when the ROC is negative.
+
+    Args:
+        prices (pd.Series): Series of asset prices.
+        window (int): Window size for computing the ROC.
+
+    Returns:
+        pd.Series: Trading signals (2: buy, 1: sell, 0: do nothing).
+    """
+    _validate_input_window(window)
+
+    roc = prices.pct_change(periods=window - 1)
+
+    signals = pd.Series(0, index=prices.index, dtype=int)
+    signals.loc[roc > 0] = 2
+    signals.loc[roc < 0] = 1
+
+    signals = signals.shift(periods=1, fill_value=0)
+    return signals
+
+
+def _rsi_signals(prices, window=14):
+    """Generate trading signals based on the Relative Strength Index (RSI).
+
+    A buy signal (2) is generated when RSI is below 30 (oversold),
+    and a sell signal (1) is generated when RSI is above 70 (overbought).
+
+    Args:
+        prices (pd.Series): Series of asset prices.
+        window (int): Window size for computing RSI.
+
+    Returns:
+        pd.Series: Trading signals (2: buy, 1: sell, 0: do nothing).
+    """
+    _validate_input_window(window)
+
+    delta = prices.diff()
+    gain = delta.where(delta > 0, 0)
+    loss = -delta.where(delta < 0, 0)
+
+    avg_gain = gain.rolling(window=window, min_periods=1).mean()
+    avg_loss = loss.rolling(window=window, min_periods=1).mean()
+
+    rs = avg_gain / avg_loss
+    rsi = 100 - (100 / (1 + rs))
+
+    signals = pd.Series(0, index=prices.index, dtype=int)
+    upper_cutoff = 70
+    lower_cutoff = 30
+    signals.loc[rsi < lower_cutoff] = 2
+    signals.loc[rsi > upper_cutoff] = 1
+
+    signals = signals.shift(periods=1, fill_value=0)
+    return signals
+
+
+def _validate_input_method(method):
+    """Validate the input method for the `generate_signals` function.
+
+    Args:
+        method (str): The signal generation method.
+
+    Raises:
+        TypeError: If the method is not a string.
+        ValueError: If the method is not one of the supported methods.
+    """
+    if not isinstance(method, str):
+        error_msg = (
+            f"Invalid type for method: expected str, got {type(method).__name__}."
+        )
+        raise TypeError(error_msg)
+
+    supported_methods = ["bollinger", "macd", "roc", "rsi"]
+
+    if method not in supported_methods:
+        error_msg = (
+            f"Invalid method '{method}'. "
+            f"Supported methods are: {', '.join(supported_methods)}."
+        )
+        raise ValueError(error_msg)
+
+
+def _validate_input_bollinger_signals(window, num_std_dev):
+    """Validate inputs for the Bollinger Bands trading signal function.
+
+    Args:
+        window (int): Window size for moving average.
+        num_std_dev (int, float): Number of standard deviations for the bands.
+    """
+    _validate_input_window(window)
+    _validate_input_num_std_dev(num_std_dev)
+
+
+def _validate_input_macd_signals(short_window, long_window, signal_window):
+    """Validate inputs for the MACD trading signal function.
+
+    Args:
+        short_window (int): Window size for the short-term EMA.
+        long_window (int): Window size for the long-term EMA.
+        signal_window (int): Window size for the signal line EMA.
+    """
+    _validate_input_window(short_window)
+    _validate_input_window(long_window)
+    _validate_input_window(signal_window)
+    _validate_window_relationships(short_window, long_window, signal_window)
+
+
+def _validate_input_window(window):
+    """Validate the window parameter.
+
+    Args:
+        window (int): Window size for moving average.
+
+    Raises:
+        TypeError: If window is not an integer.
+        ValueError: If window is not greater than 1.
+    """
+    if not isinstance(window, int):
+        error_msg = f"'window' must be an integer, got {type(window).__name__}."
+        raise TypeError(error_msg)
+
+    if window <= 1:
+        error_msg = f"'window' must be greater than 1, got {window}."
+        raise ValueError(error_msg)
+
+
+def _validate_input_num_std_dev(num_std_dev):
+    """Validate the num_std_dev parameter for Bollinger Bands.
+
+    Args:
+        num_std_dev (int, float): Number of standard deviations for the bands.
+
+    Raises:
+        TypeError: If num_std_dev is not a number.
+        ValueError: If num_std_dev is not positive.
+    """
+    if not isinstance(num_std_dev, int | float):
+        error_msg = f"'num_std_dev' must be a number, got {type(num_std_dev).__name__}."
+        raise TypeError(error_msg)
+
+    if num_std_dev <= 0:
+        error_msg = f"'num_std_dev' must be a positive number, got {num_std_dev}."
+        raise ValueError(error_msg)
+
+
+def _validate_window_relationships(short_window, long_window, signal_window):
+    """Validate logical relationships between MACD windows.
+
+    Args:
+        short_window (int): Window size for the short-term EMA.
+        long_window (int): Window size for the long-term EMA.
+        signal_window (int): Window size for the signal line EMA.
+
+    Raises:
+        ValueError: If:
+            - `short_window` is greater than or equal to `long_window`.
+            - `signal_window` is greater than `short_window`.
+    """
+    if short_window >= long_window:
+        error_msg = (
+            "'short_window' must be less than 'long_window', "
+            f"got short_window={short_window} and long_window={long_window}."
+        )
+        raise ValueError(error_msg)
+
+    if signal_window > short_window:
+        error_msg = (
+            "'signal_window' must be less than or equal to 'short_window',"
+            f"got signal_window={signal_window} and short_window={short_window}."
+        )
+        raise ValueError(error_msg)

src/backtest_bay/config.py

@@ -15,7 +15,7 @@
 START_DATES = ["2022-01-01"]
 END_DATES = ["2025-01-01"]
 INTERVALS = ["1d"]
-STRATEGIES = ["flip", "bollinger"]
+STRATEGIES = ["bollinger", "macd", "roc", "rsi"]
 
 INITIAL_CASH = 1000
 TAC = 0.05