Compare commits

..

No commits in common. "c992905bf69f8cf89aed14b0c3e7b35cffad4929" and "d7b06fbe248675f28d15f68ab36148e03c9137a7" have entirely different histories.

3 changed files with 65 additions and 167 deletions

View File

@ -200,8 +200,8 @@ class TimeSeries(TimeSeriesCore):
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, annual_compounded_returns: bool = True,
return_period_unit: Literal["years", "months", "days"] = "years", interval_type: Literal["years", "months", "days"] = "years",
return_period_value: int = 1, interval_value: int = 1,
date_format: str = None, date_format: str = None,
) -> float: ) -> float:
"""Method to calculate returns for a certain time-period as on a particular date """Method to calculate returns for a certain time-period as on a particular date
@ -239,10 +239,10 @@ class TimeSeries(TimeSeriesCore):
compounding : bool, optional compounding : bool, optional
Whether the return should be compounded annually. Whether the return should be compounded annually.
return_period_unit : 'years', 'months', 'days' interval_type : 'years', 'months', 'days'
The type of time period to use for return calculation. The type of time period to use for return calculation.
return_period_value : int interval_value : int
The value of the specified interval type over which returns needs to be calculated. The value of the specified interval type over which returns needs to be calculated.
date_format: str date_format: str
@ -268,7 +268,7 @@ class TimeSeries(TimeSeriesCore):
as_on_delta, prior_delta = _preprocess_match_options(as_on_match, prior_match, closest) 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}) prev_date = as_on - relativedelta(**{interval_type: interval_value})
current = _find_closest_date(self.data, as_on, closest_max_days, as_on_delta, if_not_found) current = _find_closest_date(self.data, as_on, closest_max_days, as_on_delta, if_not_found)
if current[1] != str("nan"): if current[1] != str("nan"):
previous = _find_closest_date(self.data, prev_date, closest_max_days, prior_delta, if_not_found) previous = _find_closest_date(self.data, prev_date, closest_max_days, prior_delta, if_not_found)
@ -278,7 +278,7 @@ class TimeSeries(TimeSeriesCore):
returns = current[1] / previous[1] returns = current[1] / previous[1]
if annual_compounded_returns: if annual_compounded_returns:
years = _interval_to_years(return_period_unit, return_period_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
@ -293,8 +293,8 @@ class TimeSeries(TimeSeriesCore):
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, annual_compounded_returns: bool = True,
return_period_unit: Literal["years", "months", "days"] = "years", interval_type: Literal["years", "months", "days"] = "years",
return_period_value: int = 1, interval_value: int = 1,
date_format: str = None, date_format: str = None,
) -> TimeSeries: ) -> TimeSeries:
"""Calculate the returns on a rolling basis. """Calculate the returns on a rolling basis.
@ -339,10 +339,10 @@ class TimeSeries(TimeSeriesCore):
compounding : bool, optional compounding : bool, optional
Should the returns be compounded annually. Should the returns be compounded annually.
return_period_unit : years | month | days interval_type : years | month | days
The interval for the return calculation. The interval for the return calculation.
return_period_value : int, optional interval_value : int, optional
The value of the interval for return calculation. The value of the interval for return calculation.
date_format : str, optional date_format : str, optional
@ -380,8 +380,8 @@ class TimeSeries(TimeSeriesCore):
returns = self.calculate_returns( returns = self.calculate_returns(
as_on=i, as_on=i,
annual_compounded_returns=annual_compounded_returns, annual_compounded_returns=annual_compounded_returns,
return_period_unit=return_period_unit, interval_type=interval_type,
return_period_value=return_period_value, interval_value=interval_value,
as_on_match=as_on_match, as_on_match=as_on_match,
prior_match=prior_match, prior_match=prior_match,
closest=closest, closest=closest,
@ -396,41 +396,22 @@ class TimeSeries(TimeSeriesCore):
self, self,
from_date: Union[datetime.date, str] = None, from_date: Union[datetime.date, str] = None,
to_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, 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", as_on_match: str = "closest",
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 = None, annual_compounded_returns: bool = None,
interval_type: Literal["years", "months", "days"] = "days",
interval_value: int = 1,
date_format: str = None, date_format: str = None,
annualize_volatility: bool = True,
traded_days: int = None,
): ):
"""Calculates the volatility of the time series.add() """Calculates the volatility of the time series.add()
The volatility is calculated as the standard deviaion of periodic returns. The volatility is calculated as the standard deviaion of periodic returns.
The periodicity of returns is based on the periodicity of underlying data. 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.
""" """
if frequency is None: if frequency is None:
@ -442,7 +423,7 @@ class TimeSeries(TimeSeriesCore):
raise ValueError(f"Invalid argument for frequency {frequency}") raise ValueError(f"Invalid argument for frequency {frequency}")
if from_date is None: if from_date is None:
from_date = self.start_date + relativedelta(**{return_period_unit: return_period_value}) from_date = self.start_date + relativedelta(**{interval_type: interval_value})
if to_date is None: if to_date is None:
to_date = self.end_date to_date = self.end_date
@ -458,17 +439,17 @@ class TimeSeries(TimeSeriesCore):
closest=closest, closest=closest,
if_not_found=if_not_found, if_not_found=if_not_found,
annual_compounded_returns=annual_compounded_returns, annual_compounded_returns=annual_compounded_returns,
return_period_unit=return_period_unit, interval_type=interval_type,
return_period_value=return_period_value, interval_value=interval_value,
) )
sd = statistics.stdev(rolling_returns.values) sd = statistics.stdev(rolling_returns.values)
if annualize_volatility: if annualize_volatility:
if traded_days is None: if traded_days is None:
traded_days = FincalOptions.traded_days traded_days = FincalOptions.traded_days
if return_period_unit == "months": if interval_type == "months":
sd *= math.sqrt(12) sd *= math.sqrt(12)
elif return_period_unit == "days": elif interval_type == "days":
sd *= math.sqrt(traded_days) sd *= math.sqrt(traded_days)
return sd return sd

