Compare commits

...

3 Commits

3 changed files with 37 additions and 4 deletions

View File

@ -5,7 +5,7 @@ from dataclasses import dataclass
from numbers import Number
from typing import Iterable, List, Literal, Mapping, Sequence, Union
from .utils import _parse_date, _preprocess_timeseries
from .utils import FincalOptions, _parse_date, _preprocess_timeseries
@dataclass(frozen=True)
@ -376,6 +376,33 @@ class TimeSeriesCore(UserDict):
def __contains__(self, key: object) -> bool:
return super().__contains__(key)
@date_parser(1)
def get(self, date: Union[str, datetime.datetime], default=None, closest=None):
if closest is None:
closest = FincalOptions.get_closest
if closest == "exact":
try:
item = self._get_item_from_date(date)
return item
except KeyError:
return default
if closest == "previous":
delta = datetime.timedelta(-1)
elif closest == "next":
delta = datetime.timedelta(1)
else:
raise ValueError(f"Invalid argument from closest {closest!r}")
while True:
try:
item = self._get_item_from_date(date)
return item
except KeyError:
date += delta
@property
def iloc(self) -> Mapping:
"""Returns an item or a set of items based on index

View File

@ -552,7 +552,10 @@ class TimeSeries(TimeSeriesCore):
return max_drawdown
def expand(
self, to_frequency: Literal["D", "W", "M", "Q", "H"], method: Literal["ffill", "bfill", "interpolate"]
self,
to_frequency: Literal["D", "W", "M", "Q", "H"],
method: Literal["ffill", "bfill", "interpolate"],
skip_weekends: bool = False,
) -> TimeSeries:
try:
to_frequency: Frequency = getattr(AllFrequencies, to_frequency)
@ -562,8 +565,10 @@ class TimeSeries(TimeSeriesCore):
if to_frequency.days >= self.frequency.days:
raise ValueError("TimeSeries can be only expanded to a higher frequency")
new_dates = create_date_series(self.start_date, self.end_date, frequency=to_frequency.symbol)
new_ts: dict = {dt: self.data.get(dt, None) for dt in new_dates}
new_dates = create_date_series(
self.start_date, self.end_date, frequency=to_frequency.symbol, skip_weekends=skip_weekends
)
new_ts: dict = {dt: self.get(dt, closest="previous")[1] for dt in new_dates}
output_ts: TimeSeries = TimeSeries(new_ts, frequency=to_frequency.symbol)
if method == "ffill":

View File

@ -10,6 +10,7 @@ class FincalOptions:
date_format: str = "%Y-%m-%d"
closest: str = "before" # after
traded_days: int = 365
get_closest: str = "exact"
def _parse_date(date: str, date_format: str = None):