You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
621 lines
21 KiB
621 lines
21 KiB
from __future__ import annotations
|
|
|
|
import datetime
|
|
import math
|
|
import statistics
|
|
from typing import Literal
|
|
|
|
from pyfacts.core import date_parser
|
|
|
|
from .pyfacts import TimeSeries, create_date_series
|
|
from .utils import _interval_to_years, _preprocess_from_to_date, covariance
|
|
|
|
# from dateutil.relativedelta import relativedelta
|
|
|
|
|
|
@date_parser(3, 4)
|
|
def sharpe_ratio(
|
|
time_series_data: TimeSeries,
|
|
risk_free_data: TimeSeries = None,
|
|
risk_free_rate: float = None,
|
|
from_date: str | datetime.datetime = None,
|
|
to_date: str | datetime.datetime = None,
|
|
frequency: Literal["D", "W", "M", "Q", "H", "Y"] = None,
|
|
return_period_unit: Literal["years", "months", "days"] = "years",
|
|
return_period_value: int = 1,
|
|
as_on_match: str = "closest",
|
|
prior_match: str = "closest",
|
|
closest: Literal["previous", "next"] = "previous",
|
|
date_format: str = None,
|
|
) -> float:
|
|
"""Calculate the Sharpe ratio of any time series
|
|
|
|
Sharpe ratio is a measure of returns per unit of risk,
|
|
where risk is measured by the standard deviation of the returns.
|
|
|
|
The formula for Sharpe ratio is:
|
|
(average asset return - risk free rate)/volatility of asset returns
|
|
|
|
Parameters
|
|
----------
|
|
time_series_data:
|
|
The time series for which Sharpe ratio needs to be calculated
|
|
|
|
risk_free_data:
|
|
Risk free rates as time series data.
|
|
This should be the time series of risk free returns,
|
|
and not the underlying asset value.
|
|
|
|
risk_free_rate:
|
|
Risk free rate to be used.
|
|
Either risk_free_data or risk_free_rate needs to be provided.
|
|
If both are provided, the time series data will be used.
|
|
|
|
from_date:
|
|
Start date from which returns should be calculated.
|
|
Defaults to the first date of the series.
|
|
|
|
to_date:
|
|
End date till which returns should be calculated.
|
|
Defaults to the last date of the series.
|
|
|
|
frequency:
|
|
The frequency at which returns should be calculated.
|
|
|
|
return_period_unit: 'years', 'months', 'days'
|
|
The type of time period to use for return calculation.
|
|
|
|
return_period_value: int
|
|
The value of the specified interval type over which returns needs to be calculated.
|
|
|
|
as_on_match: str, optional
|
|
The mode of matching the as_on_date. Refer closest.
|
|
|
|
prior_match: str, optional
|
|
The mode of matching the prior_date. Refer closest.
|
|
|
|
closest: str, optional
|
|
The mode of matching the closest date.
|
|
Valid values are 'exact', 'previous', 'next' and next.
|
|
|
|
The date format to use for this operation.
|
|
Should be passed as a datetime library compatible string.
|
|
Sets the date format only for this operation. To set it globally, use FincalOptions.date_format
|
|
|
|
Returns
|
|
-------
|
|
Value of Sharpe ratio as a float.
|
|
|
|
Raises
|
|
------
|
|
ValueError
|
|
If risk free data or risk free rate is not provided.
|
|
"""
|
|
|
|
interval_days = math.ceil(_interval_to_years(return_period_unit, return_period_value) * 365)
|
|
|
|
if from_date is None:
|
|
from_date = time_series_data.start_date + datetime.timedelta(days=interval_days)
|
|
if to_date is None:
|
|
to_date = time_series_data.end_date
|
|
|
|
if risk_free_data is None and risk_free_rate is None:
|
|
raise ValueError("At least one of risk_free_data or risk_free rate is required")
|
|
elif risk_free_data is not None:
|
|
risk_free_rate = risk_free_data.mean()
|
|
|
|
common_params = {
|
|
"from_date": from_date,
|
|
"to_date": to_date,
|
|
"frequency": frequency,
|
|
"return_period_unit": return_period_unit,
|
|
"return_period_value": return_period_value,
|
|
"as_on_match": as_on_match,
|
|
"prior_match": prior_match,
|
|
"closest": closest,
|
|
"date_format": date_format,
|
|
}
|
|
average_rr = time_series_data.average_rolling_return(**common_params, annual_compounded_returns=True)
|
|
|
|
excess_returns = average_rr - risk_free_rate
|
|
sd = time_series_data.volatility(
|
|
**common_params,
|
|
annualize_volatility=True,
|
|
)
|
|
|
|
sharpe_ratio_value = excess_returns / sd
|
|
return sharpe_ratio_value
|
|
|
|
|
|
@date_parser(2, 3)
|
|
def beta(
|
|
asset_data: TimeSeries,
|
|
market_data: TimeSeries,
|
|
from_date: str | datetime.datetime = None,
|
|
to_date: str | datetime.datetime = None,
|
|
frequency: Literal["D", "W", "M", "Q", "H", "Y"] = None,
|
|
return_period_unit: Literal["years", "months", "days"] = "years",
|
|
return_period_value: int = 1,
|
|
as_on_match: str = "closest",
|
|
prior_match: str = "closest",
|
|
closest: Literal["previous", "next"] = "previous",
|
|
date_format: str = None,
|
|
) -> float:
|
|
"""Beta is a measure of sensitivity of asset returns to market returns
|
|
|
|
The formula for beta is:
|
|
|
|
Parameters
|
|
----------
|
|
asset_data: TimeSeries
|
|
The time series data of the asset
|
|
|
|
market_data: TimeSeries
|
|
The time series data of the relevant market index
|
|
|
|
from_date:
|
|
Start date from which returns should be calculated.
|
|
Defaults to the first date of the series.
|
|
|
|
to_date:
|
|
End date till which returns should be calculated.
|
|
Defaults to the last date of the series.
|
|
|
|
frequency:
|
|
The frequency at which returns should be calculated.
|
|
|
|
return_period_unit: 'years', 'months', 'days'
|
|
The type of time period to use for return calculation.
|
|
|
|
return_period_value: int
|
|
The value of the specified interval type over which returns needs to be calculated.
|
|
|
|
as_on_match: str, optional
|
|
The mode of matching the as_on_date. Refer closest.
|
|
|
|
prior_match: str, optional
|
|
The mode of matching the prior_date. Refer closest.
|
|
|
|
closest: str, optional
|
|
The mode of matching the closest date.
|
|
Valid values are 'exact', 'previous', 'next' and next.
|
|
|
|
The date format to use for this operation.
|
|
Should be passed as a datetime library compatible string.
|
|
Sets the date format only for this operation. To set it globally, use FincalOptions.date_format
|
|
|
|
Returns
|
|
-------
|
|
The value of beta as a float.
|
|
"""
|
|
interval_years = _interval_to_years(return_period_unit, return_period_value)
|
|
interval_days = math.ceil(interval_years * 365)
|
|
|
|
annual_compounded_returns = True if interval_years > 1 else False
|
|
|
|
if from_date is None:
|
|
from_date = asset_data.start_date + datetime.timedelta(days=interval_days)
|
|
if to_date is None:
|
|
to_date = asset_data.end_date
|
|
|
|
common_params = {
|
|
"from_date": from_date,
|
|
"to_date": to_date,
|
|
"frequency": frequency,
|
|
"return_period_unit": return_period_unit,
|
|
"return_period_value": return_period_value,
|
|
"as_on_match": as_on_match,
|
|
"prior_match": prior_match,
|
|
"closest": closest,
|
|
"date_format": date_format,
|
|
"annual_compounded_returns": annual_compounded_returns,
|
|
}
|
|
|
|
asset_rr = asset_data.calculate_rolling_returns(**common_params)
|
|
market_rr = market_data.calculate_rolling_returns(**common_params)
|
|
|
|
cov = covariance(asset_rr.values, market_rr.values)
|
|
market_var = statistics.variance(market_rr.values)
|
|
|
|
beta = cov / market_var
|
|
return beta
|
|
|
|
|
|
@date_parser(4, 5)
|
|
def jensens_alpha(
|
|
asset_data: TimeSeries,
|
|
market_data: TimeSeries,
|
|
risk_free_data: TimeSeries = None,
|
|
risk_free_rate: float = None,
|
|
from_date: str | datetime.datetime = None,
|
|
to_date: str | datetime.datetime = None,
|
|
frequency: Literal["D", "W", "M", "Q", "H", "Y"] = None,
|
|
return_period_unit: Literal["years", "months", "days"] = "years",
|
|
return_period_value: int = 1,
|
|
as_on_match: str = "closest",
|
|
prior_match: str = "closest",
|
|
closest: Literal["previous", "next"] = "previous",
|
|
date_format: str = None,
|
|
) -> float:
|
|
"""
|
|
This function calculates the Jensen's alpha for a time series.
|
|
The formula for Jensen's alpha is:
|
|
Ri - Rf + B x (Rm - Rf)
|
|
where:
|
|
Ri = Realized return of the portfolio or investment
|
|
Rf = The risk free rate during the return time frame
|
|
B = Beta of the portfolio or investment
|
|
Rm = Realized return of the market index
|
|
|
|
Parameters
|
|
----------
|
|
asset_data: TimeSeries
|
|
The time series data of the asset
|
|
|
|
market_data: TimeSeries
|
|
The time series data of the relevant market index
|
|
|
|
risk_free_data:
|
|
Risk free rates as time series data.
|
|
This should be the time series of risk free returns,
|
|
and not the underlying asset value.
|
|
|
|
risk_free_rate:
|
|
Risk free rate to be used.
|
|
Either risk_free_data or risk_free_rate needs to be provided.
|
|
If both are provided, the time series data will be used.
|
|
|
|
from_date:
|
|
Start date from which returns should be calculated.
|
|
Defaults to the first date of the series.
|
|
|
|
to_date:
|
|
End date till which returns should be calculated.
|
|
Defaults to the last date of the series.
|
|
|
|
frequency:
|
|
The frequency at which returns should be calculated.
|
|
|
|
return_period_unit: 'years', 'months', 'days'
|
|
The type of time period to use for return calculation.
|
|
|
|
return_period_value: int
|
|
The value of the specified interval type over which returns needs to be calculated.
|
|
|
|
as_on_match: str, optional
|
|
The mode of matching the as_on_date. Refer closest.
|
|
|
|
prior_match: str, optional
|
|
The mode of matching the prior_date. Refer closest.
|
|
|
|
closest: str, optional
|
|
The mode of matching the closest date.
|
|
Valid values are 'exact', 'previous', 'next' and next.
|
|
|
|
The date format to use for this operation.
|
|
Should be passed as a datetime library compatible string.
|
|
Sets the date format only for this operation. To set it globally, use FincalOptions.date_format
|
|
|
|
Returns
|
|
-------
|
|
The value of Jensen's alpha as a float.
|
|
"""
|
|
|
|
interval_years = _interval_to_years(return_period_unit, return_period_value)
|
|
interval_days = math.ceil(interval_years * 365)
|
|
|
|
if from_date is None:
|
|
from_date = asset_data.start_date + datetime.timedelta(days=interval_days)
|
|
if to_date is None:
|
|
to_date = asset_data.end_date
|
|
|
|
common_params = {
|
|
"from_date": from_date,
|
|
"to_date": to_date,
|
|
"frequency": frequency,
|
|
"return_period_unit": return_period_unit,
|
|
"return_period_value": return_period_value,
|
|
"as_on_match": as_on_match,
|
|
"prior_match": prior_match,
|
|
"closest": closest,
|
|
"date_format": date_format,
|
|
}
|
|
|
|
num_days = (to_date - from_date).days
|
|
compound_realised_returns = True if num_days > 365 else False
|
|
realized_return = asset_data.calculate_returns(
|
|
as_on=to_date,
|
|
return_period_unit="days",
|
|
return_period_value=num_days,
|
|
annual_compounded_returns=compound_realised_returns,
|
|
as_on_match=as_on_match,
|
|
prior_match=prior_match,
|
|
closest=closest,
|
|
date_format=date_format,
|
|
)
|
|
market_return = market_data.calculate_returns(
|
|
as_on=to_date,
|
|
return_period_unit="days",
|
|
return_period_value=num_days,
|
|
annual_compounded_returns=compound_realised_returns,
|
|
as_on_match=as_on_match,
|
|
prior_match=prior_match,
|
|
closest=closest,
|
|
date_format=date_format,
|
|
)
|
|
beta_value = beta(asset_data=asset_data, market_data=market_data, **common_params)
|
|
|
|
if risk_free_data is None and risk_free_rate is None:
|
|
raise ValueError("At least one of risk_free_data or risk_free rate is required")
|
|
elif risk_free_data is not None:
|
|
risk_free_rate = risk_free_data.mean()
|
|
|
|
jensens_alpha = realized_return[1] - risk_free_rate + beta_value * (market_return[1] - risk_free_rate)
|
|
return jensens_alpha
|
|
|
|
|
|
@date_parser(2, 3)
|
|
def correlation(
|
|
data1: TimeSeries,
|
|
data2: TimeSeries,
|
|
from_date: str | datetime.datetime = None,
|
|
to_date: str | datetime.datetime = None,
|
|
frequency: Literal["D", "W", "M", "Q", "H", "Y"] = None,
|
|
return_period_unit: Literal["years", "months", "days"] = "years",
|
|
return_period_value: int = 1,
|
|
as_on_match: str = "closest",
|
|
prior_match: str = "closest",
|
|
closest: Literal["previous", "next"] = "previous",
|
|
date_format: str = None,
|
|
) -> float:
|
|
"""Calculate the correlation between two assets
|
|
|
|
correlation calculation is done based on rolling returns.
|
|
It must be noted that correlation is not calculated directly on the asset prices.
|
|
The asset prices used to calculate returns and correlation is then calculated based on these returns.
|
|
Hence this function requires all parameters for rolling returns calculations.
|
|
|
|
Parameters
|
|
----------
|
|
data1: TimeSeries
|
|
The first time series data
|
|
|
|
data2: TimeSeries
|
|
The second time series data
|
|
|
|
from_date:
|
|
Start date from which returns should be calculated.
|
|
Defaults to the first date of the series.
|
|
|
|
to_date:
|
|
End date till which returns should be calculated.
|
|
Defaults to the last date of the series.
|
|
|
|
frequency:
|
|
The frequency at which returns should be calculated.
|
|
|
|
return_period_unit: 'years', 'months', 'days'
|
|
The type of time period to use for return calculation.
|
|
|
|
return_period_value: int
|
|
The value of the specified interval type over which returns needs to be calculated.
|
|
|
|
as_on_match: str, optional
|
|
The mode of matching the as_on_date. Refer closest.
|
|
|
|
prior_match: str, optional
|
|
The mode of matching the prior_date. Refer closest.
|
|
|
|
closest: str, optional
|
|
The mode of matching the closest date.
|
|
Valid values are 'exact', 'previous', 'next' and next.
|
|
|
|
The date format to use for this operation.
|
|
Should be passed as a datetime library compatible string.
|
|
Sets the date format only for this operation. To set it globally, use FincalOptions.date_format
|
|
|
|
Returns
|
|
-------
|
|
The value of beta as a float.
|
|
|
|
Raises
|
|
------
|
|
ValueError:
|
|
* If frequency of both TimeSeries do not match
|
|
* If both time series do not have data between the from date and to date
|
|
"""
|
|
interval_years = _interval_to_years(return_period_unit, return_period_value)
|
|
interval_days = math.ceil(interval_years * 365)
|
|
|
|
annual_compounded_returns = True if interval_years > 1 else False
|
|
|
|
if from_date is None:
|
|
from_date = data1.start_date + datetime.timedelta(days=interval_days)
|
|
if to_date is None:
|
|
to_date = data1.end_date
|
|
|
|
if data1.frequency != data2.frequency:
|
|
raise ValueError("Correlation calculation requires both time series to be of same frequency")
|
|
|
|
if from_date < data2.start_date or to_date > data2.end_date:
|
|
raise ValueError("Data between from_date and to_date must be present in both time series")
|
|
|
|
common_params = {
|
|
"from_date": from_date,
|
|
"to_date": to_date,
|
|
"frequency": frequency,
|
|
"return_period_unit": return_period_unit,
|
|
"return_period_value": return_period_value,
|
|
"as_on_match": as_on_match,
|
|
"prior_match": prior_match,
|
|
"closest": closest,
|
|
"date_format": date_format,
|
|
"annual_compounded_returns": annual_compounded_returns,
|
|
}
|
|
|
|
asset_rr = data1.calculate_rolling_returns(**common_params)
|
|
market_rr = data2.calculate_rolling_returns(**common_params)
|
|
|
|
cor = statistics.correlation(asset_rr.values, market_rr.values)
|
|
return cor
|
|
|
|
|
|
@date_parser(3, 4)
|
|
def sortino_ratio(
|
|
time_series_data: TimeSeries,
|
|
risk_free_data: TimeSeries = None,
|
|
risk_free_rate: float = None,
|
|
from_date: str | datetime.datetime = None,
|
|
to_date: str | datetime.datetime = None,
|
|
frequency: Literal["D", "W", "M", "Q", "H", "Y"] = None,
|
|
return_period_unit: Literal["years", "months", "days"] = "years",
|
|
return_period_value: int = 1,
|
|
as_on_match: str = "closest",
|
|
prior_match: str = "closest",
|
|
closest: Literal["previous", "next"] = "previous",
|
|
date_format: str = None,
|
|
) -> float:
|
|
"""Calculate the Sortino ratio of any time series
|
|
|
|
Sortino ratio is a variation of the Sharpe ratio,
|
|
where risk is measured as standard deviation of negative returns only.
|
|
Since deviation on the positive side is not undesirable, hence sortino ratio excludes positive deviations.
|
|
|
|
The formula for Sortino ratio is:
|
|
(average asset return - risk free rate)/volatility of negative asset returns
|
|
|
|
Parameters
|
|
----------
|
|
time_series_data:
|
|
The time series for which Sharpe ratio needs to be calculated
|
|
|
|
risk_free_data:
|
|
Risk free rates as time series data.
|
|
This should be the time series of risk free returns,
|
|
and not the underlying asset value.
|
|
|
|
risk_free_rate:
|
|
Risk free rate to be used.
|
|
Either risk_free_data or risk_free_rate needs to be provided.
|
|
If both are provided, the time series data will be used.
|
|
|
|
from_date:
|
|
Start date from which returns should be calculated.
|
|
Defaults to the first date of the series.
|
|
|
|
to_date:
|
|
End date till which returns should be calculated.
|
|
Defaults to the last date of the series.
|
|
|
|
frequency:
|
|
The frequency at which returns should be calculated.
|
|
|
|
return_period_unit: 'years', 'months', 'days'
|
|
The type of time period to use for return calculation.
|
|
|
|
return_period_value: int
|
|
The value of the specified interval type over which returns needs to be calculated.
|
|
|
|
as_on_match: str, optional
|
|
The mode of matching the as_on_date. Refer closest.
|
|
|
|
prior_match: str, optional
|
|
The mode of matching the prior_date. Refer closest.
|
|
|
|
closest: str, optional
|
|
The mode of matching the closest date.
|
|
Valid values are 'exact', 'previous', 'next' and next.
|
|
|
|
The date format to use for this operation.
|
|
Should be passed as a datetime library compatible string.
|
|
Sets the date format only for this operation. To set it globally, use FincalOptions.date_format
|
|
|
|
Returns
|
|
-------
|
|
Value of Sortino ratio as a float.
|
|
|
|
Raises
|
|
------
|
|
ValueError
|
|
If risk free data or risk free rate is not provided.
|
|
"""
|
|
|
|
interval_days = math.ceil(_interval_to_years(return_period_unit, return_period_value) * 365)
|
|
|
|
# if from_date is None:
|
|
# from_date = time_series_data.start_date + relativedelta(**{return_period_unit: return_period_value})
|
|
# if to_date is None:
|
|
# to_date = time_series_data.end_date
|
|
from_date, to_date = _preprocess_from_to_date(
|
|
from_date,
|
|
to_date,
|
|
time_series_data,
|
|
False,
|
|
return_period_unit,
|
|
return_period_value,
|
|
as_on_match,
|
|
prior_match,
|
|
closest,
|
|
)
|
|
|
|
if risk_free_data is None and risk_free_rate is None:
|
|
raise ValueError("At least one of risk_free_data or risk_free rate is required")
|
|
elif risk_free_data is not None:
|
|
risk_free_rate = risk_free_data.mean()
|
|
|
|
common_params = {
|
|
"from_date": from_date,
|
|
"to_date": to_date,
|
|
"frequency": frequency,
|
|
"return_period_unit": return_period_unit,
|
|
"return_period_value": return_period_value,
|
|
"as_on_match": as_on_match,
|
|
"prior_match": prior_match,
|
|
"closest": closest,
|
|
"date_format": date_format,
|
|
}
|
|
average_rr_ts = time_series_data.calculate_rolling_returns(
|
|
**common_params, annual_compounded_returns=False, if_not_found="nan"
|
|
)
|
|
average_rr = statistics.mean(filter(lambda x: str(x) != "nan", average_rr_ts.values))
|
|
annualized_average_rr = (1 + average_rr) ** (365 / interval_days) - 1
|
|
|
|
excess_returns = annualized_average_rr - risk_free_rate
|
|
my_list = [i for i in average_rr_ts.values if i < 0]
|
|
sd = statistics.stdev(my_list) # [i for i in average_rr_ts.values if i < 0])
|
|
sd *= math.sqrt(365 / interval_days)
|
|
|
|
sortino_ratio_value = excess_returns / sd
|
|
return sortino_ratio_value
|
|
|
|
|
|
@date_parser(3, 4)
|
|
def moving_average(
|
|
time_series_data: TimeSeries,
|
|
moving_average_period_unit: Literal["years", "months", "days"],
|
|
moving_average_period_value: int,
|
|
from_date: str | datetime.datetime = None,
|
|
to_date: str | datetime.datetime = None,
|
|
as_on_match: str = "closest",
|
|
prior_match: str = "closest",
|
|
closest: Literal["previous", "next"] = "previous",
|
|
date_format: str = None,
|
|
) -> TimeSeries:
|
|
|
|
from_date, to_date = _preprocess_from_to_date(
|
|
from_date,
|
|
to_date,
|
|
time_series_data,
|
|
False,
|
|
return_period_unit=moving_average_period_unit,
|
|
return_period_value=moving_average_period_value,
|
|
as_on_match=as_on_match,
|
|
prior_match=prior_match,
|
|
closest=closest,
|
|
)
|
|
|
|
dates = create_date_series(from_date, to_date, time_series_data.frequency.symbol)
|
|
|
|
for date in dates:
|
|
start_date = date - datetime.timedelta(**{moving_average_period_unit: moving_average_period_value})
|
|
time_series_data[start_date:date]
|
|
|