Coverage for src/jquantstats/_plots/_data/_rolling.py: 100%
89 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-06 04:52 +0000
1"""Rolling risk/return metric line charts (Sharpe, Sortino, volatility, beta)."""
3from __future__ import annotations
5import math
6from typing import TYPE_CHECKING
8import plotly.graph_objects as go
9import polars as pl
11from jquantstats.exceptions import NoBenchmarkError
13from ._styling import _apply_base_layout, _apply_figsize, _ticker_colors
15if TYPE_CHECKING:
16 from jquantstats._protocol import DataLike
19def _rolling_beta_expr(asset: str, bench_col: str, window: int) -> pl.Expr:
20 """Trailing-window OLS beta of *asset* against *bench_col*.
22 Beta is ``cov(asset, bench) / var(bench)``, expanded into rolling means so
23 the whole estimate is a single Polars expression.
25 Args:
26 asset: Asset column name.
27 bench_col: Benchmark column name.
28 window: Trailing window size in rows.
30 Returns:
31 An expression aliased ``beta``.
32 """
33 mean_x = pl.col(asset).rolling_mean(window_size=window)
34 mean_y = pl.col(bench_col).rolling_mean(window_size=window)
35 mean_xy = (pl.col(asset) * pl.col(bench_col)).rolling_mean(window_size=window)
36 mean_y2 = (pl.col(bench_col) ** 2).rolling_mean(window_size=window)
37 return ((mean_xy - mean_x * mean_y) / (mean_y2 - mean_y**2)).alias("beta")
40class _RollingPlotsMixin:
41 """Rolling-window metric plots for :class:`DataPlots`."""
43 __slots__ = ()
45 _data: DataLike
47 def _beta_assets(self, df: pl.DataFrame, date_col: str, bench_col: str) -> list[str]:
48 """Asset columns to plot beta for.
50 Prefers the explicit ``returns`` frame when the data exposes one, and
51 otherwise falls back to every column of *df* that is neither the date
52 nor the benchmark.
54 Args:
55 df: The combined index/returns/benchmark frame.
56 date_col: Name of the date column.
57 bench_col: Name of the benchmark column.
59 Returns:
60 The asset column names.
61 """
62 returns_df = getattr(self._data, "returns", None)
63 if returns_df is not None:
64 return list(returns_df.columns)
65 return [c for c in df.columns if c != date_col and c != bench_col]
67 def rolling_sharpe(
68 self,
69 rolling_period: int = 126,
70 periods_per_year: int = 252,
71 title: str = "Rolling Sharpe Ratio",
72 ) -> go.Figure:
73 """Rolling annualised Sharpe ratio over time.
75 Computes ``rolling_mean / rolling_std * sqrt(periods_per_year)`` with a
76 trailing window of *rolling_period* observations for every column in the
77 dataset (assets and benchmark when present).
79 Args:
80 rolling_period: Trailing window size. Defaults to 126 (6 months).
81 periods_per_year: Annualisation factor. Defaults to 252.
82 title: Chart title. Defaults to ``"Rolling Sharpe Ratio"``.
84 Returns:
85 go.Figure: Interactive Plotly line chart.
87 """
88 df = self._data.all
89 date_col = df.columns[0]
90 tickers = [c for c in df.columns if c != date_col]
91 colors = _ticker_colors(tickers)
92 scale = math.sqrt(periods_per_year)
94 rolling = df.with_columns(
95 [
96 (
97 pl.col(t).rolling_mean(window_size=rolling_period)
98 / pl.col(t).rolling_std(window_size=rolling_period)
99 * scale
100 ).alias(t)
101 for t in tickers
102 ]
103 )
105 fig = go.Figure()
106 for ticker in tickers:
107 fig.add_trace(
108 go.Scatter(
109 x=rolling[date_col],
110 y=rolling[ticker],
111 mode="lines",
112 name=ticker,
113 line={"color": colors[ticker], "width": 1.5},
114 hovertemplate=f"{ticker}: %{{y:.2f}}",
115 )
116 )
118 fig.add_hline(y=0, line_width=1, line_color="gray", line_dash="dash")
119 _apply_base_layout(fig, title)
120 fig.update_yaxes(title_text=f"Sharpe ({rolling_period}-period rolling)")
121 return fig
123 def rolling_sortino(
124 self,
125 rolling_period: int = 126,
126 periods_per_year: int = 252,
127 title: str = "Rolling Sortino Ratio",
128 ) -> go.Figure:
129 """Rolling annualised Sortino ratio over time.
131 Computes ``rolling_mean / rolling_downside_std * sqrt(periods_per_year)``
132 where downside deviation considers only negative returns.
134 Args:
135 rolling_period: Trailing window size. Defaults to 126 (6 months).
136 periods_per_year: Annualisation factor. Defaults to 252.
137 title: Chart title. Defaults to ``"Rolling Sortino Ratio"``.
139 Returns:
140 go.Figure: Interactive Plotly line chart.
142 """
143 df = self._data.all
144 date_col = df.columns[0]
145 tickers = [c for c in df.columns if c != date_col]
146 colors = _ticker_colors(tickers)
147 scale = math.sqrt(periods_per_year)
149 exprs = []
150 for t in tickers:
151 mean_r = pl.col(t).rolling_mean(window_size=rolling_period)
152 downside = (
153 pl.when(pl.col(t) < 0)
154 .then(pl.col(t) ** 2)
155 .otherwise(0.0)
156 .rolling_mean(window_size=rolling_period)
157 .sqrt()
158 )
159 exprs.append((mean_r / downside * scale).alias(t))
161 rolling = df.with_columns(exprs)
163 fig = go.Figure()
164 for ticker in tickers:
165 fig.add_trace(
166 go.Scatter(
167 x=rolling[date_col],
168 y=rolling[ticker],
169 mode="lines",
170 name=ticker,
171 line={"color": colors[ticker], "width": 1.5},
172 hovertemplate=f"{ticker}: %{{y:.2f}}",
173 )
174 )
176 fig.add_hline(y=0, line_width=1, line_color="gray", line_dash="dash")
177 _apply_base_layout(fig, title)
178 fig.update_yaxes(title_text=f"Sortino ({rolling_period}-period rolling)")
179 return fig
181 def rolling_volatility(
182 self,
183 rolling_period: int = 126,
184 periods_per_year: int = 252,
185 title: str = "Rolling Volatility",
186 ) -> go.Figure:
187 """Rolling annualised volatility over time.
189 Computes ``rolling_std * sqrt(periods_per_year)`` for every column in
190 the dataset.
192 Args:
193 rolling_period: Trailing window size. Defaults to 126 (6 months).
194 periods_per_year: Annualisation factor. Defaults to 252.
195 title: Chart title. Defaults to ``"Rolling Volatility"``.
197 Returns:
198 go.Figure: Interactive Plotly line chart.
200 """
201 df = self._data.all
202 date_col = df.columns[0]
203 tickers = [c for c in df.columns if c != date_col]
204 colors = _ticker_colors(tickers)
205 scale = math.sqrt(periods_per_year)
207 rolling = df.with_columns(
208 [(pl.col(t).rolling_std(window_size=rolling_period) * scale).alias(t) for t in tickers]
209 )
211 fig = go.Figure()
212 for ticker in tickers:
213 fig.add_trace(
214 go.Scatter(
215 x=rolling[date_col],
216 y=rolling[ticker],
217 mode="lines",
218 name=ticker,
219 line={"color": colors[ticker], "width": 1.5},
220 hovertemplate=f"{ticker}: %{{y:.2%}}",
221 )
222 )
224 _apply_base_layout(fig, title)
225 fig.update_yaxes(title_text=f"Volatility ({rolling_period}-period rolling)", tickformat=".0%")
226 return fig
228 def rolling_beta(
229 self,
230 rolling_period: int = 126,
231 rolling_period2: int | None = 252,
232 title: str = "Rolling Beta",
233 figsize: tuple[int, int] | None = None,
234 ) -> go.Figure:
235 """Rolling beta versus the benchmark.
237 Plots one line per asset per window size. Beta is estimated via the
238 standard OLS formula: ``cov(asset, bench) / var(bench)`` computed over
239 a trailing window.
241 Args:
242 rolling_period: Primary trailing window size. Defaults to 126.
243 rolling_period2: Optional second window size overlaid on the same
244 chart. Defaults to 252. Pass ``None`` to omit.
245 title: Chart title. Defaults to ``"Rolling Beta"``.
246 figsize: Optional ``(width, height)`` in pixels.
248 Returns:
249 go.Figure: Interactive Plotly line chart.
251 Raises:
252 AttributeError: If no benchmark columns are present in the data.
254 """
255 df = self._data.all
256 date_col = df.columns[0]
258 benchmark_df = getattr(self._data, "benchmark", None)
259 if benchmark_df is None:
260 raise NoBenchmarkError
262 bench_col = benchmark_df.columns[0]
263 assets = self._beta_assets(df, date_col, bench_col)
264 colors = _ticker_colors(assets)
265 windows = [w for w in (rolling_period, rolling_period2) if w is not None]
266 line_styles = ["solid", "dash"]
268 fig = go.Figure()
269 for asset in assets:
270 for w, dash in zip(windows, line_styles, strict=False):
271 beta_df = df.with_columns(_rolling_beta_expr(asset, bench_col, w))
272 label = f"{asset} ({w}d)"
273 fig.add_trace(
274 go.Scatter(
275 x=beta_df[date_col],
276 y=beta_df["beta"],
277 mode="lines",
278 name=label,
279 line={"color": colors[asset], "width": 1.5, "dash": dash},
280 hovertemplate=f"{label}: %{{y:.2f}}",
281 )
282 )
284 fig.add_hline(y=1, line_width=1, line_color="gray", line_dash="dash")
285 _apply_base_layout(fig, title)
286 _apply_figsize(fig, figsize)
287 fig.update_yaxes(title_text="Beta")
288 return fig