Update version of library yfinance. Add functions to validate input of backtest_signals.py. Add doc strings for generate_signals.py.

Author: niklasniedermeier

environment.yml

@@ -26,7 +26,7 @@ dependencies:
   - plotly >=5.2.0,<6
 
   # Data dependencies
-  - yfinance
+  - yfinance >=0.2.54
 
   # Install project
   - pip: [-e ., pdbp, kaleido]

src/backtest_bay/analysis/backtest_signals.py

@@ -4,6 +4,34 @@
 
 
 def backtest_signals(data, signals, initial_cash, tac, trade_pct, price_col="Close"):
+    """Backtest trading signals to simulate portfolio performance.
+
+    Args:
+        data (pd.DataFrame): DataFrame containing asset price data.
+            - Must include a column specified by `price_col` (default: 'Close').
+            - The index should be datetime or sequential for portfolio tracking.
+        signals (pd.Series): Series of trading signals.
+            - 2: Buy Signal
+            - 1: Sell Signal
+            - 0: Do Nothing
+        initial_cash (int, float): Initial cash available for trading.
+        tac (int, float): Transaction cost as a percentage (e.g., 0.05 for 5%).
+        trade_pct (float): Percentage of 'initial_cash' to trade per signal.
+        price_col (str): Column name for the asset's price. Default is 'Close'.
+
+    Returns:
+        pd.DataFrame: Portfolio performance over time with columns:
+            - 'price': The price of the stock.
+            - 'signal': Trading signal used (2: Buy, 1: Sell, 0: Do Nothing).
+            - 'shares': Number of shares.
+            - 'holdings': Total value of shares (price * shares)
+            - 'cash': Cash.
+            - 'assets': Portfolio value (cash + holdings).
+    """
+    _validate_backtest_signals_input(
+        data, signals, initial_cash, tac, trade_pct, price_col
+    )
+
     prices = data[price_col].squeeze()
     cash, holdings, shares = initial_cash, 0.0, 0
     assets = cash + holdings
