Coverage for src/jquantstats/_stats/_basic_core.py: 100%
144 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-06 04:52 +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 # ── Win / loss metrics ────────────────────────────────────────────────────
219 @columnwise_stat
220 def payoff_ratio(self, series: pl.Series) -> float:
221 """Measure the payoff ratio.
223 The payoff ratio is calculated as average win / abs(average loss).
225 Args:
226 series (pl.Series): The series to calculate payoff ratio for.
228 Returns:
229 float: The payoff ratio value.
231 """
232 avg_win = self._mean_positive_expr(series)
233 avg_loss = float(np.abs(self._mean_negative_expr(series)))
234 return avg_win / avg_loss
236 @columnwise_stat
237 def profit_ratio(self, series: pl.Series) -> float:
238 """Measure the profit ratio.
240 The profit ratio is calculated as win ratio / loss ratio.
242 Args:
243 series (pl.Series): The series to calculate profit ratio for.
245 Returns:
246 float: The profit ratio value.
249 Returns NaN when:
250 ``float("nan")`` when the series has no wins or no losses.
251 """
252 wins = series.filter(series >= 0)
253 losses = self._negative(series)
255 # Filtering can legitimately leave no wins or no losses for one-sided return series.
256 if wins.is_empty() or losses.is_empty():
257 return float("nan") # indeterminate: no wins or no losses
259 win_mean = _mean(wins)
260 loss_mean = _mean(losses)
261 win_ratio = float(np.abs(win_mean / wins.count()))
262 loss_ratio = float(np.abs(loss_mean / losses.count()))
264 return win_ratio / loss_ratio
266 @columnwise_stat
267 def profit_factor(self, series: pl.Series) -> float:
268 """Measure the profit factor.
270 The profit factor is calculated as wins / loss.
272 Args:
273 series (pl.Series): The series to calculate profit factor for.
275 Returns:
276 float: The profit factor value.
278 """
279 wins = self._positive(series)
280 losses = self._negative(series)
281 wins_sum = wins.sum()
282 losses_sum = losses.sum()
284 return float(np.abs(float(wins_sum) / float(losses_sum)))
286 # ── Risk metrics ──────────────────────────────────────────────────────────
288 @columnwise_stat
289 def value_at_risk(self, series: pl.Series, sigma: float = 1.0, alpha: float = 0.05) -> float:
290 """Calculate the daily value-at-risk.
292 Uses variance-covariance calculation with confidence level.
294 Args:
295 series (pl.Series): The series to calculate value at risk for.
296 alpha (float, optional): Confidence level. Defaults to 0.05.
297 sigma (float, optional): Standard deviation multiplier. Defaults to 1.0.
299 Returns:
300 float: The value at risk.
302 """
303 mean_val = _mean(series)
304 std_val = cast(float, series.std())
305 mu = mean_val
306 sigma *= std_val if std_val is not None else 0.0
308 return self._gaussian_quantile(alpha, mu, sigma)
310 @columnwise_stat
311 def _conditional_value_at_risk_impl(self, series: pl.Series, sigma: float = 1.0, alpha: float = 0.05) -> float:
312 """Inner per-series implementation of conditional value-at-risk."""
313 mean_val = _mean(series)
314 std_val = cast(float, series.std())
315 mu = mean_val
316 sigma *= std_val if std_val is not None else 0.0
318 var = self._gaussian_quantile(alpha, mu, sigma)
320 # Compute mean of returns less than or equal to VaR
321 # Cast to Any or pl.Series to suppress Ty error
322 # Cast the mask to pl.Expr to satisfy type checker
323 mask = cast(Iterable[bool], series < var)
324 return _mean(series.filter(mask))
326 def conditional_value_at_risk(
327 self, sigma: float = 1.0, confidence: float = 0.95, **kwargs: float
328 ) -> dict[str, float]:
329 """Calculate the conditional value-at-risk (CVaR / Expected Shortfall).
331 Also known as CVaR or expected shortfall, calculated for each numeric column.
333 Args:
334 sigma (float, optional): Standard deviation multiplier. Defaults to 1.0.
335 confidence (float, optional): Confidence level (e.g. 0.95 for 95 %).
336 Converted internally to ``alpha = 1 - confidence``. Defaults to 0.95.
337 alpha (float, optional): Tail probability (lower tail). ``alpha`` is the
338 probability mass in the *loss* tail, so ``alpha = 1 - confidence``.
339 For example, a 95 % confidence level corresponds to ``alpha = 0.05``
340 (the default).
341 **kwargs: Legacy keyword arguments. Passing ``confidence`` (e.g.
342 ``confidence=0.95``) is accepted for backwards compatibility with
343 QuantStats but emits a `DeprecationWarning`. Use
344 ``alpha = 1 - confidence`` instead.
346 Returns:
347 dict[str, float]: The conditional value at risk per asset column.
349 Raises:
350 TypeError: If unexpected keyword arguments are passed.
352 """
353 return self._conditional_value_at_risk_impl(sigma=sigma, alpha=1.0 - confidence)
355 @staticmethod
356 def _drawdown_with_baseline(series: pl.Series) -> pl.Series:
357 """Compute drawdown series with a phantom zero-return baseline prepended.
359 Matches the quantstats convention: a negative first return is treated as
360 a drawdown from the initial capital of 1.0, not as the new high-water mark.
361 """
362 extended = pl.concat([pl.Series([0.0]), series.cast(pl.Float64)])
363 nav = (1.0 + extended).cum_prod()
364 hwm = nav.cum_max()
365 # The phantom baseline pins nav[0] = 1.0, so hwm >= 1.0 throughout and
366 # the 1e-10 floor is purely defensive (unreachable); a -100 % return
367 # correctly reports as a full drawdown of 1.0 here.
368 dd = ((hwm - nav) / hwm.clip(lower_bound=1e-10)).clip(lower_bound=0.0)
369 return dd[1:] # drop phantom point
371 @staticmethod
372 def _ulcer_index_series(series: pl.Series) -> float:
373 """Compute ulcer index for a single returns series."""
374 dd = _BasicCoreMixin._drawdown_with_baseline(series)
375 n = series.len()
376 return float(np.sqrt(float((dd**2).sum()) / (n - 1)))
378 @columnwise_stat
379 def ulcer_index(self, series: pl.Series) -> float:
380 """Calculate the Ulcer Index (downside risk measurement).
382 Measures the depth and duration of drawdowns as the root mean square
383 of squared drawdowns: sqrt(sum(dd²) / (n - 1)).
385 Args:
386 series (pl.Series): The series to calculate ulcer index for.
388 Returns:
389 float: Ulcer Index value.
391 """
392 return self._ulcer_index_series(series)
394 @columnwise_stat
395 def ulcer_performance_index(self, series: pl.Series, rf: float = 0.0) -> float:
396 """Calculate the Ulcer Performance Index (UPI).
398 Risk-adjusted return using Ulcer Index as the risk measure:
399 (compounded_return - rf) / ulcer_index.
401 Args:
402 series (pl.Series): The series to calculate UPI for.
403 rf (float): Risk-free rate. Defaults to 0.
405 Returns:
406 float: Ulcer Performance Index.
409 Returns NaN when:
410 ``float("nan")`` when the ulcer index is zero (no drawdowns).
411 """
412 comp = _comp_return(series)
413 ui = self._ulcer_index_series(series)
414 return float("nan") if ui == 0 else (comp - rf) / ui
416 @columnwise_stat
417 def serenity_index(self, series: pl.Series, rf: float = 0.0) -> float:
418 """Calculate the Serenity Index.
420 Combines the Ulcer Index with a CVaR-based pitfall measure:
421 (sum_returns - rf) / (ulcer_index * pitfall), where
422 pitfall = -CVaR(drawdowns) / std(returns).
424 Args:
425 series (pl.Series): The series to calculate serenity index for.
426 rf (float): Risk-free rate. Defaults to 0.
428 Returns:
429 float: Serenity Index.
432 Returns NaN when:
433 ``float("nan")`` when the returns have zero (or undefined) standard
434 deviation or the denominator ``ulcer_index * pitfall`` is zero.
435 """
436 std_val = cast(float, series.std())
437 if not std_val:
438 return float("nan") # indeterminate: zero variance
440 # Negate drawdowns to match quantstats sign convention (negative = below peak)
441 dd_neg = -self._drawdown_with_baseline(series)
442 mu = _mean(dd_neg)
443 sigma = cast(float, dd_neg.std())
444 var_threshold = self._gaussian_quantile(0.05, mu, sigma)
445 mask = cast(Iterable[bool], dd_neg < var_threshold)
446 cvar_val = _mean(dd_neg.filter(mask))
448 pitfall = -cvar_val / std_val
449 ui = self._ulcer_index_series(series)
450 denominator = ui * pitfall
451 return float("nan") if denominator == 0 else (float(series.sum()) - rf) / denominator
453 @columnwise_stat
454 def win_rate(self, series: pl.Series) -> float:
455 """Calculate the win ratio for a period.
457 Args:
458 series (pl.Series): The series to calculate win rate for.
460 Returns:
461 float: The win rate value.
463 """
464 num_pos = self._positive(series).count()
465 num_nonzero = series.filter(series != 0).count()
466 return float(num_pos / num_nonzero)