Compare commits
3 Commits
b34c14d778
...
3b96d231f4
Author | SHA1 | Date | |
---|---|---|---|
3b96d231f4 | |||
8be3359437 | |||
ee48446c6e |
@ -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
|
||||
|
@ -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":
|
||||
|
@ -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):
|
||||
|
Loading…
Reference in New Issue
Block a user