Expanded the volatility function, added annualisation

fincal/fincal.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import datetime
+import math
+import statistics
 from typing import Iterable, List, Literal, Mapping, Union
 
 from dateutil.relativedelta import relativedelta
@@ -190,7 +192,7 @@ class TimeSeries(TimeSeriesCore):
         closest: Literal["previous", "next", "exact"] = "previous",
         closest_max_days: int = -1,
         if_not_found: Literal["fail", "nan"] = "fail",
-        compounding: bool = True,
+        annual_compounded_returns: bool = True,
         interval_type: Literal["years", "months", "days"] = "years",
         interval_value: int = 1,
         date_format: str = None,
@@ -268,7 +270,7 @@ class TimeSeries(TimeSeriesCore):
             return as_on, float("NaN")
 
         returns = current[1] / previous[1]
-        if compounding:
+        if annual_compounded_returns:
             years = _interval_to_years(interval_type, interval_value)
             returns = returns ** (1 / years)
         return (current[0] if return_actual_date else as_on), returns - 1
@@ -283,7 +285,7 @@ class TimeSeries(TimeSeriesCore):
         prior_match: str = "closest",
         closest: Literal["previous", "next", "exact"] = "previous",
         if_not_found: Literal["fail", "nan"] = "fail",
-        compounding: bool = True,
+        annual_compounded_returns: bool = True,
         interval_type: Literal["years", "months", "days"] = "years",
         interval_value: int = 1,
         date_format: str = None,
