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

1"""Composite ratios, streak, outlier and autocorrelation statistics. 

2 

3`_BasicStatsMixin` extends `_BasicCoreMixin` with the higher-level 

4metrics that combine the core statistics (e.g. CPC index, common-sense 

5ratio, Kelly criterion) together with streak, outlier and 

6autocorrelation analytics. It remains the public mixin consumed by 

7`Stats`. 

8""" 

9 

10from __future__ import annotations 

11 

12import math 

13from typing import cast 

14 

15import numpy as np 

16import polars as pl 

17 

18from ._basic_core import _BasicCoreMixin 

19from ._core import _mean, columnwise_stat 

20 

21 

22class _BasicStatsMixin(_BasicCoreMixin): 

23 """Mixin providing basic return/risk and win/loss financial statistics. 

24 

25 Extends `_BasicCoreMixin` with composite ratios (CPC, common-sense, 

26 Kelly), streak and outlier analytics, and autocorrelation metrics. 

27 It is the public mixin consumed by `Stats`. 

28 """ 

29 

30 @columnwise_stat 

31 def autocorr_penalty(self, series: pl.Series) -> float: 

32 """Calculate the autocorrelation penalty for risk-adjusted metrics. 

33 

34 Computes a penalty factor that accounts for autocorrelation in returns, 

35 which can inflate Sharpe and Sortino ratios. 

36 

37 Args: 

38 series (pl.Series): The series to calculate autocorrelation penalty for. 

39 

40 Returns: 

41 float: Autocorrelation penalty factor (>= 1). 

42 

43 """ 

44 arr = series.drop_nulls().to_numpy() 

45 num = len(arr) 

46 coef = float(np.abs(np.corrcoef(arr[:-1], arr[1:])[0, 1])) 

47 x = np.arange(1, num) 

48 corr = ((num - x) / num) * (coef**x) 

49 return float(np.sqrt(1 + 2 * np.sum(corr))) 

50 

51 @staticmethod 

52 def _max_consecutive(mask: pl.Series) -> int: 

53 """Return the longest run of True values in a boolean mask. 

54 

55 Args: 

56 mask (pl.Series): Boolean series (True = qualifying period). 

57 

58 Returns: 

59 int: Length of the longest consecutive True run. 

60 

61 """ 

62 group_ids = mask.rle_id() 

63 df = pl.DataFrame({"v": mask.cast(pl.Int32), "g": group_ids}) 

64 result = ( 

65 df.with_columns((pl.int_range(pl.len()).over("g") + 1).alias("rank")) 

66 .select((pl.col("v") * pl.col("rank")).max()) 

67 .item() 

68 ) 

69 return int(result) if result is not None else 0 

70 

71 @columnwise_stat 

72 def consecutive_wins(self, series: pl.Series) -> int: 

73 """Calculate the maximum number of consecutive winning periods. 

74 

75 Args: 

76 series (pl.Series): The series to calculate consecutive wins for. 

77 

78 Returns: 

79 int: Maximum number of consecutive winning periods. 

80 

81 """ 

82 return self._max_consecutive(series > 0) 

83 

84 @columnwise_stat 

85 def consecutive_losses(self, series: pl.Series) -> int: 

86 """Calculate the maximum number of consecutive losing periods. 

87 

88 Args: 

89 series (pl.Series): The series to calculate consecutive losses for. 

90 

91 Returns: 

92 int: Maximum number of consecutive losing periods. 

93 

94 """ 

95 return self._max_consecutive(series < 0) 

96 

97 @columnwise_stat 

98 def risk_of_ruin(self, series: pl.Series) -> float: 

99 """Calculate the risk of ruin (probability of losing all capital). 

100 

101 Uses the formula: ((1 - win_rate) / (1 + win_rate)) ^ n, 

102 where n is the number of periods. 

103 

104 Args: 

105 series (pl.Series): The series to calculate risk of ruin for. 

106 

107 Returns: 

108 float: The risk of ruin probability. 

109 

110 """ 

