Coverage for src / jquantstats / _stats / _rolling.py: 100%

1"""Rolling-window statistical metrics for financial returns data.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, cast 

6 

7import numpy as np 

8import polars as pl 

9 

10from ._core import to_frame 

11 

12# ── Rolling statistics mixin ───────────────────────────────────────────────── 

13 

14 

15class _RollingStatsMixin: 

16 """Mixin class providing rolling-window financial statistics methods. 

17 

18 Separates rolling-window computations from the core point-in-time metrics 

19 in :mod:`~jquantstats._stats._core`. The concrete 

20 :class:`~jquantstats._stats.Stats` dataclass inherits from both. 

21 

22 Attributes (provided by the concrete subclass): 

23 data: The :class:`~jquantstats._data.Data` object. 

24 all: Combined DataFrame for efficient column selection. 

25 """ 

26 

27 if TYPE_CHECKING: 

28 from ._protocol import DataLike 

29 

30 data: DataLike 

31 all: pl.DataFrame | None 

32 

33 @to_frame 

34 def rolling_sortino( 

35 self, series: pl.Expr, rolling_period: int = 126, periods_per_year: int | float | None = None 

36 ) -> pl.Expr: 

37 """Calculate the rolling Sortino ratio. 

38 

39 Args: 

40 series (pl.Expr): The expression to calculate rolling Sortino ratio for. 

41 rolling_period (int, optional): The rolling window size. Defaults to 126. 

42 periods_per_year (int, optional): Number of periods per year. Defaults to 252. 

43 

44 Returns: 

45 pl.Expr: The rolling Sortino ratio expression. 

46 

47 """ 

48 ppy = periods_per_year or self.data._periods_per_year 

49 

50 mean_ret = series.rolling_mean(window_size=rolling_period) 

51 

52 # Rolling downside deviation (squared negative returns averaged over window) 

53 downside = series.map_elements(lambda x: x**2 if x < 0 else 0.0, return_dtype=pl.Float64).rolling_mean( 

54 window_size=rolling_period 

55 ) 

56 

57 # Avoid division by zero 

58 sortino = mean_ret / downside.sqrt().fill_nan(0).fill_null(0) 

59 return cast(pl.Expr, sortino * (ppy**0.5)) 

60 

61 def rolling_sharpe( 

62 self, 

63 window: int | None = None, 

64 periods: int | float | None = None, 

65 rolling_period: int | None = None, 

66 periods_per_year: int | float | None = None, 

67 ) -> pl.DataFrame: 

68 """Calculate the rolling Sharpe ratio. 

69 

70 Accepts both the analytics-style (``window``, ``periods``) and the 

71 legacy-style (``rolling_period``, ``periods_per_year``) keyword 

72 arguments so that callers using either convention continue to work. 

73 

74 Args: 

75 window: Rolling window size (analytics style). Defaults to 126. 

76 periods: Periods per year for annualisation (analytics style). 

77 rolling_period: Alias for ``window`` (legacy style). 

78 periods_per_year: Alias for ``periods`` (legacy style). 

79 

80 Returns: 

81 pl.DataFrame: Date column(s) plus one annualised rolling Sharpe 

82 column per asset. 

83 

84 Raises: 

85 ValueError: If the effective window size is not a positive integer. 

86 

87 """ 

88 actual_window = window if window is not None else (rolling_period if rolling_period is not None else 126) 

89 actual_periods = periods or periods_per_year or self.data._periods_per_year 

90 if not isinstance(actual_window, int) or actual_window <= 0: 

91 raise ValueError("window must be a positive integer") # noqa: TRY003 

92 scale = float(np.sqrt(actual_periods)) 

93 return cast(pl.DataFrame, self.all).select( 

94 [pl.col(name) for name in self.data.date_col] 

95 + [ 

96 ( 

97 pl.col(col).rolling_mean(window_size=actual_window) 

98 / pl.col(col).rolling_std(window_size=actual_window) 

99 * scale 

100 ).alias(col) 

101 for col, _ in self.data.items() 

102 ] 

103 ) 

104 

105 def rolling_volatility( 

106 self, 

107 window: int | None = None, 

108 periods: int | float | None = None, 

109 annualize: bool = True, 

110 rolling_period: int | None = None, 

111 periods_per_year: int | float | None = None, 

112 ) -> pl.DataFrame: 

113 """Calculate the rolling volatility of returns. 

114 

115 Accepts both the analytics-style (``window``, ``periods``, 

116 ``annualize``) and the legacy-style (``rolling_period``, 

117 ``periods_per_year``) keyword arguments. 

118 

119 Args: 

120 window: Rolling window size (analytics style). Defaults to 126. 

121 periods: Periods per year for annualisation (analytics style). 

122 annualize: Multiply by ``sqrt(periods)`` when True (default). 

123 rolling_period: Alias for ``window`` (legacy style). 

124 periods_per_year: Alias for ``periods`` (legacy style). 

125 

126 Returns: 

127 pl.DataFrame: Date column(s) plus one rolling volatility column 

128 per asset. 

129 

130 Raises: 

131 ValueError: If the effective window size is not a positive integer. 

132 TypeError: If the effective periods value is not numeric. 

133 

134 """ 

135 actual_window = window if window is not None else (rolling_period if rolling_period is not None else 126) 

136 actual_periods = periods or periods_per_year or self.data._periods_per_year 

137 if not isinstance(actual_window, int) or actual_window <= 0: 

138 raise ValueError("window must be a positive integer") # noqa: TRY003 

139 if not isinstance(actual_periods, int | float): 

140 raise TypeError 

141 factor = float(np.sqrt(actual_periods)) if annualize else 1.0 

142 return cast(pl.DataFrame, self.all).select( 

143 [pl.col(name) for name in self.data.date_col] 

144 + [(pl.col(col).rolling_std(window_size=actual_window) * factor).alias(col) for col, _ in self.data.items()] 

145 )