Greg Pomerantz
c4e7b05f09
- portfolio._rebalance_dates: Grouper labels groups with period-END
timestamps (Sundays, month/quarter/year ends) that never exist in a
trading-day index, so 1W/1ME/QE/YE silently never rebalanced (only
1D and buy-and-hold worked). Now takes the first trading day of each
group.
- metrics.beta_alpha: guard var(bench) == 0 (flat benchmark) instead of
dividing by zero.
- data._read_csv_clean: the downloader has written Go nil tokens
('%!f(<nil>)') into price cells (89 symbols, one bad day each), which
made pandas infer the whole column as string and crash every
pct_change on any portfolio containing them. Coerce all columns to
numeric (nil tokens -> NaN); parquet cache rebuilt (0 string cols).
- new tests/test_portfolio.py (18 checks: rebalance dates land on
trading days and actually fire, flat-benchmark beta, nil-token
columns), wired into run_tests.sh (now 5 sections).
- full suite: 30 + 22 + 18 + 38 AppTest + 12 browser e2e, all passing
126 lines
4.3 KiB
Python
126 lines
4.3 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")
|
|
var_b = float(np.var(b))
|
|
beta = float(np.cov(r, b)[0, 1] / var_b) if var_b > 0 else 0.0
|
|
# r and b are already rf-adjusted
|
|
alpha_daily = r.mean() - beta * b.mean()
|
|
return 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
|