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

1"""Performance and risk-adjusted return metrics for financial data.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from typing import TYPE_CHECKING, cast 

7 

8import numpy as np 

9import polars as pl 

10from scipy.stats import norm 

11 

12from ._core import _to_float, columnwise_stat, to_frame 

13from ._internals import _annualization_factor, _comp_return, _downside_deviation, _nav_series 

14 

15# ── Performance statistics mixin ───────────────────────────────────────────── 

16 

17 

18class _PerformanceStatsMixin: 

19 """Mixin providing performance, drawdown, and benchmark/factor metrics. 

20 

21 Covers: Sharpe ratio, Sortino ratio, adjusted Sortino, drawdown series, 

22 max drawdown, prices, R-squared, information ratio, and Greeks (alpha/beta). 

23 

24 Attributes (provided by the concrete subclass): 

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

26 all: Combined DataFrame for efficient column selection. 

27 """ 

28 

29 if TYPE_CHECKING: 

30 from ._protocol import DataLike 

31 

32 data: DataLike 

33 all: pl.DataFrame | None 

34 

35 def autocorr_penalty(self) -> dict[str, float]: 

36 """Defined on _BasicStatsMixin.""" 

37 

38 def geometric_mean(self) -> dict[str, float]: 

39 """Defined on _BasicStatsMixin.""" 

40 

41 # ── Sharpe & Sortino ────────────────────────────────────────────────────── 

42 

43 @columnwise_stat 

44 def sharpe(self, series: pl.Series, periods: int | float | None = None) -> float: 

45 """Calculate the Sharpe ratio of asset returns. 

46 

47 Args: 

48 series (pl.Series): The series to calculate Sharpe ratio for. 

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

50 

51 Returns: 

52 float: The Sharpe ratio value. 

53 

54 """ 

55 periods = periods or self.data._periods_per_year 

56 

57 std_val = series.std(ddof=1) 

58 mean_val = series.mean() 

59 divisor = cast(float, std_val) if std_val is not None else 0.0 

60 mean_f = cast(float, mean_val) if mean_val is not None else 0.0 

61 

62 _eps = np.finfo(np.float64).eps 

63 if divisor <= _eps * max(abs(mean_f), _eps) * 10: 

64 return float("nan") 

65 

66 res = mean_f / divisor 

67 factor = periods or 1 

68 return float(res * _annualization_factor(factor)) 

69 

70 @columnwise_stat 

71 def sharpe_variance(self, series: pl.Series, periods: int | float | None = None) -> float: 

72 r"""Calculate the asymptotic variance of the Sharpe Ratio. 

73 

74 .. math:: 

75 \text{Var}(SR) = \frac{1 + \frac{S \cdot SR}{2} + \frac{(K - 3) \cdot SR^2}{4}}{T} 

76 

77 where: 

78 - \(S\) is the skewness of returns 

79 - \(K\) is the kurtosis of returns 

80 - \(SR\) is the Sharpe ratio (unannualized) 

81 - \(T\) is the number of observations 

82 

83 Args: 

84 series (pl.Series): The series to calculate Sharpe ratio variance for. 

85 periods (int | float, optional): Number of periods per year. Defaults to data periods. 

86 

87 Returns: 

88 float: The asymptotic variance of the Sharpe ratio. 

89 If number of periods per year is provided or inferred from the data, the result is annualized. 

90 

91 """ 

92 t = series.count() 

93 mean_val = cast(float, series.mean()) 

94 std_val = cast(float, series.std(ddof=1)) 

95 if mean_val is None or std_val is None or std_val == 0: 

96 return float(np.nan) 

97 sr = mean_val / std_val 

98 

99 skew_val = series.skew(bias=False) 

100 kurt_val = series.kurtosis(bias=False) 

101 

102 if skew_val is None or kurt_val is None: 

103 return float(np.nan) 

104 # Base variance calculation using unannualized Sharpe ratio 

105 # Formula: (1 + skew*SR/2 + (kurt-3)*SR²/4) / T 

106 base_variance = (1 + (float(skew_val) * sr) / 2 + ((float(kurt_val) - 3) / 4) * sr**2) / t 

107 # Annualize by scaling with the number of periods 

108 periods = periods or self.data._periods_per_year 

109 factor = periods or 1 

110 return float(base_variance * _annualization_factor(factor, sqrt=False)) 

111 

112 @columnwise_stat 

113 def probabilistic_sharpe_ratio(self, series: pl.Series) -> float: 

114 r"""Calculate the probabilistic sharpe ratio (PSR). 

