- No fixed risk-free rate: the statistics page now nets Sharpe, Sortino and CAPM alpha against BIL (SPDR 1-3 Month T-Bill) daily total returns from the data bundle — the same reference the Fund Lab uses for all its alphas (pre-2007 dates fill 0, as in fundlab.decompose). - metrics: sharpe/sortino/beta_alpha/summary accept a daily rf SERIES (or scalar annual rate as before) via a shared excess() helper. - Also fixes a latent double-count: the old scalar-rf alpha subtracted rf twice (once in the returns, once in the intercept term); alpha is now mean(excess fund) - beta * mean(excess bench), the standard CAPM intercept on excess returns. - Page caption states the T-bill reference (or warns if BIL is absent). - Tests: series-rf identities (sharpe/sortino/beta/alpha), benchmark self-row beta 1 / alpha 0, caption check.
125 lines
4.2 KiB
Python
125 lines
4.2 KiB
Python
"""Performance statistics on price/return series (daily, 252 days/yr).
|
|
|
|
All functions accept a price series (or DataFrame) and return scalars or
|
|
Series. Kept deliberately dependency-light (pandas/numpy only) so every
|
|
metric is transparent and tweakable. `empyrical-reloaded` is a fine
|
|
drop-in for more metrics if you ever want them.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
ANN = 252
|
|
|
|
|
|
def daily_returns(price: pd.Series | pd.DataFrame) -> pd.Series | pd.DataFrame:
|
|
return price.pct_change().fillna(0.0)
|
|
|
|
|
|
def excess(returns: pd.Series, rf: float | pd.Series | None) -> pd.Series:
|
|
"""Net returns against the risk-free rate.
|
|
|
|
rf may be a scalar ANNUAL rate (subtracted as rf/252 per day), a
|
|
DAILY rate series aligned by index (e.g. BIL total returns, filled 0
|
|
where absent), or None/0 for no adjustment."""
|
|
if rf is None or isinstance(rf, (int, float)):
|
|
return returns - rf / ANN
|
|
return returns - rf.reindex(returns.index).fillna(0.0)
|
|
|
|
|
|
def total_return(price: pd.Series | pd.DataFrame) -> float:
|
|
return float(price.iloc[-1] / price.iloc[0] - 1.0)
|
|
|
|
|
|
def annualized_return(price: pd.Series | pd.DataFrame) -> float:
|
|
n = len(price)
|
|
return float((price.iloc[-1] / price.iloc[0]) ** (ANN / n) - 1.0)
|
|
|
|
|
|
def annualized_vol(returns: pd.Series | pd.DataFrame) -> float:
|
|
return float(returns.std() * np.sqrt(ANN))
|
|
|
|
|
|
def sharpe(returns: pd.Series, rf: float | pd.Series | None = 0.0) -> float:
|
|
r = excess(returns, rf)
|
|
sd = r.std()
|
|
return float(r.mean() / sd * np.sqrt(ANN)) if sd > 0 else 0.0
|
|
|
|
|
|
def sortino(returns: pd.Series, rf: float | pd.Series | None = 0.0) -> float:
|
|
r = excess(returns, rf)
|
|
dd = float(np.sqrt(np.mean(np.minimum(r, 0.0) ** 2)))
|
|
return float(r.mean() / dd * np.sqrt(ANN)) if dd > 0 else 0.0
|
|
|
|
|
|
def max_drawdown(price: pd.Series | pd.DataFrame) -> float:
|
|
peak = price.cummax()
|
|
return float((price / peak - 1.0).min())
|
|
|
|
|
|
def calmar(price: pd.Series | pd.DataFrame) -> float:
|
|
mdd = max_drawdown(price)
|
|
return float(annualized_return(price) / -mdd) if mdd < 0 else 0.0
|
|
|
|
|
|
def beta_alpha(returns: pd.Series, bench: pd.Series,
|
|
rf: float | pd.Series | None = 0.0):
|
|
"""CAPM regression. Returns (beta, annualized_alpha)."""
|
|
r = excess(returns, rf).dropna()
|
|
b = excess(bench, rf).dropna()
|
|
r, b = r.align(b, join="inner")
|
|
beta = np.cov(r, b)[0, 1] / np.var(b)
|
|
# r and b are already rf-adjusted
|
|
alpha_daily = r.mean() - beta * b.mean()
|
|
return float(beta), float(alpha_daily * ANN)
|
|
|
|
|
|
def summary(price: pd.Series, bench: pd.Series | None = None,
|
|
rf: float | pd.Series | None = 0.0) -> dict[str, float]:
|
|
r = daily_returns(price)
|
|
out = {
|
|
"total_return": total_return(price),
|
|
"return": annualized_return(price),
|
|
"vol": annualized_vol(r),
|
|
"sharpe": sharpe(r, rf),
|
|
"sortino": sortino(r, rf),
|
|
"max_dd": max_drawdown(price),
|
|
"calmar": calmar(price),
|
|
}
|
|
if bench is not None:
|
|
b, a = beta_alpha(r, daily_returns(bench), rf)
|
|
out["beta"] = b
|
|
out["alpha"] = a
|
|
out["ann_return_bench"] = annualized_return(bench)
|
|
return out
|
|
|
|
|
|
def xcorr(prices: dict[str, pd.Series]) -> pd.DataFrame:
|
|
"""Pairwise Pearson cross-correlation of daily returns.
|
|
|
|
prices: {label: price series}. Series of different lengths (e.g. a
|
|
young ETF next to a 30-year index) correlate over their common
|
|
history — pandas corr() is pairwise-complete over NaN."""
|
|
r = pd.DataFrame({n: s.pct_change() for n, s in prices.items()})
|
|
return r.corr().round(2)
|
|
|
|
|
|
def format_summary_table(summaries: dict[str, dict[str, float]]) -> pd.DataFrame:
|
|
"""{label: summary_dict} -> transposed table, percentages pre-formatted.
|
|
|
|
Benchmark columns may carry a ' [<benchmark>] ' suffix when there are
|
|
several benchmarks; all of them are formatted by prefix.
|
|
"""
|
|
df = pd.DataFrame(summaries).T
|
|
pct = ("total_return", "return", "vol", "max_dd", "alpha", "ann_return_bench")
|
|
two = ("sharpe", "sortino", "calmar", "beta")
|
|
for c in df.columns:
|
|
base = c.split(" [")[0]
|
|
if base in pct:
|
|
df[c] = df[c].map(lambda v: f"{v:,.1%}")
|
|
elif base in two:
|
|
df[c] = df[c].map(lambda v: f"{v:.2f}")
|
|
return df
|