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 

91 None, which infers the annualisation factor from the data. 

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

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

94 ``sqrt(periods_per_year)``. Defaults to ``False``, returning the raw 

95 information ratio, which matches the value returned by 

96 ``qs.stats.information_ratio``. Set to ``True`` to annualise. 

97 

98 Returns: 

99 float: The information ratio value. 

100 

101 """ 

102 if self._data.benchmark is None: 

103 raise NoBenchmarkError 

104 

105 ppy = periods_per_year or self._data._periods_per_year 

106 

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

108 all_series = self.all 

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

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

111 

112 mean_f = _mean(active) 

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

114 

115 try: 

116 std_f = std_val if std_val is not None else 1.0 

117 ir = mean_f / std_f 

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

119 except ZeroDivisionError: 

120 return 0.0 

121 

122 @columnwise_stat 

123 def greeks( 

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

125 ) -> dict[str, float]: 

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

127 

128 Args: 

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

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

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

132 

133 Returns: 

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

135 

136 

137 Returns NaN when: 

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

139 is zero. 

140 """ 

141 ppy = periods_per_year or self._data._periods_per_year 

142 

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

144 benchmark_col = benchmark or benchmark_data.columns[0] 

145 

146 # Evaluate both series and benchmark as Series 

147 all_data = self.all 

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

149 matrix = dframe.to_numpy() 

150 

151 # Get actual Series 

152 strategy_np = matrix[:, 0] 

153 benchmark_np = matrix[:, 1] 

154 

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

156 cov_matrix = np.cov(strategy_np, benchmark_np) 

157 

158 cov = cov_matrix[0, 1] 

159 var_benchmark = cov_matrix[1, 1] 

160 

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

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

163 

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

165 

166 @columnwise_stat 

167 def treynor_ratio( 

168 self, 

169 series: pl.Series, 

170 periods: int | float | None = None, 

171 benchmark: str | None = None, 

172 ) -> float: 

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

174 

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

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

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

178 

179 Args: 

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

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

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

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

184 first benchmark column. 

185 

186 Returns: 

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

188 is unavailable. 

189 

190 Raises: 

191 AttributeError: If no benchmark data is attached. 

192 

193 Returns NaN when: 

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

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

196 """ 

197 if self._data.benchmark is None: 

198 raise NoBenchmarkError 

199 

200 ppy = periods or self._data._periods_per_year 

201 

202 benchmark_data = self._data.benchmark 

203 benchmark_col = benchmark or benchmark_data.columns[0] 

204 

205 all_data = self.all 

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

207 matrix = dframe.to_numpy() 

208 strategy_np = matrix[:, 0] 

209 benchmark_np = matrix[:, 1] 

210 

211 cov_matrix = np.cov(strategy_np, benchmark_np) 

212 var_benchmark = cov_matrix[1, 1] 

213 if var_benchmark == 0: 

214 return float("nan") 

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

216 if beta == 0: 

217 return float("nan") 

218 

219 n = len(series) 

220 if n == 0: 

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

222 nav_final = 1.0 + _comp_return(series) 

223 if nav_final <= 0: 

224 return float("nan") 

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

226 return cagr / beta