Coverage for src/basanos/math/_engine_performance.py: 100%

1"""Performance / parameter-sweep mixin for `BasanosEngine`. 

2 

3Provides the Sharpe-ratio sweep helpers (`sharpe_at_shrink`, 

4`sharpe_at_window_factors`, `naive_sharpe`) as a reusable mixin so that 

5``optimizer.py`` stays focused on the position-solving facade. Each helper 

6rebuilds a sibling engine with a modified configuration, so it constructs a new 

7`BasanosEngine` via a deferred import (avoiding a circular import at module 

8load time). 

9 

10Classes in this module are **private implementation details**. The public API 

11is `BasanosEngine`, which inherits from `_PerformanceMixin`. 

12""" 

13 

14from __future__ import annotations 

15 

16from typing import TYPE_CHECKING 

17 

18import polars as pl 

19 

20from ._config import SlidingWindowConfig 

21 

22if TYPE_CHECKING: 

23 from ._engine_protocol import _EngineProtocol 

24 

25 

26class _PerformanceMixin: 

27 """Mixin providing portfolio-Sharpe sweep helpers for `BasanosEngine`. 

28 

29 The consuming class must satisfy `_EngineProtocol` (it uses ``assets``, 

30 ``prices``, ``mu``, and ``cfg``). 

31 """ 

32 

33 def sharpe_at_shrink(self: _EngineProtocol, shrink: float) -> float: 

34 r"""Return the annualised portfolio Sharpe ratio for the given shrinkage weight. 

35 

36 Constructs a new `BasanosEngine` with all parameters identical to 

37 ``self`` except that ``cfg.shrink`` is replaced by ``shrink``, then 

38 returns the annualised Sharpe ratio of the resulting portfolio. 

39 

40 This is the canonical single-argument callable required by the benchmarks 

41 specification: ``f(λ) → Sharpe``. Use it to sweep λ across ``[0, 1]`` 

42 and measure whether correlation adjustment adds value over the 

43 signal-proportional baseline (λ = 0) or the unregularised limit (λ = 1). 

44 

45 Corner cases: 

46 * **λ = 0** — the shrunk matrix equals the identity, so the 

47 optimiser treats all assets as uncorrelated and positions are 

48 purely signal-proportional (no correlation adjustment). 

49 * **λ = 1** — the raw EWMA correlation matrix is used without 

50 shrinkage. 

51 

52 Args: 

53 shrink: Retention weight λ ∈ [0, 1]. See 

54 `shrink` for full documentation. 

55 

56 Returns: 

57 Annualised Sharpe ratio of the portfolio returns as a ``float``. 

58 Returns ``float("nan")`` when the Sharpe ratio cannot be computed 

59 (e.g. zero-variance returns). 

60 

61 Raises: 

62 ValidationError: When ``shrink`` is outside [0, 1] (delegated to 

63 `BasanosConfig` field validation). 

64 

65 Examples: 

66 >>> import numpy as np 

67 >>> import polars as pl 

68 >>> from basanos.math.optimizer import BasanosConfig, BasanosEngine 

69 >>> dates = pl.Series("date", list(range(200))) 

70 >>> rng = np.random.default_rng(0) 

71 >>> prices = pl.DataFrame({"date": dates, "A": rng.lognormal(size=200), "B": rng.lognormal(size=200)}) 

72 >>> mu = pl.DataFrame({"date": dates, "A": rng.normal(size=200), "B": rng.normal(size=200)}) 

73 >>> cfg = BasanosConfig(vola=10, corr=20, clip=3.0, shrink=0.5, aum=1e6) 

74 >>> engine = BasanosEngine(prices=prices, mu=mu, cfg=cfg) 

75 >>> s = engine.sharpe_at_shrink(0.5) 

76 >>> isinstance(s, float) 

77 True 

78 """ 

79 from .optimizer import BasanosEngine # deferred to avoid a circular import 

80 

81 new_cfg = self.cfg.replace(shrink=shrink) 

82 engine = BasanosEngine(prices=self.prices, mu=self.mu, cfg=new_cfg) 

83 return float(engine.portfolio.stats.sharpe().get("returns") or float("nan")) 

84 

85 def sharpe_at_window_factors(self: _EngineProtocol, window: int, n_factors: int) -> float: 

