Find last trading day of month
-
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 installedpandas_market_calendars
. However, I 'm a bit puzzled as to how to correctly add a market calendar tocerebro
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.
-
@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?
-
@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}")
-
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 aspandas.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 adict
calledtrading_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 atimer
. Theallow
keyword is important, here. It makes use of myPeriodEndTradingDays
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 adict
calledrebal
def notify_timer(self, timer, when, *args, **kwargs): self.rebal[kwargs.get("timername")] = True
Lastly, in my
next
method, I can check ifself.rebal==True
. Don't forget to setself.rebal=False
at the end of thenext
method!I am happy to take any suggestions for improvement!