111 num_pos = self._positive(series).count() 

112 num_nonzero = series.filter(series != 0).count() 

113 wins = float(num_pos / num_nonzero) 

114 n = series.len() 

115 return ((1 - wins) / (1 + wins)) ** n 

116 

117 @columnwise_stat 

118 def tail_ratio(self, series: pl.Series, cutoff: float = 0.95) -> float: 

119 """Calculate the tail ratio (right tail / left tail). 

120 

121 Measures the ratio between the upper and lower tails of the return 

122 distribution: abs(quantile(cutoff) / quantile(1 - cutoff)). 

123 

124 Args: 

125 series (pl.Series): The series to calculate tail ratio for. 

126 cutoff (float): Percentile cutoff for tail analysis. Defaults to 0.95. 

127 

128 Returns: 

129 float: Tail ratio. 

130 

131 

132 Returns NaN when: 

133 ``float("nan")`` when either quantile is missing or the lower quantile 

134 is zero. 

135 """ 

136 upper = cast(float, series.quantile(cutoff, interpolation="linear")) 

137 lower = cast(float, series.quantile(1 - cutoff, interpolation="linear")) 

138 if upper is None or lower is None or lower == 0: 

139 return float("nan") # indeterminate: zero or missing quantile 

140 return float(np.abs(upper / lower)) 

141 

142 def cpc_index(self) -> dict[str, float]: 

143 """Calculate the CPC Index (Profit Factor * Win Rate * Win-Loss Ratio). 

144 

145 Returns: 

146 dict[str, float]: Dictionary mapping asset names to CPC Index values. 

147 

148 """ 

149 pf = self.profit_factor() 

150 wr = self.win_rate() 

151 wlr = self.payoff_ratio() 

152 return {col: pf[col] * wr[col] * wlr[col] for col in pf} 

153 

154 def common_sense_ratio(self) -> dict[str, float]: 

155 """Calculate the Common Sense Ratio (Profit Factor * Tail Ratio). 

156 

157 Returns: 

158 dict[str, float]: Dictionary mapping asset names to Common Sense Ratio values. 

159 

160 """ 

161 pf = self.profit_factor() 

162 tr = self.tail_ratio() 

163 return {col: pf[col] * tr[col] for col in pf} 

164 

165 def outliers(self, quantile: float = 0.95) -> dict[str, pl.Series]: 

166 """Return only the returns above a quantile threshold. 

167 

168 Args: 

169 quantile (float): Upper quantile threshold. Defaults to 0.95. 

170 

171 Returns: 

172 dict[str, pl.Series]: Filtered series per asset containing only 

173 returns above the quantile. 

174 

175 """ 

176 result = {} 

177 for col, series in self._data.items(): 

178 threshold = cast(float, series.quantile(quantile, interpolation="linear")) 

179 result[col] = series.filter(series > threshold).drop_nulls() 

180 return result 

181 

182 def remove_outliers(self, quantile: float = 0.95) -> dict[str, pl.Series]: 

183 """Return returns with values above a quantile threshold removed. 

184 

185 Args: 

186 quantile (float): Upper quantile threshold. Defaults to 0.95. 

187 

188 Returns: 

189 dict[str, pl.Series]: Filtered series per asset containing only 

190 returns below the quantile. 

191 

192 """ 

193 result = {} 

194 for col, series in self._data.items(): 

195 threshold = cast(float, series.quantile(quantile, interpolation="linear")) 

196 result[col] = series.filter(series < threshold) 

197 return result 

198 

199 @columnwise_stat 

200 def outlier_win_ratio(self, series: pl.Series, quantile: float = 0.99) -> float: 