115 

116 Args: 

117 series (pl.Series): The series to calculate probabilistic Sharpe ratio for. 

118 

119 Returns: 

120 float: Probabilistic Sharpe Ratio. 

121 

122 Note: 

123 PSR is the probability that the observed Sharpe ratio is greater than a 

124 given benchmark Sharpe ratio. 

125 

126 """ 

127 t = series.count() 

128 

129 # Calculate observed unannualized Sharpe ratio 

130 mean_val = cast(float, series.mean()) 

131 std_val = cast(float, series.std(ddof=1)) 

132 if mean_val is None or std_val is None or std_val == 0: 

133 return float(np.nan) 

134 # Unannualized observed Sharpe ratio 

135 observed_sr = mean_val / std_val 

136 

137 skew_val = series.skew(bias=False) 

138 kurt_val = series.kurtosis(bias=False) 

139 

140 if skew_val is None or kurt_val is None: 

141 return float(np.nan) 

142 

143 benchmark_sr = 0.0 

144 # Calculate variance using unannualized benchmark Sharpe ratio 

145 var_bench_sr = (1 + (float(skew_val) * benchmark_sr) / 2 + ((float(kurt_val) - 3) / 4) * benchmark_sr**2) / t 

146 

147 if var_bench_sr <= 0: 

148 return float(np.nan) # pragma: no cover 

149 return float(norm.cdf((observed_sr - benchmark_sr) / np.sqrt(var_bench_sr))) 

150 

151 @columnwise_stat 

152 def hhi_positive(self, series: pl.Series) -> float: 

153 r"""Calculate the Herfindahl-Hirschman Index (HHI) for positive returns. 

154 

155 This quantifies how concentrated the positive returns are in a series. 

156 

157 .. math:: 

158 w^{\plus} = \frac{r_{t}^{\plus}}{\sum{r_{t}^{\plus}}} \\ 

159 HHI^{\plus} = \frac{N_{\plus} \sum{(w^{\plus})^2} - 1}{N_{\plus} - 1} 

160 

161 where: 

162 - \(r_{t}^{\plus}\) are the positive returns 

163 - \(N_{\plus}\) is the number of positive returns 

164 - \(w^{\plus}\) are the weights of positive returns 

165 

166 Args: 

167 series (pl.Series): The series to calculate HHI for. 

168 

169 Returns: 

170 float: The HHI value for positive returns. Returns NaN if fewer than 3 

171 positive returns are present. 

172 

173 Note: 

174 Values range from 0 (perfectly diversified gains) to 1 (all gains 

175 concentrated in a single period). 

176 """ 

177 positive_returns = series.filter(series > 0).drop_nans() 

178 if positive_returns.len() <= 2: 

179 return float(np.nan) 

180 weight = positive_returns / positive_returns.sum() 

181 return float((weight.len() * (weight**2).sum() - 1) / (weight.len() - 1)) 

182 

183 @columnwise_stat 

184 def hhi_negative(self, series: pl.Series) -> float: 

185 r"""Calculate the Herfindahl-Hirschman Index (HHI) for negative returns. 

186 

187 This quantifies how concentrated the negative returns are in a series. 

188 

189 .. math:: 

190 w^{\minus} = \frac{r_{t}^{\minus}}{\sum{r_{t}^{\minus}}} \\ 

191 HHI^{\minus} = \frac{N_{\minus} \sum{(w^{\minus})^2} - 1}{N_{\minus} - 1} 

192 

193 where: 

194 - \(r_{t}^{\minus}\) are the negative returns 

