Coverage for src/jquantstats/_stats/_rolling.py: 100%
83 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-06 04:52 +0000
1"""Rolling-window statistical metrics for financial returns data."""
3from __future__ import annotations
5import math
6from typing import TYPE_CHECKING
8import numpy as np
9import polars as pl
11from ..exceptions import NoBenchmarkError, NonPositiveWindowError
12from ._core import _to_float
13from ._drawdown import _DrawdownMixin
14from ._internals import _annualization_factor
16if TYPE_CHECKING:
17 from ..data import Data
19# ── Rolling statistics mixin ─────────────────────────────────────────────────
22class _RollingStatsMixin:
23 """Mixin class providing rolling-window financial statistics methods.
25 Separates rolling-window computations from the core point-in-time metrics
26 in `_core`. The concrete `Stats` class inherits from both.
27 """
29 _data: Data
30 all: pl.DataFrame
32 if TYPE_CHECKING:
33 from .._protocol import DataLike
35 data: DataLike
37 def implied_volatility(self, periods: int = 252, annualize: bool = True) -> pl.DataFrame | dict[str, float]:
38 """Calculate implied volatility using log returns.
40 Uses log returns (ln(1 + r)) instead of simple returns for mathematical
41 correctness with continuous compounding.
43 When ``annualize=True`` (default), returns a rolling DataFrame of
44 annualised log-return volatility: ``rolling_std(periods) * sqrt(periods)``.
45 When ``annualize=False``, returns a scalar standard deviation per asset.
47 Args:
48 periods (int): Rolling window size and annualisation factor. Defaults to 252.
49 annualize (bool): Whether to annualize and return a rolling series.
50 Defaults to True.
52 Returns:
53 pl.DataFrame: Rolling annualised implied volatility (one column per
54 asset) when ``annualize=True``.
55 dict[str, float]: Scalar log-return std per asset when
56 ``annualize=False``.
58 """
59 if annualize:
60 scale = _annualization_factor(periods)
61 return self.all.select(
62 [pl.col(name) for name in self._data.date_col]
63 + [
64 ((1.0 + pl.col(col)).log(math.e).rolling_std(window_size=periods) * scale).alias(col)
65 for col, _ in self._data.items()
66 ]
67 )
68 return {
69 col: _to_float((1.0 + series.cast(pl.Float64)).log(math.e).cast(pl.Float64).std())
70 for col, series in self._data.items()
71 }
73 @staticmethod
74 def _pct_rank_series(s: pl.Series) -> float:
75 """Percentile rank of the last element among all elements (pandas average method).
77 Args:
78 s (pl.Series): Window of price values.
80 Returns:
81 float: Rank of s[-1] in [0, 100].
83 """
84 arr = s.to_numpy()
85 current = arr[-1]
86 n = len(arr)
87 below = float(np.sum(arr < current))
88 equal = float(np.sum(arr == current))
89 return (below + (equal + 1) / 2) / n * 100.0
91 def pct_rank(self, window: int = 60) -> pl.DataFrame:
92 """Calculate the rolling percentile rank of prices within a window.
94 Converts returns to a cumulative price series, then for each period
95 returns the percentile rank (0-100) of the current price within the
96 trailing ``window`` prices. Matches ``qs.stats.pct_rank`` (pandas
97 ``rank(pct=True)`` with ``method='average'``).
99 Args:
100 window (int): Rolling window size. Defaults to 60.
102 Returns:
103 pl.DataFrame: Date column(s) plus one percentile-rank column per asset.
105 Raises:
106 ValueError: If window is not a positive integer.
108 """
109 if not isinstance(window, int) or window <= 0:
110 raise NonPositiveWindowError("window")
112 cols: list[pl.Expr | pl.Series] = [pl.col(name) for name in self._data.date_col]
113 for col, series in self._data.items():
114 prices = _DrawdownMixin.prices(series)
115 ranked = prices.rolling_map(
116 function=self._pct_rank_series,
117 window_size=window,
118 ).alias(col)
119 cols.append(ranked)
121 return self.all.select(cols)
123 def rolling_sortino(
124 self,
125 rolling_period: int = 126,
126 periods_per_year: int | float | None = None,
127 ) -> pl.DataFrame:
128 """Calculate the rolling Sortino ratio.
130 Args:
131 rolling_period: Rolling window size. Defaults to 126.
132 periods_per_year: Periods per year for annualisation.
134 Returns:
135 pl.DataFrame: Date column(s) plus one annualised rolling Sortino
136 column per asset.
138 Raises:
139 ValueError: If rolling_period is not a positive integer.
141 """
142 if not isinstance(rolling_period, int) or rolling_period <= 0:
143 raise NonPositiveWindowError("rolling_period")
144 ppy = periods_per_year or self._data._periods_per_year
145 scale = _annualization_factor(ppy)
146 exprs: list[pl.Expr] = []
147 for col, _ in self._data.items():
148 mean_ret = pl.col(col).rolling_mean(window_size=rolling_period)
149 negative_squared = pl.when(pl.col(col) < 0).then(pl.col(col) ** 2).otherwise(0.0)
150 downside = negative_squared.rolling_mean(window_size=rolling_period)
151 exprs.append(((mean_ret / downside.sqrt()) * scale).alias(col))
152 return self.all.select([pl.col(name) for name in self._data.date_col] + exprs)
154 def rolling_sharpe(
155 self,
156 rolling_period: int = 126,
157 periods_per_year: int | float | None = None,
158 ) -> pl.DataFrame:
159 """Calculate the rolling Sharpe ratio.
161 Args:
162 rolling_period: Rolling window size. Defaults to 126.
163 periods_per_year: Periods per year for annualisation.
165 Returns:
166 pl.DataFrame: Date column(s) plus one annualised rolling Sharpe
167 column per asset.
169 Raises:
170 ValueError: If rolling_period is not a positive integer.
172 """
173 actual_window = rolling_period
174 actual_periods = periods_per_year or self._data._periods_per_year
175 if not isinstance(actual_window, int) or actual_window <= 0:
176 raise NonPositiveWindowError("rolling_period")
177 scale = _annualization_factor(actual_periods)
178 return self.all.select(
179 [pl.col(name) for name in self._data.date_col]
180 + [
181 (
182 pl.col(col).rolling_mean(window_size=actual_window)
183 / pl.col(col).rolling_std(window_size=actual_window)
184 * scale
185 ).alias(col)
186 for col, _ in self._data.items()
187 ]
188 )
190 def rolling_greeks(
191 self,
192 rolling_period: int = 126,
193 periods_per_year: int | float | None = None,
194 benchmark: str | None = None,
195 ) -> pl.DataFrame:
196 """Rolling alpha and beta versus the benchmark.
198 Computes rolling alpha (annualised) and beta for each asset against the
199 benchmark using a trailing window. Beta is estimated via the standard
200 OLS formula: ``cov(asset, bench) / var(bench)``. Alpha is the
201 per-period intercept annualised by multiplying by *periods_per_year*.
203 Args:
204 rolling_period (int): Trailing window size. Defaults to 126.
205 periods_per_year (int | float, optional): Periods per year used to
206 annualise alpha. Defaults to the value inferred from the data.
207 benchmark (str, optional): Benchmark column name. Defaults to the
208 first benchmark column.
210 Returns:
211 pl.DataFrame: Date column(s) followed by ``{asset}_alpha`` and
212 ``{asset}_beta`` columns for every asset.
214 Raises:
215 AttributeError: If no benchmark data is attached.
216 ValueError: If *rolling_period* is not a positive integer.
217 """
218 if self._data.benchmark is None:
219 raise NoBenchmarkError
220 if not isinstance(rolling_period, int) or rolling_period <= 0:
221 raise NonPositiveWindowError("rolling_period")
223 ppy = periods_per_year or self._data._periods_per_year
224 all_df = self.all
225 bench_col = benchmark or self._data.benchmark.columns[0]
227 w = rolling_period
228 exprs: list[pl.Expr] = []
229 for col, _ in self._data.items():
230 mean_x = pl.col(col).rolling_mean(window_size=w)
231 mean_y = pl.col(bench_col).rolling_mean(window_size=w)
232 mean_xy = (pl.col(col) * pl.col(bench_col)).rolling_mean(window_size=w)
233 mean_y2 = (pl.col(bench_col) ** 2).rolling_mean(window_size=w)
235 bench_var = mean_y2 - mean_y**2
236 bench_cov = mean_xy - mean_x * mean_y
238 # beta = cov(asset, bench) / var(bench); NaN when var(bench) = 0
239 beta_expr = (bench_cov / bench_var).alias(f"{col}_beta")
240 # alpha (per period) = mean(asset) - beta * mean(bench), annualised
241 alpha_expr = ((mean_x - (bench_cov / bench_var) * mean_y) * ppy).alias(f"{col}_alpha")
243 exprs.extend([beta_expr, alpha_expr])
245 return all_df.select([pl.col(name) for name in self._data.date_col] + exprs)
247 def rolling_volatility(
248 self,
249 rolling_period: int = 126,
250 periods_per_year: int | float | None = None,
251 annualize: bool = True,
252 ) -> pl.DataFrame:
253 """Calculate the rolling volatility of returns.
255 Args:
256 rolling_period: Rolling window size. Defaults to 126.
257 periods_per_year: Periods per year for annualisation.
258 annualize: Multiply by ``sqrt(periods_per_year)`` when True (default).
260 Returns:
261 pl.DataFrame: Date column(s) plus one rolling volatility column
262 per asset.
264 Raises:
265 ValueError: If rolling_period is not a positive integer.
266 TypeError: If periods_per_year is not numeric.
268 """
269 actual_window = rolling_period
270 actual_periods = periods_per_year or self._data._periods_per_year
271 if not isinstance(actual_window, int) or actual_window <= 0:
272 raise NonPositiveWindowError("rolling_period")
273 if not isinstance(actual_periods, int | float):
274 raise TypeError
275 factor = _annualization_factor(actual_periods) if annualize else 1.0
276 return self.all.select(
277 [pl.col(name) for name in self._data.date_col]
278 + [
279 (pl.col(col).rolling_std(window_size=actual_window) * factor).alias(col)
280 for col, _ in self._data.items()
281 ]
282 )