ml-finance-python

edhec_risk_kit_108.py

(6806B)

 import pandas as pd
 import numpy as np
 
 def get_ffme_returns():
     """
     Load the Fama-French Dataset for the returns of the Top and Bottom Deciles by MarketCap
     """
     me_m = pd.read_csv("data/Portfolios_Formed_on_ME_monthly_EW.csv",
                        header=0, index_col=0, na_values=-99.99)
     rets = me_m[['Lo 10', 'Hi 10']]
     rets.columns = ['SmallCap', 'LargeCap']
     rets = rets/100
     rets.index = pd.to_datetime(rets.index, format="%Y%m").to_period('M')
     return rets
 
 
 def get_hfi_returns():
     """
     Load and format the EDHEC Hedge Fund Index Returns
     """
     hfi = pd.read_csv("data/edhec-hedgefundindices.csv",
                       header=0, index_col=0, parse_dates=True)
     hfi = hfi/100
     hfi.index = hfi.index.to_period('M')
     return hfi
 
 def get_ind_returns():
     """
     Load and format the Ken French 30 Industry Portfolios Value Weighted Monthly Returns
     """
     ind = pd.read_csv("data/ind30_m_vw_rets.csv", header=0, index_col=0)/100
     ind.index = pd.to_datetime(ind.index, format="%Y%m").to_period('M')
     ind.columns = ind.columns.str.strip()
     return ind
 
 
 def skewness(r):
     """
     Alternative to scipy.stats.skew()
     Computes the skewness of the supplied Series or DataFrame
     Returns a float or a Series
     """
     demeaned_r = r - r.mean()
     # use the population standard deviation, so set dof=0
     sigma_r = r.std(ddof=0)
     exp = (demeaned_r**3).mean()
     return exp/sigma_r**3
 
 
 def kurtosis(r):
     """
     Alternative to scipy.stats.kurtosis()
     Computes the kurtosis of the supplied Series or DataFrame
     Returns a float or a Series
     """
     demeaned_r = r - r.mean()
     # use the population standard deviation, so set dof=0
     sigma_r = r.std(ddof=0)
     exp = (demeaned_r**4).mean()
     return exp/sigma_r**4
 
 
 def annualize_rets(r, periods_per_year):
     """
     Annualizes a set of returns
     We should infer the periods per year
     but that is currently left as an exercise
     to the reader :-)
     """
     compounded_growth = (1+r).prod()
     n_periods = r.shape[0]
     return compounded_growth**(periods_per_year/n_periods)-1
 
 
 def annualize_vol(r, periods_per_year):
     """
     Annualizes the vol of a set of returns
     We should infer the periods per year
     but that is currently left as an exercise
     to the reader :-)
     """
     return r.std()*(periods_per_year**0.5)
 
 
 def sharpe_ratio(r, riskfree_rate, periods_per_year):
     """
     Computes the annualized sharpe ratio of a set of returns
     """
     # convert the annual riskfree rate to per period
     rf_per_period = (1+riskfree_rate)**(1/periods_per_year)-1
     excess_ret = r - rf_per_period
     ann_ex_ret = annualize_rets(excess_ret, periods_per_year)
     ann_vol = annualize_vol(r, periods_per_year)
     return ann_ex_ret/ann_vol
 
 
 import scipy.stats
 def is_normal(r, level=0.01):
     """
     Applies the Jarque-Bera test to determine if a Series is normal or not
     Test is applied at the 1% level by default
     Returns True if the hypothesis of normality is accepted, False otherwise
     """
     if isinstance(r, pd.DataFrame):
         return r.aggregate(is_normal)
     else:
         statistic, p_value = scipy.stats.jarque_bera(r)
         return p_value > level
 
 
 def drawdown(return_series: pd.Series):
     """Takes a time series of asset returns.
        returns a DataFrame with columns for
        the wealth index, 
        the previous peaks, and 
        the percentage drawdown
     """
     wealth_index = 1000*(1+return_series).cumprod()
     previous_peaks = wealth_index.cummax()
     drawdowns = (wealth_index - previous_peaks)/previous_peaks
     return pd.DataFrame({"Wealth": wealth_index, 
                          "Previous Peak": previous_peaks, 
                          "Drawdown": drawdowns})
 
 
 def semideviation(r):
     """
     Returns the semideviation aka negative semideviation of r
     r must be a Series or a DataFrame, else raises a TypeError
     """
     if isinstance(r, pd.Series):
         is_negative = r < 0
         return r[is_negative].std(ddof=0)
     elif isinstance(r, pd.DataFrame):
         return r.aggregate(semideviation)
     else:
         raise TypeError("Expected r to be a Series or DataFrame")
 
 
 def var_historic(r, level=5):
     """
     Returns the historic Value at Risk at a specified level
     i.e. returns the number such that "level" percent of the returns
     fall below that number, and the (100-level) percent are above
     """
     if isinstance(r, pd.DataFrame):
         return r.aggregate(var_historic, level=level)
     elif isinstance(r, pd.Series):
         return -np.percentile(r, level)
     else:
         raise TypeError("Expected r to be a Series or DataFrame")
 
 
 def cvar_historic(r, level=5):
     """
     Computes the Conditional VaR of Series or DataFrame
     """
     if isinstance(r, pd.Series):
         is_beyond = r <= var_historic(r, level=level)
         return -r[is_beyond].mean()
     elif isinstance(r, pd.DataFrame):
         return r.aggregate(cvar_historic, level=level)
     else:
         raise TypeError("Expected r to be a Series or DataFrame")
 
 
 from scipy.stats import norm
 def var_gaussian(r, level=5, modified=False):
     """
     Returns the Parametric Gauusian VaR of a Series or DataFrame
     If "modified" is True, then the modified VaR is returned,
     using the Cornish-Fisher modification
     """
     # compute the Z score assuming it was Gaussian
     z = norm.ppf(level/100)
     if modified:
         # modify the Z score based on observed skewness and kurtosis
         s = skewness(r)
         k = kurtosis(r)
         z = (z +
                 (z**2 - 1)*s/6 +
                 (z**3 -3*z)*(k-3)/24 -
                 (2*z**3 - 5*z)*(s**2)/36
             )
     return -(r.mean() + z*r.std(ddof=0))
 
 
 def portfolio_return(weights, returns):
     """
     Computes the return on a portfolio from constituent returns and weights
     weights are a numpy array or Nx1 matrix and returns are a numpy array or Nx1 matrix
     """
     return weights.T @ returns
 
 
 def portfolio_vol(weights, covmat):
     """
     Computes the vol of a portfolio from a covariance matrix and constituent weights
     weights are a numpy array or N x 1 maxtrix and covmat is an N x N matrix
     """
     return (weights.T @ covmat @ weights)**0.5
 
 
 def plot_ef2(n_points, er, cov):
     """
     Plots the 2-asset efficient frontier
     """
     if er.shape[0] != 2 or er.shape[0] != 2:
         raise ValueError("plot_ef2 can only plot 2-asset frontiers")
     weights = [np.array([w, 1-w]) for w in np.linspace(0, 1, n_points)]
     rets = [portfolio_return(w, er) for w in weights]
     vols = [portfolio_vol(w, cov) for w in weights]
     ef = pd.DataFrame({
         "Returns": rets, 
         "Volatility": vols
     })
     return ef.plot.line(x="Volatility", y="Returns", style=".-")