Coverage for src/jquantstats/_stats/_basic_core.py: 100%
159 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-23 04:11 +0000
1"""Core descriptive, return, volatility and risk statistics.
3The `_BasicCoreMixin` here holds the leaf statistics that the composite
4ratios in `_basic` build on, plus the shared static helpers.
5"""
7from __future__ import annotations
9from collections.abc import Iterable
10from typing import TYPE_CHECKING, cast
12import numpy as np
13import polars as pl
14from scipy.stats import norm
16from ._core import _mean, columnwise_stat
17from ._internals import _annualization_factor, _comp_return
19if TYPE_CHECKING:
20 from ..data import Data
22# ── Basic statistics mixin ───────────────────────────────────────────────────
25class _BasicCoreMixin:
26 """Mixin providing basic return/risk and win/loss financial statistics.
28 Covers: basic statistics (skew, kurtosis, avg return/win/loss), volatility,
29 win/loss metrics (payoff ratio, profit factor), and risk metrics (VaR, CVaR,
30 win rate, kelly criterion, best/worst, exposure).
31 """
33 _data: Data
34 all: pl.DataFrame
36 if TYPE_CHECKING:
37 from .._protocol import DataLike
39 data: DataLike
41 @staticmethod
42 def _positive(series: pl.Series) -> pl.Series:
43 """Return only the positive values in *series*."""
44 return series.filter(series > 0)
46 @staticmethod
47 def _negative(series: pl.Series) -> pl.Series:
48 """Return only the negative values in *series*."""
49 return series.filter(series < 0)
51 @staticmethod
52 def _mean_positive_expr(series: pl.Series) -> float:
53 """Return the mean of all positive values in *series*, or NaN if none exist."""
54 return _mean(_BasicCoreMixin._positive(series))
56 @staticmethod
57 def _mean_negative_expr(series: pl.Series) -> float:
58 """Return the mean of all negative values in *series*, or NaN if none exist."""
59 return _mean(_BasicCoreMixin._negative(series))
61 @staticmethod
62 def _gaussian_quantile(alpha: float, mu: float, sigma: float) -> float:
63 """Gaussian inverse-CDF (``norm.ppf``) returning NaN for a zero-scale input.
65 ``norm.ppf(alpha, mu, 0.0)`` already returns ``nan`` for a degenerate
66 (zero-variance) distribution — but it emits an ``invalid value
67 encountered in multiply`` RuntimeWarning while doing so (``inf * 0``
68 internally). Degenerate scale arises for a single observation (undefined
69 std) or a constant series. Short-circuiting to ``float("nan")`` keeps the
70 exact same result while suppressing the spurious warning; downstream
71 masking relies on this NaN (Polars treats ``x < nan`` as ``True``).
72 """
73 return float("nan") if sigma == 0.0 else float(norm.ppf(alpha, mu, sigma))
75 # ── Basic statistics ──────────────────────────────────────────────────────
77 @columnwise_stat
78 def skew(self, series: pl.Series) -> int | float | None:
79 """Calculate skewness (asymmetry) for each numeric column.
81 Args:
82 series (pl.Series): The series to calculate skewness for.
84 Returns:
85 float: The skewness value.
87 """
88 return series.skew(bias=False)
90 @columnwise_stat
91 def kurtosis(self, series: pl.Series) -> int | float | None:
92 """Calculate the kurtosis of returns.
94 The degree to which a distribution peak compared to a normal distribution.
96 Args:
97 series (pl.Series): The series to calculate kurtosis for.
99 Returns:
100 float: The kurtosis value.
102 """
103 return series.kurtosis(bias=False)
105 @columnwise_stat
106 def avg_return(self, series: pl.Series) -> float:
107 """Calculate average return per non-zero value.
109 Args:
110 series (pl.Series): The series to calculate average return for.
112 Returns:
113 float: The average return value.
115 """
116 return _mean(series.filter(series.is_not_null() & (series != 0)))
118 @columnwise_stat
119 def avg_win(self, series: pl.Series) -> float:
120 """Calculate the average winning return/trade for an asset.
122 Args:
123 series (pl.Series): The series to calculate average win for.
125 Returns:
126 float: The average winning return.
128 """
129 return self._mean_positive_expr(series)
131 @columnwise_stat
132 def avg_loss(self, series: pl.Series) -> float:
133 """Calculate the average loss return/trade for a period.
135 Args:
136 series (pl.Series): The series to calculate average loss for.
138 Returns:
139 float: The average loss return.
141 """
142 return self._mean_negative_expr(series)
144 @columnwise_stat
145 def comp(self, series: pl.Series) -> float:
146 """Calculate the total compounded return over the full period.
148 Computed as product(1 + r) - 1.
150 Args:
151 series (pl.Series): The series to calculate compounded return for.
153 Returns:
154 float: Total compounded return.
156 """
157 return _comp_return(series)
159 @columnwise_stat
160 def geometric_mean(self, series: pl.Series, periods: int | float | None = None, annualize: bool = False) -> float:
161 """Calculate the geometric mean of returns.
163 Computed as the per-period geometric average: (∏(1 + rᵢ))^(1/n) - 1.
164 When annualized, raises to the power of periods_per_year instead of 1/n.
166 Args:
167 series (pl.Series): The series to calculate geometric mean for.
168 periods (int | float, optional): Periods per year for annualization. Defaults to periods_per_year.
169 annualize (bool): Whether to annualize the result. Defaults to False.
171 Returns:
172 float: The geometric mean return.
175 Returns NaN when:
176 ``float("nan")`` when the series has no non-null observations or the
177 compounded return ``product(1 + r)`` is non-positive.
178 """
179 clean = series.drop_nulls().cast(pl.Float64)
180 n = clean.len()
181 if n == 0:
182 return float("nan") # indeterminate: no observations
183 compound = float((1.0 + clean).product())
184 if compound <= 0:
185 return float("nan") # indeterminate: non-positive compound return
186 exponent = (periods or self._data._periods_per_year) / n if annualize else (1.0 / n)
187 return float(compound**exponent) - 1.0
189 # ── Volatility & risk ─────────────────────────────────────────────────────
191 @columnwise_stat
192 def volatility(self, series: pl.Series, periods: int | float | None = None, annualize: bool = True) -> float:
193 """Calculate the volatility of returns.
195 - Std dev of returns
196 - Annualized by sqrt(periods) if `annualize` is True.
198 Args:
199 series (pl.Series): The series to calculate volatility for.
200 periods (int, optional): Number of periods per year. Defaults to 252.
201 annualize (bool, optional): Whether to annualize the result. Defaults to True.
203 Returns:
204 float: The volatility value.
206 """
207 raw_periods = periods or self._data._periods_per_year
209 # Ensure it's numeric
210 if not isinstance(raw_periods, int | float):
211 raise TypeError(f"Expected int or float for periods, got {type(raw_periods).__name__}") # noqa: TRY003
213 factor = _annualization_factor(raw_periods) if annualize else 1.0
214 std_val = cast(float, series.std())
215 return (std_val if std_val is not None else 0.0) * factor
217 @columnwise_stat
218 def mad(self, series: pl.Series, periods: int | float | None = None, annualize: bool = True) -> float:
219 """Calculate the Mean Absolute Deviation (MAD) of returns.
221 MAD is the mean of absolute deviations from the mean return:
222 mean(|r - mean(r)|). It is a robust measure of dispersion less
223 sensitive to outliers than standard deviation.
225 - Annualized by sqrt(periods) if `annualize` is True.
227 Args:
228 series (pl.Series): The series to calculate MAD for.
229 periods (int, optional): Number of periods per year. Defaults to periods_per_year.
230 annualize (bool, optional): Whether to annualize the result. Defaults to True.
232 Returns:
233 float: The MAD value.
234 """
235 raw_periods = periods or self._data._periods_per_year
237 if not isinstance(raw_periods, int | float):
238 raise TypeError(f"Expected int or float for periods, got {type(raw_periods).__name__}") # noqa: TRY003
240 factor = _annualization_factor(raw_periods) if annualize else 1.0
241 mean_val = _mean(series)
242 mad_val = cast(float, (series - mean_val).abs().mean())
243 return (mad_val if mad_val is not None else 0.0) * factor
245 # ── Win / loss metrics ────────────────────────────────────────────────────
247 @columnwise_stat
248 def payoff_ratio(self, series: pl.Series) -> float:
249 """Measure the payoff ratio.
251 The payoff ratio is calculated as average win / abs(average loss).
253 Args:
254 series (pl.Series): The series to calculate payoff ratio for.
256 Returns:
257 float: The payoff ratio value.
259 """
260 avg_win = self._mean_positive_expr(series)
261 avg_loss = float(np.abs(self._mean_negative_expr(series)))
262 return avg_win / avg_loss
264 @columnwise_stat
265 def profit_ratio(self, series: pl.Series) -> float:
266 """Measure the profit ratio.
268 The profit ratio is calculated as win ratio / loss ratio.
270 Args:
271 series (pl.Series): The series to calculate profit ratio for.
273 Returns:
274 float: The profit ratio value.
277 Returns NaN when:
278 ``float("nan")`` when the series has no wins or no losses.
279 """
280 wins = series.filter(series >= 0)
281 losses = self._negative(series)
283 # Filtering can legitimately leave no wins or no losses for one-sided return series.
284 if wins.is_empty() or losses.is_empty():
285 return float("nan") # indeterminate: no wins or no losses
287 win_mean = _mean(wins)
288 loss_mean = _mean(losses)
289 win_ratio = float(np.abs(win_mean / wins.count()))
290 loss_ratio = float(np.abs(loss_mean / losses.count()))
292 return win_ratio / loss_ratio
294 @columnwise_stat
295 def profit_factor(self, series: pl.Series) -> float:
296 """Measure the profit factor.
298 The profit factor is calculated as wins / loss.
300 Args:
301 series (pl.Series): The series to calculate profit factor for.
303 Returns:
304 float: The profit factor value.
306 """
307 wins = self._positive(series)
308 losses = self._negative(series)
309 wins_sum = wins.sum()
310 losses_sum = losses.sum()
312 return float(np.abs(float(wins_sum) / float(losses_sum)))
314 # ── Risk metrics ──────────────────────────────────────────────────────────
316 @columnwise_stat
317 def value_at_risk(self, series: pl.Series, sigma: float = 1.0, alpha: float = 0.05) -> float:
318 """Calculate the daily value-at-risk.
320 Uses variance-covariance calculation with confidence level.
322 Args:
323 series (pl.Series): The series to calculate value at risk for.
324 alpha (float, optional): Confidence level. Defaults to 0.05.
325 sigma (float, optional): Standard deviation multiplier. Defaults to 1.0.
327 Returns:
328 float: The value at risk.
330 """
331 mean_val = _mean(series)
332 std_val = cast(float, series.std())
333 mu = mean_val
334 sigma *= std_val if std_val is not None else 0.0
336 return self._gaussian_quantile(alpha, mu, sigma)
338 @columnwise_stat
339 def _conditional_value_at_risk_impl(self, series: pl.Series, sigma: float = 1.0, alpha: float = 0.05) -> float:
340 """Inner per-series implementation of conditional value-at-risk."""
341 mean_val = _mean(series)
342 std_val = cast(float, series.std())
343 mu = mean_val
344 sigma *= std_val if std_val is not None else 0.0
346 var = self._gaussian_quantile(alpha, mu, sigma)
348 # Compute mean of returns less than or equal to VaR
349 # Cast to Any or pl.Series to suppress Ty error
350 # Cast the mask to pl.Expr to satisfy type checker
351 mask = cast(Iterable[bool], series < var)
352 return _mean(series.filter(mask))
354 def conditional_value_at_risk(
355 self,
356 sigma: float = 1.0,
357 confidence: float | None = None,
358 alpha: float | None = None,
359 ) -> dict[str, float]:
360 """Calculate the conditional value-at-risk (CVaR / Expected Shortfall).
362 Also known as CVaR or expected shortfall, calculated for each numeric column.
364 The tail can be specified either way round: ``confidence`` matches the
365 QuantStats spelling, ``alpha`` matches `value_at_risk` on this same
366 object. They are two names for one quantity (``alpha = 1 - confidence``),
367 so passing both is an error rather than a silent preference.
369 Args:
370 sigma (float, optional): Standard deviation multiplier. Defaults to 1.0.
371 confidence (float, optional): Confidence level (e.g. 0.95 for 95 %).
372 Converted internally to ``alpha = 1 - confidence``. Mutually
373 exclusive with ``alpha``.
374 alpha (float, optional): Tail probability in the *loss* tail (e.g. 0.05
375 for 95 % confidence). Mutually exclusive with ``confidence``.
376 Both defaulting to ``None`` selects ``alpha = 0.05``.
378 Returns:
379 dict[str, float]: The conditional value at risk per asset column.
381 Raises:
382 ValueError: If both ``confidence`` and ``alpha`` are given.
384 """
385 if confidence is not None and alpha is not None:
386 raise ValueError( # noqa: TRY003
387 f"Pass either confidence or alpha, not both "
388 f"(got confidence={confidence!r}, alpha={alpha!r}); alpha = 1 - confidence"
389 )
390 if confidence is not None:
391 alpha = 1.0 - confidence
392 elif alpha is None:
393 alpha = 0.05
395 return self._conditional_value_at_risk_impl(sigma=sigma, alpha=alpha)
397 @staticmethod
398 def _drawdown_with_baseline(series: pl.Series) -> pl.Series:
399 """Compute drawdown series with a phantom zero-return baseline prepended.
401 Matches the quantstats convention: a negative first return is treated as
402 a drawdown from the initial capital of 1.0, not as the new high-water mark.
403 """
404 extended = pl.concat([pl.Series([0.0]), series.cast(pl.Float64)])
405 nav = (1.0 + extended).cum_prod()
406 hwm = nav.cum_max()
407 # The phantom baseline pins nav[0] = 1.0, so hwm >= 1.0 throughout and
408 # the 1e-10 floor is purely defensive (unreachable); a -100 % return
409 # correctly reports as a full drawdown of 1.0 here.
410 dd = ((hwm - nav) / hwm.clip(lower_bound=1e-10)).clip(lower_bound=0.0)
411 return dd[1:] # drop phantom point
413 @staticmethod
414 def _ulcer_index_series(series: pl.Series) -> float:
415 """Compute ulcer index for a single returns series."""
416 dd = _BasicCoreMixin._drawdown_with_baseline(series)
417 n = series.len()
418 return float(np.sqrt(float((dd**2).sum()) / (n - 1)))
420 @columnwise_stat
421 def ulcer_index(self, series: pl.Series) -> float:
422 """Calculate the Ulcer Index (downside risk measurement).
424 Measures the depth and duration of drawdowns as the root mean square
425 of squared drawdowns: sqrt(sum(dd²) / (n - 1)).
427 Args:
428 series (pl.Series): The series to calculate ulcer index for.
430 Returns:
431 float: Ulcer Index value.
433 """
434 return self._ulcer_index_series(series)
436 @columnwise_stat
437 def ulcer_performance_index(self, series: pl.Series, rf: float = 0.0) -> float:
438 """Calculate the Ulcer Performance Index (UPI).
440 Risk-adjusted return using Ulcer Index as the risk measure:
441 (compounded_return - rf) / ulcer_index.
443 Args:
444 series (pl.Series): The series to calculate UPI for.
445 rf (float): Risk-free rate. Defaults to 0.
447 Returns:
448 float: Ulcer Performance Index.
451 Returns NaN when:
452 ``float("nan")`` when the ulcer index is zero (no drawdowns).
453 """
454 comp = _comp_return(series)
455 ui = self._ulcer_index_series(series)
456 return float("nan") if ui == 0 else (comp - rf) / ui
458 @columnwise_stat
459 def serenity_index(self, series: pl.Series, rf: float = 0.0) -> float:
460 """Calculate the Serenity Index.
462 Combines the Ulcer Index with a CVaR-based pitfall measure:
463 (sum_returns - rf) / (ulcer_index * pitfall), where
464 pitfall = -CVaR(drawdowns) / std(returns).
466 Args:
467 series (pl.Series): The series to calculate serenity index for.
468 rf (float): Risk-free rate. Defaults to 0.
470 Returns:
471 float: Serenity Index.
474 Returns NaN when:
475 ``float("nan")`` when the returns have zero (or undefined) standard
476 deviation or the denominator ``ulcer_index * pitfall`` is zero.
477 """
478 std_val = cast(float, series.std())
479 if not std_val:
480 return float("nan") # indeterminate: zero variance
482 # Negate drawdowns to match quantstats sign convention (negative = below peak)
483 dd_neg = -self._drawdown_with_baseline(series)
484 mu = _mean(dd_neg)
485 sigma = cast(float, dd_neg.std())
486 var_threshold = self._gaussian_quantile(0.05, mu, sigma)
487 mask = cast(Iterable[bool], dd_neg < var_threshold)
488 cvar_val = _mean(dd_neg.filter(mask))
490 pitfall = -cvar_val / std_val
491 ui = self._ulcer_index_series(series)
492 denominator = ui * pitfall
493 return float("nan") if denominator == 0 else (float(series.sum()) - rf) / denominator
495 @columnwise_stat
496 def win_rate(self, series: pl.Series) -> float:
497 """Calculate the win ratio for a period.
499 Args:
500 series (pl.Series): The series to calculate win rate for.
502 Returns:
503 float: The win rate value.
505 """
506 num_pos = self._positive(series).count()
507 num_nonzero = series.filter(series != 0).count()
508 return float(num_pos / num_nonzero)