Find last trading day of month

andi

I wanted to make use of backtrader's trading calendar functionality, especially I wanted to use the last_monthday method (https://www.backtrader.com/docu/tradingcalendar/tradingcalendar/). So I pip installed pandas_market_calendars. However, I 'm a bit puzzled as to how to correctly add a market calendar to cerebro instance.

I need German trading days, so I figured that EUREX will be the calendar to choose.

...
cerebro.addcalendar("EUREX")
results = cerebro.run()  # This line (107) throws an error

How am I supposed to correctly add a market calendar?

Here's the traceback...

  File "C:/Users/D292498/PycharmProjects/pybt/src/bt/main.py", line 107, in <module>
    results = cerebro.run()
  File "C:\Users\D292498\AppData\Local\conda\conda\envs\pybt\lib\site-packages\backtrader\cerebro.py", line 1127, in run
    runstrat = self.runstrategies(iterstrat)
  File "C:\Users\D292498\AppData\Local\conda\conda\envs\pybt\lib\site-packages\backtrader\cerebro.py", line 1298, in runstrategies
    self._runnext(runstrats)
  File "C:\Users\D292498\AppData\Local\conda\conda\envs\pybt\lib\site-packages\backtrader\cerebro.py", line 1542, in _runnext
    drets.append(d.next(ticks=False))
  File "C:\Users\D292498\AppData\Local\conda\conda\envs\pybt\lib\site-packages\backtrader\feed.py", line 407, in next
    ret = self.load()
  File "C:\Users\D292498\AppData\Local\conda\conda\envs\pybt\lib\site-packages\backtrader\feed.py", line 523, in load
    retff = ff(self, *fargs, **fkwargs)
  File "C:\Users\D292498\AppData\Local\conda\conda\envs\pybt\lib\site-packages\backtrader\resamplerfilter.py", line 518, in __call__
    onedge, docheckover = self._dataonedge(data)  # for subdays
  File "C:\Users\D292498\AppData\Local\conda\conda\envs\pybt\lib\site-packages\backtrader\resamplerfilter.py", line 322, in _dataonedge
    ret = data._calendar.last_monthday(data.datetime.date())
  File "C:\Users\D292498\AppData\Local\conda\conda\envs\pybt\lib\site-packages\backtrader\tradingcal.py", line 94, in last_monthday
    return day.month != self._nextday(day)[0].month
  File "C:\Users\D292498\AppData\Local\conda\conda\envs\pybt\lib\site-packages\backtrader\tradingcal.py", line 250, in _nextday
    i = self.dcache.searchsorted(day)
  File "C:\Users\D292498\AppData\Local\conda\conda\envs\pybt\lib\site-packages\pandas\core\indexes\extension.py", line 253, in searchsorted
    return self._data.searchsorted(value, side=side, sorter=sorter)
  File "C:\Users\D292498\AppData\Local\conda\conda\envs\pybt\lib\site-packages\pandas\core\arrays\_mixins.py", line 190, in searchsorted
    value = self._validate_searchsorted_value(value)
  File "C:\Users\D292498\AppData\Local\conda\conda\envs\pybt\lib\site-packages\pandas\core\arrays\datetimelike.py", line 636, in _validate_searchsorted_value
    return self._validate_scalar(value, allow_listlike=True, setitem=False)
  File "C:\Users\D292498\AppData\Local\conda\conda\envs\pybt\lib\site-packages\pandas\core\arrays\datetimelike.py", line 564, in _validate_scalar
    raise TypeError(msg)
TypeError: value should be a 'Timestamp', 'NaT', or array of those. Got 'date' instead.

run-out

@andi The error message suggests your dates are not properly formatted somehow. Could you share a few lines with datas, and also how you are putting them into cerebro?

andi

@run-out
This is a bit weird, because if I don't add the calendar, the script runs smoothly. I am actually loading a csv file that contains data, which I downloaded from Bloomberg.

Here's an excerpt from the csv file:

date,open,high,low,close,volume
2001-01-03,62.64,64.5,61.8,64.5,1087185.0
2001-01-04,65.23,65.23,63.67,63.67,1160550.0
2001-01-05,64.18,64.7,63.48,63.79,1789392.0
2001-01-08,63.68,64.3,63.65,63.73,562735.0
2001-01-09,64.54,64.6,63.67,63.81,943175.0
2001-01-10,64.06,64.11,62.92,63.48,944343.0
2001-01-11,63.9,64.44,63.4,64.44,678861.0
2001-01-12,64.83,65.42,64.68,64.89,621234.0

class BloombergCSV(bt.feeds.GenericCSVData):
    params = (
        ("fromdate", datetime.datetime(1990, 1, 1)),
        ("todate", datetime.date.today()),
        ("nullvalue", float("NaN")),
        ("dtformat", "%Y-%m-%d"),
        ("tmformat", "%H.%M.%S"),
        ("datetime", 0),
        ("open", 1),
        ("high", 2),
        ("low", 3),
        ("close", 4),
        ("volume", 5),
        ("time", -1),
        ("openinterest", -1),
    )

Excerpt from main code:

if __name__ == "__main__":
    print("--- Initializing backtest...")
    cerebro = bt.Cerebro()

    # --- data feeds ---
    dax = BloombergCSV(
        dataname="./resources/DAXEX.csv", fromdate=datetime.datetime(2020, 1, 1)
    )
    cerebro.adddata(data=dax, name="DAX")
    cerebro.resampledata(dax, name="DAX_monthly", timeframe=bt.TimeFrame.Months)

    # --- strategy ---
    cerebro.addstrategy(SMAMomentum, period=10, print_log=True)

    # --- analyzer ---
    cerebro.addanalyzer(CashMarket, _name="cashmarket")

    # --- execution ---
    print(f"Starting Portfolio Value: {cerebro.broker.getvalue():,.2f}")
    results = cerebro.run()
    print(f"Final Portfolio Value: {cerebro.broker.getvalue():,.2f}")

andi

I came up with the following work-around:

Extracting the trading days from my data feeds
Identifying the month-end dates
Creating a timer that throws an alarm via the notify_timer method in my strategy class.

Extracting the trading days from my data feeds:

I stored my pricing data in csv files. So I created a simple function that reads in the csv file and extracts the days.
However you get your trading dates, make sure to save them as pandas.DatetimeIndex.

def get_trading_days(
    csv_file: Path, fromdate: datetime.date, todate: datetime.date
) -> pd.DatetimeIndex:
    """
    Extracts trading days from a csv-file.
    
    Parameters
    ----------
    csv_file
        Path to csv-file.
    fromdate
        First selected trading date.
    todate
        Last selected trading date.

    Returns
    -------
    pandas.DatetimeIndex
        DatetimeIndex comprising the selected trading dates.
    """
    path = Path("./resources").joinpath(csv_file)
    df = pd.read_csv(path, index_col=0)
    df.index = pd.to_datetime(df.index)

    return df.loc[(df.index >= fromdate) & (df.index <= todate)].index

Identifying month-end dates and creating a timer class:

I created a PeriodEndTradingDays class. This class provides a private method called _extract_period_end_dates. The extracted dates will be serving the timer.

# Timer.py
import datetime
from typing import Literal

import pandas as pd


class PeriodEndTradingDays:
    """
    Creates a timer, which notifies if the current trading day is the last trading
    day of the period.

    Parameters
    ----------
    dt_index
        Datetime index comprising all trading days.
    frequency
        Determines the frequency used in order to extract the last trading days. 
        Could be weekly ('W'), monthly ('M'), quarterly ('Q'), or yearly ('Y'). 
    """

    def __init__(
        self, dt_index: pd.DatetimeIndex, frequency: Literal["W", "M", "Q", "Y"]
    ):
        self.dt_index = dt_index
        self.freq = frequency
        self.period_end_dates = self._extract_period_end_dates()

    def __call__(self, d):
        if d in self.period_end_dates:
            return True
        else:
            return False

    def _extract_period_end_dates(self):
        """Extracts the last trading days for a given frequency."""
        if self.freq.lower() == "w":
            mask = pd.Series(self.dt_index.week) != pd.Series(self.dt_index.week).shift(
                -1
            )
        elif self.freq.lower() == "m":
            mask = pd.Series(self.dt_index.month) != pd.Series(
                self.dt_index.month
            ).shift(-1)
        elif self.freq.lower() == "q":
            mask = pd.Series(self.dt_index.quarter) != pd.Series(
                self.dt_index.quarter
            ).shift(-1)
        else:
            mask = pd.Series(self.dt_index.year) != pd.Series(self.dt_index.year).shift(
                -1
            )

        period_end_trading_dates = self.dt_index[mask.values]
        return [
            datetime.date(year=x.year, month=x.month, day=x.day)
            for x in period_end_trading_dates
        ]

In my main.py script, I am saving the trading days for all my feeds in a dict called trading_days.

portfolio = ["DAX Index", "SPX Index"]
fromdate = datetime.datetime(2019, 12, 20)
todate = datetime.datetime(2021, 7, 31)

trading_days = {
    ticker: DataFeeds.get_trading_days(
        f"{ticker}.csv", fromdate=fromdate, todate=todate
    )
    for ticker in portfolio
}

Later, when adding my feeds to cerebro, I am also adding a timer. The allow keyword is important, here. It makes use of my PeriodEndTradingDays class. You can find more about manually created timers here: https://www.backtrader.com/docu/timers/timers/

for ticker in portfolio:
    cerebro.adddata(data=data_feeds[ticker], name=ticker)

    cerebro.add_timer(
        when=bt.timer.SESSION_END,
        allow=PeriodEndTradingDays(dt_index=trading_days[ticker], frequency="M"),
        strats=True,
        timername=ticker,
    )

Finally, in my strategy class, I overwrite the notify_timer method. The timer's output (True/False) will be saved in a dict called rebal

def notify_timer(self, timer, when, *args, **kwargs):
    self.rebal[kwargs.get("timername")] = True

Lastly, in my next method, I can check if self.rebal==True. Don't forget to set self.rebal=False at the end of the next method!

I am happy to take any suggestions for improvement!