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 numbers import Number
from typing import Iterable, List, Literal, Mapping, Sequence, Union 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) @dataclass(frozen=True)
@ -376,6 +376,33 @@ class TimeSeriesCore(UserDict):
def __contains__(self, key: object) -> bool: def __contains__(self, key: object) -> bool:
return super().__contains__(key) 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 @property
def iloc(self) -> Mapping: def iloc(self) -> Mapping:
"""Returns an item or a set of items based on index """Returns an item or a set of items based on index

View File

@ -552,7 +552,10 @@ class TimeSeries(TimeSeriesCore):
return max_drawdown return max_drawdown
def expand( 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: ) -> TimeSeries:
try: try:
to_frequency: Frequency = getattr(AllFrequencies, to_frequency) to_frequency: Frequency = getattr(AllFrequencies, to_frequency)
@ -562,8 +565,10 @@ class TimeSeries(TimeSeriesCore):
if to_frequency.days >= self.frequency.days: if to_frequency.days >= self.frequency.days:
raise ValueError("TimeSeries can be only expanded to a higher frequency") 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_dates = create_date_series(
new_ts: dict = {dt: self.data.get(dt, None) for dt in new_dates} 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) output_ts: TimeSeries = TimeSeries(new_ts, frequency=to_frequency.symbol)
if method == "ffill": if method == "ffill":

View File

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