@@ -24,6 +52,19 @@ def backtest_signals(data, signals, initial_cash, tac, trade_pct, price_col="Clo
 
 
 def _execute_trade(signal, cash, price, shares, trade_vol, tac):
+    """Execute a trade based on the trading signal.
+
+    Args:
+        signal (int): Current trading signal.
+        cash (int, float): Current cash.
+        price (float): Current price of the stock.
+        shares (int): Current shares.
+        trade_vol (float): Volume of portfolio to trade.
+        tac (int, float): Transaction cost.
+
+    Returns:
+        tuple: Updated cash and shares after the trade.
+    """
     buy_signal = 2
     sell_signal = 1
 
@@ -37,6 +78,20 @@ def _execute_trade(signal, cash, price, shares, trade_vol, tac):
 
 
 def _execute_buy(cash, price, shares, trade_vol, tac):
+    """Execute a buy trade.
+
+    Args:
+        cash (int, float): Current cash.
+        price (float): Current price of the asset.
+        shares (int): Current shares.
+        trade_vol (float): Volume of portfolio to trade.
+        tac (float): Transaction cost.
+
+    Returns:
+        tuple: Updated cash and shares after the buy trade.
+            - cash (float): Remaining cash after the trade.
+            - shares (int): Updated number of shares held.
+    """
     buy_shares = math.floor(trade_vol / (price * (1 + tac)))
     cost = buy_shares * price * (1 + tac)
 
@@ -49,12 +104,36 @@ def _execute_buy(cash, price, shares, trade_vol, tac):
 
 
 def _is_buy_trade_affordable(buy_shares, cost, cash):
+    """Check if the buy trade is affordable.
+
+    Args:
+        buy_shares (int): Number of shares to buy.
+        cost (int, float): Total cost of the shares.
+        cash (int, float): Current cash.
+
+    Returns:
+        bool: True if the trade is affordable, False otherwise.
+    """
     is_trade_vol_enough = buy_shares >= 1
     is_cash_enough = cash >= cost
     return is_trade_vol_enough and is_cash_enough
 
 
 def _execute_sell(cash, price, shares, trade_vol, tac):
+    """Execute a sell trade.
+
+    Args:
+        cash (float): Current cash.
+        price (float): Current price.
+        shares (int):  Current shares.
+        trade_vol (float): Volume of portfolio to trade.
+        tac (float): Transaction cost.
+
+    Returns:
+        tuple: Updated cash and shares after the sell trade.
+            - cash (float): Updated cash.
+            - shares (int): Updated shares.
+    """
     sell_shares = math.floor(trade_vol / (price * (1 - tac)))
 
     if not _is_sell_trade_affordable(shares):
@@ -69,10 +148,133 @@ def _execute_sell(cash, price, shares, trade_vol, tac):
 
 
 def _is_sell_trade_affordable(shares):
+    """Checks if there are enough shares to sell."""
     return shares >= 1
 
 
 def _update_portfolio(cash, shares, price):
+    """Updates holdings and assets after trade."""
     holdings = shares * price
     assets = cash + holdings
     return assets, holdings
+
+
+def _validate_backtest_signals_input(
+    data, signals, initial_cash, tac, trade_pct, price_col
+):
+    """Validates input for backtesting signals."""
+    _validate_data(data, price_col)
+    _validate_signals(signals, data, price_col)
+    _validate_initial_cash(initial_cash)
+    _validate_tac(tac)
+    _validate_trade_pct(trade_pct)
+
+
+def _validate_data(data, price_col):
+    """Validate the input data for backtesting.
+
+    Args:
+        data (pd.DataFrame): DataFrame containing stock data.
+        price_col (str): Column name for the stock price.
+
+    Raises:
+        TypeError: If data is not a pandas DataFrame.
+        ValueError: If the price column is missing or contains non-numeric values.
+    """
+    if not isinstance(data, pd.DataFrame):
+        error_msg = f"data must be a pandas DataFrame, got {type(data).__name__}."
+        raise TypeError(error_msg)
+
+    if price_col not in data.columns:
+        error_msg = f"data must contain a '{price_col}' column."
+        raise ValueError(error_msg)
+
+    if not pd.api.types.is_numeric_dtype(data[price_col].squeeze()):
+        error_msg = f"The '{price_col}' column must contain numeric values."
+        raise ValueError(error_msg)
+
+
+def _validate_signals(signals, data, price_col):
+    """Validate the trading signals.
+
+    Args:
+        signals (pd.Series): Trading signals for backtesting.
+        data (pd.DataFrame): DataFrame containing stock data.
+        price_col (str): Column name for the stock price.
+
+    Raises:
+        TypeError: If signals is not a pandas Series.
+        ValueError: If signals contain invalid values.
+        ValueError: If signals do not match the length of the price column.
+    """
+    if not isinstance(signals, pd.Series):
+        error_msg = f"signals must be a pandas Series, got {type(signals).__name__}."
+        raise TypeError(error_msg)
+
+    if not all(isinstance(signal, int) for signal in signals):
+        error_msg = "signals must contain only integers."
+        raise ValueError(error_msg)
+
+    if not all(signal in [0, 1, 2] for signal in signals):
+        error_msg = "signals must contain only 0 (Hold), 1 (Sell), or 2 (Buy)."
+        raise ValueError(error_msg)
+
+    if len(signals) != len(data[price_col]):
+        error_msg = f"signals must have the same length as the '{price_col}' column."
+        raise ValueError(error_msg)
+
+
+def _validate_initial_cash(initial_cash):
+    """Validate the initial cash value for backtesting.
+
+    Args:
+        initial_cash (int or float): The initial cash available for trading.
+
+    Raises:
+        TypeError: If initial_cash is not an integer or float.
+        ValueError: If initial_cash is not positive.
+    """
+    if not isinstance(initial_cash, int | float):
+        error_msg = f"initial_cash must be a number, got {type(initial_cash).__name__}."
+        raise TypeError(error_msg)
+    if initial_cash <= 0:
+        error_msg = "initial_cash must be a positive number."
+        raise ValueError(error_msg)
+
+
+def _validate_tac(tac):
+    """Validate the transaction cost (tac) for backtesting.
+
+    Args:
+        tac (int or float): Transaction cost as a percentage.
+
+    Raises:
+        TypeError: If tac is not an integer or float.
+        ValueError: If tac is negative or greater than 1.
+    """
+    if not isinstance(tac, int | float):
+        error_msg = f"tac must be a number, got {type(tac).__name__}."
+        raise TypeError(error_msg)
+
+    if not (0 <= tac <= 1):
+        error_msg = "tac must be between 0 and 1."
+        raise ValueError(error_msg)
+
+
+def _validate_trade_pct(trade_pct):
+    """Validate the trade percentage (trade_pct) for backtesting.
+
+    Args:
+        trade_pct (float): Trade percentage of total assets per trade.
+
+    Raises:
+        TypeError: If trade_pct is not a float.
+        ValueError: If trade_pct is not between 0 and 1.
+    """
+    if not isinstance(trade_pct, float):
+        error_msg = f"trade_pct must be a float, got {type(trade_pct).__name__}."
+        raise TypeError(error_msg)
+
+    if not (0 < trade_pct <= 1):
+        error_msg = "trade_pct must be between 0 and 1. Zero is not possible."
+        raise ValueError(error_msg)

src/backtest_bay/analysis/generate_signals.py

@@ -8,8 +8,22 @@ def generate_signals(data, method, **kwargs):
         data (pd.DataFrame): DataFrame containing asset price data
             (must include 'Close' column).
         method (str): The signal generation method. Currently supported:
-             - 'bollinger_bands': Uses Bollinger Bands for signal generation.
+             - 'bollinger'
+             - 'macd'
+             - 'roc'
+             - 'rsi'
         **kwargs: Additional parameters specific to the chosen method.
+            - 'bollinger':
+                - window (int): Window size for moving average.
+                - num_std_dev (float, int): Standard deviation multiplier for bands.
+            - 'macd':
+                - short_window (int): Window size for the fast EMA.
+                - long_window (int): Window size for the slow EMA.
+                - signal_window (int): Window size for the signal line EMA.
+            - 'roc':
+                - window (int): Window size for calculating the rate of change.
+            - 'rsi':
+                - window (int): Window size for calculating RSI.
 
     Returns:
         pd.Series: Trading signals (2: buy, 1: sell, 0: do nothing).

src/backtest_bay/analysis/task_backtest_signals.py

@@ -15,22 +15,19 @@
     id_backtest = (
         f"{row.stock}_{row.start_date}_{row.end_date}_" f"{row.interval}_{row.strategy}"
     )
+
     data_path = BLD / f"{row.stock}_{row.start_date}_{row.end_date}_{row.interval}.pkl"
-    data = pd.read_pickle(data_path)
     produces = BLD / f"{id_backtest}.pkl"
 
-    signals = generate_signals(data, row.strategy)
-
     @pytask.task(id=id_backtest)
-    def task_backtest_signals(
-        data=data,
-        signals=signals,
-        initial_cash=INITIAL_CASH,
-        tac=TAC,
-        investment_pct=TRADE_PCT,
-        scripts=scripts,
-        data_path=data_path,
-        produces=produces,
-    ):
-        portfolio = backtest_signals(data, signals, initial_cash, tac, investment_pct)
+    def task_backtest(scripts=scripts, data_path=data_path, produces=produces, row=row):
+        data = pd.read_pickle(data_path)
+        signals = generate_signals(data=data, method=row.strategy)
+        portfolio = backtest_signals(
+            data=data,
+            signals=signals,
+            initial_cash=INITIAL_CASH,
+            tac=TAC,
+            trade_pct=TRADE_PCT,
+        )
         portfolio.to_pickle(produces)

src/backtest_bay/config.py

@@ -12,7 +12,7 @@
 
 # Configure input data
 STOCKS = ["AAPL", "MSFT"]
-START_DATES = ["2022-01-01"]
+START_DATES = ["2019-01-01"]
 END_DATES = ["2025-01-01"]
 INTERVALS = ["1d"]
 STRATEGIES = ["bollinger", "macd", "roc", "rsi"]

src/backtest_bay/data/task_download_data.py

@@ -14,19 +14,12 @@
 
     produces = BLD / f"{id_download}.pkl"
 
-    if produces.exists():
-        continue
-
     @pytask.task(id=id_download)
-    def task_download_data(
-        symbol=row.stock,
-        start_date=row.start_date,
-        end_date=row.end_date,
-        interval=row.interval,
-        depends_on=scripts,
-        produces=produces,
-    ):
+    def task_download_data(depends_on=scripts, produces=produces, row=row):
         data = download_data(
-            symbol=symbol, start_date=start_date, end_date=end_date, interval=interval
+            symbol=row.stock,
+            start_date=row.start_date,
+            end_date=row.end_date,
+            interval=row.interval,
         )
         data.to_pickle(produces)

tests/analysis/test_backtest_signals.py

@@ -1,4 +1,3 @@
-import numpy as np
 import pandas as pd
 import pandas.testing as pdt
 
@@ -99,7 +98,7 @@ def test_backtest_portfolio_correct_calculation():
     data = pd.DataFrame({"Close": [10, 5, 10, 8, 10]}, index=index)
 
     # Only hold
-    signals = np.array([0, 0, 0, 0, 0])
+    signals = pd.Series([0, 0, 0, 0, 0])
     portfolio = backtest_signals(
         data, signals, initial_cash=100, tac=0, trade_pct=1.0, price_col="Close"
     )
@@ -117,7 +116,7 @@ def test_backtest_portfolio_correct_calculation():
     pdt.assert_frame_equal(portfolio, expected_portfolio)
 
     # Buy once
-    signals = np.array([0, 2, 0, 0, 0])
+    signals = pd.Series([0, 2, 0, 0, 0])
     portfolio = backtest_signals(
         data, signals, initial_cash=100, tac=0, trade_pct=1.0, price_col="Close"
     )
@@ -135,7 +134,7 @@ def test_backtest_portfolio_correct_calculation():
     pdt.assert_frame_equal(portfolio, expected_portfolio)
 
     # Buy and sell once
-    signals = np.array([0, 2, 0, 0, 1])
+    signals = pd.Series([0, 2, 0, 0, 1])
     portfolio = backtest_signals(
         data, signals, initial_cash=100, tac=0, trade_pct=1.0, price_col="Close"
     )