Coverage for src/jquantstats/_stats/_reporting.py: 100%
110 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-06 04:52 +0000
1"""Temporal reporting metrics.
3Capture ratios live in `_capture.py` and the aggregating `summary` /
4`annual_breakdown` pair in `_summary.py`; all three are composed into `Stats`.
5"""
7from __future__ import annotations
9from typing import TYPE_CHECKING
11import polars as pl
13from ._core import _drawdown_series, _to_float, columnwise_stat
14from ._internals import _comp_return
16if TYPE_CHECKING:
17 from ..data import Data
19# ── Reporting statistics mixin ───────────────────────────────────────────────
22class _ReportingStatsMixin:
23 """Mixin providing temporal reporting metrics.
25 Covers: periods per year, average drawdown, CAGR, expected return, RAR,
26 Calmar ratio, recovery factor, max drawdown duration, and monthly win rate.
28 Cross-mixin dependencies:
29 - _BasicStatsMixin: exposure
30 """
32 _data: Data
33 all: pl.DataFrame
35 if TYPE_CHECKING:
36 from .._protocol import DataLike
38 data: DataLike
40 def exposure(self) -> dict[str, float]:
41 """Defined on _BasicStatsMixin."""
43 @property
44 def periods_per_year(self) -> float:
45 """Estimate the number of periods per year from the data index spacing.
47 Returns:
48 float: Estimated number of observations per calendar year.
49 """
50 return self._data._periods_per_year
52 @columnwise_stat
53 def avg_drawdown(self, series: pl.Series) -> float:
54 """Average drawdown across all underwater periods.
56 Returns 0.0 when there are no underwater periods.
58 Matches the QuantStats sign convention: drawdown is expressed as a
59 negative fraction (e.g. ``-0.2`` for 20% below peak).
61 Args:
62 series (pl.Series): Series of additive daily returns.
64 Returns:
65 float: Mean drawdown in [-1, 0].
66 """
67 dd = _drawdown_series(series)
68 in_dd = dd.filter(dd > 0)
69 # A series that never falls below its high-water mark has an average drawdown of exactly 0.0.
70 if in_dd.is_empty():
71 return 0.0
72 return -_to_float(in_dd.mean())
74 @columnwise_stat
75 def cagr(
76 self,
77 series: pl.Series,
78 rf: float = 0.0,
79 compounded: bool = True,
80 periods: int | float | None = None,
81 ) -> float:
82 """Calculate the Compound Annual Growth Rate (CAGR) of excess returns.
84 CAGR represents the geometric mean annual growth rate, providing a
85 smoothed annualized return that accounts for compounding effects.
87 Args:
88 series (pl.Series): Series of additive daily returns.
89 rf (float): Annualized risk-free rate. Defaults to 0.0.
90 compounded (bool): Whether to compound returns. Defaults to True.
91 periods: Periods per year for annualisation. Defaults to ``periods_per_year``.
93 Returns:
94 float: CAGR of excess returns.
96 Returns NaN when:
97 ``float("nan")`` when the series is empty.
98 """
99 raw_periods = periods or self._data._periods_per_year
100 n = len(series)
101 if n == 0:
102 return float("nan") # pragma: no cover
103 excess = series.cast(pl.Float64) - rf / raw_periods
104 total = _comp_return(excess) if compounded else _to_float(excess.sum())
105 years = n / raw_periods
106 return float(abs(1.0 + total) ** (1.0 / years) - 1.0)
108 def expected_return(
109 self,
110 aggregate: str | None = None,
111 compounded: bool = True,
112 ) -> dict[str, float]:
113 """Expected return with optional period aggregation.
115 Returns the arithmetic mean of per-period returns. When *aggregate* is
116 provided the returns are first compounded (or summed) within each
117 calendar period, and the mean is taken over those period returns.
119 Args:
120 aggregate (str | None): Period to aggregate to before computing the
121 mean. Accepted values: ``'weekly'``, ``'monthly'``,
122 ``'quarterly'``, ``'annual'`` / ``'yearly'``. Defaults to
123 ``None`` (raw per-period mean).
124 compounded (bool): Compound returns within each period when
125 *aggregate* is set. Defaults to ``True``.
127 Returns:
128 dict[str, float]: Mean return per asset for the specified period.
130 Raises:
131 ValueError: If *aggregate* is an unrecognised string.
133 Note:
134 Requires a temporal (Date / Datetime) index when *aggregate* is not
135 ``None``; falls back to the raw per-period mean otherwise.
137 Returns NaN when:
138 Entries are ``float("nan")`` when an asset has no non-null
139 observations.
140 """
141 _freq_map: dict[str, str] = {
142 "weekly": "1w",
143 "monthly": "1mo",
144 "quarterly": "3mo",
145 "annual": "1y",
146 "yearly": "1y",
147 }
149 def _geomean(s: pl.Series) -> float:
150 """Per-period geometric mean: (product(1 + r))^(1/n) - 1."""
151 n = s.count()
152 if n == 0:
153 return float("nan")
154 return float(_to_float((1.0 + s.cast(pl.Float64)).product()) ** (1.0 / n) - 1.0)
156 def _raw_expected_returns() -> dict[str, float]:
157 """Return the geometric mean of each raw return series."""
158 return {col: _geomean(series.drop_nulls()) for col, series in self._data.items()}
160 if aggregate is None:
161 return _raw_expected_returns()
163 if aggregate.lower() not in _freq_map:
164 raise ValueError(f"aggregate must be one of {list(_freq_map)}, got {aggregate!r}") # noqa: TRY003
166 all_df = self.all
167 date_col_name = self._data.date_col[0] if self._data.date_col else None
168 if date_col_name is None or not all_df[date_col_name].dtype.is_temporal():
169 return _raw_expected_returns()
171 trunc = _freq_map[aggregate.lower()]
172 agg_expr = ((1.0 + pl.col("ret")).product() - 1.0) if compounded else pl.col("ret").sum()
174 result: dict[str, float] = {}
175 for col, series in self._data.items():
176 df = (
177 pl.DataFrame({"date": all_df[date_col_name], "ret": series})
178 .drop_nulls()
179 .with_columns(pl.col("date").dt.truncate(trunc).alias("period"))
180 )
181 period_rets = df.group_by("period").agg(agg_expr.alias("ret"))["ret"]
182 result[col] = _geomean(period_rets)
183 return result
185 def rar(self, periods: int | float = 252) -> dict[str, float]:
186 """Risk-Adjusted Return: CAGR divided by exposure.
188 Measures annualised return per unit of market participation time,
189 matching the quantstats convention.
191 Args:
192 periods: Periods per year for CAGR annualisation. Defaults to ``periods_per_year``.
194 Returns:
195 dict[str, float]: RAR per asset.
196 """
197 cagr = self.cagr(periods=periods)
198 exp = self.exposure()
199 return {col: cagr[col] / exp[col] for col in cagr}
201 @columnwise_stat
202 def calmar(self, series: pl.Series, periods: int | float | None = None) -> float:
203 """Calmar ratio (CAGR divided by maximum drawdown).
205 Returns ``nan`` when the maximum drawdown is zero.
207 Args:
208 series (pl.Series): Series of additive daily returns.
209 periods: Annualisation factor. Defaults to ``periods_per_year``.
211 Returns:
212 float: Calmar ratio, or ``nan`` if max drawdown is zero.
213 """
214 raw_periods = float(periods or self._data._periods_per_year)
215 max_dd = _to_float(_drawdown_series(series).max())
216 if max_dd <= 0:
217 return float("nan")
218 n = len(series)
219 comp_return = _comp_return(series)
220 cagr = float((1.0 + comp_return) ** (raw_periods / n)) - 1.0
221 return cagr / max_dd
223 @columnwise_stat
224 def recovery_factor(self, series: pl.Series) -> float:
225 """Recovery factor (total return divided by maximum drawdown).
227 Matches the quantstats convention: total return is the simple sum of
228 returns, not compounded. Returns ``nan`` when the maximum drawdown
229 is zero.
231 Args:
232 series (pl.Series): Series of additive daily returns.
234 Returns:
235 float: Recovery factor, or ``nan`` if max drawdown is zero.
236 """
237 max_dd = _to_float(_drawdown_series(series).max())
238 if max_dd <= 0:
239 return float("nan")
240 total_return = _to_float(series.sum())
241 return abs(total_return) / max_dd
243 def max_drawdown_duration(self) -> dict[str, float | int | None]:
244 """Maximum drawdown duration in calendar days (or periods) per asset.
246 When the index is a temporal column (``Date`` / ``Datetime``) the
247 duration is expressed as calendar days spanned by the longest
248 underwater run. For integer-indexed data each row counts as one
249 period.
251 Returns:
252 dict[str, float | int | None]: Asset → max drawdown duration.
253 Returns 0 when there are no underwater periods.
254 """
255 all_df = self.all
256 date_col_name = self._data.date_col[0] if self._data.date_col else None
257 has_date = date_col_name is not None and all_df[date_col_name].dtype.is_temporal()
258 result: dict[str, float | int | None] = {}
259 for col, series in self._data.items():
260 nav = 1.0 + series.cast(pl.Float64).cum_sum()
261 hwm = nav.cum_max()
262 in_dd = nav < hwm
264 if not in_dd.any():
265 result[col] = 0
266 continue
268 if has_date and date_col_name is not None:
269 frame = pl.DataFrame({"date": all_df[date_col_name], "in_dd": in_dd})
270 else:
271 frame = pl.DataFrame({"date": pl.Series(list(range(len(series))), dtype=pl.Int64), "in_dd": in_dd})
273 frame = frame.with_columns(pl.col("in_dd").rle_id().alias("run_id"))
274 dd_runs = (
275 frame.filter(pl.col("in_dd"))
276 .group_by("run_id")
277 .agg([pl.col("date").min().alias("start"), pl.col("date").max().alias("end")])
278 )
280 if has_date:
281 dd_runs = dd_runs.with_columns(
282 ((pl.col("end") - pl.col("start")).dt.total_days() + 1).alias("duration")
283 )
284 else:
285 dd_runs = dd_runs.with_columns((pl.col("end") - pl.col("start") + 1).alias("duration"))
287 result[col] = int(_to_float(dd_runs["duration"].max()))
288 return result
290 def monthly_win_rate(self) -> dict[str, float]:
291 """Fraction of calendar months with a positive compounded return per asset.
293 Requires a temporal (Date / Datetime) index. Returns ``nan`` per
294 asset when no temporal index is present.
296 Returns:
297 dict[str, float]: Monthly win rate in [0, 1] per asset.
299 Returns NaN when:
300 Entries are ``float("nan")`` when no temporal index is present or an
301 asset has no non-null observations.
302 """
303 all_df = self.all
304 date_col_name = self._data.date_col[0] if self._data.date_col else None
305 if date_col_name is None or not all_df[date_col_name].dtype.is_temporal():
306 return {col: float("nan") for col, _ in self._data.items()}
308 result: dict[str, float] = {}
309 for col, _ in self._data.items():
310 df = (
311 all_df.select([date_col_name, col])
312 .drop_nulls()
313 .with_columns(
314 [
315 pl.col(date_col_name).dt.year().alias("_year"),
316 pl.col(date_col_name).dt.month().alias("_month"),
317 ]
318 )
319 )
320 monthly = (
321 df.group_by(["_year", "_month"])
322 .agg((pl.col(col) + 1.0).product().alias("gross"))
323 .with_columns((pl.col("gross") - 1.0).alias("monthly_return"))
324 )
325 n_total = len(monthly)
326 if n_total == 0:
327 result[col] = float("nan")
328 else:
329 n_positive = int((monthly["monthly_return"] > 0).sum())
330 result[col] = n_positive / n_total
331 return result