Coverage for src/jquantstats/_plots/_specs/_rolling.py: 100%
92 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-23 04:11 +0000
1"""Spec builders for the rolling-window and per-year risk charts.
3Both facades are served from here. `Data.plots` computes its own rolling
4metrics from the returns frame; `Portfolio.plots` asks its stats facade for
5them. The marks are the same either way, which is why they share a module.
6"""
8from __future__ import annotations
10import math
11from typing import TYPE_CHECKING
13import polars as pl
15from jquantstats.exceptions import NoBenchmarkError
17from .._spec import Axis, BarSeries, Dash, FigureSpec, HoverSpec, LineSeries, Panel, RefLine, TickFormat
18from .._style import ticker_colors
20if TYPE_CHECKING:
21 from jquantstats._protocol import DataLike
23 from .._protocol import PortfolioLike
25__all__ = [
26 "annual_sharpe_spec",
27 "portfolio_rolling_sharpe_spec",
28 "portfolio_rolling_volatility_spec",
29 "rolling_beta_expr",
30 "rolling_beta_spec",
31 "rolling_sharpe_spec",
32 "rolling_sortino_spec",
33 "rolling_volatility_spec",
34 "validate_window",
35]
37# Rolling metrics are noisier than the cumulative curves, so they are drawn
38# finer to keep several overlapping series legible.
39_ROLLING_WIDTH = 1.5
41# The portfolio facade's rolling charts are finer still and take the backend's
42# palette rather than naming colours.
43_PORTFOLIO_WIDTH = 1
46def _split_columns(frame: pl.DataFrame) -> tuple[str, list[str]]:
47 """Separate the date column from the value columns.
49 Args:
50 frame: A frame whose first column is the date axis.
52 Returns:
53 tuple[str, list[str]]: The date column name and every other column.
55 """
56 date_col = frame.columns[0]
57 return date_col, [c for c in frame.columns if c != date_col]
60def validate_window(window: int) -> None:
61 """Reject a non-positive or non-integer rolling window.
63 Lives in the builder rather than a renderer so both backends reject the
64 same inputs with the same message.
66 Args:
67 window: The candidate rolling-window size.
69 Raises:
70 ValueError: If *window* is not a positive integer.
72 """
73 if not isinstance(window, int) or window <= 0:
74 raise ValueError(f"window must be a positive integer, got {window!r}") # noqa: TRY003
77def rolling_beta_expr(asset: str, bench_col: str, window: int) -> pl.Expr:
78 """Trailing-window OLS beta of *asset* against *bench_col*.
80 Beta is ``cov(asset, bench) / var(bench)``, expanded into rolling means so
81 the whole estimate is a single Polars expression.
83 Args:
84 asset: Asset column name.
85 bench_col: Benchmark column name.
86 window: Trailing window size in rows.
88 Returns:
89 pl.Expr: An expression aliased ``beta``.
91 """
92 mean_x = pl.col(asset).rolling_mean(window_size=window)
93 mean_y = pl.col(bench_col).rolling_mean(window_size=window)
94 mean_xy = (pl.col(asset) * pl.col(bench_col)).rolling_mean(window_size=window)
95 mean_y2 = (pl.col(bench_col) ** 2).rolling_mean(window_size=window)
96 return ((mean_xy - mean_x * mean_y) / (mean_y2 - mean_y**2)).alias("beta")
99def _metric_lines(
100 frame: pl.DataFrame,
101 date_col: str,
102 tickers: list[str],
103 value_format: TickFormat,
104) -> tuple[LineSeries, ...]:
105 """Build one line per ticker from an already-computed metric frame.
107 Args:
108 frame: The frame holding the rolling metric, one column per ticker.
109 date_col: Name of the date column.
110 tickers: Columns to draw, in order.
111 value_format: How tooltip values are rendered.
113 Returns:
114 tuple[LineSeries, ...]: One series per ticker.
116 """
117 colors = ticker_colors(tickers)
118 return tuple(
119 LineSeries(
120 name=ticker,
121 x=frame[date_col],
122 y=frame[ticker],
123 color=colors[ticker],
124 width=_ROLLING_WIDTH,
125 hover=HoverSpec(label=ticker, value_format=value_format, date_header=False),
126 )
127 for ticker in tickers
128 )
131def rolling_sharpe_spec(data: DataLike, rolling_period: int, periods_per_year: int, title: str) -> FigureSpec:
132 """Describe the rolling Sharpe-ratio chart.
134 Args:
135 data: The dataset to plot.
136 rolling_period: Trailing window size in rows.
137 periods_per_year: Annualisation factor.
138 title: Chart title.
140 Returns:
141 FigureSpec: One line per column, with a break-even marker.
143 """
144 df = data.all
145 date_col, tickers = _split_columns(df)
146 scale = math.sqrt(periods_per_year)
148 rolling = df.with_columns(
149 [
150 (
151 pl.col(t).rolling_mean(window_size=rolling_period)
152 / pl.col(t).rolling_std(window_size=rolling_period)
153 * scale
154 ).alias(t)
155 for t in tickers
156 ]
157 )
159 panel = Panel(
160 lines=_metric_lines(rolling, date_col, tickers, "float2"),
161 ref_lines=(RefLine(value=0, dash="dash"),),
162 yaxis=Axis(title=f"Sharpe ({rolling_period}-period rolling)"),
163 )
164 return FigureSpec(title=title, panels=(panel,))
167def rolling_sortino_spec(data: DataLike, rolling_period: int, periods_per_year: int, title: str) -> FigureSpec:
168 """Describe the rolling Sortino-ratio chart.
170 Sortino divides by downside deviation rather than total volatility, so only
171 negative returns contribute to the denominator.
173 Args:
174 data: The dataset to plot.
175 rolling_period: Trailing window size in rows.
176 periods_per_year: Annualisation factor.
177 title: Chart title.
179 Returns:
180 FigureSpec: One line per column, with a break-even marker.
182 """
183 df = data.all
184 date_col, tickers = _split_columns(df)
185 scale = math.sqrt(periods_per_year)
187 exprs = []
188 for t in tickers:
189 mean_r = pl.col(t).rolling_mean(window_size=rolling_period)
190 downside = (
191 pl.when(pl.col(t) < 0).then(pl.col(t) ** 2).otherwise(0.0).rolling_mean(window_size=rolling_period).sqrt()
192 )
193 exprs.append((mean_r / downside * scale).alias(t))
195 rolling = df.with_columns(exprs)
197 panel = Panel(
198 lines=_metric_lines(rolling, date_col, tickers, "float2"),
199 ref_lines=(RefLine(value=0, dash="dash"),),
200 yaxis=Axis(title=f"Sortino ({rolling_period}-period rolling)"),
201 )
202 return FigureSpec(title=title, panels=(panel,))
205def rolling_volatility_spec(data: DataLike, rolling_period: int, periods_per_year: int, title: str) -> FigureSpec:
206 """Describe the rolling-volatility chart.
208 Args:
209 data: The dataset to plot.
210 rolling_period: Trailing window size in rows.
211 periods_per_year: Annualisation factor.
212 title: Chart title.
214 Returns:
215 FigureSpec: One line per column. Volatility cannot be negative, so
216 there is no break-even marker.
218 """
219 df = data.all
220 date_col, tickers = _split_columns(df)
221 scale = math.sqrt(periods_per_year)
223 rolling = df.with_columns([(pl.col(t).rolling_std(window_size=rolling_period) * scale).alias(t) for t in tickers])
225 panel = Panel(
226 lines=_metric_lines(rolling, date_col, tickers, "percent2"),
227 yaxis=Axis(title=f"Volatility ({rolling_period}-period rolling)", tick_format="percent0"),
228 )
229 return FigureSpec(title=title, panels=(panel,))
232def _beta_assets(data: DataLike, df: pl.DataFrame, date_col: str, bench_col: str) -> list[str]:
233 """Asset columns to plot beta for.
235 Prefers the explicit ``returns`` frame when the data exposes one, and
236 otherwise falls back to every column that is neither the date nor the
237 benchmark.
239 Args:
240 data: The dataset being plotted.
241 df: The combined index/returns/benchmark frame.
242 date_col: Name of the date column.
243 bench_col: Name of the benchmark column.
245 Returns:
246 list[str]: The asset column names.
248 """
249 returns_df = getattr(data, "returns", None)
250 if returns_df is not None:
251 return list(returns_df.columns)
252 return [c for c in df.columns if c != date_col and c != bench_col]
255def rolling_beta_spec(
256 data: DataLike,
257 rolling_period: int,
258 rolling_period2: int | None,
259 title: str,
260 figsize: tuple[int, int] | None,
261) -> FigureSpec:
262 """Describe the rolling-beta chart.
264 Args:
265 data: The dataset to plot. Must carry a benchmark.
266 rolling_period: Primary trailing window size.
267 rolling_period2: Optional second window, overlaid dashed, or None.
268 title: Chart title.
269 figsize: Optional ``(width, height)`` in pixels.
271 Returns:
272 FigureSpec: One line per asset per window, with a marker at beta = 1.
274 Raises:
275 NoBenchmarkError: If the data carries no benchmark columns.
277 """
278 df = data.all
279 date_col, _ = _split_columns(df)
281 benchmark_df = getattr(data, "benchmark", None)
282 if benchmark_df is None:
283 raise NoBenchmarkError
285 bench_col = benchmark_df.columns[0]
286 assets = _beta_assets(data, df, date_col, bench_col)
287 colors = ticker_colors(assets)
288 windows = [w for w in (rolling_period, rolling_period2) if w is not None]
290 lines = []
291 for asset in assets:
292 # The shorter window is solid and the longer one dashed, so the pair
293 # for one asset reads as the same colour at two horizons.
294 dashes: tuple[Dash, ...] = ("solid", "dash")
295 for window, dash in zip(windows, dashes, strict=False):
296 beta_df = df.with_columns(rolling_beta_expr(asset, bench_col, window))
297 label = f"{asset} ({window}d)"
298 lines.append(
299 LineSeries(
300 name=label,
301 x=beta_df[date_col],
302 y=beta_df["beta"],
303 color=colors[asset],
304 width=_ROLLING_WIDTH,
305 dash=dash,
306 hover=HoverSpec(label=label, value_format="float2", date_header=False),
307 )
308 )
310 panel = Panel(
311 lines=tuple(lines),
312 # Beta of 1 means moving with the benchmark, which is the reference
313 # worth marking rather than zero.
314 ref_lines=(RefLine(value=1, dash="dash"),),
315 yaxis=Axis(title="Beta"),
316 )
317 return FigureSpec(title=title, panels=(panel,), figsize=figsize)
320def _portfolio_metric_lines(rolling: pl.DataFrame) -> tuple[LineSeries, ...]:
321 """Build one line per non-date column of a portfolio metric frame.
323 These charts name no colours, taking the backend's palette instead, and
324 carry no tooltips.
326 Args:
327 rolling: A frame with an optional ``date`` column and one column per
328 asset.
330 Returns:
331 tuple[LineSeries, ...]: One series per asset column.
333 """
334 dates = rolling["date"] if "date" in rolling.columns else None
335 return tuple(
336 LineSeries(name=col, x=dates, y=rolling[col], width=_PORTFOLIO_WIDTH)
337 for col in rolling.columns
338 if col != "date"
339 )
342def portfolio_rolling_sharpe_spec(portfolio: PortfolioLike, window: int) -> FigureSpec:
343 """Describe the portfolio's rolling Sharpe-ratio chart.
345 Args:
346 portfolio: The portfolio to plot.
347 window: Rolling-window size in periods.
349 Returns:
350 FigureSpec: One line per asset, with a break-even marker.
352 Raises:
353 ValueError: If *window* is not a positive integer.
355 """
356 validate_window(window)
357 panel = Panel(
358 lines=_portfolio_metric_lines(portfolio.stats.rolling_sharpe(rolling_period=window)),
359 ref_lines=(RefLine(value=0, dash="dash"),),
360 yaxis=Axis(title="Sharpe ratio"),
361 )
362 return FigureSpec(title=f"Rolling Sharpe Ratio ({window}-period window)", panels=(panel,))
365def portfolio_rolling_volatility_spec(portfolio: PortfolioLike, window: int) -> FigureSpec:
366 """Describe the portfolio's rolling-volatility chart.
368 Args:
369 portfolio: The portfolio to plot.
370 window: Rolling-window size in periods.
372 Returns:
373 FigureSpec: One line per asset. Volatility cannot be negative, so there
374 is no break-even marker.
376 Raises:
377 ValueError: If *window* is not a positive integer.
379 """
380 validate_window(window)
381 panel = Panel(
382 lines=_portfolio_metric_lines(portfolio.stats.rolling_volatility(rolling_period=window)),
383 yaxis=Axis(title="Annualised volatility"),
384 )
385 return FigureSpec(title=f"Rolling Volatility ({window}-period window)", panels=(panel,))
388def annual_sharpe_spec(portfolio: PortfolioLike) -> FigureSpec:
389 """Describe the per-calendar-year Sharpe breakdown.
391 Args:
392 portfolio: The portfolio to plot.
394 Returns:
395 FigureSpec: A grouped bar chart, one bar per year per asset.
397 """
398 breakdown = portfolio.stats.annual_breakdown()
399 sharpe_rows = breakdown.filter(pl.col("metric") == "sharpe")
400 asset_cols = [c for c in sharpe_rows.columns if c not in ("year", "metric")]
402 panel = Panel(
403 bars=tuple(BarSeries(name=asset, x=sharpe_rows["year"], y=sharpe_rows[asset]) for asset in asset_cols),
404 ref_lines=(RefLine(value=0),),
405 xaxis=Axis(title="Year"),
406 yaxis=Axis(title="Sharpe ratio"),
407 )
408 return FigureSpec(
409 title="Annual Sharpe Ratio by Year",
410 panels=(panel,),
411 # Calendar years, so no range selector; and this chart never fixed a
412 # height, letting the container size it.
413 date_range_selector=False,
414 height=None,
415 bar_mode="group",
416 )