Compare commits

...

5 Commits

  1. 41
      fincal/core.py
  2. 15
      fincal/exceptions.py
  3. 185
      fincal/fincal.py
  4. 13
      fincal/utils.py

41
fincal/core.py

@ -2,7 +2,7 @@ import datetime
from collections import UserDict, UserList
from dataclasses import dataclass
from numbers import Number
from typing import Iterable, List, Literal, Sequence
from typing import Iterable, List, Literal, Sequence, Tuple
from .utils import _parse_date, _preprocess_timeseries
@ -189,29 +189,40 @@ class TimeSeriesCore(UserDict):
self._end_date = None
@property
def dates(self):
def dates(self) -> Series:
"""Get a list of all the dates in the TimeSeries object"""
if self._dates is None or len(self._dates) != len(self.data):
self._dates = list(self.data.keys())
return Series(self._dates, "date")
@property
def values(self):
def values(self) -> Series:
"""Get a list of all the Values in the TimeSeries object"""
if self._values is None or len(self._values) != len(self.data):
self._values = list(self.data.values())
return Series(self._values, "number")
@property
def start_date(self):
def start_date(self) -> datetime.datetime:
"""The first date in the TimeSeries object"""
return self.dates[0]
@property
def end_date(self):
def end_date(self) -> datetime.datetime:
"""The last date in the TimeSeries object"""
return self.dates[-1]
def _get_printable_slice(self, n: int):
"""Returns a slice of the dataframe from beginning and end"""
"""Helper function for __repr__ and __str__
Returns a slice of the dataframe from beginning and end.
"""
printable = {}
iter_f = iter(self.data)
@ -322,7 +333,21 @@ class TimeSeriesCore(UserDict):
return self.data.items()
@property
def iloc(self):
"""Returns an item or a set of items based on index"""
def iloc(self) -> List[Tuple[datetime.datetime, float]]:
"""Returns an item or a set of items based on index
supports slicing using numerical index.
Accepts integers or Python slice objects
Usage
-----
>>> ts = TimeSeries(data, frequency='D')
>>> ts.iloc[0] # get the first value
>>> ts.iloc[-1] # get the last value
>>> ts.iloc[:3] # get the first 3 values
>>> ts.illoc[-3:] # get the last 3 values
>>> ts.iloc[5:10] # get five values starting from the fifth value
>>> ts.iloc[::2] # get every alternate date
"""
return _IndexSlicer(self)

15
fincal/exceptions.py

@ -1,6 +1,21 @@
import datetime
from typing import Literal
class DateNotFoundError(Exception):
"""Exception to be raised when date is not found"""
def __init__(self, message, date):
message = f"{message}: {date}"
super().__init__(message)
class DateOutOfRangeError(Exception):
"""Exception to be raised when provided date is outside the range of dates in the time series"""
def __init__(self, date: datetime.datetime, type: Literal['min', 'max']) -> None:
if type == 'min':
message = f"Provided date {date} is before the first date in the TimeSeries"
if type == 'max':
message = f"Provided date {date} is after the last date in the TimeSeries"
super().__init__(message)

185
fincal/fincal.py