View File

@ -200,7 +200,7 @@ class TestFincalBasic:
assert time_series.iloc[:3] is not None assert time_series.iloc[:3] is not None
assert time_series.iloc[5:7] is not None assert time_series.iloc[5:7] is not None
assert isinstance(time_series.iloc[0], tuple) assert isinstance(time_series.iloc[0], tuple)
assert isinstance(time_series.iloc[10:20], TimeSeries) assert isinstance(time_series.iloc[10:20], list)
assert len(time_series.iloc[10:20]) == 10 assert len(time_series.iloc[10:20]) == 10
def test_key_slicing(self): def test_key_slicing(self):
@ -236,65 +236,57 @@ 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( returns = ts.calculate_returns(
"2021-01-01", annual_compounded_returns=False, return_period_unit="years", return_period_value=1 "2021-01-01", annual_compounded_returns=False, interval_type="years", interval_value=1
) )
assert returns[1] == 2.4 assert returns[1] == 2.4
returns = ts.calculate_returns( returns = ts.calculate_returns(
"2020-04-01", annual_compounded_returns=False, return_period_unit="months", return_period_value=3 "2020-04-01", annual_compounded_returns=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( returns = ts.calculate_returns(
"2020-04-01", annual_compounded_returns=True, return_period_unit="months", return_period_value=3 "2020-04-01", annual_compounded_returns=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( returns = ts.calculate_returns(
"2020-04-01", annual_compounded_returns=False, return_period_unit="days", return_period_value=90 "2020-04-01", annual_compounded_returns=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( returns = ts.calculate_returns(
"2020-04-01", annual_compounded_returns=True, return_period_unit="days", return_period_value=90 "2020-04-01", annual_compounded_returns=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( returns = ts.calculate_returns(
"2020-04-10", annual_compounded_returns=True, return_period_unit="days", return_period_value=90 "2020-04-10", annual_compounded_returns=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", return_period_unit="days", return_period_value=90, as_on_match="exact") ts.calculate_returns("2020-04-10", interval_type="days", interval_value=90, as_on_match="exact")
with pytest.raises(DateNotFoundError): with pytest.raises(DateNotFoundError):
ts.calculate_returns("2020-04-10", return_period_unit="days", return_period_value=90, prior_match="exact") ts.calculate_returns("2020-04-10", interval_type="days", interval_value=90, prior_match="exact")
def test_date_formats(self): def test_date_formats(self):
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( ts.calculate_returns("2020-04-10", annual_compounded_returns=True, interval_type="days", interval_value=90)
"2020-04-10", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
)
returns1 = ts.calculate_returns( returns1 = ts.calculate_returns("2020-04-10", interval_type="days", interval_value=90, date_format="%Y-%m-%d")
"2020-04-10", return_period_unit="days", return_period_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", return_period_unit="days", return_period_value=90)
assert round(returns1[1], 4) == round(returns2[1], 4) == 5.727 assert round(returns1[1], 4) == round(returns2[1], 4) == 5.727
FincalOptions.date_format = "%m-%d-%Y" FincalOptions.date_format = "%m-%d-%Y"
with pytest.raises(ValueError): with pytest.raises(ValueError):
ts.calculate_returns( ts.calculate_returns("2020-04-10", annual_compounded_returns=True, interval_type="days", interval_value=90)
"2020-04-10", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
)
returns1 = ts.calculate_returns( returns1 = ts.calculate_returns("2020-04-10", interval_type="days", interval_value=90, date_format="%Y-%m-%d")
"2020-04-10", return_period_unit="days", return_period_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", return_period_unit="days", return_period_value=90)
assert round(returns1[1], 4) == round(returns2[1], 4) == 5.727 assert round(returns1[1], 4) == round(returns2[1], 4) == 5.727
def test_limits(self): def test_limits(self):
ts = TimeSeries(self.data, frequency="M") ts = TimeSeries(self.data, frequency="M")
FincalOptions.date_format = "%Y-%m-%d" FincalOptions.date_format = "%Y-%m-%d"
with pytest.raises(DateNotFoundError): with pytest.raises(DateNotFoundError):
ts.calculate_returns("2020-04-25", return_period_unit="days", return_period_value=90, closest_max_days=10) ts.calculate_returns("2020-04-25", interval_type="days", interval_value=90, closest_max_days=10)
class TestVolatility: class TestVolatility:

View File

@ -3,37 +3,18 @@ import math
import random import random
import pytest import pytest
from dateutil.relativedelta import relativedelta
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 TimeSeries, create_date_series
from fincal.utils import FincalOptions from fincal.utils import FincalOptions
def create_prices(s0: float, mu: float, sigma: float, num_prices: int) -> list: def create_prices(s0: float, mu: float, sigma: float, num_prices: int) -> list:
"""Generates a price following a geometric brownian motion process based on the input of the arguments. """Generates a price following a geometric brownian motion process based on the input of the arguments:
- s0: Asset inital price.
Since this function is used only to generate data for tests, the seed is fixed as 1234. - mu: Interest rate expressed annual terms.
Many of the tests rely on exact values generated using this seed. - sigma: Volatility expressed annual terms.
If the seed is changed, those tests will fail. - seed: seed for the random number generator
- num_prices: number of prices to generate
Parameters:
------------
s0: float
Asset inital price.
mu: float
Interest rate expressed annual terms.
sigma: float
Volatility expressed annual terms.
num_prices: int
number of prices to generate
Returns:
--------
Returns a list of values generated using GBM algorithm
""" """
random.seed(1234) # WARNING! Changing the seed will cause most tests to fail random.seed(1234) # WARNING! Changing the seed will cause most tests to fail
@ -47,124 +28,68 @@ def create_prices(s0: float, mu: float, sigma: float, num_prices: int) -> list:
return all_values return all_values
def create_test_timeseries( def create_data():
frequency: Frequency, num: int = 1000, skip_weekends: bool = False, mu: float = 0.1, sigma: float = 0.05 """Creates TimeSeries data"""
) -> TimeSeries:
"""Creates TimeSeries data
Parameters: dates = create_date_series("2017-01-01", "2020-10-31", "D", skip_weekends=True)
----------- values = create_prices(1000, 0.1, 0.05, 1000)
frequency: Frequency ts = TimeSeries(dict(zip(dates, values)), frequency="D")
The frequency of the time series data to be generated.
num: int
Number of date: value pairs to be generated.
skip_weekends: bool
Whether weekends (saturday, sunday) should be skipped.
Gets used only if the frequency is daily.
mu: float
Mean return for the values.
sigma: float
standard deviation of the values.
Returns:
--------
Returns a TimeSeries object
"""
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))
}
end_date = start_date + relativedelta(**timedelta_dict)
dates = create_date_series(start_date, end_date, frequency.symbol, skip_weekends=skip_weekends)
values = create_prices(1000, mu, sigma, num)
ts = TimeSeries(dict(zip(dates, values)), frequency=frequency.symbol)
return ts return ts
class TestReturns: class TestReturns:
def test_returns_calc(self): def test_returns_calc(self):
ts = create_test_timeseries() ts = create_data()
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, interval_type="years", interval_value=1
) )
assert round(returns[1], 6) == 0.112913 assert round(returns[1], 6) == 0.112913
returns = ts.calculate_returns( returns = ts.calculate_returns(
"2020-04-01", annual_compounded_returns=False, return_period_unit="months", return_period_value=3 "2020-04-01", annual_compounded_returns=False, interval_type="months", interval_value=3
) )
assert round(returns[1], 6) == 0.015908 assert round(returns[1], 6) == 0.015908
returns = ts.calculate_returns( returns = ts.calculate_returns(
"2020-04-01", annual_compounded_returns=True, return_period_unit="months", return_period_value=3 "2020-04-01", annual_compounded_returns=True, interval_type="months", interval_value=3
) )
assert round(returns[1], 6) == 0.065167 assert round(returns[1], 6) == 0.065167
returns = ts.calculate_returns( returns = ts.calculate_returns(
"2020-04-01", annual_compounded_returns=False, return_period_unit="days", return_period_value=90 "2020-04-01", annual_compounded_returns=False, interval_type="days", interval_value=90
) )
assert round(returns[1], 6) == 0.017673 assert round(returns[1], 6) == 0.017673
returns = ts.calculate_returns( returns = ts.calculate_returns(
"2020-04-01", annual_compounded_returns=True, return_period_unit="days", return_period_value=90 "2020-04-01", annual_compounded_returns=True, interval_type="days", interval_value=90
) )
assert round(returns[1], 6) == 0.073632 assert round(returns[1], 6) == 0.073632
with pytest.raises(DateNotFoundError): with pytest.raises(DateNotFoundError):
ts.calculate_returns("2020-04-04", return_period_unit="days", return_period_value=90, as_on_match="exact") ts.calculate_returns("2020-04-04", interval_type="days", interval_value=90, as_on_match="exact")
with pytest.raises(DateNotFoundError): with pytest.raises(DateNotFoundError):
ts.calculate_returns("2020-04-04", return_period_unit="months", return_period_value=3, prior_match="exact") ts.calculate_returns("2020-04-04", interval_type="months", interval_value=3, prior_match="exact")
def test_date_formats(self): def test_date_formats(self):
ts = create_test_timeseries() ts = create_data()
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("2020-04-10", annual_compounded_returns=True, interval_type="days", interval_value=90)
"2020-04-10", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
)
returns1 = ts.calculate_returns( returns1 = ts.calculate_returns("2020-04-01", interval_type="days", interval_value=90, date_format="%Y-%m-%d")
"2020-04-01", return_period_unit="days", return_period_value=90, date_format="%Y-%m-%d" returns2 = ts.calculate_returns("01-04-2020", interval_type="days", interval_value=90)
)
returns2 = ts.calculate_returns("01-04-2020", return_period_unit="days", return_period_value=90)
assert round(returns1[1], 6) == round(returns2[1], 6) == 0.073632 assert round(returns1[1], 6) == round(returns2[1], 6) == 0.073632
FincalOptions.date_format = "%m-%d-%Y" FincalOptions.date_format = "%m-%d-%Y"
with pytest.raises(ValueError): with pytest.raises(ValueError):
ts.calculate_returns( ts.calculate_returns("2020-04-01", annual_compounded_returns=True, interval_type="days", interval_value=90)
"2020-04-01", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
)
returns1 = ts.calculate_returns( returns1 = ts.calculate_returns("2020-04-01", interval_type="days", interval_value=90, date_format="%Y-%m-%d")
"2020-04-01", return_period_unit="days", return_period_value=90, date_format="%Y-%m-%d" returns2 = ts.calculate_returns("04-01-2020", interval_type="days", interval_value=90)
)
returns2 = ts.calculate_returns("04-01-2020", return_period_unit="days", return_period_value=90)
assert round(returns1[1], 6) == round(returns2[1], 6) == 0.073632 assert round(returns1[1], 6) == round(returns2[1], 6) == 0.073632
def test_limits(self): def test_limits(self):
ts = create_data()
FincalOptions.date_format = "%Y-%m-%d" FincalOptions.date_format = "%Y-%m-%d"
ts = create_test_timeseries()
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", interval_type="days", interval_value=90, closest_max_days=10)
class TestVolatility:
def test_daily_ts(self):
ts = create_test_timeseries(AllFrequencies.D)
assert len(ts) == 1000
sd = ts.volatility(annualize_volatility=False)
assert round(sd, 6) == 0.002622
sd = ts.volatility()
assert round(sd, 6) == 0.050098
sd = ts.volatility(annual_compounded_returns=True)
assert round(sd, 4) == 37.9329
sd = ts.volatility(return_period_unit="months", annual_compounded_returns=True)
assert round(sd, 4) == 0.6778
sd = ts.volatility(return_period_unit="years")
assert round(sd, 6) == 0.023164
sd = ts.volatility(from_date="2017-10-01", to_date="2019-08-31", annualize_volatility=True)
assert round(sd, 6) == 0.050559