195 - \(N_{\minus}\) is the number of negative returns 

196 - \(w^{\minus}\) are the weights of negative returns 

197 

198 Args: 

199 series (pl.Series): The returns series to calculate HHI for. 

200 

201 Returns: 

202 float: The HHI value for negative returns. Returns NaN if fewer than 3 

203 negative returns are present. 

204 

205 Note: 

206 Values range from 0 (perfectly diversified losses) to 1 (all losses 

207 concentrated in a single period). 

208 """ 

209 negative_returns = series.filter(series < 0).drop_nans() 

210 if negative_returns.len() <= 2: 

211 return float(np.nan) 

212 weight = negative_returns / negative_returns.sum() 

213 return float((weight.len() * (weight**2).sum() - 1) / (weight.len() - 1)) 

214 

215 @columnwise_stat 

216 def sortino(self, series: pl.Series, periods: int | float | None = None) -> float: 

217 """Calculate the Sortino ratio. 

218 

219 The Sortino ratio is the mean return divided by downside deviation. 

220 Based on Red Rock Capital's Sortino ratio paper. 

221 

222 Args: 

223 series (pl.Series): The series to calculate Sortino ratio for. 

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

225 

226 Returns: 

227 float: The Sortino ratio value. 

228 

229 """ 

230 periods = periods or self.data._periods_per_year 

231 downside_deviation = _downside_deviation(series) 

232 mean_val = cast(float, series.mean()) 

233 mean_f = mean_val if mean_val is not None else 0.0 

234 if downside_deviation == 0.0: 

235 if mean_f > 0: 

236 return float("inf") 

237 elif mean_f < 0: # pragma: no cover # unreachable: no negatives ⟹ mean ≥ 0 

238 return float("-inf") 

239 else: 

240 return float("nan") 

241 ratio = mean_f / downside_deviation 

242 return float(ratio * _annualization_factor(periods)) 

243 

244 @columnwise_stat 

245 def omega( 

246 self, 

247 series: pl.Series, 

248 rf: float = 0.0, 

249 required_return: float = 0.0, 

250 periods: int | float | None = None, 

251 ) -> float: 

252 """Calculate the Omega ratio. 

253 

254 The Omega ratio is the probability-weighted ratio of gains to losses 

255 relative to a threshold return. It is computed as the sum of returns 

256 above the threshold divided by the absolute sum of returns below it. 

257 

258 Args: 

259 series (pl.Series): The series to calculate Omega ratio for. 

260 rf (float): Annualised risk-free rate. Defaults to 0.0. 

261 required_return (float): Annualised minimum acceptable return 

262 threshold. Defaults to 0.0. 

263 periods (int | float | None): Number of periods per year. Defaults 

264 to the value inferred from the data. 

265 

266 Returns: 

267 float: The Omega ratio, or NaN when the denominator is zero or 

268 when ``required_return <= -1``. 

269 

270 Note: 

271 See https://en.wikipedia.org/wiki/Omega_ratio for details. 

272 

273 """ 

274 if required_return <= -1: 

275 return float("nan") 

276 

277 periods = periods or self.data._periods_per_year 

278 

279 # Subtract per-period risk-free rate from returns when rf is non-zero. 

280 if rf != 0.0: 

281 rf_per_period = float((1.0 + rf) ** (1.0 / periods) - 1.0) 

282 series = series - rf_per_period 

283 

284 # Convert annualised required return to a per-period threshold. 

285 return_threshold = float((1.0 + required_return) ** (1.0 / periods) - 1.0) 

286 

287 returns_less_thresh = series - return_threshold 

288 

289 numer = float(returns_less_thresh.filter(returns_less_thresh > 0.0).sum()) 

290 denom = float(-returns_less_thresh.filter(returns_less_thresh < 0.0).sum()) 

291 

292 if denom <= 0.0: 

293 return float("nan") 

294 return numer / denom 

295 

296 # ── Cumulative returns ──────────────────────────────────────────────────── 

297 

298 @to_frame 

299 def compsum(self, series: pl.Expr) -> pl.Expr: 

300 """Calculate the rolling compounded (cumulative) returns. 

