Benchmark Alignment and Relative Strategy Performance Diagnostics
Code Machine Learning for Trading
Summary
The document outlines ways to compare strategy returns with two external benchmarks: daily SPY total returns and Fama-French market returns formed by adding the market risk premium and risk-free rate. It provides date filtering and alignment helpers, including a month-based join for datasets whose monthly timestamps use different conventions. The caller must ensure compatible return frequencies and align the observations before calculating diagnostics.
The diagnostic calculations include information ratio, beta, correlation, and annualized tracking error, with optional subperiod summaries of strategy returns, annualized return, Sharpe ratio, and maximum drawdown. They drop paired missing values and return undefined metrics when too few observations are available or variance is zero. These measures describe relative behavior, not whether a strategy is investable or likely to persist. The document supplies implementation guidance rather than empirical findings, and its annualization depends on the period frequency supplied by the user.
Key ideas
- SPY close-to-close returns and Fama-French market returns provide alternative comparison series.
- Daily timestamps can be joined directly, while monthly series may require matching by calendar month.
- Benchmark diagnostics summarize relative return, beta, correlation, and tracking error.
- Subperiod calculations can show how strategy performance varies across defined date buckets.
- Frequency, missing observations, and zero variance affect the validity of the reported metrics.
Tags
Full text
# external_benchmarks.py
```py
"""External benchmark loaders for strategy-analysis notebooks.
Provides SPY (broad-equity ETF) and FF-market (Ken French Mkt-RF + RF)
return series on the canonical schema (`timestamp`, `benchmark_return`).
Both flavors return polars DataFrames; alignment with strategy returns
is the caller's responsibility.
The diagnostic helper `compute_benchmark_diagnostics` returns the
information ratio, beta, correlation, and tracking error of a strategy
return series against a benchmark return series.
"""
from __future__ import annotations
import datetime as _dt
import numpy as np
import polars as pl
from data import load_etfs, load_ff_factors
def _date_or_none(value: str | _dt.date | _dt.datetime | None) -> _dt.date | None:
if value is None:
return None
if isinstance(value, _dt.date) and not isinstance(value, _dt.datetime):
return value
if isinstance(value, _dt.datetime):
return value.date()
return _dt.date.fromisoformat(str(value)[:10])
def load_spy_returns(
start: str | _dt.date | None = None,
end: str | _dt.date | None = None,
) -> pl.DataFrame:
"""SPY daily total returns (close-to-close pct change).
Returns columns: `timestamp` (Date), `benchmark_return` (Float64).
"""
df = load_etfs(symbols=["SPY"]).sort("timestamp").select("timestamp", "close")
df = (
df.with_columns(
benchmark_return=pl.col("close").pct_change(),
)
.drop("close")
.drop_nulls("benchmark_return")
)
s, e = _date_or_none(start), _date_or_none(end)
if s is not None:
df = df.filter(pl.col("timestamp") >= s)
if e is not None:
df = df.filter(pl.col("timestamp") <= e)
return df
def load_ff_market_returns(
start: str | _dt.date | None = None,
end: str | _dt.date | None = None,
frequency: str = "daily",
) -> pl.DataFrame:
"""Fama-French market return (Mkt-RF + RF, i.e. nominal market).
`frequency` is `"daily"` or `"monthly"`. Returns columns:
`timestamp` (Date), `benchmark_return` (Float64).
"""
df = load_ff_factors(dataset="ff5", frequency=frequency).sort("timestamp")
df = df.with_columns(
benchmark_return=pl.col("Mkt-RF") + pl.col("RF"),
).select("timestamp", "benchmark_return")
s, e = _date_or_none(start), _date_or_none(end)
if s is not None:
df = df.filter(pl.col("timestamp") >= s)
if e is not None:
df = df.filter(pl.col("timestamp") <= e)
return df
def align_to_strategy(
strategy_df: pl.DataFrame,
benchmark_df: pl.DataFrame,
timestamp_col: str = "ts",
strategy_col: str = "strategy",
benchmark_col: str = "benchmark_return",
) -> pl.DataFrame:
"""Inner-join a benchmark series onto strategy returns.
Both inputs must carry the canonical timestamp column (Date). The
result has columns `[timestamp_col, strategy_col, benchmark_col]`.
Caller-side alignment for monthly cadences should resample first.
"""
bench = benchmark_df.with_columns(
pl.col("timestamp").cast(pl.Date).alias(timestamp_col)
).select(timestamp_col, benchmark_col)
return strategy_df.join(bench, on=timestamp_col, how="inner").sort(timestamp_col)
def align_to_strategy_monthly(
strategy_df: pl.DataFrame,
benchmark_df: pl.DataFrame,
timestamp_col: str = "ts",
strategy_col: str = "strategy",
benchmark_col: str = "benchmark_return",
) -> pl.DataFrame:
"""Month-anchored inner join (FF monthly is first-of-month; benchmark
series often end-of-month). Joins on (year, month) key.
"""
s = strategy_df.with_columns(
_y=pl.col(timestamp_col).dt.year(),
_m=pl.col(timestamp_col).dt.month(),
)
b = benchmark_df.with_columns(
_y=pl.col("timestamp").dt.year(),
_m=pl.col("timestamp").dt.month(),
).select("_y", "_m", benchmark_col)
return s.join(b, on=["_y", "_m"], how="inner").drop("_y", "_m").sort(timestamp_col)
def compute_subperiod_diagnostics(
df: pl.DataFrame,
buckets: list[tuple[str, _dt.date | str, _dt.date | str]],
*,
timestamp_col: str = "ts",
strategy_col: str = "strategy",
benchmark_col: str | None = "benchmark",
periods_per_year: int = 252,
) -> pl.DataFrame:
"""Per-bucket diagnostics for sub-period decomposition tables.
`df` must carry an aligned `(timestamp, strategy[, benchmark])` triple.
`buckets` is a list of `(label, start_date, end_date)` triples; date
bounds are inclusive at both ends. Per-bucket metrics: `n`,
`cum_return`, `ann_return`, `sharpe`, `max_drawdown`. When
`benchmark_col` is provided, the table also includes `beta_vs_bm` and
`info_ratio_vs_bm`.
"""
rows: list[dict[str, object]] = []
for label, start, end in buckets:
s = _date_or_none(start)
e = _date_or_none(end)
sub = df.filter((pl.col(timestamp_col) >= s) & (pl.col(timestamp_col) <= e))
n = sub.height
if n == 0:
rows.append({"bucket": label, "n": 0})
continue
s_arr = sub[strategy_col].to_numpy()
cum = float(np.prod(1 + s_arr) - 1)
if periods_per_year > 0:
ann = (1 + cum) ** (periods_per_year / max(n, 1)) - 1
else:
ann = float("nan")
std = float(np.std(s_arr, ddof=1)) if n > 1 else 0.0
sharpe = (
float(np.mean(s_arr) / std * np.sqrt(periods_per_year)) if std > 0 else float("nan")
)
equity = np.cumprod(1 + s_arr)
peak = np.maximum.accumulate(equity)
max_dd = float(np.min(equity / peak - 1))
row: dict[str, object] = {
"bucket": label,
"n": n,
"cum_return": cum,
"ann_return": ann,
"sharpe": sharpe,
"max_drawdown": max_dd,
}
if benchmark_col is not None and benchmark_col in sub.columns:
b_arr = sub[benchmark_col].to_numpy()
diag = compute_benchmark_diagnostics(s_arr, b_arr, periods_per_year)
row["beta_vs_bm"] = diag["beta"]
row["info_ratio_vs_bm"] = diag["info_ratio"]
rows.append(row)
return pl.DataFrame(rows)
def compute_benchmark_diagnostics(
strategy: np.ndarray,
benchmark: np.ndarray,
periods_per_year: int = 252,
) -> dict[str, float | int]:
"""Information ratio, beta, correlation, tracking error vs benchmark.
Inputs are aligned period returns; lengths must match. NaNs are
dropped pairwise. Returns NaN for any metric whose inputs are
degenerate (all-NaN or zero variance).
"""
s = np.asarray(strategy, dtype=float)
b = np.asarray(benchmark, dtype=float)
if s.shape != b.shape:
raise ValueError(f"shape mismatch: strategy {s.shape} vs benchmark {b.shape}")
mask = np.isfinite(s) & np.isfinite(b)
s, b = s[mask], b[mask]
n = int(s.size)
out: dict[str, float | int] = {"n": n}
if n < 2:
return {
**out,
"info_ratio": float("nan"),
"beta": float("nan"),
"correlation": float("nan"),
"tracking_error": float("nan"),
}
excess = s - b
te_period = float(np.std(excess, ddof=1))
te_ann = te_period * np.sqrt(periods_per_year)
if te_period > 0:
ir = float(np.mean(excess) * periods_per_year / te_ann)
else:
ir = float("nan")
var_b = float(np.var(b, ddof=1))
beta = float(np.cov(s, b, ddof=1)[0, 1] / var_b) if var_b > 0 else float("nan")
if np.std(s, ddof=1) > 0 and np.std(b, ddof=1) > 0:
corr = float(np.corrcoef(s, b)[0, 1])
else:
corr = float("nan")
return {
"n": n,
"info_ratio": ir,
"beta": beta,
"correlation": corr,
"tracking_error": te_ann,
}
```Shown in full with attribution under the source's licence. Licence: MIT
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.