Merge branch 'master' of http://192.168.0.114:3000/buddy/fincal

Added other type hints for variables

fincal/core.py

@@ -236,12 +236,12 @@ class TimeSeriesCore(UserDict):
         self.data = dict(data)
         if len(self.data) != len(data):
             print("Warning: The input data contains duplicate dates which have been ignored.")
-        self.frequency = getattr(AllFrequencies, frequency)
-        self.iter_num = -1
-        self._dates = None
-        self._values = None
-        self._start_date = None
-        self._end_date = None
+        self.frequency: Frequency = getattr(AllFrequencies, frequency)
+        self.iter_num: int = -1
+        self._dates: list = None
+        self._values: list = None
+        self._start_date: datetime.datetime = None
+        self._end_date: datetime.datetime = None
 
     @property
     def dates(self) -> Series:

fincal/fincal.py

@@ -3,11 +3,11 @@ from __future__ import annotations
 import datetime
 import math
 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 .core import AllFrequencies, Series, TimeSeriesCore, date_parser
+from .core import AllFrequencies, Frequency, Series, TimeSeriesCore, date_parser
 from .utils import (
     FincalOptions,
     _find_closest_date,
@@ -16,6 +16,12 @@ from .utils import (
 )
 
 
+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],
@@ -115,11 +121,11 @@ class TimeSeries(TimeSeriesCore):
 
         super().__init__(data, frequency, date_format)
 
-    def info(self):
+    def info(self) -> str:
         """Summary info about the TimeSeries object"""
 
-        total_dates = len(self.data.keys())
-        res_string = "First date: {}\nLast date: {}\nNumber of rows: {}"
+        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]:
@@ -138,7 +144,7 @@ class TimeSeries(TimeSeriesCore):
             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)
 
         new_ts = dict()
@@ -171,7 +177,7 @@ class TimeSeries(TimeSeriesCore):
             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.append(self.end_date)
 
@@ -517,24 +523,58 @@ class TimeSeries(TimeSeriesCore):
         rr = self.calculate_rolling_returns(**kwargs)
         return statistics.mean(rr.values)
 
-    def max_drawdown(self):
-        max_val_dict = {}
+    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 = 0
-        prev_date = list(self.data)[0]
+        prev_val: float = 0
+        prev_date: datetime.datetime = list(self.data)[0]
 
         for dt, val in self.data.items():
             if val > prev_val:
-                max_val_dict[dt] = (dt, val, 0)
+                drawdowns[dt] = (dt, val, 0)
                 prev_date, prev_val = dt, val
             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 = dict(start_date=max_drawdown[1][0], end_date=max_drawdown[0], drawdown=max_drawdown[1][2])
+        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 = [

tests/test_fincal2.py

@@ -1,12 +1,13 @@
 import datetime
 import math
 import random
+from unittest import skip
 
 import pytest
 from dateutil.relativedelta import relativedelta
 from fincal.core import AllFrequencies, Frequency
 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
 
 
@@ -77,7 +78,9 @@ def create_test_timeseries(
 
     start_date = datetime.datetime(2017, 1, 1)
     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)
     dates = create_date_series(start_date, end_date, frequency.symbol, skip_weekends=skip_weekends)
@@ -88,7 +91,7 @@ def create_test_timeseries(
 
 class TestReturns:
     def test_returns_calc(self):
-        ts = create_test_timeseries()
+        ts = create_test_timeseries(AllFrequencies.D, skip_weekends=True)
         returns = ts.calculate_returns(
             "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")
 
     def test_date_formats(self):
-        ts = create_test_timeseries()
+        ts = create_test_timeseries(AllFrequencies.D, skip_weekends=True)
         FincalOptions.date_format = "%d-%m-%Y"
         with pytest.raises(ValueError):
             ts.calculate_returns(
@@ -147,7 +150,7 @@ class TestReturns:
 
     def test_limits(self):
         FincalOptions.date_format = "%Y-%m-%d"
-        ts = create_test_timeseries()
+        ts = create_test_timeseries(AllFrequencies.D)
         with pytest.raises(DateNotFoundError):
             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,
         )
         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