@@ -370,7 +372,7 @@ class TimeSeries(TimeSeriesCore):
         for i in dates:
             returns = self.calculate_returns(
                 as_on=i,
-                compounding=compounding,
+                annual_compounded_returns=annual_compounded_returns,
                 interval_type=interval_type,
                 interval_value=interval_value,
                 as_on_match=as_on_match,
@@ -382,6 +384,60 @@ class TimeSeries(TimeSeriesCore):
         rolling_returns.sort()
         return self.__class__(rolling_returns, self.frequency.symbol)
 
+    @date_parser(1, 2)
+    def volatility(
+        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 = None,
+        interval_type: Literal["years", "months", "days"] = "days",
+        interval_value: int = 1,
+        date_format: str = None,
+        annualize_volatility: bool = True,
+    ):
+        """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.
+        """
+
+        if frequency is None:
+            frequency = self.frequency
+        else:
+            try:
+                frequency = getattr(AllFrequencies, frequency)
+            except AttributeError:
+                raise ValueError(f"Invalid argument for frequency {frequency}")
+
+        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,
+            interval_type=interval_type,
+            interval_value=interval_value,
+        )
+        sd = statistics.stdev(rolling_returns.values)
+        if annualize_volatility:
+            if interval_type == "months":
+                sd *= math.sqrt(12)
+            elif interval_type == "days":
+                sd *= math.sqrt(252)
+
+        return sd
+
 
 if __name__ == "__main__":
     date_series = [

test2.py

@@ -1,37 +1,58 @@
-# type: ignore
+import pandas as pd
-
+
-if __name__ == "__main__":
+from fincal.fincal import TimeSeries, create_date_series
-
+
-    import datetime
+dfd = pd.read_csv("test_files/nav_history_daily - Copy.csv")
-    import time
+dfd = dfd[dfd["amfi_code"] == 118825].reset_index(drop=True)
-
+ts = TimeSeries([(i.date, i.nav) for i in dfd.itertuples()], frequency="D")
-    import pandas as pd
+repr(ts)
-
+# print(ts[['2022-01-31', '2021-05-28']])
-    from fincal.fincal import TimeSeries
+
-
+# rr = ts.calculate_rolling_returns(from_date='2021-01-01', to_date='2022-01-01', frequency='D', interval_type='days', interval_value=30, compounding=False)
-    df = pd.read_csv('test_files/msft.csv')
+
-    df = df.sort_values(by='Date')  # type: ignore
+
-    data_list = [(i.Date, i.Close) for i in df.itertuples()]
+# data = [
-
+#     ("2020-01-01", 10),
-    start = time.time()
+#     ("2020-02-01", 12),
-    ts_data = TimeSeries(data_list, frequency='D', date_format='%d-%m-%Y')
+#     ("2020-03-01", 14),
-    print(f"Instantiation took {round((time.time() - start)*1000, 2)} ms")
+#     ("2020-04-01", 16),
-    # ts_data.fill_missing_days()
+#     ("2020-05-01", 18),
-    start = time.time()
+#     ("2020-06-01", 20),
-    # ts_data.calculate_returns(as_on=datetime.datetime(2022, 1, 4), closest='next', years=1)
+#     ("2020-07-01", 22),
-    rr = ts_data.calculate_rolling_returns(datetime.datetime(1994, 1, 1),
+#     ("2020-08-01", 24),
-                                        datetime.datetime(2022, 2, 17),
+#     ("2020-09-01", 26),
-                                        frequency='D',
+#     ("2020-10-01", 28),
-                                        as_on_match='next',
+#     ("2020-11-01", 30),
-                                        prior_match='previous',
+#     ("2020-12-01", 32),
-                                        closest='previous',
+#     ("2021-01-01", 34),
-                                        years=1)
+# ]
-
+
-    # ffill_data = ts_data.bfill()
+# ts = TimeSeries(data, frequency="M")
-    print(f"Calculation took {round((time.time() - start)*1000, 2)} ms")
+# rr = ts.calculate_rolling_returns(
-    rr.sort()
+#     "2020-02-01",
-    for i in rr[:10]:
+#     "2021-01-01",
-        print(i)
+#     if_not_found="nan",
-    # print(ffill_data)
+#     compounding=False,
-    # print(ts_data)
+#     interval_type="months",
-    # print(repr(ts_data))
+#     interval_value=1,
+#     as_on_match="exact",
+# )
+
+# for i in rr:
+#     print(i)
+
+# returns = ts.calculate_returns(
+#     "2020-04-25",
+#     return_actual_date=True,
+#     closest_max_days=15,
+#     compounding=True,
+#     interval_type="days",
+#     interval_value=90,
+#     closest="previous",
+#     if_not_found="fail",
+# )
+
+# print(returns)
+
+volatility = ts.volatility(start_date="2018-01-01", end_date="2021-01-01")
+print(volatility)

tests/test_fincal.py

@@ -199,17 +199,17 @@ class TestReturns:
 
     def test_returns_calc(self):
         ts = TimeSeries(self.data, frequency="M")
-        returns = ts.calculate_returns("2021-01-01", compounding=False, interval_type="years", interval_value=1)
+        returns = ts.calculate_returns("2021-01-01", annual_compounded_returns=False, interval_type="years", interval_value=1)
         assert returns[1] == 2.4
-        returns = ts.calculate_returns("2020-04-01", compounding=False, interval_type="months", interval_value=3)
+        returns = ts.calculate_returns("2020-04-01", annual_compounded_returns=False, interval_type="months", interval_value=3)
         assert round(returns[1], 4) == 0.6
-        returns = ts.calculate_returns("2020-04-01", compounding=True, interval_type="months", interval_value=3)
+        returns = ts.calculate_returns("2020-04-01", annual_compounded_returns=True, interval_type="months", interval_value=3)
         assert round(returns[1], 4) == 5.5536
-        returns = ts.calculate_returns("2020-04-01", compounding=False, interval_type="days", interval_value=90)
+        returns = ts.calculate_returns("2020-04-01", annual_compounded_returns=False, interval_type="days", interval_value=90)
         assert round(returns[1], 4) == 0.6
-        returns = ts.calculate_returns("2020-04-01", compounding=True, interval_type="days", interval_value=90)
+        returns = ts.calculate_returns("2020-04-01", annual_compounded_returns=True, interval_type="days", interval_value=90)
         assert round(returns[1], 4) == 5.727
-        returns = ts.calculate_returns("2020-04-10", compounding=True, interval_type="days", interval_value=90)
+        returns = ts.calculate_returns("2020-04-10", annual_compounded_returns=True, interval_type="days", interval_value=90)
         assert round(returns[1], 4) == 5.727
         with pytest.raises(DateNotFoundError):
             ts.calculate_returns("2020-04-10", interval_type="days", interval_value=90, as_on_match="exact")
@@ -220,7 +220,7 @@ class TestReturns:
         ts = TimeSeries(self.data, frequency="M")
         FincalOptions.date_format = "%d-%m-%Y"
         with pytest.raises(ValueError):
-            ts.calculate_returns("2020-04-10", compounding=True, interval_type="days", interval_value=90)
+            ts.calculate_returns("2020-04-10", annual_compounded_returns=True, interval_type="days", interval_value=90)
 
         returns1 = ts.calculate_returns("2020-04-10", interval_type="days", interval_value=90, date_format="%Y-%m-%d")
         returns2 = ts.calculate_returns("10-04-2020", interval_type="days", interval_value=90)
@@ -228,7 +228,7 @@ class TestReturns:
 
         FincalOptions.date_format = "%m-%d-%Y"
         with pytest.raises(ValueError):
-            ts.calculate_returns("2020-04-10", compounding=True, interval_type="days", interval_value=90)
+            ts.calculate_returns("2020-04-10", annual_compounded_returns=True, interval_type="days", interval_value=90)
 
         returns1 = ts.calculate_returns("2020-04-10", interval_type="days", interval_value=90, date_format="%Y-%m-%d")
         returns2 = ts.calculate_returns("04-10-2020", interval_type="days", interval_value=90)