Compare commits

...

3 Commits

  1. 61
      fincal/fincal.py
  2. 40
      tests/test_fincal.py
  3. 131
      tests/test_fincal2.py

61
fincal/fincal.py

@ -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,
interval_type: Literal["years", "months", "days"] = "years", return_period_unit: Literal["years", "months", "days"] = "years",
interval_value: int = 1, return_period_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.
interval_type : 'years', 'months', 'days' return_period_unit : 'years', 'months', 'days'
The type of time period to use for return calculation. The type of time period to use for return calculation.
interval_value : int return_period_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(**{interval_type: interval_value}) prev_date = as_on - relativedelta(**{return_period_unit: return_period_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(interval_type, interval_value) years = _interval_to_years(return_period_unit, return_period_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,
interval_type: Literal["years", "months", "days"] = "years", return_period_unit: Literal["years", "months", "days"] = "years",
interval_value: int = 1, return_period_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.
interval_type : years | month | days return_period_unit : years | month | days
The interval for the return calculation. The interval for the return calculation.
interval_value : int, optional return_period_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,
interval_type=interval_type, return_period_unit=return_period_unit,
interval_value=interval_value, return_period_value=return_period_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,22 +396,41 @@ 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:
@ -423,7 +442,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(**{interval_type: interval_value}) from_date = self.start_date + relativedelta(**{return_period_unit: return_period_value})
if to_date is None: if to_date is None:
to_date = self.end_date to_date = self.end_date
@ -439,17 +458,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,
interval_type=interval_type, return_period_unit=return_period_unit,
interval_value=interval_value, return_period_value=return_period_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 interval_type == "months": if return_period_unit == "months":
sd *= math.sqrt(12) sd *= math.sqrt(12)
elif interval_type == "days": elif return_period_unit == "days":
sd *= math.sqrt(traded_days) sd *= math.sqrt(traded_days)
return sd return sd

40
tests/test_fincal.py

@ -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], list) assert isinstance(time_series.iloc[10:20], TimeSeries)
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,57 +236,65 @@ 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, interval_type="years", interval_value=1 "2021-01-01", annual_compounded_returns=False, return_period_unit="years", return_period_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, interval_type="months", interval_value=3 "2020-04-01", annual_compounded_returns=False, return_period_unit="months", return_period_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, interval_type="months", interval_value=3 "2020-04-01", annual_compounded_returns=True, return_period_unit="months", return_period_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, interval_type="days", interval_value=90 "2020-04-01", annual_compounded_returns=False, return_period_unit="days", return_period_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, interval_type="days", interval_value=90 "2020-04-01", annual_compounded_returns=True, return_period_unit="days", return_period_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, interval_type="days", interval_value=90 "2020-04-10", annual_compounded_returns=True, return_period_unit="days", return_period_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", return_period_unit="days", return_period_value=90, as_on_match="exact")
with pytest.raises(DateNotFoundError): with pytest.raises(DateNotFoundError):
ts.calculate_returns("2020-04-10", interval_type="days", interval_value=90, prior_match="exact") ts.calculate_returns("2020-04-10", return_period_unit="days", return_period_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("2020-04-10", annual_compounded_returns=True, interval_type="days", interval_value=90) ts.calculate_returns(
"2020-04-10", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
)
returns1 = ts.calculate_returns("2020-04-10", interval_type="days", interval_value=90, date_format="%Y-%m-%d") returns1 = ts.calculate_returns(
returns2 = ts.calculate_returns("10-04-2020", interval_type="days", interval_value=90) "2020-04-10", return_period_unit="days", return_period_value=90, date_format="%Y-%m-%d"
)
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("2020-04-10", annual_compounded_returns=True, interval_type="days", interval_value=90) ts.calculate_returns(
"2020-04-10", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
)
returns1 = ts.calculate_returns("2020-04-10", interval_type="days", interval_value=90, date_format="%Y-%m-%d") returns1 = ts.calculate_returns(
returns2 = ts.calculate_returns("04-10-2020", interval_type="days", interval_value=90) "2020-04-10", return_period_unit="days", return_period_value=90, date_format="%Y-%m-%d"
)
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", interval_type="days", interval_value=90, closest_max_days=10) ts.calculate_returns("2020-04-25", return_period_unit="days", return_period_value=90, closest_max_days=10)
class TestVolatility: class TestVolatility:

131
tests/test_fincal2.py

@ -3,18 +3,37 @@ 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.
- mu: Interest rate expressed annual terms. Since this function is used only to generate data for tests, the seed is fixed as 1234.
- sigma: Volatility expressed annual terms. Many of the tests rely on exact values generated using this seed.
- seed: seed for the random number generator If the seed is changed, those tests will fail.
- 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
@ -28,68 +47,124 @@ def create_prices(s0: float, mu: float, sigma: float, num_prices: int) -> list:
return all_values return all_values
def create_data(): def create_test_timeseries(
"""Creates TimeSeries data""" frequency: Frequency, num: int = 1000, skip_weekends: bool = False, mu: float = 0.1, sigma: float = 0.05
) -> TimeSeries:
"""Creates TimeSeries data
Parameters:
-----------
frequency: Frequency
The frequency of the time series data to be generated.
num: int
Number of date: value pairs to be generated.
dates = create_date_series("2017-01-01", "2020-10-31", "D", skip_weekends=True) skip_weekends: bool
values = create_prices(1000, 0.1, 0.05, 1000) Whether weekends (saturday, sunday) should be skipped.
ts = TimeSeries(dict(zip(dates, values)), frequency="D") 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_data() ts = create_test_timeseries()
returns = ts.calculate_returns( returns = ts.calculate_returns(
"2020-01-01", annual_compounded_returns=False, interval_type="years", interval_value=1 "2020-01-01", annual_compounded_returns=False, return_period_unit="years", return_period_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, interval_type="months", interval_value=3 "2020-04-01", annual_compounded_returns=False, return_period_unit="months", return_period_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, interval_type="months", interval_value=3 "2020-04-01", annual_compounded_returns=True, return_period_unit="months", return_period_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, interval_type="days", interval_value=90 "2020-04-01", annual_compounded_returns=False, return_period_unit="days", return_period_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, interval_type="days", interval_value=90 "2020-04-01", annual_compounded_returns=True, return_period_unit="days", return_period_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", interval_type="days", interval_value=90, as_on_match="exact") ts.calculate_returns("2020-04-04", return_period_unit="days", return_period_value=90, as_on_match="exact")
with pytest.raises(DateNotFoundError): with pytest.raises(DateNotFoundError):
ts.calculate_returns("2020-04-04", interval_type="months", interval_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_data() ts = create_test_timeseries()
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", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
)
returns1 = ts.calculate_returns("2020-04-01", interval_type="days", interval_value=90, date_format="%Y-%m-%d") returns1 = ts.calculate_returns(
returns2 = ts.calculate_returns("01-04-2020", interval_type="days", interval_value=90) "2020-04-01", return_period_unit="days", return_period_value=90, date_format="%Y-%m-%d"
)
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("2020-04-01", annual_compounded_returns=True, interval_type="days", interval_value=90) ts.calculate_returns(
"2020-04-01", annual_compounded_returns=True, return_period_unit="days", return_period_value=90
)
returns1 = ts.calculate_returns("2020-04-01", interval_type="days", interval_value=90, date_format="%Y-%m-%d") returns1 = ts.calculate_returns(
returns2 = ts.calculate_returns("04-01-2020", interval_type="days", interval_value=90) "2020-04-01", return_period_unit="days", return_period_value=90, date_format="%Y-%m-%d"
)
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", interval_type="days", interval_value=90, closest_max_days=10) ts.calculate_returns("2020-11-25", return_period_unit="days", return_period_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

Loading…
Cancel
Save