Greg Pomerantz
4f36bc7aea
- data.py: non-blocking load_bundle(); background watcher thread refreshes the parquet cache (5s scan, 30s min rebuild cadence); refresh()/ up_to_date()/generation() - statistics tab: one table per benchmark (vs <label>), plain column names (beta/alpha/return/vol...), selectable+reorderable stat list in settings.json - correlation tab: per-portfolio components-vs-benchmarks + all-portfolios-vs-benchmarks; numbered columns - global date range (window radio + start/end boxes) applied to all tabs; metrics.xcorr(); equity window radio gains YTD/3M/1M
112 lines
3.7 KiB
Python
112 lines
3.7 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 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 = 0.0) -> float:
|
|
r = returns - rf / ANN
|
|
sd = r.std()
|
|
return float(r.mean() / sd * np.sqrt(ANN)) if sd > 0 else 0.0
|
|
|
|
|
|
def sortino(returns: pd.Series, rf: float = 0.0) -> float:
|
|
r = returns - rf / ANN
|
|
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 = 0.0):
|
|
"""CAPM regression. Returns (beta, annualized_alpha)."""
|
|
r = (returns - rf / ANN).dropna()
|
|
b = (bench - rf / ANN).dropna()
|
|
r, b = r.align(b, join="inner")
|
|
beta = np.cov(r, b)[0, 1] / np.var(b)
|
|
alpha_daily = r.mean() - (rf / ANN + beta * (b.mean() - rf / ANN))
|
|
return float(beta), float(alpha_daily * ANN)
|
|
|
|
|
|
def summary(price: pd.Series, bench: pd.Series | None = None,
|
|
rf: float = 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
|