86 r"""Return the annualised portfolio Sharpe ratio for the given sliding-window parameters. 

87 

88 Constructs a new `BasanosEngine` with ``covariance_mode`` set to 

89 ``"sliding_window"`` and the supplied ``window`` / ``n_factors``, keeping 

90 all other configuration identical to ``self``. 

91 

92 Use this method to sweep ``(W, k)`` and compare the sliding-window 

93 estimator against the EWMA baseline (via `sharpe_at_shrink`). 

94 

95 Args: 

96 window: Rolling window length $W \geq 1$. 

97 Rule of thumb: $W \geq 2 \cdot n_{\text{assets}}$. 

98 n_factors: Number of latent factors $k \geq 1$. 

99 

100 Returns: 

101 Annualised Sharpe ratio of the portfolio returns as a ``float``. 

102 Returns ``float("nan")`` when the Sharpe ratio cannot be computed 

103 (e.g. not enough history to fill the first window). 

104 

105 Raises: 

106 ValidationError: When ``window`` or ``n_factors`` fail field 

107 constraints (delegated to `BasanosConfig`). 

108 

109 Examples: 

110 >>> import numpy as np 

111 >>> import polars as pl 

112 >>> from basanos.math.optimizer import BasanosConfig, BasanosEngine 

113 >>> dates = pl.Series("date", list(range(200))) 

114 >>> rng = np.random.default_rng(0) 

115 >>> prices = pl.DataFrame({"date": dates, "A": rng.lognormal(size=200), "B": rng.lognormal(size=200)}) 

116 >>> mu = pl.DataFrame({"date": dates, "A": rng.normal(size=200), "B": rng.normal(size=200)}) 

117 >>> cfg = BasanosConfig(vola=10, corr=20, clip=3.0, shrink=0.5, aum=1e6) 

118 >>> engine = BasanosEngine(prices=prices, mu=mu, cfg=cfg) 

119 >>> s = engine.sharpe_at_window_factors(window=40, n_factors=2) 

120 >>> isinstance(s, float) 

121 True 

122 """ 

123 from .optimizer import BasanosEngine # deferred to avoid a circular import 

124 

125 new_cfg = self.cfg.replace( 

126 covariance_config=SlidingWindowConfig(window=window, n_factors=n_factors), 

127 ) 

128 engine = BasanosEngine(prices=self.prices, mu=self.mu, cfg=new_cfg) 

129 return float(engine.portfolio.stats.sharpe().get("returns") or float("nan")) 

130 

131 @property 

132 def naive_sharpe(self: _EngineProtocol) -> float: 

133 r"""Sharpe ratio of the naïve equal-weight signal (μ = 1 for every asset/timestamp). 

134 

135 Replaces the expected-return signal ``mu`` with a constant matrix of 

136 ones, then runs the optimiser with the current configuration and returns 

137 the annualised Sharpe ratio of the resulting portfolio. 

138 

139 This provides the baseline answer to *"does the signal add value?"*: 

140 a real signal should produce a higher Sharpe than the naïve benchmark. 

141 Combined with `sharpe_at_shrink`, this yields a three-way 

142 comparison: 

143 

144 +--------------------+----------------------------------------------+ 

145 | Benchmark | What it measures | 

146 +====================+==============================================+ 

147 | ``naive_sharpe`` | No signal skill; pure correlation routing | 

148 +--------------------+----------------------------------------------+ 

149 | ``sharpe_at_shrink(0.0)`` | Signal skill, no correlation adj. | 

150 +--------------------+----------------------------------------------+ 

151 | ``sharpe_at_shrink(cfg.shrink)`` | Signal + correlation adj. | 

152 +--------------------+----------------------------------------------+ 

153 

154 Returns: 

155 Annualised Sharpe ratio of the equal-weight portfolio as a ``float``. 

156 Returns ``float("nan")`` when the Sharpe ratio cannot be computed. 

157 

158 Examples: 

159 >>> import numpy as np 

160 >>> import polars as pl 

161 >>> from basanos.math.optimizer import BasanosConfig, BasanosEngine 

162 >>> dates = pl.Series("date", list(range(200))) 

163 >>> rng = np.random.default_rng(0) 

164 >>> prices = pl.DataFrame({"date": dates, "A": rng.lognormal(size=200), "B": rng.lognormal(size=200)}) 

165 >>> mu = pl.DataFrame({"date": dates, "A": rng.normal(size=200), "B": rng.normal(size=200)}) 

166 >>> cfg = BasanosConfig(vola=10, corr=20, clip=3.0, shrink=0.5, aum=1e6) 

167 >>> engine = BasanosEngine(prices=prices, mu=mu, cfg=cfg) 

168 >>> s = engine.naive_sharpe 

169 >>> isinstance(s, float) 

170 True 

171 """ 

172 from .optimizer import BasanosEngine # deferred to avoid a circular import 

173 

174 naive_mu = self.mu.with_columns(pl.lit(1.0).alias(asset) for asset in self.assets) 

175 engine = BasanosEngine(prices=self.prices, mu=naive_mu, cfg=self.cfg) 

176 return float(engine.portfolio.stats.sharpe().get("returns") or float("nan"))