Coverage for src/jquantstats/_stats/_drawdown.py: 100%
97 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-23 04:11 +0000
1"""Drawdown and cumulative-return metrics for financial returns data."""
3from __future__ import annotations
5from typing import TYPE_CHECKING, cast
7import polars as pl
9from ._core import columnwise_stat, to_frame
10from ._internals import _nav_series
12if TYPE_CHECKING:
13 from ..data import Data
16def _drawdown_underwater(series: pl.Series) -> pl.Series:
17 """Return only the underwater (negative) drawdown values as positive fractions.
19 Args:
20 series: A Polars Series of returns values.
22 Returns:
23 A Polars Series of positive drawdown values (e.g., 0.15 for 15% drawdown)
24 for periods where the strategy is underwater. Empty if never underwater.
25 """
26 nav = _nav_series(series)
27 hwm = nav.cum_max()
28 dd = nav / hwm - 1 # negative or zero
29 return -dd.filter(dd < 0) # positive values only
32# ── Drawdown statistics mixin ─────────────────────────────────────────────────
35class _DrawdownMixin:
36 """Mixin providing cumulative-return and drawdown metrics.
38 Covers: compounded cumulative returns (``compsum``), the drawdown series,
39 price (NAV) conversion, maximum drawdown, and per-episode drawdown details.
40 """
42 _data: Data
43 all: pl.DataFrame
45 if TYPE_CHECKING:
46 from .._protocol import DataLike
48 data: DataLike
50 # ── Cumulative returns ────────────────────────────────────────────────────
52 @to_frame
53 def compsum(self, series: pl.Series) -> pl.Series:
54 """Calculate the rolling compounded (cumulative) returns.
56 Computed as cumprod(1 + r) - 1 for each period.
58 Args:
59 series (pl.Series): The series to calculate cumulative returns for.
61 Returns:
62 pl.Series: Cumulative compounded returns per period.
64 """
65 return (1.0 + series).cum_prod() - 1.0
67 # ── Drawdown ──────────────────────────────────────────────────────────────
69 @to_frame
70 def drawdown(self, series: pl.Series) -> pl.Series:
71 """Calculate the drawdown series for returns.
73 Args:
74 series (pl.Series): The series to calculate drawdown for.
76 Returns:
77 pl.Series: The drawdown series.
79 """
80 equity = self.prices(series)
81 d = (equity / equity.cum_max()) - 1
82 return -d
84 @staticmethod
85 def prices(series: pl.Series) -> pl.Series:
86 """Convert returns series to price series.
88 Args:
89 series (pl.Series): The returns series to convert.
91 Returns:
92 pl.Series: The price series.
94 """
95 return _nav_series(series)
97 @staticmethod
98 def max_drawdown_single_series(series: pl.Series) -> float:
99 """Compute the maximum drawdown for a single returns series.
101 Args:
102 series: A Polars Series of returns values.
104 Returns:
105 float: The maximum drawdown as a positive fraction (e.g. 0.2 for 20%).
106 """
107 price = _DrawdownMixin.prices(series)
108 peak = price.cum_max()
109 drawdown = price / peak - 1
110 dd_min = cast(float, drawdown.min())
111 return dd_min if dd_min is not None else 0.0
113 @columnwise_stat
114 def max_drawdown(self, series: pl.Series) -> float:
115 """Calculate the maximum drawdown for each column.
117 Args:
118 series (pl.Series): The series to calculate maximum drawdown for.
120 Returns:
121 float: The maximum drawdown value.
123 """
124 return _DrawdownMixin.max_drawdown_single_series(series)
126 @columnwise_stat
127 def expected_drawdown(self, series: pl.Series) -> float:
128 """Calculate the average drawdown during underwater periods.
130 This metric averages the drawdown only over periods where the strategy
131 is underwater (drawdown > 0), unlike `avg_drawdown` which averages
132 over all periods (including zeros).
134 Args:
135 series (pl.Series): The series to calculate expected drawdown for.
137 Returns:
138 float: The expected drawdown as a positive fraction (e.g. 0.15 for 15%).
139 Returns 0.0 when there are no underwater periods.
141 """
142 underwater = _drawdown_underwater(series)
143 if underwater.is_empty():
144 return 0.0
145 return float(cast(float, underwater.mean()))
147 @columnwise_stat
148 def drawdown_value_at_risk(self, series: pl.Series, alpha: float = 0.05) -> float:
149 """Calculate the Drawdown Value-at-Risk (DD VaR) at confidence level alpha.
151 DD VaR is the alpha-quantile of the underwater drawdown distribution.
152 For alpha=0.05, it represents the drawdown level that only 5% of
153 underwater periods exceed.
155 Args:
156 series (pl.Series): The series to calculate DD VaR for.
157 alpha (float): Tail probability (e.g., 0.05 for 95% confidence).
158 Must be in (0, 1). Defaults to 0.05.
160 Returns:
161 float: The DD VaR as a positive fraction (e.g., 0.15 for 15%).
163 Raises:
164 ValueError: If alpha is not in (0, 1).
166 """
167 if not 0.0 < alpha < 1.0:
168 raise ValueError("alpha must be in (0, 1)") # noqa: TRY003
169 underwater = _drawdown_underwater(series)
170 if underwater.is_empty():
171 return float("nan")
172 return float(cast(float, underwater.quantile(1.0 - alpha, interpolation="linear")))
174 @columnwise_stat
175 def conditional_drawdown_at_risk(self, series: pl.Series, alpha: float = 0.05) -> float:
176 """Calculate the Conditional Drawdown at Risk (CDaR) at confidence level alpha.
178 Also known as Expected Drawdown Shortfall. It is the expected drawdown
179 given that the drawdown exceeds the DD VaR threshold.
181 Args:
182 series (pl.Series): The series to calculate CDaR for.
183 alpha (float): Tail probability (e.g., 0.05 for 95% confidence).
184 Must be in (0, 1). Defaults to 0.05.
186 Returns:
187 float: The CDaR as a positive fraction (e.g., 0.20 for 20%).
189 Raises:
190 ValueError: If alpha is not in (0, 1).
192 """
193 if not 0.0 < alpha < 1.0:
194 raise ValueError("alpha must be in (0, 1)") # noqa: TRY003
195 underwater = _drawdown_underwater(series)
196 if underwater.is_empty():
197 return float("nan")
198 var_threshold = underwater.quantile(1.0 - alpha, interpolation="linear")
199 tail = underwater.filter(underwater >= var_threshold)
200 if tail.is_empty(): # pragma: no cover
201 return float("nan")
202 return float(cast(float, tail.mean()))
204 @columnwise_stat
205 def tail_drawdown_ratio(self, series: pl.Series, alpha: float = 0.05) -> float:
206 """Calculate the Tail Drawdown Ratio (CDaR / Expected Drawdown).
208 This is the drawdown analog of the tail ratio. It measures how much
209 worse the tail drawdowns are compared to the average underwater drawdown.
211 Args:
212 series (pl.Series): The series to calculate tail drawdown ratio for.
213 alpha (float): Tail probability (e.g., 0.05 for 95% confidence).
214 Must be in (0, 1). Defaults to 0.05.
216 Returns:
217 float: The tail drawdown ratio (>= 1.0). Returns NaN if no drawdowns.
219 Raises:
220 ValueError: If alpha is not in (0, 1).
222 """
223 if not 0.0 < alpha < 1.0:
224 raise ValueError("alpha must be in (0, 1)") # noqa: TRY003
225 underwater = _drawdown_underwater(series)
226 if underwater.is_empty():
227 return float("nan")
228 expected = float(cast(float, underwater.mean()))
229 if expected == 0: # pragma: no cover
230 return float("nan")
231 var_threshold = underwater.quantile(1.0 - alpha, interpolation="linear")
232 tail = underwater.filter(underwater >= var_threshold)
233 if tail.is_empty(): # pragma: no cover
234 return float("nan")
235 cdar = float(cast(float, tail.mean()))
236 return cdar / expected
238 def drawdown_details(self) -> dict[str, pl.DataFrame]:
239 """Return detailed statistics for each individual drawdown period.
241 For each contiguous underwater episode, records the start date, valley
242 (worst point), recovery date, total duration, maximum drawdown, and
243 recovery duration.
245 Returns:
246 dict[str, pl.DataFrame]: Per-asset DataFrames with columns
247 ``start``, ``valley``, ``end``, ``duration``, ``max_drawdown``,
248 ``recovery_duration``.
250 Note:
251 ``end`` and ``recovery_duration`` are ``null`` for drawdown periods
252 that have not yet recovered by the last observation.
253 ``max_drawdown`` is a negative fraction (e.g. ``-0.2`` for 20%).
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()
259 result: dict[str, pl.DataFrame] = {}
260 for col, series in self._data.items():
261 nav = _nav_series(series)
262 hwm = nav.cum_max()
263 in_dd = nav < hwm
264 dd_pct = nav / hwm - 1 # negative or zero
266 if has_date and date_col_name is not None:
267 dates = all_df[date_col_name]
268 else:
269 dates = pl.Series(list(range(len(series))), dtype=pl.Int64)
271 date_dtype = dates.dtype
273 frame = (
274 pl.DataFrame({"date": dates, "nav": nav, "dd_pct": dd_pct, "in_dd": in_dd})
275 .with_row_index("row_idx")
276 .with_columns(pl.col("in_dd").rle_id().cast(pl.Int64).alias("run_id"))
277 )
279 dd_frame = frame.filter(pl.col("in_dd"))
281 # A monotonic NAV has no underwater rows, so drawdown_details should return an empty typed frame.
282 if dd_frame.is_empty():
283 result[col] = pl.DataFrame(
284 {
285 "start": pl.Series([], dtype=date_dtype),
286 "valley": pl.Series([], dtype=date_dtype),
287 "end": pl.Series([], dtype=date_dtype),
288 "duration": pl.Series([], dtype=pl.Int64),
289 "max_drawdown": pl.Series([], dtype=pl.Float64),
290 "recovery_duration": pl.Series([], dtype=pl.Int64),
291 }
292 )
293 continue
295 # Per-period stats: start, last_dd_date, valley, max drawdown
296 dd_periods = (
297 dd_frame.group_by("run_id")
298 .agg(
299 [
300 pl.col("date").first().alias("start"),
301 pl.col("date").last().alias("last_dd_date"),
302 pl.col("date").sort_by("nav").first().alias("valley"),
303 pl.col("dd_pct").min().alias("max_drawdown"),
304 ]
305 )
306 .sort("start")
307 )
309 # First date of each non-drawdown run → recovery date for the preceding drawdown run
310 non_dd_starts = (
311 frame.filter(~pl.col("in_dd"))
312 .group_by("run_id")
313 .agg(pl.col("date").first().alias("end"))
314 .with_columns((pl.col("run_id") - 1).alias("run_id"))
315 )
317 dd_periods = dd_periods.join(non_dd_starts.select(["run_id", "end"]), on="run_id", how="left")
319 # Compute durations
320 if has_date:
321 dd_periods = dd_periods.with_columns(
322 [
323 pl.when(pl.col("end").is_not_null())
324 .then((pl.col("end") - pl.col("start")).dt.total_days())
325 .otherwise((pl.col("last_dd_date") - pl.col("start")).dt.total_days() + 1)
326 .cast(pl.Int64)
327 .alias("duration"),
328 pl.when(pl.col("end").is_not_null())
329 .then((pl.col("end") - pl.col("valley")).dt.total_days().cast(pl.Int64))
330 .otherwise(pl.lit(None, dtype=pl.Int64))
331 .alias("recovery_duration"),
332 ]
333 )
334 else:
335 dd_periods = dd_periods.with_columns(
336 [
337 pl.when(pl.col("end").is_not_null())
338 .then((pl.col("end") - pl.col("start")).cast(pl.Int64))
339 .otherwise((pl.col("last_dd_date") - pl.col("start") + 1).cast(pl.Int64))
340 .alias("duration"),
341 pl.when(pl.col("end").is_not_null())
342 .then((pl.col("end") - pl.col("valley")).cast(pl.Int64))
343 .otherwise(pl.lit(None, dtype=pl.Int64))
344 .alias("recovery_duration"),
345 ]
346 )
348 result[col] = dd_periods.select(["start", "valley", "end", "duration", "max_drawdown", "recovery_duration"])
350 return result