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 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
|
||||||
|
@ -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":
|
||||||
|
@ -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):
|
||||||
|
Loading…
Reference in New Issue
Block a user