201 """Calculate the outlier winners ratio. 

202 

203 Ratio of the high-quantile return to the mean positive return, 

204 showing how much outlier wins contribute to overall performance. 

205 

206 Args: 

207 series (pl.Series): The series to calculate outlier win ratio for. 

208 quantile (float): Quantile for the outlier threshold. Defaults to 0.99. 

209 

210 Returns: 

211 float: Outlier win ratio. 

212 

213 

214 Returns NaN when: 

215 ``float("nan")`` when the mean of non-negative returns is zero. 

216 """ 

217 positive_mean = _mean(series.filter(series >= 0)) 

218 if positive_mean == 0: 

219 return float("nan") # indeterminate: zero mean of positive returns 

220 quantile_val = cast(float, series.quantile(quantile, interpolation="linear")) 

221 return float(quantile_val / positive_mean) 

222 

223 @columnwise_stat 

224 def outlier_loss_ratio(self, series: pl.Series, quantile: float = 0.01) -> float: 

225 """Calculate the outlier losers ratio. 

226 

227 Ratio of the low-quantile return to the mean negative return, 

228 showing how much outlier losses contribute to overall risk. 

229 

230 Args: 

231 series (pl.Series): The series to calculate outlier loss ratio for. 

232 quantile (float): Quantile for the outlier threshold. Defaults to 0.01. 

233 

234 Returns: 

235 float: Outlier loss ratio. 

236 

237 

238 Returns NaN when: 

239 ``float("nan")`` when the mean of negative returns is zero. 

240 """ 

241 negative_mean = self._mean_negative_expr(series) 

242 if negative_mean == 0: # pragma: no cover 

243 return float("nan") # indeterminate: zero mean of negative returns 

244 quantile_val = cast(float, series.quantile(quantile, interpolation="linear")) 

245 return float(quantile_val / negative_mean) 

246 

247 @columnwise_stat 

248 def gain_to_pain_ratio(self, series: pl.Series) -> float: 

249 """Calculate Jack Schwager's Gain-to-Pain Ratio. 

250 

251 The ratio is calculated as total return / sum of losses (in absolute value). 

252 

253 Args: 

254 series (pl.Series): The series to calculate gain to pain ratio for. 

255 

256 Returns: 

257 float: The gain to pain ratio value. 

258 

259 

260 Returns NaN when: 

261 ``float("nan")`` when there are no losses (the denominator is zero). 

262 """ 

263 total_gain = series.sum() 

264 total_pain = self._negative(series).abs().sum() 

265 try: 

266 return float(float(total_gain) / float(total_pain)) 

267 except ZeroDivisionError: 

268 return float("nan") # indeterminate: no losses (denominator is zero) 

269 

270 @columnwise_stat 

271 def risk_return_ratio(self, series: pl.Series) -> float: 

272 """Calculate the return/risk ratio. 

273 

274 This is equivalent to the Sharpe ratio without a risk-free rate. 

275 

276 Args: 

277 series (pl.Series): The series to calculate risk return ratio for. 

278 

279 Returns: 

280 float: The risk return ratio value. 

281 

282 """ 

283 mean_val = _mean(series) 

284 std_val = cast(float, series.std()) 

285 return mean_val / (std_val if std_val is not None else 1.0) 

286 

287 def kelly_criterion(self) -> dict[str, float]: 

288 """Calculate the optimal capital allocation per column. 

289 

290 Uses the Kelly Criterion formula: f* = [(b * p) - q] / b 

291 where: 

292 - b = payoff ratio 

293 - p = win rate 

294 - q = 1 - p. 

295 

296 Returns: 

297 dict[str, float]: Dictionary mapping asset names to Kelly criterion values. 

298 

299 """ 

300 b = self.payoff_ratio() 

301 p = self.win_rate() 

302 

303 return {col: ((b[col] * p[col]) - (1 - p[col])) / b[col] for col in b} 

304 

305 @columnwise_stat 

306 def best(self, series: pl.Series) -> float | None: 

307 """Find the maximum return per column (best period). 

308 

309 Args: 

310 series (pl.Series): The series to find the best return for. 

311 

312 Returns: 

313 float: The maximum return value. 

314 

315 """ 

