Coverage for src/jquantstats/_stats/_basic_core.py: 100%
164 statements
« prev ^ index » next coverage.py v7.17.1, created at 2026-11-03 15: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 typing import TYPE_CHECKING, cast
11import numpy as np
12import polars as pl
13from scipy.stats import norm
15from ._core import _mean, columnwise_stat
16from ._internals import _annualization_factor, _comp_return
18if TYPE_CHECKING:
19 from ..data import Data
21# ── Basic statistics mixin ───────────────────────────────────────────────────
24class _BasicCoreMixin:
25 """Mixin providing basic return/risk and win/loss financial statistics.
27 Covers: basic statistics (skew, kurtosis, avg return/win/loss), volatility,
28 win/loss metrics (payoff ratio, profit factor), and risk metrics (VaR, CVaR,
29 win rate, kelly criterion, best/worst, exposure).
30 """
32 _data: Data
33 all: pl.DataFrame
35 if TYPE_CHECKING:
36 from .._protocol import DataLike
38 data: DataLike
40 @staticmethod
41 def _positive(series: pl.Series) -> pl.Series:
42 """Return only the positive values in *series*."""
43 return series.filter(series > 0)
45 @staticmethod
46 def _negative(series: pl.Series) -> pl.Series:
47 """Return only the negative values in *series*."""
48 return series.filter(series < 0)
50 @staticmethod
51 def _mean_positive_expr(series: pl.Series) -> float:
52 """Return the mean of all positive values in *series*, or NaN if none exist."""
53 return _mean(_BasicCoreMixin._positive(series))
55 @staticmethod
56 def _mean_negative_expr(series: pl.Series) -> float:
57 """Return the mean of all negative values in *series*, or NaN if none exist."""
58 return _mean(_BasicCoreMixin._negative(series))
60 @staticmethod
61 def _gaussian_quantile(alpha: float, mu: float, sigma: float) -> float:
62 """Gaussian inverse-CDF (``norm.ppf``) returning NaN for a zero-scale input.
64 ``norm.ppf(alpha, mu, 0.0)`` already returns ``nan`` for a degenerate
65 (zero-variance) distribution — but it emits an ``invalid value
66 encountered in multiply`` RuntimeWarning while doing so (``inf * 0``
67 internally). Degenerate scale arises for a single observation (undefined
68 std) or a constant series. Short-circuiting to ``float("nan")`` keeps the
69 exact same result while suppressing the spurious warning; downstream
70 masking relies on this NaN (Polars treats ``x < nan`` as ``True``).
71 """
72 return float("nan") if sigma == 0.0 else float(norm.ppf(alpha, mu, sigma))
74 # ── Basic statistics ──────────────────────────────────────────────────────
76 @columnwise_stat
77 def skew(self, series: pl.Series) -> int | float | None:
78 """Calculate skewness (asymmetry) for each numeric column.
80 Args:
81 series (pl.Series): The series to calculate skewness for.
83 Returns:
84 float: The skewness value.
86 """
87 return series.skew(bias=False)
89 @columnwise_stat
90 def kurtosis(self, series: pl.Series) -> int | float | None:
91 """Calculate the kurtosis of returns.
93 The degree to which a distribution peak compared to a normal distribution.
95 Args:
96 series (pl.Series): The series to calculate kurtosis for.
98 Returns:
99 float: The kurtosis value.
101 """
102 return series.kurtosis(bias=False)
104 @columnwise_stat
105 def avg_return(self, series: pl.Series) -> float:
106 """Calculate average return per non-zero value.
108 Args:
109 series (pl.Series): The series to calculate average return for.
111 Returns:
112 float: The average return value.
114 """
115 return _mean(series.filter(series.is_not_null() & (series != 0)))
117 @columnwise_stat
118 def avg_win(self, series: pl.Series) -> float:
119 """Calculate the average winning return/trade for an asset.
121 Args:
122 series (pl.Series): The series to calculate average win for.
124 Returns:
125 float: The average winning return.
127 """
128 return self._mean_positive_expr(series)
130 @columnwise_stat
131 def avg_loss(self, series: pl.Series) -> float:
132 """Calculate the average loss return/trade for a period.
134 Args:
135 series (pl.Series): The series to calculate average loss for.
137 Returns:
138 float: The average loss return.
140 """
141 return self._mean_negative_expr(series)
143 @columnwise_stat
144 def comp(self, series: pl.Series) -> float:
145 """Calculate the total compounded return over the full period.
147 Computed as product(1 + r) - 1.
149 Args:
150 series (pl.Series): The series to calculate compounded return for.
152 Returns:
153 float: Total compounded return.
155 """
156 return _comp_return(series)
158 @columnwise_stat
159 def geometric_mean(self, series: pl.Series, periods: int | float | None = None, annualize: bool = False) -> float:
160 """Calculate the geometric mean of returns.
162 Computed as the per-period geometric average: (∏(1 + rᵢ))^(1/n) - 1.
163 When annualized, raises to the power of periods_per_year instead of 1/n.
165 Args:
166 series (pl.Series): The series to calculate geometric mean for.
167 periods (int | float, optional): Periods per year for annualization. Defaults to periods_per_year.
168 annualize (bool): Whether to annualize the result. Defaults to False.
170 Returns:
171 float: The geometric mean return.
174 Returns NaN when:
175 ``float("nan")`` when the series has no non-null observations or the
176 compounded return ``product(1 + r)`` is non-positive.
177 """
178 clean = series.drop_nulls().cast(pl.Float64)
179 n = clean.len()
180 if n == 0:
181 return float("nan") # indeterminate: no observations
182 compound = float((1.0 + clean).product())
183 if compound <= 0:
184 return float("nan") # indeterminate: non-positive compound return
185 exponent = (periods or self._data._periods_per_year) / n if annualize else (1.0 / n)
186 return float(compound**exponent) - 1.0
188 # ── Volatility & risk ─────────────────────────────────────────────────────
190 @columnwise_stat
191 def volatility(self, series: pl.Series, periods: int | float | None = None, annualize: bool = True) -> float:
192 """Calculate the volatility of returns.
194 - Std dev of returns
195 - Annualized by sqrt(periods) if `annualize` is True.
197 Args:
198 series (pl.Series): The series to calculate volatility for.
199 periods (int, optional): Number of periods per year. Defaults to 252.
200 annualize (bool, optional): Whether to annualize the result. Defaults to True.
202 Returns:
203 float: The volatility value.
205 """
206 raw_periods = periods or self._data._periods_per_year
208 # Ensure it's numeric
209 if not isinstance(raw_periods, int | float):
210 raise TypeError(f"Expected int or float for periods, got {type(raw_periods).__name__}") # noqa: TRY003
212 factor = _annualization_factor(raw_periods) if annualize else 1.0
213 std_val = cast(float, series.std())
214 return (std_val if std_val is not None else 0.0) * factor
216 @columnwise_stat
217 def mad(self, series: pl.Series, periods: int | float | None = None, annualize: bool = True) -> float:
218 """Calculate the Mean Absolute Deviation (MAD) of returns.
220 MAD is the mean of absolute deviations from the mean return:
221 mean(|r - mean(r)|). It is a robust measure of dispersion less
222 sensitive to outliers than standard deviation.
224 - Annualized by sqrt(periods) if `annualize` is True.
226 Args:
227 series (pl.Series): The series to calculate MAD for.
228 periods (int, optional): Number of periods per year. Defaults to periods_per_year.
229 annualize (bool, optional): Whether to annualize the result. Defaults to True.
231 Returns:
232 float: The MAD value.
233 """
234 raw_periods = periods or self._data._periods_per_year
236 if not isinstance(raw_periods, int | float):
237 raise TypeError(f"Expected int or float for periods, got {type(raw_periods).__name__}") # noqa: TRY003
239 factor = _annualization_factor(raw_periods) if annualize else 1.0
240 mean_val = _mean(series)
241 mad_val = cast(float, (series - mean_val).abs().mean())
242 return (mad_val if mad_val is not None else 0.0) * factor
244 # ── Win / loss metrics ────────────────────────────────────────────────────
246 @columnwise_stat
247 def payoff_ratio(self, series: pl.Series) -> float:
248 """Measure the payoff ratio.
250 The payoff ratio is calculated as average win / abs(average loss).
252 Args:
253 series (pl.Series): The series to calculate payoff ratio for.
255 Returns:
256 float: The payoff ratio value.
258 """
259 avg_win = self._mean_positive_expr(series)
260 avg_loss = float(np.abs(self._mean_negative_expr(series)))
261 return avg_win / avg_loss
263 @columnwise_stat
264 def profit_ratio(self, series: pl.Series) -> float:
265 """Measure the profit ratio.
267 The profit ratio is calculated as win ratio / loss ratio.
269 Args:
270 series (pl.Series): The series to calculate profit ratio for.
272 Returns:
273 float: The profit ratio value.
276 Returns NaN when:
277 ``float("nan")`` when the series has no wins or no losses.
278 """
279 wins = series.filter(series >= 0)
280 losses = self._negative(series)
282 # Filtering can legitimately leave no wins or no losses for one-sided return series.
283 if wins.is_empty() or losses.is_empty():
284 return float("nan") # indeterminate: no wins or no losses
286 win_mean = _mean(wins)
287 loss_mean = _mean(losses)
288 win_ratio = float(np.abs(win_mean / wins.count()))
289 loss_ratio = float(np.abs(loss_mean / losses.count()))
291 return win_ratio / loss_ratio
293 @columnwise_stat
294 def profit_factor(self, series: pl.Series) -> float:
295 """Measure the profit factor.
297 The profit factor is calculated as wins / loss.
299 Args:
300 series (pl.Series): The series to calculate profit factor for.
302 Returns:
303 float: The profit factor value.
305 """
306 wins = self._positive(series)
307 losses = self._negative(series)
308 wins_sum = wins.sum()
309 losses_sum = losses.sum()
311 return float(np.abs(float(wins_sum) / float(losses_sum)))
313 # ── Risk metrics ──────────────────────────────────────────────────────────
315 @columnwise_stat
316 def value_at_risk(self, series: pl.Series, sigma: float = 1.0, alpha: float = 0.05) -> float:
317 """Calculate the daily value-at-risk.
319 Uses variance-covariance calculation with confidence level.
321 Args:
322 series (pl.Series): The series to calculate value at risk for.
323 alpha (float, optional): Confidence level. Defaults to 0.05.
324 sigma (float, optional): Standard deviation multiplier. Defaults to 1.0.
326 Returns:
327 float: The value at risk.
329 """
330 mean_val = _mean(series)
331 std_val = cast(float, series.std())
332 mu = mean_val
333 sigma *= std_val if std_val is not None else 0.0
335 return self._gaussian_quantile(alpha, mu, sigma)
337 @staticmethod
338 def _gaussian_expected_shortfall(alpha: float, mu: float, sigma: float) -> float:
339 """Closed-form expected shortfall ``E[X | X <= VaR_alpha]`` of ``N(mu, sigma)``.
341 This is the estimator that belongs with `value_at_risk`'s
342 variance-covariance method. A degenerate (zero-scale) distribution has
343 no tail beyond its single point, so the shortfall is ``mu`` itself.
344 """
345 if sigma == 0.0:
346 return mu
347 return float(mu - sigma * norm.pdf(norm.ppf(alpha)) / alpha)
349 @columnwise_stat
350 def _conditional_value_at_risk_impl(
351 self, series: pl.Series, sigma: float = 1.0, alpha: float = 0.05, method: str = "parametric"
352 ) -> float:
353 """Inner per-series implementation of conditional value-at-risk."""
354 series = series.drop_nulls().drop_nans()
355 if series.is_empty():
356 return float("nan")
357 if method == "historical":
358 threshold = cast(float, series.quantile(alpha, interpolation="linear"))
359 return _mean(series.filter(series <= threshold))
360 std_val = cast(float | None, series.std())
361 if std_val is None:
362 return float("nan") # indeterminate: a single observation has no spread
363 return self._gaussian_expected_shortfall(alpha, _mean(series), sigma * std_val)
365 def conditional_value_at_risk(
366 self,
367 sigma: float = 1.0,
368 confidence: float | None = None,
369 alpha: float | None = None,
370 method: str = "parametric",
371 ) -> dict[str, float]:
372 """Calculate the conditional value-at-risk (CVaR / Expected Shortfall).
374 Also known as CVaR or expected shortfall, calculated for each numeric column.
376 The tail can be specified either way round: ``confidence`` matches the
377 QuantStats spelling, ``alpha`` matches `value_at_risk` on this same
378 object. They are two names for one quantity (``alpha = 1 - confidence``),
379 so passing both is an error rather than a silent preference.
381 Args:
382 sigma (float, optional): Standard deviation multiplier. Defaults to 1.0.
383 confidence (float, optional): Confidence level (e.g. 0.95 for 95 %).
384 Converted internally to ``alpha = 1 - confidence``. Mutually
385 exclusive with ``alpha``.
386 alpha (float, optional): Tail probability in the *loss* tail (e.g. 0.05
387 for 95 % confidence). Mutually exclusive with ``confidence``.
388 Both defaulting to ``None`` selects ``alpha = 0.05``.
389 method (str, optional): ``"parametric"`` (default) is the closed-form
390 normal expected shortfall, consistent with `value_at_risk`.
391 ``"historical"`` averages the observations at or below the
392 empirical ``alpha`` quantile, which captures fat tails but needs
393 enough data to be meaningful.
395 Returns:
396 dict[str, float]: The conditional value at risk per asset column.
398 Raises:
399 ValueError: If both ``confidence`` and ``alpha`` are given, or
400 ``method`` is not ``"parametric"`` or ``"historical"``.
402 """
403 if method not in ("parametric", "historical"):
404 raise ValueError(f"method must be 'parametric' or 'historical', got {method!r}") # noqa: TRY003
405 if confidence is not None and alpha is not None:
406 raise ValueError( # noqa: TRY003
407 f"Pass either confidence or alpha, not both "
408 f"(got confidence={confidence!r}, alpha={alpha!r}); alpha = 1 - confidence"
409 )
410 if confidence is not None:
411 alpha = 1.0 - confidence
412 elif alpha is None:
413 alpha = 0.05
415 return self._conditional_value_at_risk_impl(sigma=sigma, alpha=alpha, method=method)
417 @staticmethod
418 def _drawdown_with_baseline(series: pl.Series) -> pl.Series:
419 """Compute drawdown series with a phantom zero-return baseline prepended.
421 Matches the quantstats convention: a negative first return is treated as
422 a drawdown from the initial capital of 1.0, not as the new high-water mark.
423 """
424 extended = pl.concat([pl.Series([0.0]), series.cast(pl.Float64)])
425 nav = (1.0 + extended).cum_prod()
426 hwm = nav.cum_max()
427 # The phantom baseline pins nav[0] = 1.0, so hwm >= 1.0 throughout and
428 # the 1e-10 floor is purely defensive (unreachable); a -100 % return
429 # correctly reports as a full drawdown of 1.0 here.
430 dd = ((hwm - nav) / hwm.clip(lower_bound=1e-10)).clip(lower_bound=0.0)
431 return dd[1:] # drop phantom point
433 @staticmethod
434 def _ulcer_index_series(series: pl.Series) -> float:
435 """Compute ulcer index for a single returns series."""
436 dd = _BasicCoreMixin._drawdown_with_baseline(series)
437 n = series.len()
438 return float(np.sqrt(float((dd**2).sum()) / (n - 1)))
440 @columnwise_stat
441 def ulcer_index(self, series: pl.Series) -> float:
442 """Calculate the Ulcer Index (downside risk measurement).
444 Measures the depth and duration of drawdowns as the root mean square
445 of squared drawdowns: sqrt(sum(dd²) / (n - 1)).
447 Args:
448 series (pl.Series): The series to calculate ulcer index for.
450 Returns:
451 float: Ulcer Index value.
453 """
454 return self._ulcer_index_series(series)
456 @columnwise_stat
457 def ulcer_performance_index(self, series: pl.Series, rf: float = 0.0) -> float:
458 """Calculate the Ulcer Performance Index (UPI).
460 Risk-adjusted return using Ulcer Index as the risk measure:
461 (compounded_return - rf) / ulcer_index.
463 Args:
464 series (pl.Series): The series to calculate UPI for.
465 rf (float): Risk-free rate. Defaults to 0.
467 Returns:
468 float: Ulcer Performance Index.
471 Returns NaN when:
472 ``float("nan")`` when the ulcer index is zero (no drawdowns).
473 """
474 comp = _comp_return(series)
475 ui = self._ulcer_index_series(series)
476 return float("nan") if ui == 0 else (comp - rf) / ui
478 @columnwise_stat
479 def serenity_index(self, series: pl.Series, rf: float = 0.0) -> float:
480 """Calculate the Serenity Index.
482 Combines the Ulcer Index with a CVaR-based pitfall measure:
483 (sum_returns - rf) / (ulcer_index * pitfall), where
484 pitfall = -CVaR(drawdowns) / std(returns), using the parametric
485 (Gaussian) CVaR as QuantStats does.
487 Args:
488 series (pl.Series): The series to calculate serenity index for.
489 rf (float): Risk-free rate. Defaults to 0.
491 Returns:
492 float: Serenity Index.
495 Returns NaN when:
496 ``float("nan")`` when the returns have zero (or undefined) standard
497 deviation or the denominator ``ulcer_index * pitfall`` is zero.
498 """
499 std_val = cast(float, series.std())
500 if not std_val:
501 return float("nan") # indeterminate: zero variance
503 # Negate drawdowns to match quantstats sign convention (negative = below peak)
504 dd_neg = -self._drawdown_with_baseline(series)
505 cvar_val = self._gaussian_expected_shortfall(0.05, _mean(dd_neg), cast(float, dd_neg.std()))
507 pitfall = -cvar_val / std_val
508 ui = self._ulcer_index_series(series)
509 denominator = ui * pitfall
510 return float("nan") if denominator == 0 else (float(series.sum()) - rf) / denominator
512 @columnwise_stat
513 def win_rate(self, series: pl.Series) -> float:
514 """Calculate the win ratio for a period.
516 Args:
517 series (pl.Series): The series to calculate win rate for.
519 Returns:
520 float: The win rate value.
522 """
523 num_pos = self._positive(series).count()
524 num_nonzero = series.filter(series != 0).count()
525 return float(num_pos / num_nonzero)