Coverage for src/jquantstats/_stats/_performance.py: 100%
144 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-06 04: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 + 0.5·base² - skew·base + (kurt-3)/4·base²) / (n-1)).
275 Args:
276 base (float): Unannualized observed ratio (e.g. Sortino).
277 series (pl.Series): The original returns series (for moments and n).
279 Returns:
280 float: Probabilistic ratio in [0, 1].
283 Returns NaN when:
284 ``float("nan")`` when moments are missing, there are fewer than two
285 observations, or the estimated variance is non-positive.
286 """
287 n = series.count()
288 skew_val = series.skew(bias=False)
289 kurt_val = series.kurtosis(bias=False)
290 if skew_val is None or kurt_val is None or n <= 1:
291 return float("nan") # indeterminate: missing moments or insufficient data
292 variance = (1 + 0.5 * base**2 - float(skew_val) * base + ((float(kurt_val) - 3) / 4) * base**2) / (n - 1)
293 if variance <= 0:
294 return float("nan") # indeterminate: non-positive variance
295 return float(norm.cdf(base / np.sqrt(variance)))
297 @columnwise_stat
298 def probabilistic_sortino_ratio(self, series: pl.Series, periods: int | float | None = None) -> float:
299 """Calculate the Probabilistic Sortino Ratio.
301 The probability that the observed Sortino ratio is greater than zero,
302 accounting for estimation uncertainty via skewness and kurtosis.
304 Args:
305 series (pl.Series): The series to calculate the ratio for.
306 periods (int | float, optional): Accepted for API compatibility; has no effect
307 since the base ratio is un-annualized.
309 Returns:
310 float: Probabilistic Sortino ratio in [0, 1].
313 Returns NaN when:
314 ``float("nan")`` when the downside deviation is zero, moments are
315 missing, or the estimated variance is non-positive.
316 """
317 downside_deviation = _downside_deviation(series)
318 mean_f = _mean(series)
319 if downside_deviation == 0.0:
320 return float("nan") # indeterminate: zero downside deviation
321 base = float(mean_f / downside_deviation)
322 return self._probabilistic_ratio_from_base(base, series)
324 @columnwise_stat
325 def probabilistic_adjusted_sortino_ratio(self, series: pl.Series, periods: int | float | None = None) -> float:
326 """Calculate the Probabilistic Adjusted Sortino Ratio.
328 The probability that the observed adjusted Sortino ratio (divided by sqrt(2)
329 for Sharpe comparability) is greater than zero, accounting for estimation
330 uncertainty via skewness and kurtosis.
332 Args:
333 series (pl.Series): The series to calculate the ratio for.
334 periods (int | float, optional): Accepted for API compatibility; has no effect
335 since the base ratio is un-annualized.
337 Returns:
338 float: Probabilistic adjusted Sortino ratio in [0, 1].
341 Returns NaN when:
342 ``float("nan")`` when the downside deviation is zero, moments are
343 missing, or the estimated variance is non-positive.
344 """
345 downside_deviation = _downside_deviation(series)
346 mean_f = _mean(series)
347 if downside_deviation == 0.0:
348 return float("nan") # indeterminate: zero downside deviation
349 base = float(mean_f / downside_deviation) / np.sqrt(2)
350 return self._probabilistic_ratio_from_base(base, series)
352 def probabilistic_ratio(
353 self,
354 base: str | Callable[[pl.Series], float] = "sharpe",
355 ) -> dict[str, float]:
356 r"""Generic probabilistic ratio for any base metric.
358 Computes the probability that the observed ratio is greater than zero,
359 accounting for estimation uncertainty via skewness and kurtosis using
360 the Lopez de Prado (2018) framework.
362 Args:
363 base: Base ratio to use. Either:
365 - A string: ``'sharpe'``, ``'sortino'``, ``'adjusted_sortino'``.
366 - A callable ``(series: pl.Series) -> float`` returning the
367 **unannualized** ratio for a single series.
369 Returns:
370 dict[str, float]: Probabilistic ratio in ``[0, 1]`` per asset.
372 Raises:
373 ValueError: If *base* is an unrecognised string.
376 Returns NaN when:
377 Entries are ``float("nan")`` when the base ratio is undefined (zero
378 standard deviation / zero downside deviation), moments are missing, or
379 the estimated variance is non-positive.
380 """
382 def _sharpe_base(s: pl.Series) -> float:
383 """Return the per-period Sharpe ratio (mean / std, ddof=1) of *s*."""
384 mean_val = _mean(s)
385 std_val = cast(float, s.std(ddof=1))
386 if not std_val or std_val == 0:
387 return float("nan")
388 return mean_val / std_val
390 def _sortino_base(s: pl.Series) -> float:
391 """Return the per-period Sortino ratio (mean / downside_dev) of *s*."""
392 downside_sum = _to_float((s.filter(s < 0) ** 2).sum())
393 downside_dev = float(np.sqrt(downside_sum / s.count()))
394 if downside_dev == 0.0:
395 return float("nan")
396 return _mean(s) / downside_dev
398 _builtin: dict[str, Callable[[pl.Series], float]] = {
399 "sharpe": _sharpe_base,
400 "sortino": _sortino_base,
401 "adjusted_sortino": lambda s: _sortino_base(s) / float(np.sqrt(2)),
402 }
404 if isinstance(base, str):
405 if base not in _builtin:
406 raise ValueError(f"base must be one of {list(_builtin)}, got {base!r}") # noqa: TRY003
407 base_fn = _builtin[base]
408 else:
409 base_fn = base
411 result: dict[str, float] = {}
412 for col, series in self._data.items():
413 base_val = base_fn(series)
414 if np.isnan(base_val):
415 result[col] = float("nan")
416 else:
417 result[col] = _RiskStatsMixin._probabilistic_ratio_from_base(base_val, series)
418 return result
420 def smart_sharpe(self, periods: int | float | None = None) -> dict[str, float]:
421 """Calculate the Smart Sharpe ratio (Sharpe with autocorrelation penalty).
423 Divides the Sharpe ratio by the autocorrelation penalty to account for
424 return autocorrelation that can artificially inflate risk-adjusted metrics.
426 Args:
427 periods (int | float, optional): Number of periods per year. Defaults to periods_per_year.
429 Returns:
430 dict[str, float]: Dictionary mapping asset names to Smart Sharpe ratios.
432 """
433 sharpe_data = self.sharpe(periods=periods)
434 penalty_data = self.autocorr_penalty()
435 return {k: sharpe_data[k] / penalty_data[k] for k in sharpe_data}
437 def smart_sortino(self, periods: int | float | None = None) -> dict[str, float]:
438 """Calculate the Smart Sortino ratio (Sortino with autocorrelation penalty).
440 Divides the Sortino ratio by the autocorrelation penalty to account for
441 return autocorrelation that can artificially inflate risk-adjusted metrics.
443 Args:
444 periods (int | float, optional): Number of periods per year. Defaults to periods_per_year.
446 Returns:
447 dict[str, float]: Dictionary mapping asset names to Smart Sortino ratios.
449 """
450 sortino_data = self.sortino(periods=periods)
451 penalty_data = self.autocorr_penalty()
452 return {k: sortino_data[k] / penalty_data[k] for k in sortino_data}
454 def adjusted_sortino(self, periods: int | float | None = None) -> dict[str, float]:
455 """Calculate Jack Schwager's adjusted Sortino ratio.
457 This adjustment allows for direct comparison to Sharpe ratio.
458 See: https://archive.is/wip/2rwFW.
460 Args:
461 periods (int, optional): Number of periods per year. Defaults to 252.
463 Returns:
464 dict[str, float]: Dictionary mapping asset names to adjusted Sortino ratios.
466 """
467 sortino_data = self.sortino(periods=periods)
468 return {k: v / np.sqrt(2) for k, v in sortino_data.items()}