Add strategy and backtest hyperparameters to config.py. Add scripts to implement simple trading strategies and a scripts to backtest the strategies.

Author: niklasniedermeier

src/backtest_bay/analysis/backtest_signals.py

@@ -0,0 +1,62 @@
+import math
+
+import pandas as pd
+
+
+def execute_buy(cash, price, shares, trade_vol, tac):
+    buy_shares = math.floor(trade_vol / (price * (1 + tac)))
+    if buy_shares < 1:
+        return cash, shares
+    cost = buy_shares * price * (1 + tac)
+    if cost > cash:
+        return cash, shares
+    cash -= cost
+    shares += buy_shares
+    return cash, shares
+
+
+def execute_sell(cash, price, shares, trade_vol, tac):
+    sell_shares = math.floor(trade_vol / (price * (1 - tac)))
+
+    if sell_shares > shares:
+        if shares < 1:
+            return cash, shares
+        sell_shares = shares
+
+    profit = sell_shares * price * (1 - tac)
+    cash += profit
+    shares -= sell_shares
+    return cash, shares
+
+
+def update_portfolio(cash, shares, price):
+    holdings = shares * price
+    assets = cash + holdings
+    return assets, holdings
+
+
+def backtest_signals(data, signals, initial_cash, tac, trade_pct, price_col="Close"):
+    prices = data[price_col].squeeze()
+    cash, holdings, shares = initial_cash, 0.0, 0
+    assets = cash + holdings
+    portfolio = []
+
+    buy_signal = 2
+    sell_signal = 1
+
+    for price, signal in zip(prices, signals, strict=False):
+        trade_vol = trade_pct * assets
+
+        if signal == buy_signal:
+            cash, shares = execute_buy(cash, price, shares, trade_vol, tac)
+
+        elif signal == sell_signal:
+            cash, shares = execute_sell(cash, price, shares, trade_vol, tac)
+
+        assets, holdings = update_portfolio(cash, shares, price)
+        portfolio.append((cash, holdings, shares, assets, signal, price))
+
+    portfolio = pd.DataFrame(
+        portfolio, columns=["cash", "holdings", "shares", "total", "signal", "price"]
+    )
+    return portfolio

src/backtest_bay/analysis/generate_signals.py

@@ -0,0 +1,72 @@
+import numpy as np
+import pandas as pd
+
+
+def generate_signals(data, method, **kwargs):
+    """Derive trading signals using the specified method and parameters.
+
+    Args:
+        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.
+        **kwargs: Additional parameters specific to the chosen method
+            (e.g., window size, number of standard deviations).
+
+    Returns:
+        pd.Series: Trading signals (2: buy, 1: sell, 0: do nothing).
+    """
+    if method == "flip":
+        signal = _flip_signals(prices=data["Close"].squeeze())
+    if method == "bollinger":
+        signal = _bollinger_signals(prices=data["Close"].squeeze(), **kwargs)
+    return signal
+
+
+def _bollinger_signals(prices, window=20, num_std_dev=2):
+    """Generate anticyclical trading signals based on Bollinger Bands.
+
+    Bollinger Bands are computed using a moving average and standard deviations
+    to identify overbought (sell signal) and oversold (buy signal) conditions.
+
+    Args:
+        prices (pd.Series): Series of asset prices.
+        window (int): Window size for Bollinger Bands calculation (default is 20).
+        num_std_dev (float): Number of standard deviations for the bands (default is 2).
+
+    Returns:
+        pd.Series: Trading signals (2: buy, 1: sell, 0: do nothing).
+    """
+    moving_avg = prices.rolling(window=window).mean().fillna(0)
+    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 = _shift_signals_to_right(signals)
+    return pd.Series(signals, index=prices.index)
+
+
+def _flip_signals(prices):
+    """Generate trading signals based on price changes from the previous price.
+
+    Args:
+        prices (np.Series): Series of asset prices without index.
+
+    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).
+    """
+    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
+    return signals
+
+
+def _shift_signals_to_right(signals, shift=1):
+    return np.concatenate(([0] * shift, signals[:-shift]))

src/backtest_bay/analysis/task_backtest_signals.py

@@ -0,0 +1,36 @@
+import pandas as pd
+import pytask
+
+from backtest_bay.analysis.backtest_signals import backtest_signals
+from backtest_bay.analysis.generate_signals import generate_signals
+from backtest_bay.config import BLD, INITIAL_CASH, PARAMS, SRC, TAC, TRADE_PCT
+
+scripts = [
+    SRC / "config.py",
+    SRC / "analysis" / "generate_signals.py",
+    SRC / "analysis" / "backtest_signals.py",
+]
+
+for row in PARAMS.itertuples(index=False):
+    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)
+        portfolio.to_pickle(produces)

src/backtest_bay/config.py

@@ -1,14 +1,28 @@
 """All the general configuration of the project."""
 
