Coverage for src/jquantstats/_stats/_core.py: 100%
55 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-23 06:13 +0000
1"""Module helpers and method decorators for statistical computations.
3Provides:
5- `_drawdown_series` — drawdown series from a returns series.
6- `_to_float` — safe Polars aggregation result → Python float.
7- `_mean` — series mean with ``None → 0.0`` fallback.
8- `_std_is_negligible` — shared "is this std numerically zero?" test for
9 mean/std ratio metrics.
10- `columnwise_stat` — decorator: apply a metric to every asset column.
11- `to_frame` — decorator: build a per-column Polars DataFrame result.
13These building blocks are shared across the stats mixin modules
14(`_basic`, `_performance`,
15`_reporting`, `_rolling`).
17Null-return convention
18----------------------
19- **Scalar metrics** return ``float("nan")`` when the series has no non-null
20 observations (use ``_mean`` for the ``None → nan`` conversion).
21- **Ratio metrics** return ``float("nan")`` when the denominator is zero
22 or indeterminate.
23- Use ``_mean`` for the ``None → nan`` conversion rather than
24 ``cast(float, ...)``.
25"""
27from __future__ import annotations
29import sys
30from collections.abc import Callable
31from datetime import timedelta
32from functools import wraps
33from typing import Any, Concatenate, ParamSpec, TypeVar, cast, overload
35import polars as pl
37P = ParamSpec("P")
38R = TypeVar("R")
40# ── Module helpers ────────────────────────────────────────────────────────────
43def _drawdown_series(series: pl.Series) -> pl.Series:
44 """Compute the drawdown percentage series from a returns series.
46 Builds a compound NAV (geometric cumulative product) from the returns
47 series and expresses drawdown as the fraction below the running high-water
48 mark. This matches the quantstats convention.
50 Args:
51 series: A Polars Series of multiplicative daily returns.
53 Returns:
54 A Polars Float64 Series whose values are in [0, 1]. A value of 0
55 means the NAV is at its all-time high; a value of 0.2 means the NAV
56 is 20 % below its previous peak.
58 Numerical edge cases:
59 The high-water mark can only fall below the ``1e-10`` floor when
60 *every* NAV value so far is below it, i.e. when the very first
61 return is (effectively) -100 %. Because ``0 <= nav <= hwm`` always
62 holds, the result stays within [0, 1] even when the floor is active.
63 Note that an exact -100 % first return yields ``nav == hwm == 0``
64 and therefore a drawdown of 0: with no baseline, the first
65 observation *is* its own high-water mark. Metrics that need the
66 quantstats convention (initial capital of 1.0 as the baseline)
67 should use ``_drawdown_with_baseline`` instead.
69 Examples:
70 >>> import polars as pl
71 >>> s = pl.Series([0.0, -0.1, 0.2])
72 >>> [round(x, 10) for x in _drawdown_series(s).to_list()]
73 [0.0, 0.1, 0.0]
74 """
75 nav = (1.0 + series.cast(pl.Float64)).cum_prod()
76 hwm = nav.cum_max()
77 # The floor keeps the division defined after a -100 % return wipes out
78 # the NAV; since 0 <= nav <= hwm the ratio stays in [0, 1] regardless.
79 hwm_safe = hwm.clip(lower_bound=1e-10)
80 return ((hwm - nav) / hwm_safe).clip(lower_bound=0.0)
83def _to_float(value: Any) -> float:
84 """Safely convert a Polars aggregation result to float.
86 Examples:
87 >>> _to_float(2.0)
88 2.0
89 >>> _to_float(None)
90 0.0
91 """
92 if value is None:
93 return 0.0
94 if isinstance(value, timedelta):
95 return value.total_seconds()
96 return float(cast(float, value))
99def _std_is_negligible(std: float | None, mean: float) -> bool:
100 """Return True when a sample standard deviation is numerically zero.
102 Mean/std ratios (Sharpe and friends) are meaningless when the measured
103 dispersion is smaller than the floating-point rounding noise of the
104 inputs: a constant series can produce a tiny non-zero ``std`` purely from
105 accumulated rounding, and dividing by it would report an absurdly large
106 ratio instead of "no dispersion". The threshold is 10 machine epsilons
107 scaled by the magnitude of the mean, with an absolute floor of one
108 epsilon for means at or near zero. Callers map this case to
109 ``float("nan")``.
111 Args:
112 std: Sample standard deviation, or ``None`` when undefined
113 (fewer than two observations).
114 mean: Sample mean of the same series, used to scale the threshold.
116 Examples:
117 >>> _std_is_negligible(None, 1.0)
118 True
119 >>> _std_is_negligible(0.0, 0.05)
120 True
121 >>> _std_is_negligible(0.01, 0.05)
122 False
123 """
124 if std is None:
125 return True
126 eps = sys.float_info.epsilon
127 return float(std) <= eps * max(abs(mean), eps) * 10.0
130def _mean(series: pl.Series) -> float:
131 """Return series mean, or ``float("nan")`` if the series is empty or all-null.
133 Use this instead of ``cast(float, series.mean())`` to avoid ``None``
134 leaking into arithmetic — consistent with the scalar-metric convention
135 that returns ``float("nan")`` when there are no non-null observations.
137 Examples:
138 >>> import polars as pl
139 >>> _mean(pl.Series([1.0, 3.0]))
140 2.0
141 >>> import math
142 >>> math.isnan(_mean(pl.Series([], dtype=pl.Float64)))
143 True
144 """
145 result = series.mean()
146 return float(cast(float, result)) if result is not None else float("nan")
149# ── Module-level decorators ──────────────────────────────────────────────────
152@overload
153def columnwise_stat(
154 func: Callable[Concatenate[Any, pl.Series, P], R], *, data_attr: str = ...
155) -> Callable[Concatenate[Any, P], dict[str, R]]: ...
158@overload
159def columnwise_stat(
160 func: None = ..., *, data_attr: str = ...
161) -> Callable[[Callable[Concatenate[Any, pl.Series, P], R]], Callable[Concatenate[Any, P], dict[str, R]]]: ...
164def columnwise_stat(
165 func: Callable[Concatenate[Any, pl.Series, P], R] | None = None, *, data_attr: str = "_data"
166) -> (
167 Callable[Concatenate[Any, P], dict[str, R]]
168 | Callable[[Callable[Concatenate[Any, pl.Series, P], R]], Callable[Concatenate[Any, P], dict[str, R]]]
169):
170 """Apply a column-wise statistical function to all numeric columns.
172 The decorated method must accept ``(self, series, *args, **kwargs)``; the
173 wrapper drops the ``series`` parameter and preserves the remaining
174 signature (via ParamSpec), returning ``{column: value}`` with the wrapped
175 method's return type as the value type.
177 Args:
178 func (Callable | None): The function to decorate.
179 data_attr: Attribute name that holds the column-wise data object.
181 Returns:
182 Callable: The decorated function.
184 """
186 def decorator(
187 inner_func: Callable[Concatenate[Any, pl.Series, P], R],
188 ) -> Callable[Concatenate[Any, P], dict[str, R]]:
189 """Wrap *inner_func* to iterate over the configured data attribute columns."""
191 @wraps(inner_func)
192 def wrapper(self: Any, *args: P.args, **kwargs: P.kwargs) -> dict[str, R]:
193 """Apply *func* to every column and return a ``{column: value}`` mapping."""
194 if not hasattr(self, data_attr):
195 msg = (
196 f"columnwise_stat requires host object to define '{data_attr}' "
197 f"(missing attribute on {type(self).__name__})."
198 )
199 raise AttributeError(msg)
200 data = getattr(self, data_attr)
201 return {col: inner_func(self, series, *args, **kwargs) for col, series in data.items()}
203 return cast("Callable[Concatenate[Any, P], dict[str, R]]", wrapper)
205 if func is None:
206 return decorator
207 return decorator(func)
210@overload
211def to_frame(
212 func: Callable[Concatenate[Any, pl.Series, P], pl.Series], *, data_attr: str = ...
213) -> Callable[Concatenate[Any, P], pl.DataFrame]: ...
216@overload
217def to_frame(
218 func: None = ..., *, data_attr: str = ...
219) -> Callable[[Callable[Concatenate[Any, pl.Series, P], pl.Series]], Callable[Concatenate[Any, P], pl.DataFrame]]: ...
222def to_frame(
223 func: Callable[Concatenate[Any, pl.Series, P], pl.Series] | None = None, *, data_attr: str = "_data"
224) -> (
225 Callable[Concatenate[Any, P], pl.DataFrame]
226 | Callable[[Callable[Concatenate[Any, pl.Series, P], pl.Series]], Callable[Concatenate[Any, P], pl.DataFrame]]
227):
228 """Apply per-column expressions and evaluates with .with_columns(...).
230 The decorated method must accept ``(self, series, *args, **kwargs)`` and
231 return a per-column Polars Series; the wrapper drops the ``series``
232 parameter and preserves the remaining signature (via ParamSpec).
234 Args:
235 func (Callable | None): The function to decorate.
236 data_attr: Attribute name that holds the column-wise data object.
238 Returns:
239 Callable: The decorated function.
241 """
243 def decorator(
244 inner_func: Callable[Concatenate[Any, pl.Series, P], pl.Series],
245 ) -> Callable[Concatenate[Any, P], pl.DataFrame]:
246 """Wrap *inner_func* to build a per-column frame from the configured data attribute."""
248 @wraps(inner_func)
249 def wrapper(self: Any, *args: P.args, **kwargs: P.kwargs) -> pl.DataFrame:
250 """Apply *func* per column and return the result as a Polars DataFrame."""
251 if not hasattr(self, data_attr):
252 msg = (
253 f"to_frame requires host object to define '{data_attr}' "
254 f"(missing attribute on {type(self).__name__})."
255 )
256 raise AttributeError(msg)
257 data = getattr(self, data_attr)
258 result: pl.DataFrame = self.all.select(
259 [pl.col(name) for name in data.date_col]
260 + [inner_func(self, series, *args, **kwargs).alias(col) for col, series in data.items()]
261 )
262 return result
264 return cast("Callable[Concatenate[Any, P], pl.DataFrame]", wrapper)
266 if func is None:
267 return decorator
268 return decorator(func)