This commit is contained in:
Gourav Kumar 2022-03-21 20:48:06 +05:30
commit b34c14d778
2 changed files with 66 additions and 19 deletions

View File

@ -3,7 +3,7 @@ from __future__ import annotations
import datetime import datetime
import math import math
import statistics import statistics
from typing import Iterable, List, Literal, Mapping, Union from typing import Iterable, List, Literal, Mapping, TypedDict, Union
from dateutil.relativedelta import relativedelta from dateutil.relativedelta import relativedelta
@ -16,6 +16,12 @@ from .utils import (
) )
class MaxDrawdown(TypedDict):
start_date: datetime.datetime
end_date: datetime.datetime
drawdown: float
@date_parser(0, 1) @date_parser(0, 1)
def create_date_series( def create_date_series(
start_date: Union[str, datetime.datetime], start_date: Union[str, datetime.datetime],
@ -115,11 +121,11 @@ class TimeSeries(TimeSeriesCore):
super().__init__(data, frequency, date_format) super().__init__(data, frequency, date_format)
def info(self): def info(self) -> str:
"""Summary info about the TimeSeries object""" """Summary info about the TimeSeries object"""
total_dates = len(self.data.keys()) total_dates: int = len(self.data.keys())
res_string = "First date: {}\nLast date: {}\nNumber of rows: {}" res_string: str = "First date: {}\nLast date: {}\nNumber of rows: {}"
return res_string.format(self.start_date, self.end_date, total_dates) return res_string.format(self.start_date, self.end_date, total_dates)
def ffill(self, inplace: bool = False, limit: int = None) -> Union[TimeSeries, None]: def ffill(self, inplace: bool = False, limit: int = None) -> Union[TimeSeries, None]:
@ -138,7 +144,7 @@ class TimeSeries(TimeSeriesCore):
Returns a TimeSeries object if inplace is False, otherwise None Returns a TimeSeries object if inplace is False, otherwise None
""" """
eomonth = True if self.frequency.days >= AllFrequencies.M.days else False 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 = create_date_series(self.start_date, self.end_date, self.frequency.symbol, eomonth)
new_ts = dict() new_ts = dict()
@ -171,7 +177,7 @@ class TimeSeries(TimeSeriesCore):
Returns a TimeSeries object if inplace is False, otherwise None Returns a TimeSeries object if inplace is False, otherwise None
""" """
eomonth = True if self.frequency.days >= AllFrequencies.M.days else False 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 = create_date_series(self.start_date, self.end_date, self.frequency.symbol, eomonth)
dates_to_fill.append(self.end_date) dates_to_fill.append(self.end_date)
@ -517,21 +523,31 @@ class TimeSeries(TimeSeriesCore):
rr = self.calculate_rolling_returns(**kwargs) rr = self.calculate_rolling_returns(**kwargs)
return statistics.mean(rr.values) return statistics.mean(rr.values)
def max_drawdown(self): def max_drawdown(self) -> MaxDrawdown:
max_val_dict = {} """Calculates the maximum fall the stock has taken between any two points.
prev_val = 0 Returns
prev_date = list(self.data)[0] -------
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(): for dt, val in self.data.items():
if val > prev_val: if val > prev_val:
max_val_dict[dt] = (dt, val, 0) drawdowns[dt] = (dt, val, 0)
prev_date, prev_val = dt, val prev_date, prev_val = dt, val
else: else:
max_val_dict[dt] = (prev_date, prev_val, val / prev_val - 1) drawdowns[dt] = (prev_date, prev_val, val / prev_val - 1)
max_drawdown = min(max_val_dict.items(), key=lambda x: x[1][2]) max_drawdown = min(drawdowns.items(), key=lambda x: x[1][2])
max_drawdown = dict(start_date=max_drawdown[1][0], end_date=max_drawdown[0], drawdown=max_drawdown[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 return max_drawdown

View File

@ -1,12 +1,13 @@
import datetime import datetime
import math import math
import random import random
from unittest import skip
import pytest import pytest
from dateutil.relativedelta import relativedelta from dateutil.relativedelta import relativedelta
from fincal.core import AllFrequencies, Frequency from fincal.core import AllFrequencies, Frequency
from fincal.exceptions import DateNotFoundError from fincal.exceptions import DateNotFoundError
from fincal.fincal import TimeSeries, create_date_series from fincal.fincal import MaxDrawdown, TimeSeries, create_date_series
from fincal.utils import FincalOptions from fincal.utils import FincalOptions
@ -77,7 +78,9 @@ def create_test_timeseries(
start_date = datetime.datetime(2017, 1, 1) start_date = datetime.datetime(2017, 1, 1)
timedelta_dict = { timedelta_dict = {
frequency.freq_type: int(frequency.value * num * (7 / 5 if frequency == "D" and skip_weekends else 1)) frequency.freq_type: int(
frequency.value * num * (7 / 5 if frequency == AllFrequencies.D and skip_weekends else 1)
)
} }
end_date = start_date + relativedelta(**timedelta_dict) end_date = start_date + relativedelta(**timedelta_dict)
dates = create_date_series(start_date, end_date, frequency.symbol, skip_weekends=skip_weekends) dates = create_date_series(start_date, end_date, frequency.symbol, skip_weekends=skip_weekends)
@ -88,7 +91,7 @@ def create_test_timeseries(
class TestReturns: class TestReturns:
def test_returns_calc(self): def test_returns_calc(self):
ts = create_test_timeseries() ts = create_test_timeseries(AllFrequencies.D, skip_weekends=True)
returns = ts.calculate_returns( returns = ts.calculate_returns(
"2020-01-01", annual_compounded_returns=False, return_period_unit="years", return_period_value=1 "2020-01-01", annual_compounded_returns=False, return_period_unit="years", return_period_value=1
) )
@ -120,7 +123,7 @@ class TestReturns:
ts.calculate_returns("2020-04-04", return_period_unit="months", return_period_value=3, prior_match="exact") ts.calculate_returns("2020-04-04", return_period_unit="months", return_period_value=3, prior_match="exact")
def test_date_formats(self): def test_date_formats(self):
ts = create_test_timeseries() ts = create_test_timeseries(AllFrequencies.D, skip_weekends=True)
FincalOptions.date_format = "%d-%m-%Y" FincalOptions.date_format = "%d-%m-%Y"
with pytest.raises(ValueError): with pytest.raises(ValueError):
ts.calculate_returns( ts.calculate_returns(
@ -147,7 +150,7 @@ class TestReturns:
def test_limits(self): def test_limits(self):
FincalOptions.date_format = "%Y-%m-%d" FincalOptions.date_format = "%Y-%m-%d"
ts = create_test_timeseries() ts = create_test_timeseries(AllFrequencies.D)
with pytest.raises(DateNotFoundError): with pytest.raises(DateNotFoundError):
ts.calculate_returns("2020-11-25", return_period_unit="days", return_period_value=90, closest_max_days=10) ts.calculate_returns("2020-11-25", return_period_unit="days", return_period_value=90, closest_max_days=10)
@ -177,3 +180,31 @@ class TestVolatility:
annualize_volatility=False, annualize_volatility=False,
) )
assert round(sd, 6) == 0.020547 assert round(sd, 6) == 0.020547
class TestDrawdown:
def test_daily_ts(self):
ts = create_test_timeseries(AllFrequencies.D, skip_weekends=True)
mdd = ts.max_drawdown()
assert isinstance(mdd, dict)
assert len(mdd) == 3
assert all(i in mdd for i in ["start_date", "end_date", "drawdown"])
expeced_response = {
"start_date": datetime.datetime(2017, 6, 6, 0, 0),
"end_date": datetime.datetime(2017, 7, 31, 0, 0),
"drawdown": -0.028293686030751997,
}
assert mdd == expeced_response
def test_weekly_ts(self):
ts = create_test_timeseries(AllFrequencies.W, mu=1, sigma=0.5)
mdd = ts.max_drawdown()
assert isinstance(mdd, dict)
assert len(mdd) == 3
assert all(i in mdd for i in ["start_date", "end_date", "drawdown"])
expeced_response = {
"start_date": datetime.datetime(2019, 2, 17, 0, 0),
"end_date": datetime.datetime(2019, 11, 17, 0, 0),
"drawdown": -0.2584760499552089,
}
assert mdd == expeced_response