+import itertools
 from pathlib import Path
 
+import pandas as pd
+
 SRC = Path(__file__).parent.resolve()
 ROOT = SRC.joinpath("..", "..").resolve()
 
 BLD = ROOT.joinpath("bld").resolve()
 
 # Configure input data
 STOCKS = ["AAPL", "MSFT"]
-START_DATES = ["2022-01-01", "2022-01-01"]
-END_DATES = ["2025-01-01", "2025-01-01"]
-INTERVALS = ["1d", "1d"]
+START_DATES = ["2022-01-01"]
+END_DATES = ["2025-01-01"]
+INTERVALS = ["1d"]
+STRATEGIES = ["flip", "bollinger"]
+
+INITIAL_CASH = 1000
+TAC = 0.05
+TRADE_PCT = 0.05
+
+# Define PARAMS using input data
+PARAMS = pd.DataFrame(
+    list(itertools.product(STOCKS, START_DATES, END_DATES, INTERVALS, STRATEGIES)),
+    columns=["stock", "start_date", "end_date", "interval", "strategy"],
+)

src/backtest_bay/data/task_download_data.py

@@ -1,29 +1,30 @@
 import pytask
 
-from backtest_bay.config import BLD, END_DATES, INTERVALS, SRC, START_DATES, STOCKS
+from backtest_bay.config import BLD, PARAMS, SRC
 from backtest_bay.data.download_data import download_data
 
-dependencies = [SRC / "config.py", SRC / "data" / "download_data.py"]
+scripts = [SRC / "config.py", SRC / "data" / "download_data.py"]
 
-for index, _ in enumerate(STOCKS):
-    _id = (
-        STOCKS[index]
-        + "_"
-        + START_DATES[index]
-        + "_"
-        + END_DATES[index]
-        + "_"
-        + INTERVALS[index]
-    )
+data_to_download = PARAMS.drop_duplicates(
+    subset=["stock", "start_date", "end_date", "interval"]
+)
 
-    @pytask.task(id=_id)
+for row in data_to_download.itertuples(index=False):
+    id_download = f"{row.stock}_{row.start_date}_{row.end_date}_{row.interval}"
+
+    produces = BLD / f"{id_download}.pkl"
+
+    if produces.exists():
+        continue
+
+    @pytask.task(id=id_download)
     def task_download_data(
-        symbol=STOCKS[index],
-        start_date=START_DATES[index],
-        end_date=END_DATES[index],
-        interval=INTERVALS[index],
-        depends_on=dependencies,
-        produces=BLD / (_id + ".pkl"),
+        symbol=row.stock,
+        start_date=row.start_date,
+        end_date=row.end_date,
+        interval=row.interval,
+        depends_on=scripts,
+        produces=produces,
     ):
         data = download_data(
             symbol=symbol, start_date=start_date, end_date=end_date, interval=interval

tests/analysis/test_generate_signals.py

@@ -0,0 +1,33 @@
+import numpy as np
+import pandas as pd
+
+from backtest_bay.analysis.generate_signals import _bollinger_signals
+
+# Tests for _bollinger_signals
+
+
+def test_bollinger_signals_constant_prices():
+    prices = pd.Series([100] * 30)
+    signals = _bollinger_signals(prices)
+    expected = pd.Series([0] * 30, index=prices.index)
+    pd.testing.assert_series_equal(signals, expected)
+
+
+def test_bollinger_signals_buy_signal():
+    prices = pd.Series([100] * 20 + [90])
+    signals = _bollinger_signals(prices)
+    buy_signal = 2
+    assert signals.iloc[-1] == buy_signal
+
+
+def test_bollinger_signals_sell_signal():
+    prices = pd.Series([100] * 20 + [110])
+    signals = _bollinger_signals(prices)
+    assert signals.iloc[-1] == 1
+
+
+def test_bollinger_signals_window_effect():
+    prices = pd.Series(np.linspace(100, 200, 25))
+    signals_small_window = _bollinger_signals(prices, window=2, num_std_dev=1)
+    signals_large_window = _bollinger_signals(prices, window=20, num_std_dev=1)
+    assert not signals_small_window.equals(signals_large_window)

tests/data/test_download_data.py

@@ -13,7 +13,6 @@
 # Tests for _validate_symbol
 def test_validate_symbol_valid_cases():
     _validate_symbol("AAPL")
-    _validate_symbol("MSFT")
 
 
 def test_validate_symbol_invalid_invalid_symbol():
@@ -98,8 +97,6 @@ def test_validate_date_format_non_string():
 
 
 # tests for _validate_date_range
-
-
 def test_valid_date_range():
     _validate_date_range("2024-01-01", "2024-12-31")
 
@@ -110,8 +107,6 @@ def test_invalid_date_range():
 
 
 # tests for _validate_output
-
-
 def test_validate_output_valid_input():
     # generate typical Yahoo Finance output format
     symbol = "AAPL"