Coverage for src/jquantstats/_stats/_performance.py: 100%
145 statements
« prev ^ index » next coverage.py v7.17.1, created at 2026-11-03 15:52 +0000
1"""Risk-adjusted return ratios for financial data.
3This module owns the ratio family: Sharpe and Sortino, Omega, and the
4probabilistic / smart / adjusted variants derived from them. Two concerns that
5previously shared this module were split out, taking it from 714 lines to ~460:
7- concentration (HHI) → :mod:`jquantstats._stats._concentration`
8- benchmark-relative and factor metrics → :mod:`jquantstats._stats._benchmark`
10What remains is deliberately kept together: the ``probabilistic_*``, ``smart_*``
11and ``adjusted_*`` methods are all defined in terms of the base ``sharpe`` and
12``sortino`` ratios above them (via ``_probabilistic_ratio_from_base``), so
13splitting them further would separate derived metrics from the ones they derive
14from.
15"""
17from __future__ import annotations
19from collections.abc import Callable
20from typing import TYPE_CHECKING, cast
22import numpy as np
23import polars as pl
24from scipy.stats import norm
26from ._core import _mean, _std_is_negligible, _to_float, columnwise_stat
27from ._internals import _annualization_factor, _downside_deviation
29if TYPE_CHECKING:
30 from ..data import Data
32# ── Risk statistics mixin ────────────────────────────────────────────────────
35class _RiskStatsMixin:
36 """Mixin providing risk-adjusted return ratios.
38 Covers: Sharpe ratio, Sortino ratio, Omega, adjusted Sortino, and the
39 probabilistic / smart variants of each.
41 Cross-mixin dependencies:
42 - _BasicStatsMixin: geometric_mean, autocorr_penalty
44 Sibling mixins split out of this one, both composed into the same ``Stats``
45 class so the public API is unchanged:
46 - _ConcentrationStatsMixin: hhi_positive, hhi_negative
47 - _BenchmarkStatsMixin: r_squared, information_ratio, greeks, treynor_ratio
48 """
50 _data: Data
51 all: pl.DataFrame
53 if TYPE_CHECKING:
54 from .._protocol import DataLike
56 data: DataLike
58 def autocorr_penalty(self) -> dict[str, float]:
59 """Defined on _BasicStatsMixin."""
61 def geometric_mean(self) -> dict[str, float]:
62 """Defined on _BasicStatsMixin."""
64 # ── Sharpe & Sortino ──────────────────────────────────────────────────────
66 @columnwise_stat
67 def sharpe(self, series: pl.Series, periods: int | float | None = None) -> float:
68 """Calculate the Sharpe ratio of asset returns.
70 Args:
71 series (pl.Series): The series to calculate Sharpe ratio for.
72 periods (int, optional): Number of periods per year. Defaults to 252.
74 Returns:
75 float: The Sharpe ratio value.
78 Returns NaN when:
79 ``float("nan")`` when the standard deviation is missing (fewer than two
80 observations) or numerically negligible.
81 """
82 periods = periods or self._data._periods_per_year
84 std_val = cast(float | None, series.std(ddof=1))
85 mean_val = series.mean()
86 mean_f = cast(float, mean_val) if mean_val is not None else 0.0
88 if _std_is_negligible(std_val, mean_f):
89 return float("nan")
91 res = mean_f / cast(float, std_val)
92 factor = periods or 1
93 return float(res * _annualization_factor(factor))
95 @columnwise_stat
96 def sharpe_variance(self, series: pl.Series, periods: int | float | None = None) -> float:
97 r"""Calculate the asymptotic variance of the Sharpe Ratio.
99 .. math::
100 \text{Var}(SR) = \frac{1 + \frac{S \cdot SR}{2} + \frac{(K - 3) \cdot SR^2}{4}}{T}
102 where:
103 - \(S\) is the skewness of returns
104 - \(K\) is the kurtosis of returns
105 - \(SR\) is the Sharpe ratio (unannualized)
106 - \(T\) is the number of observations
108 Args:
109 series (pl.Series): The series to calculate Sharpe ratio variance for.
110 periods (int | float, optional): Number of periods per year. Defaults to data periods.
112 Returns:
113 float: The asymptotic variance of the Sharpe ratio.
114 If number of periods per year is provided or inferred from the data, the result is annualized.
117 Returns NaN when:
118 ``float("nan")`` when the standard deviation is zero/missing or
119 skewness/kurtosis cannot be computed.
120 """
121 t = series.count()
122 mean_val = _mean(series)
123 std_val = cast(float, series.std(ddof=1))
124 if std_val is None or std_val == 0:
125 return float("nan") # indeterminate: zero or missing standard deviation
126 sr = mean_val / std_val
128 skew_val = series.skew(bias=False)
129 kurt_val = series.kurtosis(bias=False)
131 if skew_val is None or kurt_val is None:
132 return float("nan") # indeterminate: missing moments
133 # Base variance calculation using unannualized Sharpe ratio
134 # Formula: (1 + skew*SR/2 + (kurt-3)*SR²/4) / T
135 base_variance = (1 + (float(skew_val) * sr) / 2 + ((float(kurt_val) - 3) / 4) * sr**2) / t
136 # Annualize by scaling with the number of periods
137 periods = periods or self._data._periods_per_year
138 factor = periods or 1
139 return float(base_variance * _annualization_factor(factor, sqrt=False))
141 @columnwise_stat
142 def probabilistic_sharpe_ratio(self, series: pl.Series) -> float:
143 r"""Calculate the probabilistic sharpe ratio (PSR).
145 Args:
146 series (pl.Series): The series to calculate probabilistic Sharpe ratio for.
148 Returns:
149 float: Probabilistic Sharpe Ratio.
151 Note:
152 PSR is the probability that the observed Sharpe ratio is greater than a
153 given benchmark Sharpe ratio.
156 Returns NaN when:
157 ``float("nan")`` when the standard deviation is zero/missing, moments
158 are missing, or the estimated Sharpe variance is non-positive.
159 """
160 t = series.count()
162 # Calculate observed unannualized Sharpe ratio
163 mean_val = _mean(series)
164 std_val = cast(float, series.std(ddof=1))
165 if std_val is None or std_val == 0:
166 return float("nan") # indeterminate: zero or missing standard deviation
167 # Unannualized observed Sharpe ratio
168 observed_sr = mean_val / std_val
170 skew_val = series.skew(bias=False)
171 kurt_val = series.kurtosis(bias=False)
173 if skew_val is None or kurt_val is None:
174 return float("nan") # indeterminate: missing moments
176 benchmark_sr = 0.0
177 # Calculate variance using unannualized benchmark Sharpe ratio
178 var_bench_sr = (1 + (float(skew_val) * benchmark_sr) / 2 + ((float(kurt_val) - 3) / 4) * benchmark_sr**2) / t
180 if var_bench_sr <= 0:
181 return float("nan") # pragma: no cover # indeterminate: non-positive variance
182 return float(norm.cdf((observed_sr - benchmark_sr) / np.sqrt(var_bench_sr)))
184 @columnwise_stat
185 def sortino(self, series: pl.Series, periods: int | float | None = None) -> float:
186 """Calculate the Sortino ratio.
188 The Sortino ratio is the mean return divided by downside deviation.
189 Based on Red Rock Capital's Sortino ratio paper.
191 Args:
192 series (pl.Series): The series to calculate Sortino ratio for.
193 periods (int, optional): Number of periods per year. Defaults to 252.
195 Returns:
196 float: The Sortino ratio value.
199 Returns NaN when:
200 ``float("nan")`` when both the mean return and the downside deviation
201 are zero.
202 """
203 periods = periods or self._data._periods_per_year
204 downside_deviation = _downside_deviation(series)
205 mean_f = _mean(series)
206 if downside_deviation == 0.0:
207 if mean_f > 0:
208 return float("inf")
209 elif mean_f < 0: # pragma: no cover # unreachable: no negatives ⟹ mean ≥ 0
210 return float("-inf")
211 else:
212 return float("nan") # indeterminate: zero mean and zero downside deviation
213 ratio = mean_f / downside_deviation
214 return float(ratio * _annualization_factor(periods))
216 @columnwise_stat
217 def omega(
218 self,
219 series: pl.Series,
220 rf: float = 0.0,
221 required_return: float = 0.0,
222 periods: int | float | None = None,
223 ) -> float:
224 """Calculate the Omega ratio.
226 The Omega ratio is the probability-weighted ratio of gains to losses
227 relative to a threshold return. It is computed as the sum of returns
228 above the threshold divided by the absolute sum of returns below it.
230 Args:
231 series (pl.Series): The series to calculate Omega ratio for.
232 rf (float): Annualised risk-free rate. Defaults to 0.0.
233 required_return (float): Annualised minimum acceptable return
234 threshold. Defaults to 0.0.
235 periods (int | float | None): Number of periods per year. Defaults
236 to the value inferred from the data.
238 Returns:
239 float: The Omega ratio, or NaN when the denominator is zero or
240 when ``required_return <= -1``.
242 Note:
243 See https://en.wikipedia.org/wiki/Omega_ratio for details.
245 """
246 if required_return <= -1:
247 return float("nan")
249 periods = periods or self._data._periods_per_year
251 # Subtract per-period risk-free rate from returns when rf is non-zero.
252 if rf != 0.0:
253 rf_per_period = float((1.0 + rf) ** (1.0 / periods) - 1.0)
254 series = series - rf_per_period
256 # Convert annualised required return to a per-period threshold.
257 return_threshold = float((1.0 + required_return) ** (1.0 / periods) - 1.0)
259 returns_less_thresh = series - return_threshold
261 numer = float(returns_less_thresh.filter(returns_less_thresh > 0.0).sum())
262 denom = float(-returns_less_thresh.filter(returns_less_thresh < 0.0).sum())
264 if denom <= 0.0:
265 return float("nan")
266 return numer / denom
268 @staticmethod
269 def _probabilistic_ratio_from_base(base: float, series: pl.Series) -> float:
270 """Compute the probabilistic ratio given an observed unannualized base ratio.
272 Uses the formula: norm.cdf(base / sigma), where
273 sigma = sqrt((1 - skew·base + (kurt-1)/4·base²) / (n-1)) with *raw*
274 kurtosis (Bailey & Lopez de Prado, 2012). Polars returns excess
275 kurtosis, so the ``(kurt-1)/4`` term is ``(excess+2)/4``; it reduces to
276 Lo's ``(1 + base²/2) / (n-1)`` for normally distributed returns.
278 Args:
279 base (float): Unannualized observed ratio (e.g. Sortino).
280 series (pl.Series): The original returns series (for moments and n).
282 Returns:
283 float: Probabilistic ratio in [0, 1].
286 Returns NaN when:
287 ``float("nan")`` when moments are missing, there are fewer than two
288 observations, or the estimated variance is non-positive.
289 """
290 n = series.count()
291 skew_val = series.skew(bias=False)
292 kurt_val = series.kurtosis(bias=False)
293 if skew_val is None or kurt_val is None or n <= 1:
294 return float("nan") # indeterminate: missing moments or insufficient data
295 raw_kurt = float(kurt_val) + 3 # polars returns excess kurtosis
296 variance = (1 - float(skew_val) * base + ((raw_kurt - 1) / 4) * base**2) / (n - 1)
297 if variance <= 0:
298 return float("nan") # indeterminate: non-positive variance
299 return float(norm.cdf(base / np.sqrt(variance)))
301 @columnwise_stat
302 def probabilistic_sortino_ratio(self, series: pl.Series, periods: int | float | None = None) -> float:
303 """Calculate the Probabilistic Sortino Ratio.
305 The probability that the observed Sortino ratio is greater than zero,
306 accounting for estimation uncertainty via skewness and kurtosis.
308 Args:
309 series (pl.Series): The series to calculate the ratio for.
310 periods (int | float, optional): Accepted for API compatibility; has no effect
311 since the base ratio is un-annualized.
313 Returns:
314 float: Probabilistic Sortino ratio in [0, 1].
317 Returns NaN when:
318 ``float("nan")`` when the downside deviation is zero, moments are
319 missing, or the estimated variance is non-positive.
320 """
321 downside_deviation = _downside_deviation(series)
322 mean_f = _mean(series)
323 if downside_deviation == 0.0:
324 return float("nan") # indeterminate: zero downside deviation
325 base = float(mean_f / downside_deviation)
326 return self._probabilistic_ratio_from_base(base, series)
328 @columnwise_stat
329 def probabilistic_adjusted_sortino_ratio(self, series: pl.Series, periods: int | float | None = None) -> float:
330 """Calculate the Probabilistic Adjusted Sortino Ratio.
332 The probability that the observed adjusted Sortino ratio (divided by sqrt(2)
333 for Sharpe comparability) is greater than zero, accounting for estimation
334 uncertainty via skewness and kurtosis.
336 Args:
337 series (pl.Series): The series to calculate the ratio for.
338 periods (int | float, optional): Accepted for API compatibility; has no effect
339 since the base ratio is un-annualized.
341 Returns:
342 float: Probabilistic adjusted Sortino ratio in [0, 1].
345 Returns NaN when:
346 ``float("nan")`` when the downside deviation is zero, moments are
347 missing, or the estimated variance is non-positive.
348 """
349 downside_deviation = _downside_deviation(series)
350 mean_f = _mean(series)
351 if downside_deviation == 0.0:
352 return float("nan") # indeterminate: zero downside deviation
353 base = float(mean_f / downside_deviation) / np.sqrt(2)
354 return self._probabilistic_ratio_from_base(base, series)
356 def probabilistic_ratio(
357 self,
358 base: str | Callable[[pl.Series], float] = "sharpe",
359 ) -> dict[str, float]:
360 r"""Generic probabilistic ratio for any base metric.
362 Computes the probability that the observed ratio is greater than zero,
363 accounting for estimation uncertainty via skewness and kurtosis using
364 the Lopez de Prado (2018) framework.
366 Args:
367 base: Base ratio to use. Either:
369 - A string: ``'sharpe'``, ``'sortino'``, ``'adjusted_sortino'``.
370 - A callable ``(series: pl.Series) -> float`` returning the
371 **unannualized** ratio for a single series.
373 Returns:
374 dict[str, float]: Probabilistic ratio in ``[0, 1]`` per asset.
376 Raises:
377 ValueError: If *base* is an unrecognised string.
380 Returns NaN when:
381 Entries are ``float("nan")`` when the base ratio is undefined (zero
382 standard deviation / zero downside deviation), moments are missing, or
383 the estimated variance is non-positive.
384 """
386 def _sharpe_base(s: pl.Series) -> float:
387 """Return the per-period Sharpe ratio (mean / std, ddof=1) of *s*."""
388 mean_val = _mean(s)
389 std_val = cast(float, s.std(ddof=1))
390 if not std_val or std_val == 0:
391 return float("nan")
392 return mean_val / std_val
394 def _sortino_base(s: pl.Series) -> float:
395 """Return the per-period Sortino ratio (mean / downside_dev) of *s*."""
396 downside_sum = _to_float((s.filter(s < 0) ** 2).sum())
397 downside_dev = float(np.sqrt(downside_sum / s.count()))
398 if downside_dev == 0.0:
399 return float("nan")
400 return _mean(s) / downside_dev
402 _builtin: dict[str, Callable[[pl.Series], float]] = {
403 "sharpe": _sharpe_base,
404 "sortino": _sortino_base,
405 "adjusted_sortino": lambda s: _sortino_base(s) / float(np.sqrt(2)),
406 }
408 if isinstance(base, str):
409 if base not in _builtin:
410 raise ValueError(f"base must be one of {list(_builtin)}, got {base!r}") # noqa: TRY003
411 base_fn = _builtin[base]
412 else:
413 base_fn = base
415 result: dict[str, float] = {}
416 for col, series in self._data.items():
417 base_val = base_fn(series)
418 if np.isnan(base_val):
419 result[col] = float("nan")
420 else:
421 result[col] = _RiskStatsMixin._probabilistic_ratio_from_base(base_val, series)
422 return result
424 def smart_sharpe(self, periods: int | float | None = None) -> dict[str, float]:
425 """Calculate the Smart Sharpe ratio (Sharpe with autocorrelation penalty).
427 Divides the Sharpe ratio by the autocorrelation penalty to account for
428 return autocorrelation that can artificially inflate risk-adjusted metrics.
430 Args:
431 periods (int | float, optional): Number of periods per year. Defaults to periods_per_year.
433 Returns:
434 dict[str, float]: Dictionary mapping asset names to Smart Sharpe ratios.
436 """
437 sharpe_data = self.sharpe(periods=periods)
438 penalty_data = self.autocorr_penalty()
439 return {k: sharpe_data[k] / penalty_data[k] for k in sharpe_data}
441 def smart_sortino(self, periods: int | float | None = None) -> dict[str, float]:
442 """Calculate the Smart Sortino ratio (Sortino with autocorrelation penalty).
444 Divides the Sortino ratio by the autocorrelation penalty to account for
445 return autocorrelation that can artificially inflate risk-adjusted metrics.
447 Args:
448 periods (int | float, optional): Number of periods per year. Defaults to periods_per_year.
450 Returns:
451 dict[str, float]: Dictionary mapping asset names to Smart Sortino ratios.
453 """
454 sortino_data = self.sortino(periods=periods)
455 penalty_data = self.autocorr_penalty()
456 return {k: sortino_data[k] / penalty_data[k] for k in sortino_data}
458 def adjusted_sortino(self, periods: int | float | None = None) -> dict[str, float]:
459 """Calculate Jack Schwager's adjusted Sortino ratio.
461 This adjustment allows for direct comparison to Sharpe ratio.
462 See: https://archive.is/wip/2rwFW.
464 Args:
465 periods (int, optional): Number of periods per year. Defaults to 252.
467 Returns:
468 dict[str, float]: Dictionary mapping asset names to adjusted Sortino ratios.
470 """
471 sortino_data = self.sortino(periods=periods)
472 return {k: v / np.sqrt(2) for k, v in sortino_data.items()}