Skip to content

Instantly share code, notes, and snippets.

@TobeTek
Created January 3, 2022 03:32
Show Gist options
  • Save TobeTek/335747c00b8cc6965d4f55765b4d1d67 to your computer and use it in GitHub Desktop.
Save TobeTek/335747c00b8cc6965d4f55765b4d1d67 to your computer and use it in GitHub Desktop.
An OOP approach to Backtrader Timers
#!/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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment