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

1"""Benchmark-relative and factor metrics. 

2 

3Split out of :mod:`jquantstats._stats._performance`, which had grown to 714 lines 

4across four unrelated concerns. This module owns the metrics that are only 

5meaningful *against a benchmark* — R-squared, information ratio, the CAPM Greeks 

6(alpha/beta) and the Treynor ratio — each of which raises 

7:class:`~jquantstats.exceptions.NoBenchmarkError` when none is configured. 

8 

9The risk-adjusted-ratio family stays in ``_performance``; concentration metrics 

10live in ``_concentration``. 

11""" 

12 

13from __future__ import annotations 

14 

15from typing import TYPE_CHECKING, cast 

16 

17import numpy as np 

18import polars as pl 

19 

20from ..exceptions import NoBenchmarkError 

21from ._core import _mean, columnwise_stat 

22from ._internals import _comp_return 

23 

24if TYPE_CHECKING: 

25 from ..data import Data 

26 

27# ── Benchmark & factor mixin ───────────────────────────────────────────────── 

28 

29 

30class _BenchmarkStatsMixin: 

31 """Mixin providing benchmark-relative and factor analytics. 

32 

33 Covers R-squared, information ratio, the CAPM Greeks (alpha/beta) and the 

34 Treynor ratio. Every metric here requires ``self._data.benchmark`` and raises 

35 :class:`~jquantstats.exceptions.NoBenchmarkError` when it is absent — that 

36 shared precondition is what makes these a coherent unit. 

37 """ 

38 

39 _data: Data 

40 all: pl.DataFrame 

41 

42 @columnwise_stat 

43 def r_squared(self, series: pl.Series, benchmark: str | None = None) -> float: 

44 """Measure the straight line fit of the equity curve. 

45 

46 Args: 

47 series (pl.Series): The series to calculate R-squared for. 

48 benchmark (str, optional): The benchmark column name. Defaults to None. 

49 

50 Returns: 

51 float: The R-squared value. 

52 

53 Raises: 

54 AttributeError: If no benchmark data is available. 

55 

56 """ 

57 if self._data.benchmark is None: 

58 raise NoBenchmarkError 

59 

60 benchmark_col = benchmark or self._data.benchmark.columns[0] 

61 

62 # Evaluate both series and benchmark as Series 

63 all_data = self.all 

64 dframe = all_data.select([series, pl.col(benchmark_col).alias("benchmark")]).drop_nulls() 

65 

66 matrix = dframe.to_numpy() 

67 # Get actual Series 

68 

69 strategy_np = matrix[:, 0] 

70 benchmark_np = matrix[:, 1] 

71 

72 corr_matrix = np.corrcoef(strategy_np, benchmark_np) 

73 r = corr_matrix[0, 1] 

74 return float(r**2) 

75 

76 @columnwise_stat 

77 def information_ratio( 

78 self, 

79 series: pl.Series, 

80 periods_per_year: int | float | None = None, 

81 benchmark: str | None = None, 

82 annualise: bool = False, 

83 ) -> float: 

84 """Calculate the information ratio. 

85 

86 This is essentially the risk return ratio of the net profits. 

87 

88 Args: 

89 series (pl.Series): The series to calculate information ratio for. 

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

91 benchmark (str, optional): The benchmark column name. Defaults to None. 

92 annualise (bool, optional): Whether to annualise the ratio by multiplying by 

93 ``sqrt(periods_per_year)``. Defaults to ``True``. Set to ``False`` to 

94 obtain the raw (non-annualised) information ratio, which matches the value 

95 returned by ``qs.stats.information_ratio``. 

96 

97 Returns: 

98 float: The information ratio value. 

99 

100 """ 

101 if self._data.benchmark is None: 

102 raise NoBenchmarkError 

103 

104 ppy = periods_per_year or self._data._periods_per_year 

105 

106 benchmark_col = benchmark or self._data.benchmark.columns[0] 

107 all_series = self.all 

108 valid_pairs = pl.DataFrame({"strategy": series, "benchmark": all_series[benchmark_col]}).drop_nulls() 

109 active = valid_pairs["strategy"] - valid_pairs["benchmark"] 

110 

111 mean_f = _mean(active) 

112 std_val = cast(float, active.std()) 

113 

114 try: 

115 std_f = std_val if std_val is not None else 1.0 

116 ir = mean_f / std_f 

117 return float(ir * (ppy**0.5) if annualise else ir) 

118 except ZeroDivisionError: 

119 return 0.0 

120 

121 @columnwise_stat 

122 def greeks( 

123 self, series: pl.Series, periods_per_year: int | float | None = None, benchmark: str | None = None 

124 ) -> dict[str, float]: 

125 """Calculate alpha and beta of the portfolio. 

126 

127 Args: 

128 series (pl.Series): The series to calculate greeks for. 

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

130 benchmark (str, optional): The benchmark column name. Defaults to None. 

131 

132 Returns: 

133 dict[str, float]: Dictionary containing alpha and beta values. 

134 

135 

136 Returns NaN when: 

137 Both alpha and beta are ``float("nan")`` when the benchmark variance 

138 is zero. 

139 """ 

140 ppy = periods_per_year or self._data._periods_per_year 

141 

142 benchmark_data = cast(pl.DataFrame, self._data.benchmark) 

143 benchmark_col = benchmark or benchmark_data.columns[0] 

144 

145 # Evaluate both series and benchmark as Series 

146 all_data = self.all 

147 dframe = all_data.select([series, pl.col(benchmark_col).alias("benchmark")]).drop_nulls() 

148 matrix = dframe.to_numpy() 

149 

150 # Get actual Series 

151 strategy_np = matrix[:, 0] 

152 benchmark_np = matrix[:, 1] 

153 

154 # 2x2 covariance matrix: [[var_strategy, cov], [cov, var_benchmark]] 

155 cov_matrix = np.cov(strategy_np, benchmark_np) 

156 

157 cov = cov_matrix[0, 1] 

158 var_benchmark = cov_matrix[1, 1] 

159 

160 beta = float(cov / var_benchmark) if var_benchmark != 0 else float("nan") 

161 alpha = float(np.mean(strategy_np) - beta * np.mean(benchmark_np)) 

162 

163 return {"alpha": float(alpha * ppy), "beta": beta} 

164 

165 @columnwise_stat 

166 def treynor_ratio( 

167 self, 

168 series: pl.Series, 

169 periods: int | float | None = None, 

170 benchmark: str | None = None, 

171 ) -> float: 

172 """Treynor ratio: annualised excess return divided by beta. 

173 

174 Measures return per unit of systematic (market) risk. Unlike the Sharpe 

175 ratio, which divides by total volatility, the Treynor ratio divides by 

176 beta — making it most meaningful for well-diversified portfolios. 

177 

178 Args: 

179 series (pl.Series): The returns series for one asset. 

180 periods (int | float, optional): Periods per year for CAGR 

181 annualisation. Defaults to the value inferred from the data. 

182 benchmark (str, optional): Benchmark column name. Defaults to the 

183 first benchmark column. 

184 

185 Returns: 

186 float: Treynor ratio, or ``nan`` when beta is zero or the benchmark 

187 is unavailable. 

188 

189 Raises: 

190 AttributeError: If no benchmark data is attached. 

191 

192 Returns NaN when: 

193 ``float("nan")`` when the benchmark variance or beta is zero, the 

194 series is empty, or the compounded NAV is non-positive. 

195 """ 

196 if self._data.benchmark is None: 

197 raise NoBenchmarkError 

198 

199 ppy = periods or self._data._periods_per_year 

200 

201 benchmark_data = self._data.benchmark 

202 benchmark_col = benchmark or benchmark_data.columns[0] 

203 

204 all_data = self.all 

205 dframe = all_data.select([series, pl.col(benchmark_col).alias("_bench")]).drop_nulls() 

206 matrix = dframe.to_numpy() 

207 strategy_np = matrix[:, 0] 

208 benchmark_np = matrix[:, 1] 

209 

210 cov_matrix = np.cov(strategy_np, benchmark_np) 

211 var_benchmark = cov_matrix[1, 1] 

212 if var_benchmark == 0: 

213 return float("nan") 

214 beta = float(cov_matrix[0, 1] / var_benchmark) 

215 if beta == 0: 

216 return float("nan") 

217 

218 n = len(series) 

219 if n == 0: 

220 return float("nan") # pragma: no cover 

221 nav_final = 1.0 + _comp_return(series) 

222 if nav_final <= 0: 

223 return float("nan") 

224 cagr = float(nav_final ** (ppy / n) - 1.0) 

225 return cagr / beta