301 

302 Computed as cumprod(1 + r) - 1 for each period. 

303 

304 Args: 

305 series (pl.Expr): The expression to calculate cumulative returns for. 

306 

307 Returns: 

308 pl.Expr: Cumulative compounded returns expression. 

309 

310 """ 

311 return (1.0 + series).cum_prod() - 1.0 

312 

313 def ghpr(self) -> dict[str, float]: 

314 """Calculate the Geometric Holding Period Return. 

315 

316 Shorthand for geometric_mean() — the per-period geometric average return. 

317 

318 Returns: 

319 dict[str, float]: Dictionary mapping asset names to GHPR values. 

320 

321 """ 

322 return self.geometric_mean() 

323 

324 # ── Drawdown ────────────────────────────────────────────────────────────── 

325 

326 @to_frame 

327 def drawdown(self, series: pl.Series) -> pl.Series: 

328 """Calculate the drawdown series for returns. 

329 

330 Args: 

331 series (pl.Series): The series to calculate drawdown for. 

332 

333 Returns: 

334 pl.Series: The drawdown series. 

335 

336 """ 

337 equity = self.prices(series) 

338 d = (equity / equity.cum_max()) - 1 

339 return -d 

340 

341 @staticmethod 

342 def prices(series: pl.Series) -> pl.Series: 

343 """Convert returns series to price series. 

344 

345 Args: 

346 series (pl.Series): The returns series to convert. 

347 

348 Returns: 

349 pl.Series: The price series. 

350 

351 """ 

352 return _nav_series(series) 

353 

354 @staticmethod 

355 def max_drawdown_single_series(series: pl.Series) -> float: 

356 """Compute the maximum drawdown for a single returns series. 

357 

358 Args: 

359 series: A Polars Series of returns values. 

360 

361 Returns: 

362 float: The maximum drawdown as a positive fraction (e.g. 0.2 for 20%). 

363 """ 

364 price = _PerformanceStatsMixin.prices(series) 

365 peak = price.cum_max() 

366 drawdown = price / peak - 1 

367 dd_min = cast(float, drawdown.min()) 

368 return dd_min if dd_min is not None else 0.0 

369 

370 @columnwise_stat 

371 def max_drawdown(self, series: pl.Series) -> float: 

372 """Calculate the maximum drawdown for each column. 

373 

374 Args: 

375 series (pl.Series): The series to calculate maximum drawdown for. 

376 

377 Returns: 

378 float: The maximum drawdown value. 

379 

380 """ 

381 return _PerformanceStatsMixin.max_drawdown_single_series(series) 

382 

383 def drawdown_details(self) -> dict[str, pl.DataFrame]: 

384 """Return detailed statistics for each individual drawdown period. 

385 

386 For each contiguous underwater episode, records the start date, valley 

387 (worst point), recovery date, total duration, maximum drawdown, and 

388 recovery duration. 

389 

390 Returns: 

391 dict[str, pl.DataFrame]: Per-asset DataFrames with columns 

392 ``start``, ``valley``, ``end``, ``duration``, ``max_drawdown``, 

393 ``recovery_duration``. 

394 

395 Note: 

396 ``end`` and ``recovery_duration`` are ``null`` for drawdown periods 

397 that have not yet recovered by the last observation. 

398 ``max_drawdown`` is a negative fraction (e.g. ``-0.2`` for 20%). 