316 val = cast(float, series.max()) 

317 return val if val is not None else None 

318 

319 @columnwise_stat 

320 def worst(self, series: pl.Series) -> float | None: 

321 """Find the minimum return per column (worst period). 

322 

323 Args: 

324 series (pl.Series): The series to find the worst return for. 

325 

326 Returns: 

327 float: The minimum return value. 

328 

329 """ 

330 val = cast(float, series.min()) 

331 return val if val is not None else None 

332 

333 @columnwise_stat 

334 def exposure(self, series: pl.Series) -> float: 

335 """Calculate the market exposure time (returns != 0). 

336 

337 Args: 

338 series (pl.Series): The series to calculate exposure for. 

339 

340 Returns: 

341 float: The exposure value. 

342 

343 """ 

344 all_data = self.all 

345 ex = series.filter(series != 0).count() / all_data.height 

346 return math.ceil(ex * 100) / 100 

347 

348 @staticmethod 

349 def _pearson_corr_shifted(series: pl.Series, lag: int) -> float: 

350 """Compute Pearson correlation between *series* and its lag-*lag* shift. 

351 

352 Args: 

353 series (pl.Series): The input series. 

354 lag (int): Number of positions to shift. 

355 

356 Returns: 

357 float: Pearson correlation coefficient, or NaN if no valid pairs remain. 

358 

359 """ 

360 shifted = series.shift(lag) 

361 paired = pl.DataFrame({"x": series, "y": shifted}).drop_nulls() 

362 # Large lags or null-only overlap can leave no aligned observations to correlate. 

363 if paired.is_empty(): 

364 return float("nan") 

365 return float(np.corrcoef(paired["x"].to_numpy(), paired["y"].to_numpy())[0, 1]) 

366 

367 @columnwise_stat 

368 def autocorr(self, series: pl.Series, lag: int = 1) -> float: 

369 """Compute lag-n autocorrelation of returns. 

370 

371 Args: 

372 series (pl.Series): The series to calculate autocorrelation for. 

373 lag (int): Number of periods to lag. Must be a positive integer. 

374 

375 Returns: 

376 float: Pearson correlation between returns and their lagged values. 

377 

378 Raises: 

379 TypeError: If *lag* is not an ``int``. 

380 ValueError: If *lag* is not a positive integer (>= 1). 

381 

382 """ 

383 if not isinstance(lag, int): 

384 msg = f"lag must be an int, got {type(lag).__name__}" 

385 raise TypeError(msg) 

386 if lag <= 0: 

387 msg = f"lag must be a positive integer, got {lag}" 

388 raise ValueError(msg) 

389 return self._pearson_corr_shifted(series, lag) 

390 

391 def acf(self, nlags: int = 20) -> pl.DataFrame: 

392 """Compute the autocorrelation function up to nlags. 

393 

394 Args: 

395 nlags (int): Maximum number of lags to include. Default is 20. 

396 

397 Returns: 

398 pl.DataFrame: DataFrame with a ``lag`` column (0..nlags) and one 

399 column per asset containing the ACF values. 

400 

401 Raises: 

402 TypeError: If *nlags* is not an ``int``. 

403 ValueError: If *nlags* is negative. 

404 

405 """ 

406 if not isinstance(nlags, int): 

407 msg = f"nlags must be an int, got {type(nlags).__name__}" 

408 raise TypeError(msg) 

409 if nlags < 0: 

410 msg = f"nlags must be non-negative, got {nlags}" 

411 raise ValueError(msg) 

412 result: dict[str, list[float]] = {"lag": list(range(nlags + 1))} 

413 for col, series in self._data.items(): 

414 acf_values: list[float] = [1.0] 

415 for k in range(1, nlags + 1): 

416 acf_values.append(self._pearson_corr_shifted(series, k)) 

417 result[col] = acf_values 

418 return pl.DataFrame(result)