Compare commits

..

No commits in common. "793d5b1ad7bffbc975e8de458f9886a36067f500" and "17b3e348a241b9279b3b56ddef92e67df952d5f8" have entirely different histories.

3 changed files with 42 additions and 119 deletions

View File

@ -1,8 +1,6 @@
from __future__ import annotations from __future__ import annotations
import datetime import datetime
import math
import statistics
from typing import Iterable, List, Literal, Mapping, Union from typing import Iterable, List, Literal, Mapping, Union
from dateutil.relativedelta import relativedelta from dateutil.relativedelta import relativedelta
@ -192,7 +190,7 @@ class TimeSeries(TimeSeriesCore):
closest: Literal["previous", "next", "exact"] = "previous", closest: Literal["previous", "next", "exact"] = "previous",
closest_max_days: int = -1, closest_max_days: int = -1,
if_not_found: Literal["fail", "nan"] = "fail", if_not_found: Literal["fail", "nan"] = "fail",
annual_compounded_returns: bool = True, compounding: bool = True,
interval_type: Literal["years", "months", "days"] = "years", interval_type: Literal["years", "months", "days"] = "years",
interval_value: int = 1, interval_value: int = 1,
date_format: str = None, date_format: str = None,
@ -270,7 +268,7 @@ class TimeSeries(TimeSeriesCore):
return as_on, float("NaN") return as_on, float("NaN")
returns = current[1] / previous[1] returns = current[1] / previous[1]
if annual_compounded_returns: if compounding:
years = _interval_to_years(interval_type, interval_value) years = _interval_to_years(interval_type, interval_value)
returns = returns ** (1 / years) returns = returns ** (1 / years)
return (current[0] if return_actual_date else as_on), returns - 1 return (current[0] if return_actual_date else as_on), returns - 1
@ -285,7 +283,7 @@ class TimeSeries(TimeSeriesCore):
prior_match: str = "closest", prior_match: str = "closest",
closest: Literal["previous", "next", "exact"] = "previous", closest: Literal["previous", "next", "exact"] = "previous",
if_not_found: Literal["fail", "nan"] = "fail", if_not_found: Literal["fail", "nan"] = "fail",
annual_compounded_returns: bool = True, compounding: bool = True,
interval_type: Literal["years", "months", "days"] = "years", interval_type: Literal["years", "months", "days"] = "years",
interval_value: int = 1, interval_value: int = 1,
date_format: str = None, date_format: str = None,
@ -372,7 +370,7 @@ class TimeSeries(TimeSeriesCore):
for i in dates: for i in dates:
returns = self.calculate_returns( returns = self.calculate_returns(
as_on=i, as_on=i,
annual_compounded_returns=annual_compounded_returns, compounding=compounding,
interval_type=interval_type, interval_type=interval_type,
interval_value=interval_value, interval_value=interval_value,
as_on_match=as_on_match, as_on_match=as_on_match,
@ -384,60 +382,6 @@ class TimeSeries(TimeSeriesCore):
rolling_returns.sort() rolling_returns.sort()
return self.__class__(rolling_returns, self.frequency.symbol) 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__": if __name__ == "__main__":
date_series = [ date_series = [

View File

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

View File

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