399 """ 

400 all_df = cast(pl.DataFrame, self.all) 

401 date_col_name = self.data.date_col[0] if self.data.date_col else None 

402 has_date = date_col_name is not None and all_df[date_col_name].dtype.is_temporal() 

403 

404 result: dict[str, pl.DataFrame] = {} 

405 for col, series in self.data.items(): 

406 nav = _nav_series(series) 

407 hwm = nav.cum_max() 

408 in_dd = nav < hwm 

409 dd_pct = nav / hwm - 1 # negative or zero 

410 

411 if has_date and date_col_name is not None: 

412 dates = all_df[date_col_name] 

413 else: 

414 dates = pl.Series(list(range(len(series))), dtype=pl.Int64) 

415 

416 date_dtype = dates.dtype 

417 

418 frame = ( 

419 pl.DataFrame({"date": dates, "nav": nav, "dd_pct": dd_pct, "in_dd": in_dd}) 

420 .with_row_index("row_idx") 

421 .with_columns(pl.col("in_dd").rle_id().cast(pl.Int64).alias("run_id")) 

422 ) 

423 

424 dd_frame = frame.filter(pl.col("in_dd")) 

425 

426 if dd_frame.is_empty(): 

427 result[col] = pl.DataFrame( 

428 { 

429 "start": pl.Series([], dtype=date_dtype), 

430 "valley": pl.Series([], dtype=date_dtype), 

431 "end": pl.Series([], dtype=date_dtype), 

432 "duration": pl.Series([], dtype=pl.Int64), 

433 "max_drawdown": pl.Series([], dtype=pl.Float64), 

434 "recovery_duration": pl.Series([], dtype=pl.Int64), 

435 } 

436 ) 

437 continue 

438 

439 # Per-period stats: start, last_dd_date, valley, max drawdown 

440 dd_periods = ( 

441 dd_frame.group_by("run_id") 

442 .agg( 

443 [ 

444 pl.col("date").first().alias("start"), 

445 pl.col("date").last().alias("last_dd_date"), 

446 pl.col("date").sort_by("nav").first().alias("valley"), 

447 pl.col("dd_pct").min().alias("max_drawdown"), 

448 ] 

449 ) 

450 .sort("start") 

451 ) 

452 

453 # First date of each non-drawdown run → recovery date for the preceding drawdown run 

454 non_dd_starts = ( 

455 frame.filter(~pl.col("in_dd")) 

456 .group_by("run_id") 

457 .agg(pl.col("date").first().alias("end")) 

458 .with_columns((pl.col("run_id") - 1).alias("run_id")) 

459 ) 

460 

461 dd_periods = dd_periods.join(non_dd_starts.select(["run_id", "end"]), on="run_id", how="left") 

462 

463 # Compute durations 

464 if has_date: 

465 dd_periods = dd_periods.with_columns( 

466 [ 

467 pl.when(pl.col("end").is_not_null()) 

468 .then((pl.col("end") - pl.col("start")).dt.total_days()) 

469 .otherwise((pl.col("last_dd_date") - pl.col("start")).dt.total_days() + 1) 

470 .cast(pl.Int64) 

471 .alias("duration"), 

472 pl.when(pl.col("end").is_not_null()) 

473 .then((pl.col("end") - pl.col("valley")).dt.total_days().cast(pl.Int64)) 

474 .otherwise(pl.lit(None, dtype=pl.Int64)) 

475 .alias("recovery_duration"), 

476 ] 

477 ) 

478 else: 

479 dd_periods = dd_periods.with_columns( 

480 [ 

481 pl.when(pl.col("end").is_not_null()) 

482 .then((pl.col("end") - pl.col("start")).cast(pl.Int64)) 

483 .otherwise((pl.col("last_dd_date") - pl.col("start") + 1).cast(pl.Int64)) 

484 .alias("duration"), 

485 pl.when(pl.col("end").is_not_null()) 

486 .then((pl.col("end") - pl.col("valley")).cast(pl.Int64)) 

487 .otherwise(pl.lit(None, dtype=pl.Int64)) 

488 .alias("recovery_duration"), 

489 ] 

490 ) 

491 

492 result[col] = dd_periods.select(["start", "valley", "end", "duration", "max_drawdown", "recovery_duration"]) 

493 

494 return result 

495 

496 @staticmethod 

497 def _probabilistic_ratio_from_base(base: float, series: pl.Series) -> float: 

498 """Compute the probabilistic ratio given an observed unannualized base ratio. 

499 

500 Uses the formula: norm.cdf(base / sigma), where 

501 sigma = sqrt((1 + 0.5·base² - skew·base + (kurt-3)/4·base²) / (n-1)). 

502 

503 Args: 

504 base (float): Unannualized observed ratio (e.g. Sortino). 

505 series (pl.Series): The original returns series (for moments and n). 

506 

507 Returns: 

508 float: Probabilistic ratio in [0, 1]. 

509 

510 """ 

511 n = series.count() 

512 skew_val = series.skew(bias=False) 

513 kurt_val = series.kurtosis(bias=False) 

514 if skew_val is None or kurt_val is None or n <= 1: 

515 return float(np.nan) 

516 variance = (1 + 0.5 * base**2 - float(skew_val) * base + ((float(kurt_val) - 3) / 4) * base**2) / (n - 1) 

517 if variance <= 0: 

518 return float(np.nan) 

519 return float(norm.cdf(base / np.sqrt(variance))) 

520 

521 @columnwise_stat 

522 def probabilistic_sortino_ratio(self, series: pl.Series, periods: int | float | None = None) -> float: 

523 """Calculate the Probabilistic Sortino Ratio. 

524 

525 The probability that the observed Sortino ratio is greater than zero, 

526 accounting for estimation uncertainty via skewness and kurtosis. 

527 

528 Args: 

529 series (pl.Series): The series to calculate the ratio for. 

530 periods (int | float, optional): Accepted for API compatibility; has no effect 

531 since the base ratio is un-annualized. 

532 

533 Returns: 

534 float: Probabilistic Sortino ratio in [0, 1]. 

535 

536 """ 

537 downside_deviation = _downside_deviation(series) 

538 mean_val = cast(float, series.mean()) 

539 mean_f = mean_val if mean_val is not None else 0.0 

540 if downside_deviation == 0.0: 

541 return float(np.nan) 

542 base = float(mean_f / downside_deviation) 

543 return self._probabilistic_ratio_from_base(base, series) 

544 

545 @columnwise_stat 

546 def probabilistic_adjusted_sortino_ratio(self, series: pl.Series, periods: int | float | None = None) -> float: 

547 """Calculate the Probabilistic Adjusted Sortino Ratio. 

548 

549 The probability that the observed adjusted Sortino ratio (divided by sqrt(2) 

550 for Sharpe comparability) is greater than zero, accounting for estimation 

551 uncertainty via skewness and kurtosis. 

552 

553 Args: 

554 series (pl.Series): The series to calculate the ratio for. 

555 periods (int | float, optional): Accepted for API compatibility; has no effect 

556 since the base ratio is un-annualized. 

557 

558 Returns: 

559 float: Probabilistic adjusted Sortino ratio in [0, 1]. 

560 

561 """ 

562 downside_deviation = _downside_deviation(series) 

563 mean_val = cast(float, series.mean()) 

564 mean_f = mean_val if mean_val is not None else 0.0 

565 if downside_deviation == 0.0: 

566 return float(np.nan) 

567 base = float(mean_f / downside_deviation) / np.sqrt(2) 

568 return self._probabilistic_ratio_from_base(base, series) 

569 

570 def probabilistic_ratio( 

571 self, 

572 base: str | Callable[[pl.Series], float] = "sharpe", 

573 ) -> dict[str, float]: 

574 r"""Generic probabilistic ratio for any base metric. 

575 

576 Computes the probability that the observed ratio is greater than zero, 

577 accounting for estimation uncertainty via skewness and kurtosis using 

578 the Lopez de Prado (2018) framework. 

579 

580 Args: 

581 base: Base ratio to use. Either: 

582 

583 - A string: ``'sharpe'``, ``'sortino'``, ``'adjusted_sortino'``. 

584 - A callable ``(series: pl.Series) -> float`` returning the 

585 **unannualized** ratio for a single series. 

586 

587 Returns: 

588 dict[str, float]: Probabilistic ratio in ``[0, 1]`` per asset. 

589 

590 Raises: 

591 ValueError: If *base* is an unrecognised string. 

592 

593 """ 

594 

595 def _sharpe_base(s: pl.Series) -> float: 

596 """Return the per-period Sharpe ratio (mean / std, ddof=1) of *s*.""" 

597 mean_val = cast(float, s.mean()) 

598 std_val = cast(float, s.std(ddof=1)) 

599 if not std_val or std_val == 0: 

600 return float("nan") 

601 return mean_val / std_val 

602 

603 def _sortino_base(s: pl.Series) -> float: 

604 """Return the per-period Sortino ratio (mean / downside_dev) of *s*.""" 

605 downside_sum = _to_float((s.filter(s < 0) ** 2).sum()) 

606 downside_dev = float(np.sqrt(downside_sum / s.count())) 

607 if downside_dev == 0.0: 

608 return float("nan") 

609 return _to_float(s.mean()) / downside_dev 

610 

611 _builtin: dict[str, Callable[[pl.Series], float]] = { 

612 "sharpe": _sharpe_base, 

613 "sortino": _sortino_base, 

614 "adjusted_sortino": lambda s: _sortino_base(s) / float(np.sqrt(2)), 

615 } 

616 

617 if isinstance(base, str): 

618 if base not in _builtin: 

619 raise ValueError(f"base must be one of {list(_builtin)}, got {base!r}") # noqa: TRY003 

620 base_fn = _builtin[base] 

621 else: 

622 base_fn = base 

623 

624 result: dict[str, float] = {} 

625 for col, series in self.data.items(): 

626 base_val = base_fn(series) 

627 if np.isnan(base_val): 

628 result[col] = float("nan") 

629 else: 

630 result[col] = _PerformanceStatsMixin._probabilistic_ratio_from_base(base_val, series) 

631 return result 

632 

633 def smart_sharpe(self, periods: int | float | None = None) -> dict[str, float]: 

634 """Calculate the Smart Sharpe ratio (Sharpe with autocorrelation penalty). 

635 

636 Divides the Sharpe ratio by the autocorrelation penalty to account for 

637 return autocorrelation that can artificially inflate risk-adjusted metrics. 

638 

639 Args: 

640 periods (int | float, optional): Number of periods per year. Defaults to periods_per_year. 

641 

642 Returns: 

643 dict[str, float]: Dictionary mapping asset names to Smart Sharpe ratios. 

644 

645 """ 

646 sharpe_data = self.sharpe(periods=periods) 

647 penalty_data = self.autocorr_penalty() 

648 return {k: sharpe_data[k] / penalty_data[k] for k in sharpe_data} 

649 

650 def smart_sortino(self, periods: int | float | None = None) -> dict[str, float]: 

651 """Calculate the Smart Sortino ratio (Sortino with autocorrelation penalty). 

652 

653 Divides the Sortino ratio by the autocorrelation penalty to account for 

654 return autocorrelation that can artificially inflate risk-adjusted metrics. 

655 

656 Args: 

657 periods (int | float, optional): Number of periods per year. Defaults to periods_per_year. 

658 

659 Returns: 

660 dict[str, float]: Dictionary mapping asset names to Smart Sortino ratios. 

661 

662 """ 

663 sortino_data = self.sortino(periods=periods) 

664 penalty_data = self.autocorr_penalty() 

665 return {k: sortino_data[k] / penalty_data[k] for k in sortino_data} 

666 

667 def adjusted_sortino(self, periods: int | float | None = None) -> dict[str, float]: 

668 """Calculate Jack Schwager's adjusted Sortino ratio. 

669 

670 This adjustment allows for direct comparison to Sharpe ratio. 

671 See: https://archive.is/wip/2rwFW. 

672 

673 Args: 

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

675 

676 Returns: 

677 dict[str, float]: Dictionary mapping asset names to adjusted Sortino ratios. 

678 

679 """ 

680 sortino_data = self.sortino(periods=periods) 

681 return {k: v / np.sqrt(2) for k, v in sortino_data.items()} 

682 

683 # ── Benchmark & factor ──────────────────────────────────────────────────── 

684 

685 @columnwise_stat 

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

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

688 

689 Args: 

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

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

692 

693 Returns: 

694 float: The R-squared value. 

695 

696 Raises: 

697 AttributeError: If no benchmark data is available. 

698 

699 """ 

700 if self.data.benchmark is None: 

701 raise AttributeError("No benchmark data available") # noqa: TRY003 

702 

703 benchmark_col = benchmark or self.data.benchmark.columns[0] 

704 

705 # Evaluate both series and benchmark as Series 

706 all_data = cast(pl.DataFrame, self.all) 

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

708 

709 # Drop nulls 

710 dframe = dframe.drop_nulls() 

711 

712 matrix = dframe.to_numpy() 

713 # Get actual Series 

714 

715 strategy_np = matrix[:, 0] 

716 benchmark_np = matrix[:, 1] 

717 

718 corr_matrix = np.corrcoef(strategy_np, benchmark_np) 

719 r = corr_matrix[0, 1] 

720 return float(r**2) 

721 

722 def r2(self) -> dict[str, float]: 

723 """Shorthand for r_squared(). 

724 

725 Returns: 

726 dict[str, float]: Dictionary mapping asset names to R-squared values. 

727 

728 """ 

729 return self.r_squared() 

730 

731 @columnwise_stat 

732 def information_ratio( 

733 self, 

734 series: pl.Series, 

735 periods_per_year: int | float | None = None, 

736 benchmark: str | None = None, 

737 annualise: bool = False, 

738 ) -> float: 

739 """Calculate the information ratio. 

740 

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

742 

743 Args: 

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

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

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

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

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

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

750 returned by ``qs.stats.information_ratio``. 

751 

752 Returns: 

753 float: The information ratio value. 

754 

755 """ 

756 ppy = periods_per_year or self.data._periods_per_year 

757 

758 benchmark_data = cast(pl.DataFrame, self.data.benchmark) 

759 benchmark_col = benchmark or benchmark_data.columns[0] 

760 

761 active = series - benchmark_data[benchmark_col] 

762 

763 mean_val = cast(float, active.mean()) 

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

765 

766 try: 

767 mean_f = mean_val if mean_val is not None else 0.0 

768 std_f = std_val if std_val is not None else 1.0 

769 ir = mean_f / std_f 

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

771 except ZeroDivisionError: 

772 return 0.0 

773 

774 @columnwise_stat 

775 def greeks( 

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

777 ) -> dict[str, float]: 

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

779 

780 Args: 

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

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

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

784 

785 Returns: 

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

787 

788 """ 

789 ppy = periods_per_year or self.data._periods_per_year 

790 

791 benchmark_data = cast(pl.DataFrame, self.data.benchmark) 

792 benchmark_col = benchmark or benchmark_data.columns[0] 

793 

794 # Evaluate both series and benchmark as Series 

795 all_data = cast(pl.DataFrame, self.all) 

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

797 

798 # Drop nulls 

799 dframe = dframe.drop_nulls() 

800 matrix = dframe.to_numpy() 

801 

802 # Get actual Series 

803 strategy_np = matrix[:, 0] 

804 benchmark_np = matrix[:, 1] 

805 

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

807 cov_matrix = np.cov(strategy_np, benchmark_np) 

808 

809 cov = cov_matrix[0, 1] 

810 var_benchmark = cov_matrix[1, 1] 

811 

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

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

814 

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

816 

817 @columnwise_stat 

818 def treynor_ratio( 

819 self, 

820 series: pl.Series, 

821 periods: int | float | None = None, 

822 benchmark: str | None = None, 

823 ) -> float: 

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

825 

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

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

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

829 

830 Args: 

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

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

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

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

835 first benchmark column. 

836 

837 Returns: 

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

839 is unavailable. 

840 

841 Raises: 

842 AttributeError: If no benchmark data is attached. 

843 """ 

844 if self.data.benchmark is None: 

845 raise AttributeError("No benchmark data available") # noqa: TRY003 

846 

847 ppy = periods or self.data._periods_per_year 

848