from __future__ import annotations import datetime import math import statistics from typing import Iterable, List, Literal, Mapping, TypedDict, Union from dateutil.relativedelta import relativedelta from .core import AllFrequencies, Frequency, Series, TimeSeriesCore, date_parser from .utils import ( FincalOptions, _find_closest_date, _interval_to_years, _preprocess_match_options, ) class MaxDrawdown(TypedDict): start_date: datetime.datetime end_date: datetime.datetime drawdown: float @date_parser(0, 1) def create_date_series( start_date: Union[str, datetime.datetime], end_date: Union[str, datetime.datetime], frequency: Literal["D", "W", "M", "Q", "H", "Y"], eomonth: bool = False, skip_weekends: bool = False, ) -> List[datetime.datetime]: """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: raise ValueError(f"eomonth cannot be set to True if frequency is higher than {AllFrequencies.M.name}") datediff = (end_date - start_date).days / frequency.days + 1 dates = [] for i in range(0, int(datediff)): diff = {frequency.freq_type: frequency.value * i} date = start_date + relativedelta(**diff) if eomonth: 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: if frequency.days > 1 or not skip_weekends: dates.append(date) elif date.weekday() < 5: dates.append(date) return Series(dates, data_type="date") class TimeSeries(TimeSeriesCore): """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) -> str: """Summary info about the TimeSeries object""" total_dates: int = len(self.data.keys()) res_string: str = "First date: {}\nLast date: {}\nNumber of rows: {}" return res_string.format(self.start_date, self.end_date, total_dates) def ffill(self, inplace: bool = False, limit: int = None) -> Union[TimeSeries, None]: """Forward fill missing dates in the time series Parameters ---------- inplace : bool Modify the time-series data in place and return None. limit : int, optional Maximum number of periods to forward fill Returns ------- Returns a TimeSeries object if inplace is False, otherwise None """ eomonth: bool = True if self.frequency.days >= AllFrequencies.M.days else False dates_to_fill = create_date_series(self.start_date, self.end_date, self.frequency.symbol, eomonth) new_ts = dict() for cur_date in dates_to_fill: try: cur_val = self.data[cur_date] except KeyError: pass new_ts.update({cur_date: cur_val}) if inplace: self.data = new_ts return None return self.__class__(new_ts, frequency=self.frequency.symbol) def bfill(self, inplace: bool = False, limit: int = None) -> Union[TimeSeries, None]: """Backward fill missing dates in the time series Parameters ---------- inplace : bool Modify the time-series data in place and return None. limit : int, optional Maximum number of periods to back fill Returns ------- Returns a TimeSeries object if inplace is False, otherwise None """ eomonth: bool = True if self.frequency.days >= AllFrequencies.M.days else False dates_to_fill = create_date_series(self.start_date, self.end_date, self.frequency.symbol, eomonth) dates_to_fill.append(self.end_date) bfill_ts = dict() for cur_date in reversed(dates_to_fill): try: cur_val = self.data[cur_date] except KeyError: pass bfill_ts.update({cur_date: cur_val}) new_ts = {k: bfill_ts[k] for k in reversed(bfill_ts)} if inplace: self.data = new_ts return None return self.__class__(new_ts, frequency=self.frequency.symbol) @date_parser(1) def calculate_returns( self, as_on: Union[str, datetime.datetime], return_actual_date: bool = True, as_on_match: str = "closest", prior_match: str = "closest", closest: Literal["previous", "next", "exact"] = "previous", closest_max_days: int = -1, if_not_found: Literal["fail", "nan"] = "fail", annual_compounded_returns: bool = True, return_period_unit: Literal["years", "months", "days"] = "years", return_period_value: int = 1, date_format: str = None, ) -> float: """Method to calculate returns for a certain time-period as on a particular date Parameters ---------- as_on : datetime.datetime The date as on which the return is to be calculated. return_actual_date : bool, default True If true, the output will contain the actual date based on which the return was calculated. Set to False to return the date passed in the as_on argument. as_on_match : str, optional The mode of matching the as_on_date. Refer closest. prior_match : str, optional The mode of matching the prior_date. Refer closest. closest : str, optional 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 * nan: Return nan as the value compounding : bool, optional Whether the return should be compounded annually. return_period_unit : 'years', 'months', 'days' The type of time period to use for return calculation. return_period_value : int The value of the specified interval type over which returns needs to be calculated. date_format: str The date format to use for this operation. Should be passed as a datetime library compatible string. Sets the date format only for this operation. To set it globally, use FincalOptions.date_format Returns ------- A tuple containing the date and float value of the returns. Raises ------ ValueError * If match mode for any of the dates is exact and the exact match is not found * If the arguments passsed for closest, as_on_match, and prior_match are invalid Example -------- >>> calculate_returns(datetime.date(2020, 1, 1), years=1) (datetime.datetime(2020, 1, 1, 0, 0), .0567) """ as_on_delta, prior_delta = _preprocess_match_options(as_on_match, prior_match, closest) prev_date = as_on - relativedelta(**{return_period_unit: return_period_value}) current = _find_closest_date(self.data, as_on, closest_max_days, as_on_delta, if_not_found) if current[1] != str("nan"): 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") returns = current[1] / previous[1] if annual_compounded_returns: years = _interval_to_years(return_period_unit, return_period_value) returns = returns ** (1 / years) return (current[0] if return_actual_date else as_on), returns - 1 @date_parser(1, 2) def calculate_rolling_returns( self, from_date: Union[datetime.date, str], to_date: Union[datetime.date, str], frequency: Literal["D", "W", "M", "Q", "H", "Y"] = None, as_on_match: str = "closest", prior_match: str = "closest", closest: Literal["previous", "next", "exact"] = "previous", if_not_found: Literal["fail", "nan"] = "fail", annual_compounded_returns: bool = True, return_period_unit: Literal["years", "months", "days"] = "years", return_period_value: int = 1, 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. return_period_unit : years | month | days The interval for the return calculation. return_period_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 """ if frequency is None: frequency = self.frequency else: try: frequency = getattr(AllFrequencies, frequency) except AttributeError: raise ValueError(f"Invalid argument for frequency {frequency}") dates = create_date_series(from_date, to_date, frequency.symbol) if frequency == AllFrequencies.D: dates = [i for i in dates if i in self.data] rolling_returns = [] for i in dates: returns = self.calculate_returns( as_on=i, annual_compounded_returns=annual_compounded_returns, return_period_unit=return_period_unit, return_period_value=return_period_value, as_on_match=as_on_match, prior_match=prior_match, closest=closest, if_not_found=if_not_found, ) rolling_returns.append(returns) rolling_returns.sort() return self.__class__(rolling_returns, self.frequency.symbol) @date_parser(1, 2) def volatility( self, from_date: Union[datetime.date, str] = None, to_date: Union[datetime.date, str] = None, annualize_volatility: bool = True, traded_days: int = None, frequency: Literal["D", "W", "M", "Q", "H", "Y"] = None, return_period_unit: Literal["years", "months", "days"] = "days", return_period_value: int = 1, as_on_match: str = "closest", prior_match: str = "closest", closest: Literal["previous", "next", "exact"] = "previous", if_not_found: Literal["fail", "nan"] = "fail", annual_compounded_returns: bool = None, date_format: str = None, ) -> float: """Calculates the volatility of the time series.add() The volatility is calculated as the standard deviaion of periodic returns. The periodicity of returns is based on the periodicity of underlying data. Parameters: ---------- from_date: datetime.datetime | str, optional Starting date for the volatility calculation. Default is the first date on which volatility can be calculated based on the interval type. to_date: datetime.datetime | str, optional Ending date for the volatility calculation. Default is the last date in the TimeSeries. annualize_volatility: bool, default True Whether the volatility number should be annualized. Multiplies the standard deviation with the square root of the number of periods in a year traded_days: bool, optional Number of traded days per year to be considered for annualizing volatility. Only used when annualizing volatility for a time series with daily frequency. If not provided, will use the value in FincalOptions.traded_days. Remaining options are passed on to rolling_return function. Returns: ------- Returns the volatility number as float Raises: ------- ValueError: If frequency string is outside valid values Also see: -------- TimeSeries.calculate_rolling_returns() """ if frequency is None: frequency = self.frequency else: try: frequency = getattr(AllFrequencies, frequency) except AttributeError: raise ValueError(f"Invalid argument for frequency {frequency}") if from_date is None: from_date = self.start_date + relativedelta(**{return_period_unit: return_period_value}) if to_date is None: to_date = self.end_date if annual_compounded_returns is None: annual_compounded_returns = False if frequency.days <= 366 else True rolling_returns = self.calculate_rolling_returns( from_date=from_date, to_date=to_date, frequency=frequency.symbol, as_on_match=as_on_match, prior_match=prior_match, closest=closest, if_not_found=if_not_found, annual_compounded_returns=annual_compounded_returns, return_period_unit=return_period_unit, return_period_value=return_period_value, ) sd = statistics.stdev(rolling_returns.values) if annualize_volatility: if traded_days is None: traded_days = FincalOptions.traded_days if return_period_unit == "months": sd *= math.sqrt(12) elif return_period_unit == "days": sd *= math.sqrt(traded_days) return sd def average_rolling_return(self, **kwargs) -> float: """Calculates the average rolling return for a given period Parameters ---------- kwargs: parameters to be passed to the calculate_rolling_returns() function Returns ------- float returns the average rolling return for a given period Also see: --------- TimeSeries.calculate_rolling_returns() """ kwargs["return_period_unit"] = kwargs.get("return_period_unit", self.frequency.freq_type) kwargs["return_period_value"] = kwargs.get("return_period_value", 1) kwargs["to_date"] = kwargs.get("to_date", self.end_date) if kwargs.get("from_date", None) is None: start_date = self.start_date + relativedelta( **{kwargs["return_period_unit"]: kwargs["return_period_value"]} ) kwargs["from_date"] = start_date rr = self.calculate_rolling_returns(**kwargs) return statistics.mean(rr.values) def max_drawdown(self) -> MaxDrawdown: """Calculates the maximum fall the stock has taken between any two points. Returns ------- MaxDrawdown Returns the start_date, end_date, and the drawdown value in decimal. """ drawdowns: dict = dict() prev_val: float = 0 prev_date: datetime.datetime = list(self.data)[0] for dt, val in self.data.items(): if val > prev_val: drawdowns[dt] = (dt, val, 0) prev_date, prev_val = dt, val else: drawdowns[dt] = (prev_date, prev_val, val / prev_val - 1) max_drawdown = min(drawdowns.items(), key=lambda x: x[1][2]) max_drawdown: MaxDrawdown = dict( start_date=max_drawdown[1][0], end_date=max_drawdown[0], drawdown=max_drawdown[1][2] ) return max_drawdown def expand( self, to_frequency: Literal["D", "W", "M", "Q", "H"], method: Literal["ffill", "bfill", "interpolate"] ) -> TimeSeries: try: to_frequency: Frequency = getattr(AllFrequencies, to_frequency) except AttributeError: raise ValueError(f"Invalid argument for to_frequency {to_frequency}") 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} output_ts: TimeSeries = TimeSeries(new_ts, frequency=to_frequency.symbol) if method == "ffill": output_ts.ffill(inplace=True) elif method == "bfill": output_ts.bfill(inplace=True) else: raise NotImplementedError(f"Method {method} not implemented") return output_ts if __name__ == "__main__": date_series = [ datetime.datetime(2020, 1, 11), datetime.datetime(2020, 1, 12), datetime.datetime(2020, 1, 13), datetime.datetime(2020, 1, 14), datetime.datetime(2020, 1, 17), datetime.datetime(2020, 1, 18), datetime.datetime(2020, 1, 19), datetime.datetime(2020, 1, 20), datetime.datetime(2020, 1, 22), ]