Coverage for rivapy/tools/datetime_grid.py: 35%
141 statements
« prev ^ index » next coverage.py v7.8.2, created at 2025-06-05 14:27 +0000
1from typing import Union, Callable
2import datetime as dt
3import abc
4import pandas as pd
5import numpy as np
6from scipy.optimize import curve_fit as curve_fit
7from scipy.interpolate import interp1d
8from rivapy.tools.enums import DayCounterType
9from rivapy.tools.datetools import DayCounter
12class DateTimeGrid:
13 def __init__(self,
14 datetime_grid: pd.DatetimeIndex=None,
15 start:Union[dt.datetime, pd.Timestamp]=None,
16 end:Union[dt.datetime, pd.Timestamp]=None,
17 freq: str='1H',
18 daycounter:Union[str, DayCounterType]=DayCounterType.Act365Fixed,
19 tz=None,
20 inclusive = 'left'):
21 """Object to handle datetimes together with their respective timegridpoints (according to a given daycount convention)
23 Args:
24 datetime_grid (pd.DatetimeIndex, optional): A grid of datetime values that is then transformed to datapoints. Defaults to None. Note that either this or a start and end date together with a frequency must be given. If a start date and a datetimegrid are specified at the same time, an exception will be thrown.
25 start (Union[dt.datetime, pd.Timestamp], optional): Start date of the datetime grid. Defaults to None.
26 end (Union[dt.datetime, pd.Timestamp], optional): Enddate of the datetimegrid. The parameter inclusive specifies whether the end date will be included into the grid or not.. Defaults to None.
27 freq (str, optional): A frequency string. Defaults to '1H'. See the documentation for the pandas function :external:py:func:`pandas.date_range` for more details of this string.
28 daycounter (Union[str, DayCounterType], optional): String or daycounterType used to compute the timepoints internally. Defaults to DayCounterType.Act365Fixed.
29 tz (str, optional): Time zone name for returning localized DatetimeIndex, see the pandas function :external:py:func:`pandas.date_range` for more details. Defaults to None.
30 inclusive (str, optional): Defines which boundary is included into the grid, see the pandas function ::external:py:func:`pandas.date_range`. Defaults to 'left'.
32 Raises:
33 ValueError: If both, datetime_grid and start, are either None or not None.
34 """
35 if (start is not None) and (datetime_grid is not None):
36 raise ValueError('Either datetime_grid or start must be None.')
37 if start is not None:
38 self.dates = pd.date_range(start, end, freq=freq, tz=tz, inclusive=inclusive).to_pydatetime()
39 else:
40 self.dates = datetime_grid
41 if self.dates is not None:
42 if start is None:
43 start = self.dates[0]
44 self.timegrid = np.array(DayCounter(daycounter).yf(start, self.dates))
45 self.shape = self.timegrid.shape
46 self.df = pd.DataFrame({'dates': self.dates, 'tg': self.timegrid})
47 else:
48 self.dates = None
49 self.timegrid = None
50 self.shape = None
51 self.df = None
54 def get_daily_subgrid(self)->'DateTimeGrid':
55 """Return a new datetime grid that is a subgrid of the current grid consisting of just daily values.
57 Returns:
58 DateTimeGrid: Reulting grid.
59 """
60 df = self.df.groupby(by=['dates']).min()
61 df = df.reset_index()
62 result = DateTimeGrid(None, None, freq='1D')
63 result.dates=np.array([d.to_pydatetime() for d in df['dates']])
64 result.timegrid = df['tg'].values
65 result.shape = result.timegrid.shape
66 result.df = pd.DataFrame({'dates': result.dates, 'tg': result.timegrid})
67 return result
69 def get_day_of_year(self):
70 if 'day_of_year' not in self.df.columns:
71 self.df['day_of_year'] = self.df.dates.dt.dayofyear
72 return self.df['day_of_year']
74 def get_day_of_week(self):
75 if 'day_of_week' not in self.df.columns:
76 self.df['day_of_week'] = self.df.dates.dt.dayofweek
77 return self.df['day_of_week']
79 def get_hour_of_day(self):
80 if 'hour_of_day' not in self.df.columns:
81 self.df['hour_of_day'] = self.df.dates.dt.hour
82 return self.df['hour_of_day']
84 def get_minute_of_day(self):
85 if 'minute_of_day' not in self.df.columns:
86 self.df['minute_of_day'] = self.df.dates.dt.minute
87 return self.df['minute_of_day']
90class __TimeGridFunction(abc.ABC):
91 @abc.abstractmethod
92 def _compute(self, d: dt.datetime)->float:
93 pass
95 def compute(self, tg: DateTimeGrid, x=None)->np.ndarray:
96 if x is None:
97 x = np.empty(tg.shape)
98 for i in range(tg.shape[0]):
99 x[i] = self._compute(tg.dates[i])
100 return x
102class _Add(__TimeGridFunction):
103 def __init__(self, f1,f2):
104 self._f1 = f1
105 self._f2 = f2
107 def _compute(self, d: dt.datetime)->float:
108 return self._f1._compute(d)+self._f2._compute(d)
110class _Mul(__TimeGridFunction):
111 def __init__(self, f1,f2):
112 self._f1 = f1
113 self._f2 = f2
115 def _compute(self, d: dt.datetime)->float:
116 return self._f1._compute(d)*self._f2._compute(d)
118class _TimeGridFunction(__TimeGridFunction):
119 """Abstract base class for all functions that are defined on datetimes
121 Args:
122 _TimeGridFunction (_type_): _description_
124 Returns:
125 _type_: _description_
126 """
127 @abc.abstractmethod
128 def _compute(self, d: dt.datetime)->float:
129 pass
131 def __add__(self, other):
132 return _Add(self, other)
134 def __mul__(self, other):
135 return _Mul(self, other)
137class MonthlyConstantFunction(_TimeGridFunction):
138 def __init__(self, values:list):
139 """Function that is constant across a month.
141 Args:
142 values (list): values[i] contains the value for the (i+1)-th month
143 """
144 self.values = values
146 def _compute(self, d: dt.datetime)->float:
147 return self.values[d.month-1]
149class HourlyConstantFunction(_TimeGridFunction):
150 def __init__(self, values:list):
151 """Function that is constant on hours.
153 Args:
154 values (list): values[i] contains the value for the i-th hour
155 """
156 self.values = values
158 def _compute(self, d: dt.datetime)->float:
159 return self.values[d.hour]
161class ParametrizedFunction:
162 def __init__(self, x: np.ndarray):
163 self.x
165 def __call__(self, x):
166 pass
167class PeriodicFunction(_TimeGridFunction):
168 def __init__(self, f: Callable, frequency: str, ignore_leap_day: bool=True, granularity='D'):
169 self.f = f
170 self.ignore_leap_day = ignore_leap_day
171 self.frequency = frequency
172 self.granularity = granularity
174 def _compute(self, d: dt.datetime)->float:
175 raise NotImplemented()
176 return self.f(d.dayofyear)
178 def compute(self, tg: DateTimeGrid, x=None)->np.ndarray:
179 if self.frequency == 'Y':
180 x = tg.get_day_of_year().values
181 if self.ignore_leap_day:
182 x = np.minimum(x, 365)
183 scaler = 1.0/365.0
184 elif self.frequency == 'W':
185 x = tg.get_day_of_week().values
186 #x = x/6.0
187 scaler = 1.0/6.0
188 else:
189 raise ValueError('Unknown frequency ' + self.frequency)
190 if self.granularity == 'H':
191 x = x + tg.get_hour_of_day().values/(24.0)
192 elif self.granularity == 'T':
193 x = x + tg.minute_of_day().values/(24.0*60.0)
194 x = scaler*x
195 return self.f(x)
197 def calibrate(self, dates: Union[pd.DatetimeIndex, DateTimeGrid], values: np.ndarray):
198 def f(x, *params):
199 for i in range(self.f.x.shape[0]):
200 self.f.x[i] = params[i]
201 return self.compute(x)
202 tg = dates
203 if not isinstance(tg, DateTimeGrid):
204 tg = DateTimeGrid(datetime_grid=dates)
205 popt, pcov = curve_fit(f, tg,values,self.f.x)
206 f.x = popt
211class InterpolatedFunction(_TimeGridFunction):
212 def __init__(self, datetime_grid: pd.DatetimeIndex, values: np.ndarray, kind: str='linear', bounds_error=False):
213 self._values = values
214 self._tg = DateTimeGrid(datetime_grid)
215 self.kind = kind
216 self._f = interp1d(self._tg.timegrid, self._values, kind=self.kind, fill_value=(values[0], values[-1]), bounds_error=bounds_error)
218 def _compute(self, d: dt.datetime)->float:
219 raise NotImplemented()
221 def compute(self, datetime_grid: pd.DatetimeIndex=None)->np.ndarray:
222 x = (datetime_grid-self._tg.dates[0]).total_seconds()/pd.Timedelta(days=365).total_seconds()
223 return self._f(x)