@ -1,7 +1,7 @@
from __future__ import annotations
import datetime
from typing import List, Literal, Union
from typing import Iterable, List, Literal, Mapping, Union
from dateutil.relativedelta import relativedelta
@ -15,9 +15,41 @@ from .utils import (
def create_date_series(
start_date: datetime.datetime, end_date: datetime.datetime, frequency: str, eomonth: bool = False
start_date: Union[str, datetime.datetime],
end_date: Union[str, datetime.datetime],
frequency: Literal["D", "W", "M", "Q", "H", "Y"],
eomonth: bool = False,
) -> List[datetime.datetime]:
"""Creates a date series using a frequency"""
"""Create a date series with a specified frequency
Parameters
----------
start_date : str | datetime.datetime
Date series will always start at this date
end_date : str | datetime.datetime
The date till which the series should extend
Depending on the other parameters, this date may or may not be present
in the final date series
frequency : D | W | M | Q | H | Y
Frequency of the date series.
The gap between each successive date will be equivalent to this frequency
eomonth : bool, optional
Speacifies if the dates in the series should be end of month dates.
Can only be used if the frequency is Monthly or lower.
Returns
-------
List[datetime.datetime]
Returns the series as a list of datetime objects
Raises
------
ValueError
If eomonth is True and frequency is higher than monthly
"""
frequency = getattr(AllFrequencies, frequency)
if eomonth and frequency.days < AllFrequencies.M.days:
@ -31,11 +63,11 @@ def create_date_series(
for i in range(0, int(datediff)):
diff = {frequency.freq_type: frequency.value * i}
date = start_date + relativedelta(**diff)
if eomonth:
if date.month == 12:
date = date.replace(day=31)
else:
date = date.replace(day=1).replace(month=date.month+1) - relativedelta(days=1)
next_month = 1 if date.month == 12 else date.month + 1
date = date.replace(day=1).replace(month=next_month) - relativedelta(days=1)
if date <= end_date:
dates.append(date)
@ -43,7 +75,40 @@ def create_date_series(
class TimeSeries(TimeSeriesCore):
"""Container for TimeSeries objects"""
"""1-Dimensional Time Series object
Parameters
----------
data : List[Iterable] | Mapping
Time Series data in the form of list of tuples.
The first element of each tuple should be a date and second element should be a value.
The following types of objects can be passed to create a TimeSeries object:
* List of tuples containing date & value
* List of lists containing date & value
* List of dictionaries containing key: value pair of date and value
* List of dictionaries with 2 keys, first representing date & second representing value
* Dictionary of key: value pairs
date_format : str, optional, default "%Y-%m-%d"
Specify the format of the date
Required only if the first argument of tuples is a string. Otherwise ignored.
frequency : str, optional, default "infer"
The frequency of the time series. Default is infer.
The class will try to infer the frequency automatically and adjust to the closest member.
Note that inferring frequencies can fail if the data is too irregular.
Valid values are {D, W, M, Q, H, Y}
"""
def __init__(
self,
data: Union[List[Iterable], Mapping],
frequency: Literal["D", "W", "M", "Q", "H", "Y"],
date_format: str = "%Y-%m-%d",
):
"""Instantiate a TimeSeriesCore object"""
super().__init__(data, frequency, date_format)
def info(self):
"""Summary info about the TimeSeries object"""
@ -125,12 +190,13 @@ class TimeSeries(TimeSeriesCore):
return_actual_date: bool = True,
as_on_match: str = "closest",
prior_match: str = "closest",
closest: str = "previous",
if_not_found: Literal['fail', 'nan'] = 'fail',
closest: Literal["previous", "next", "exact"] = "previous",
closest_max_days: int = -1,
if_not_found: Literal["fail", "nan"] = "fail",
compounding: bool = True,
interval_type: Literal['years', 'months', 'days'] = 'years',
interval_type: Literal["years", "months", "days"] = "years",
interval_value: int = 1,
date_format: str = None
date_format: str = None,
) -> float:
"""Method to calculate returns for a certain time-period as on a particular date
@ -153,6 +219,12 @@ class TimeSeries(TimeSeriesCore):
The mode of matching the closest date.
Valid values are 'exact', 'previous', 'next' and next.
closest_max_days: int, default -1
The maximum acceptable gap between the provided date arguments and actual date.
Pass -1 for no limit.
Note: There's a hard max limit of 1000 days due to Python's limits on recursion.
This can be overridden by importing the sys module.
if_not_found : 'fail' | 'nan'
What to do when required date is not found:
* fail: Raise a ValueError
@ -185,17 +257,18 @@ class TimeSeries(TimeSeriesCore):
Example
--------
>>> calculate_returns(datetime.date(2020, 1, 1), years=1)
(datetime.datetime(2020, 1, 1, 0, 0), .0567)
"""
as_on = _parse_date(as_on, date_format)
as_on_delta, prior_delta = _preprocess_match_options(as_on_match, prior_match, closest)
prev_date = as_on - relativedelta(**{interval_type: interval_value})
current = _find_closest_date(self.data, as_on, as_on_delta, if_not_found)
previous = _find_closest_date(self.data, prev_date, prior_delta, if_not_found)
current = _find_closest_date(self.data, as_on, closest_max_days, as_on_delta, if_not_found)
previous = _find_closest_date(self.data, prev_date, closest_max_days, prior_delta, if_not_found)
if current[1] == str('nan') or previous[1] == str('nan'):
return as_on, float('NaN')
if current[1] == str("nan") or previous[1] == str("nan"):
return as_on, float("NaN")
returns = current[1] / previous[1]
if compounding:
@ -207,17 +280,81 @@ class TimeSeries(TimeSeriesCore):
self,
from_date: Union[datetime.date, str],
to_date: Union[datetime.date, str],
frequency: str = None,
frequency: Literal["D", "W", "M", "Q", "H", "Y"] = None,
as_on_match: str = "closest",
prior_match: str = "closest",
closest: str = "previous",
if_not_found: Literal['fail', 'nan'] = 'fail',
closest: Literal["previous", "next", "exact"] = "previous",
if_not_found: Literal["fail", "nan"] = "fail",
compounding: bool = True,
interval_type: Literal['years', 'months', 'days'] = 'years',
interval_type: Literal["years", "months", "days"] = "years",
interval_value: int = 1,
date_format: str = None
) -> List[tuple]:
"""Calculates the rolling return"""
date_format: str = None,
) -> TimeSeries:
"""Calculate the returns on a rolling basis.
This is a wrapper function around the calculate_returns function.
Parameters
----------
from_date : datetime.date | str
Start date for the return calculation.
to_date : datetime.date | str
End date for the returns calculation.
frequency : str, optional
Frequency at which the returns should be calcualated.
Valid values are {D, W, M, Q, H, Y}
as_on_match : str, optional
The match mode to be used for the as on date.
If not specified, the value for the closes parameter will be used.
prior_match : str, optional
The match mode to be used for the prior date, i.e., the date against which the return will be calculated.
If not specified, the value for the closes parameter will be used.
closest : previous | next | exact
The default match mode for dates.
* Previous: look for the immediate previous available date
* Next: look for the immediate next available date
* Exact: Only look for the exact date passed in the input
if_not_found : fail | nan
Specifies what should be done if the date is not found.
* fail: raise a DateNotFoundError.
* nan: return nan as the value.
Note, this will return float('NaN') and not 'nan' as string.
Note, this function will always raise an error if it is not possible to find a matching date.`
For instance, if the input date is before the starting of the first date of the time series,
but match mode is set to previous. A DateOutOfRangeError will be raised in such cases.
compounding : bool, optional
Should the returns be compounded annually.
interval_type : years | month | days
The interval for the return calculation.
interval_value : int, optional
The value of the interval for return calculation.
date_format : str, optional
A datetime library compatible format string.
If not specified, will use the setting in FincalOptions.date_format.
Returns
-------
Returs the rolling returns as a TimeSeries object.
Raises
------
ValueError
- If an invalid argument is passed for frequency parameter.
See also
--------
TimeSeries.calculate_returns
"""
from_date = _parse_date(from_date, date_format)
to_date = _parse_date(to_date, date_format)
@ -244,7 +381,7 @@ class TimeSeries(TimeSeriesCore):
as_on_match=as_on_match,
prior_match=prior_match,
closest=closest,
if_not_found=if_not_found
if_not_found=if_not_found,
)
rolling_returns.append(returns)
rolling_returns.sort()

13
fincal/utils.py

@ -2,7 +2,7 @@ import datetime
from dataclasses import dataclass
from typing import Iterable, List, Literal, Mapping, Sequence, Tuple, Union
from .exceptions import DateNotFoundError
from .exceptions import DateNotFoundError, DateOutOfRangeError
@dataclass
@ -85,15 +85,20 @@ def _preprocess_match_options(as_on_match: str, prior_match: str, closest: str)
return as_on_delta, prior_delta
def _find_closest_date(data, date, delta, if_not_found):
def _find_closest_date(data, date, limit_days, delta, if_not_found):
"""Helper function to find data for the closest available date"""
if delta.days < 0 and date < min(data):
raise DateOutOfRangeError(date, 'min')
if delta.days > 0 and date > max(data):
raise DateOutOfRangeError(date, 'max')
row = data.get(date, None)
if row is not None:
return date, row
if delta:
return _find_closest_date(data, date + delta, delta, if_not_found)
if delta and limit_days != 0:
return _find_closest_date(data, date + delta, limit_days-1, delta, if_not_found)
if if_not_found == "fail":
raise DateNotFoundError("Data not found for date", date)

Loading…
Cancel
Save