"""
Portfolio and asset performance/risk metric functions.
This module provides a collection of statistical and financial functions for
analyzing return series, including higher-moment statistics (skewness,
kurtosis), return compounding and annualization, risk-adjusted performance
(Sharpe ratio), drawdown analysis, and Value-at-Risk / Conditional
Value-at-Risk estimation (historic and Gaussian/Cornish-Fisher).
All functions that accept a returns argument ``r`` are designed to work
transparently with either a :class:`pandas.Series` (a single return series)
or a :class:`pandas.DataFrame` (multiple return series, one per column),
returning a scalar or a :class:`pandas.Series` of per-column results
accordingly.
"""
import numpy as np
import pandas as pd
from scipy.stats import norm
[docs]
def skewness(r: pd.Series | pd.DataFrame) -> float | pd.Series:
"""
Compute the skewness of the supplied returns.
Alternative to :func:`scipy.stats.skew` that uses the population
standard deviation (``ddof=0``).
Parameters
----------
r : pandas.Series or pandas.DataFrame
Series or DataFrame of returns.
Returns
-------
float or pandas.Series
The skewness of `r`. Returns a float if `r` is a Series, or a
Series of per-column skewness values if `r` is a DataFrame.
"""
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
[docs]
def kurtosis(r: pd.Series | pd.DataFrame) -> float | pd.Series:
"""
Compute the kurtosis of the supplied returns.
Alternative to :func:`scipy.stats.kurtosis` that uses the population
standard deviation (``ddof=0``).
Parameters
----------
r : pandas.Series or pandas.DataFrame
Series or DataFrame of returns.
Returns
-------
float or pandas.Series
The kurtosis of `r`. Returns a float if `r` is a Series, or a
Series of per-column kurtosis values if `r` is a DataFrame.
"""
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
[docs]
def compound(r: pd.Series | pd.DataFrame) -> float | pd.Series:
"""
Compound the set of returns in `r`.
Parameters
----------
r : pandas.Series or pandas.DataFrame
Series or DataFrame of periodic returns.
Returns
-------
float or pandas.Series
The compounded (total) return over the whole period. Returns a
float if `r` is a Series, or a Series of per-column compounded
returns if `r` is a DataFrame.
"""
return np.expm1(np.log1p(r).sum())
[docs]
def annualize_rets(r: pd.Series | pd.DataFrame, periods_per_year: int) -> float | pd.Series:
"""
Annualize a set of returns.
Notes
-----
We should infer the periods per year, but that is currently left as an
exercise to the reader :-)
Parameters
----------
r : pandas.Series or pandas.DataFrame
Series or DataFrame of periodic returns.
periods_per_year : int
Number of periods per year (e.g. 252 for daily returns, 12 for
monthly returns).
Returns
-------
float or pandas.Series
The annualized return. Returns a float if `r` is a Series, or a
Series of per-column annualized returns if `r` is a DataFrame.
"""
compounded_growth = (1 + r).prod()
n_periods = r.shape[0]
return compounded_growth**(periods_per_year / n_periods) - 1
[docs]
def annualize_vol(r: pd.Series | pd.DataFrame, periods_per_year: int) -> float | pd.Series:
"""
Annualize the volatility of a set of returns.
Notes
-----
We should infer the periods per year, but that is currently left as an
exercise to the reader :-)
Parameters
----------
r : pandas.Series or pandas.DataFrame
Series or DataFrame of periodic returns.
periods_per_year : int
Number of periods per year (e.g. 252 for daily returns, 12 for
monthly returns).
Returns
-------
float or pandas.Series
The annualized volatility. Returns a float if `r` is a Series, or
a Series of per-column annualized volatilities if `r` is a
DataFrame.
"""
return r.std() * (periods_per_year**0.5)
[docs]
def sharpe_ratio(
r: pd.Series | pd.DataFrame,
riskfree_rate: float,
periods_per_year: int,
) -> float | pd.Series:
"""
Compute the annualized Sharpe ratio of a set of returns.
Parameters
----------
r : pandas.Series or pandas.DataFrame
Series or DataFrame of periodic returns.
riskfree_rate : float
Annual risk-free rate, expressed as a decimal (e.g. ``0.03`` for
3%). It's converted internally to a per-period rate using
`periods_per_year`.
periods_per_year : int
Number of periods per year (e.g. 252 for daily returns, 12 for
monthly returns).
Returns
-------
float or pandas.Series
The annualized Sharpe ratio. Returns a float if `r` is a Series,
or a Series of per-column Sharpe ratios if `r` is a DataFrame.
"""
# 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
[docs]
def drawdown(return_series: pd.Series) -> pd.DataFrame:
"""
Compute the wealth index, running maximum, and drawdown of a return series.
Parameters
----------
return_series : pandas.Series
Time series of asset returns.
Returns
-------
pandas.DataFrame
DataFrame indexed like `return_series`, with columns:
- ``Wealth``: the cumulative wealth index, starting at 1000.
- ``Previous Peak``: the running maximum of the wealth index.
- ``Drawdown``: the percentage drawdown from the previous peak.
"""
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,
})
[docs]
def semideviation(r: pd.Series | pd.DataFrame) -> float | pd.Series:
"""
Compute the semideviation (negative semideviation) of `r`.
Parameters
----------
r : pandas.Series or pandas.DataFrame
Series or DataFrame of returns.
Returns
-------
float or pandas.Series
The semideviation, computed as the population standard deviation
(``ddof=0``) of the negative values in `r`. Returns a float if `r`
is a Series, or a Series of per-column semideviations if `r` is a
DataFrame.
Raises
------
TypeError
If `r` is neither a :class:`pandas.Series` nor a
:class:`pandas.DataFrame`.
"""
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")
[docs]
def var_historic(r: pd.Series | pd.DataFrame, level: float = 5) -> float | pd.Series:
"""
Compute the historic Value at Risk (VaR) at a specified level.
Returns the number such that `level` percent of the returns fall
below that number (as a positive loss figure), and ``100 - level``
percent are above it.
Parameters
----------
r : pandas.Series or pandas.DataFrame
Series or DataFrame of returns.
level : float, default 5
The percentile level, in percent (e.g. ``5`` for the 5th
percentile / 95% VaR).
Returns
-------
float or pandas.Series
The historic VaR, expressed as a positive number. Returns a float
if `r` is a Series, or a Series of per-column VaR values if `r`
is a DataFrame.
Raises
------
TypeError
If `r` is neither a :class:`pandas.Series` nor a
:class:`pandas.DataFrame`.
"""
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")
[docs]
def cvar_historic(r: pd.Series | pd.DataFrame, level: float = 5) -> float | pd.Series:
"""
Compute the historic Conditional Value at Risk (CVaR) of `r`.
The CVaR (also known as expected shortfall) is the average of the
returns that fall beyond the historic VaR threshold.
Parameters
----------
r : pandas.Series or pandas.DataFrame
Series or DataFrame of returns.
level : float, default 5
The percentile level, in percent, used to compute the underlying
VaR threshold (see :func:`var_historic`).
Returns
-------
float or pandas.Series
The historic CVaR, expressed as a positive number. Returns a
float if `r` is a Series, or a Series of per-column CVaR values
if `r` is a DataFrame.
Raises
------
TypeError
If `r` is neither a :class:`pandas.Series` nor a
:class:`pandas.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")
[docs]
def var_gaussian(
r: pd.Series | pd.DataFrame,
level: float = 5,
*,
modified: bool = False,
) -> float | pd.Series:
"""
Compute the parametric Gaussian VaR of `r`.
Parameters
----------
r : pandas.Series or pandas.DataFrame
Series or DataFrame of returns.
level : float, default 5
The percentile level, in percent (e.g. ``5`` for the 5th
percentile / 95% VaR).
modified : bool, default False
If True, apply the Cornish-Fisher modification to the Z-score
using the observed skewness and kurtosis of `r`, returning the
modified VaR instead of the plain Gaussian VaR.
Returns
-------
float or pandas.Series
The parametric (optionally Cornish-Fisher modified) Gaussian VaR,
expressed as a positive number. Returns a float if `r` is a
Series, or a Series of per-column VaR values if `r` is a
DataFrame.
"""
# 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))
[docs]
def summary_stats(
r: pd.DataFrame,
riskfree_rate: float = 0.0,
periods_per_year: int = 252,
) -> pd.DataFrame:
"""
Compute aggregated summary statistics for the returns in `r`.
Parameters
----------
r : pandas.DataFrame
DataFrame of periodic returns, one column per asset/strategy.
riskfree_rate : float, default 0.0
Annual risk-free rate, expressed as a decimal, used to compute
the Sharpe ratio (see :func:`sharpe_ratio`).
periods_per_year : int, default 252
Number of periods per year used for annualizing returns,
volatility, and the Sharpe ratio.
Returns
-------
pandas.DataFrame
A single-row DataFrame (indexed by ``'Strategy'``) with one column
per summary statistic: ``Annualized Return``, ``Annualized Vol``,
``Sharpe Ratio``, ``Max Drawdown``, ``Skewness``, ``Kurtosis``,
``Cornish-Fisher VaR (5%)``, and ``Historic CVaR (5%)``.
"""
ann_r = r.aggregate(annualize_rets, periods_per_year=periods_per_year)
ann_vol = r.aggregate(annualize_vol, periods_per_year=periods_per_year)
ann_sr = r.aggregate(sharpe_ratio, riskfree_rate=riskfree_rate, periods_per_year=periods_per_year)
dd = r.aggregate(lambda r: drawdown(r).Drawdown.min())
skew = r.aggregate(skewness)
kurt = r.aggregate(kurtosis)
cf_var5 = r.aggregate(var_gaussian, modified=True)
hist_cvar5 = r.aggregate(cvar_historic)
return pd.DataFrame({
"Annualized Return": ann_r,
"Annualized Vol": ann_vol,
"Sharpe Ratio": ann_sr,
"Max Drawdown": dd,
"Skewness": skew,
"Kurtosis": kurt,
"Cornish-Fisher VaR (5%)": cf_var5,
"Historic CVaR (5%)": hist_cvar5,
}, index=['Strategy'])
[docs]
def portfolio_return_daily_ts(
portfolio_weights: pd.DataFrame,
asset_returns: pd.DataFrame,
) -> pd.Series:
"""
Calculate daily portfolio returns from weights and asset returns.
The two DataFrames are aligned on their common dates (index) and
common assets (columns) before computing the weighted sum of returns.
Parameters
----------
portfolio_weights : pandas.DataFrame
DataFrame with dates as index and assets as columns, containing
the portfolio weight of each asset on each date.
asset_returns : pandas.DataFrame
DataFrame with dates as index and assets as columns, containing
the periodic return of each asset on each date.
Returns
-------
pandas.Series
Series of daily portfolio returns, indexed by the dates common to
both `portfolio_weights` and `asset_returns`.
"""
# Align indices
common_dates = portfolio_weights.index.intersection(asset_returns.index)
common_stocks = portfolio_weights.columns.intersection(asset_returns.columns)
aligned_weights = portfolio_weights.loc[common_dates, common_stocks]
aligned_returns = asset_returns.loc[common_dates, common_stocks]
# Calculate daily portfolio returns
return (aligned_weights * aligned_returns).sum(axis=1)
