Compare commits
4 Commits
ad68dcd930
...
60f25de710
Author | SHA1 | Date | |
---|---|---|---|
60f25de710 | |||
347634f536 | |||
eeda4064b2 | |||
00ac66219b |
@ -1,4 +1,5 @@
|
|||||||
import datetime
|
import datetime
|
||||||
|
import inspect
|
||||||
from collections import UserDict, UserList
|
from collections import UserDict, UserList
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from numbers import Number
|
from numbers import Number
|
||||||
@ -16,6 +17,34 @@ class Frequency:
|
|||||||
symbol: str
|
symbol: str
|
||||||
|
|
||||||
|
|
||||||
|
def date_parser(*pos):
|
||||||
|
def parse_dates(func):
|
||||||
|
def wrapper_func(*args, **kwargs):
|
||||||
|
date_format = kwargs.get("date_format", None)
|
||||||
|
args = list(args)
|
||||||
|
sig = inspect.signature(func)
|
||||||
|
params = [i[0] for i in sig.parameters.items()]
|
||||||
|
|
||||||
|
for j in pos:
|
||||||
|
kwarg = params[j]
|
||||||
|
date = kwargs.get(kwarg, None)
|
||||||
|
in_args = False
|
||||||
|
if date is None:
|
||||||
|
date = args[j]
|
||||||
|
in_args = True
|
||||||
|
|
||||||
|
parsed_date = _parse_date(date, date_format)
|
||||||
|
if not in_args:
|
||||||
|
kwargs[kwarg] = parsed_date
|
||||||
|
else:
|
||||||
|
args[j] = parsed_date
|
||||||
|
return func(*args, **kwargs)
|
||||||
|
|
||||||
|
return wrapper_func
|
||||||
|
|
||||||
|
return parse_dates
|
||||||
|
|
||||||
|
|
||||||
class AllFrequencies:
|
class AllFrequencies:
|
||||||
D = Frequency("daily", "days", 1, 1, "D")
|
D = Frequency("daily", "days", 1, 1, "D")
|
||||||
W = Frequency("weekly", "days", 7, 7, "W")
|
W = Frequency("weekly", "days", 7, 7, "W")
|
||||||
@ -157,7 +186,7 @@ class TimeSeriesCore(UserDict):
|
|||||||
def __init__(
|
def __init__(
|
||||||
self, data: List[Iterable], frequency: Literal["D", "W", "M", "Q", "H", "Y"], date_format: str = "%Y-%m-%d"
|
self, data: List[Iterable], frequency: Literal["D", "W", "M", "Q", "H", "Y"], date_format: str = "%Y-%m-%d"
|
||||||
):
|
):
|
||||||
"""Instantiate a TimeSeries object
|
"""Instantiate a TimeSeriesCore object
|
||||||
|
|
||||||
Parameters
|
Parameters
|
||||||
----------
|
----------
|
||||||
|
@ -5,7 +5,7 @@ from typing import Iterable, List, Literal, Mapping, Union
|
|||||||
|
|
||||||
from dateutil.relativedelta import relativedelta
|
from dateutil.relativedelta import relativedelta
|
||||||
|
|
||||||
from .core import AllFrequencies, TimeSeriesCore
|
from .core import AllFrequencies, TimeSeriesCore, date_parser
|
||||||
from .utils import (
|
from .utils import (
|
||||||
_find_closest_date,
|
_find_closest_date,
|
||||||
_interval_to_years,
|
_interval_to_years,
|
||||||
@ -184,6 +184,7 @@ class TimeSeries(TimeSeriesCore):
|
|||||||
|
|
||||||
return self.__class__(new_ts, frequency=self.frequency.symbol)
|
return self.__class__(new_ts, frequency=self.frequency.symbol)
|
||||||
|
|
||||||
|
@date_parser(1)
|
||||||
def calculate_returns(
|
def calculate_returns(
|
||||||
self,
|
self,
|
||||||
as_on: Union[str, datetime.datetime],
|
as_on: Union[str, datetime.datetime],
|
||||||
@ -260,7 +261,7 @@ class TimeSeries(TimeSeriesCore):
|
|||||||
(datetime.datetime(2020, 1, 1, 0, 0), .0567)
|
(datetime.datetime(2020, 1, 1, 0, 0), .0567)
|
||||||
"""
|
"""
|
||||||
|
|
||||||
as_on = _parse_date(as_on, date_format)
|
# as_on = _parse_date(as_on, date_format)
|
||||||
as_on_delta, prior_delta = _preprocess_match_options(as_on_match, prior_match, closest)
|
as_on_delta, prior_delta = _preprocess_match_options(as_on_match, prior_match, closest)
|
||||||
|
|
||||||
prev_date = as_on - relativedelta(**{interval_type: interval_value})
|
prev_date = as_on - relativedelta(**{interval_type: interval_value})
|
||||||
@ -276,6 +277,7 @@ class TimeSeries(TimeSeriesCore):
|
|||||||
returns = returns ** (1 / years)
|
returns = returns ** (1 / years)
|
||||||
return (current[0] if return_actual_date else as_on), returns - 1
|
return (current[0] if return_actual_date else as_on), returns - 1
|
||||||
|
|
||||||
|
@date_parser(1, 2)
|
||||||
def calculate_rolling_returns(
|
def calculate_rolling_returns(
|
||||||
self,
|
self,
|
||||||
from_date: Union[datetime.date, str],
|
from_date: Union[datetime.date, str],
|
||||||
@ -356,8 +358,8 @@ class TimeSeries(TimeSeriesCore):
|
|||||||
TimeSeries.calculate_returns
|
TimeSeries.calculate_returns
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from_date = _parse_date(from_date, date_format)
|
# from_date = _parse_date(from_date, date_format)
|
||||||
to_date = _parse_date(to_date, date_format)
|
# to_date = _parse_date(to_date, date_format)
|
||||||
|
|
||||||
if frequency is None:
|
if frequency is None:
|
||||||
frequency = self.frequency
|
frequency = self.frequency
|
||||||
|
@ -13,7 +13,7 @@ class FincalOptions:
|
|||||||
|
|
||||||
def _parse_date(date: str, date_format: str = None):
|
def _parse_date(date: str, date_format: str = None):
|
||||||
"""Parses date and handles errors"""
|
"""Parses date and handles errors"""
|
||||||
|
# print(date, date_format)
|
||||||
if isinstance(date, (datetime.datetime, datetime.date)):
|
if isinstance(date, (datetime.datetime, datetime.date)):
|
||||||
return datetime.datetime.fromordinal(date.toordinal())
|
return datetime.datetime.fromordinal(date.toordinal())
|
||||||
|
|
||||||
|
Loading…
Reference in New Issue
Block a user