An OOP approach to Backtrader Timers

custom_backtrader_timer.py

#!/usr/bin/env python
# -*- coding: utf-8; py-indent-offset:4 -*-

###############################################################################
#
# Copyright (C) 2021 Emmanuel Katchy
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
###############################################################################

import datetime
from pandas.core.indexes.base import ensure_index
import pandas_market_calendars as pmcal
import pandas as pd
import pytz

dtime = datetime.datetime

class CustFreq:
    def __init__(self, start_date, end_date, *args, **kwargs):
        start_date = start_date.replace(tzinfo=pytz.UTC)
        end_date = end_date.replace(tzinfo=pytz.UTC)
        self.cal = pmcal.get_calendar(kwargs.get("cal","NYSE"))
        self.freq = kwargs.get("freq","D")
        self.valid_days = self.cal.valid_days(start_date, end_date)
        self.timer_valid_days = []
        self.bd_dts = []

        cal_range = pd.bdate_range(start_date, end_date)

        cal_df = pd.DataFrame(index=cal_range, data=cal_range)
        cal_range = list(cal_df.resample(self.freq).groups)

        self.timer_valid_days = list(map(pd.Timestamp.to_pydatetime,cal_range))

        self.generate_market_dates()

    def _get_nearest_date(self, dt):
        later = list(filter(lambda d: d > dt, self.valid_days))
        if len(later) == 0: 
            return None
        closest_date = min(later)
        return closest_date

    def generate_market_dates(self):
        self.bd_dts = [] # Dates that need to be changed to match calendar
        for indx,i in enumerate(self.timer_valid_days):
            nd = self._get_nearest_date(i)
            if i != nd:
                self.bd_dts.append((i, nd))
            else:
                pass
        self._replace_out_of_cal_vals()

    
    def _replace_out_of_cal_vals(self):
        for (b,g) in self.bd_dts:
            indx = self.timer_valid_days.index(b)
            self.timer_valid_days.remove(b)
            self.timer_valid_days.insert(indx, g)
        
        self._pdtimestamp_to_date()

    def _pdtimestamp_to_date(self):
        self.timer_valid_days = list(map(pd.Timestamp, self.timer_valid_days))
        self.timer_valid_days = list(map(pd.Timestamp.date, self.timer_valid_days))

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