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

1"""Signal-evaluation mixin for BasanosEngine. 

2 

3Provides information-coefficient (IC) metrics as a reusable mixin so that 

4``optimizer.py`` stays focused on the core position-solving logic. 

5 

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

7is `BasanosEngine`, which inherits from 

8`_SignalEvaluatorMixin`. 

9""" 

10 

11from __future__ import annotations 

12 

13from typing import TYPE_CHECKING 

14 

15import numpy as np 

16import polars as pl 

17from scipy.stats import spearmanr 

18 

19if TYPE_CHECKING: 

20 from ._engine_protocol import _EngineProtocol 

21 

22 

23def _correlate(signal: np.ndarray, fwd_ret: np.ndarray, *, use_rank: bool) -> float: 

24 """Correlate two finite vectors, returning ``nan`` for degenerate input. 

25 

26 Both `np.corrcoef` and `scipy.stats.spearmanr` emit warnings (numpy 

27 ``invalid value encountered in divide``; scipy ``ConstantInputWarning``) 

28 and yield ``nan`` when either input has zero variance (a constant column). 

29 That degenerate case is expected — an IC is undefined when a signal or the 

30 forward returns are flat — so it is short-circuited to ``nan`` here before 

31 the correlation is attempted, keeping the test log and any downstream 

32 caller free of the warning. 

33 

34 Args: 

35 signal: Finite signal values for the assets valid at this timestamp. 

36 fwd_ret: Matching finite forward returns. 

37 use_rank: When ``True`` use Spearman rank correlation; otherwise Pearson. 

38 

39 Returns: 

40 float: The correlation, or ``nan`` when either input is constant. 

41 """ 

42 if signal.std() == 0.0 or fwd_ret.std() == 0.0: 

43 return float("nan") 

44 if use_rank: 

45 corr, _ = spearmanr(signal, fwd_ret) 

46 return float(corr) 

47 return float(np.corrcoef(signal, fwd_ret)[0, 1]) 

48 

49 

50class _SignalEvaluatorMixin: 

51 """Mixin providing cross-sectional information-coefficient (IC) metrics. 

52 

53 The consuming class must satisfy `_EngineProtocol`, 

54 i.e. it must expose: 

55 

56 * ``assets`` — list of asset column names 

57 * ``prices`` — Polars DataFrame with a ``'date'`` column 

58 * ``mu`` — Polars DataFrame of expected-return signals 

59 """ 

60 

61 def _ic_series(self: _EngineProtocol, use_rank: bool, h: int = 1) -> pl.DataFrame: 

62 """Compute the cross-sectional IC time series. 

63 

64 For each timestamp *t* (from 0 to T-1-h), correlates the signal vector 

65 ``mu[t, :]`` with the *h*-period forward return vector 

66 ``prices[t+h, :] / prices[t, :] - 1`` across all assets where both 

67 quantities are finite. When fewer than two valid asset pairs are 

68 available, the IC value is set to ``NaN``. 

69 

70 Args: 

71 use_rank: When ``True`` the Spearman rank correlation is used 

72 (Rank IC); when ``False`` the Pearson correlation is used (IC). 

73 h: Forward-return horizon in periods. ``h=1`` (default) gives the 

74 classic one-period IC; ``h=5`` evaluates signal quality against 

75 five-period returns. Must be >= 1. 

76 

77 Returns: 

78 pl.DataFrame: Two-column frame with ``date`` (signal date) and 

79 either ``ic`` or ``rank_ic``. 

80 

81 Raises: 

82 ValueError: If *h* < 1. 

83 """ 

84 if h < 1: 

85 msg = f"h must be >= 1, got {h}" 

86 raise ValueError(msg) 

87 

88 assets = self.assets 

89 prices_np = self.prices.select(assets).to_numpy().astype(float) 

90 mu_np = self.mu.select(assets).to_numpy().astype(float) 

91 dates = self.prices["date"].to_list() 

92 

93 col_name = "rank_ic" if use_rank else "ic" 

94 ic_values: list[float] = [] 

95 ic_dates = [] 

96 

97 for t in range(len(dates) - h): 

98 fwd_ret = prices_np[t + h] / prices_np[t] - 1.0 

99 signal = mu_np[t] 

100 

101 # Both signal and forward return must be finite 

102 mask = np.isfinite(signal) & np.isfinite(fwd_ret) 

103 n_valid = int(mask.sum()) 

104 

105 ic_values.append( 

106 _correlate(signal[mask], fwd_ret[mask], use_rank=use_rank) if n_valid >= 2 else float("nan") 

107 ) 

108 ic_dates.append(dates[t]) 

109 

110 return pl.DataFrame({"date": ic_dates, col_name: pl.Series(ic_values, dtype=pl.Float64)}) 

111 

112 def ic(self: _EngineProtocol, h: int = 1) -> pl.DataFrame: 

113 """Cross-sectional Pearson Information Coefficient (IC) time series. 

114 

115 For each timestamp *t*, computes the Pearson correlation between the 

116 signal ``mu[t, :]`` and the *h*-period forward return 

117 ``prices[t+h, :] / prices[t, :] - 1`` across all assets where both 

118 quantities are finite. 

119 

120 An IC value close to +1 means the signal ranked assets in the same 

121 order as forward returns; close to -1 means the opposite; near 0 means 

122 no predictive relationship. 

123 

124 Args: 

125 h: Forward-return horizon in periods (default 1). 

126 

127 Returns: 

128 pl.DataFrame: Frame with columns ``['date', 'ic']``. ``date`` is 

129 the timestamp at which the signal was observed. ``ic`` is a 

130 ``Float64`` series (``NaN`` when fewer than 2 valid asset pairs 

131 are available for a given timestamp). 

132 

133 See Also: 

134 `rank_ic` — Spearman variant, more robust to outliers. 

135 `ic_mean`, `ic_std`, `icir` — summary 

136 statistics. 

137 """ 

138 return self._ic_series(use_rank=False, h=h) 

139 

140 def rank_ic(self: _EngineProtocol, h: int = 1) -> pl.DataFrame: 

141 """Cross-sectional Spearman Rank Information Coefficient time series. 

142 

143 Identical to `ic` but uses the Spearman rank correlation 

144 instead of the Pearson correlation, making it more robust to fat-tailed 

145 return distributions and outliers. 

146 

147 Args: 

148 h: Forward-return horizon in periods (default 1). 

149 

150 Returns: 

151 pl.DataFrame: Frame with columns ``['date', 'rank_ic']``. 

152 ``rank_ic`` is a ``Float64`` series. 

153 

154 See Also: 

155 `ic` — Pearson variant. 

156 `rank_ic_mean`, `rank_ic_std` — summary 

157 statistics. 

158 """ 

159 return self._ic_series(use_rank=True, h=h) 

160 

161 def ic_mean(self: _EngineProtocol, h: int = 1) -> float: 

162 """Mean of the IC time series, ignoring NaN values. 

163 

164 Args: 

165 h: Forward-return horizon in periods (default 1). 

166 

167 Returns: 

168 float: Arithmetic mean of all finite IC values, or ``NaN`` if 

169 no finite values exist. 

170 """ 

171 arr = self._ic_series(use_rank=False, h=h)["ic"].drop_nulls().to_numpy() 

172 finite = arr[np.isfinite(arr)] 

173 return float(np.mean(finite)) if len(finite) > 0 else float("nan") 

174 

175 def ic_std(self: _EngineProtocol, h: int = 1) -> float: 

176 """Standard deviation of the IC time series, ignoring NaN values. 

177 

178 Uses ``ddof=1`` (sample standard deviation). 

179 

180 Args: 

181 h: Forward-return horizon in periods (default 1). 

182 

183 Returns: 

184 float: Sample standard deviation of all finite IC values, or 

185 ``NaN`` if fewer than 2 finite values exist. 

186 """ 

187 arr = self._ic_series(use_rank=False, h=h)["ic"].drop_nulls().to_numpy() 

188 finite = arr[np.isfinite(arr)] 

189 return float(np.std(finite, ddof=1)) if len(finite) > 1 else float("nan") 

190 

191 def icir(self: _EngineProtocol, h: int = 1) -> float: 

192 """Information Coefficient Information Ratio (ICIR). 

193 

194 Defined as ``IC mean / IC std``. A higher absolute ICIR indicates a 

195 more consistent signal: the mean IC is large relative to its 

196 variability. 

197 

198 Args: 

199 h: Forward-return horizon in periods (default 1). 

200 

201 Returns: 

202 float: ``ic_mean / ic_std``, or ``NaN`` when ``ic_std`` is zero 

203 or non-finite. 

204 """ 

205 ic_df = self._ic_series(use_rank=False, h=h) 

206 arr = ic_df["ic"].drop_nulls().to_numpy() 

207 finite = arr[np.isfinite(arr)] 

208 mean = float(np.mean(finite)) if len(finite) > 0 else float("nan") 

209 std = float(np.std(finite, ddof=1)) if len(finite) > 1 else float("nan") 

210 if not np.isfinite(std) or std == 0.0: 

211 return float("nan") 

212 return float(mean / std) 

213 

214 def rank_ic_mean(self: _EngineProtocol, h: int = 1) -> float: 

215 """Mean of the Rank IC time series, ignoring NaN values. 

216 

217 Args: 

218 h: Forward-return horizon in periods (default 1). 

219 

220 Returns: 

221 float: Arithmetic mean of all finite Rank IC values, or ``NaN`` 

222 if no finite values exist. 

223 """ 

224 arr = self._ic_series(use_rank=True, h=h)["rank_ic"].drop_nulls().to_numpy() 

225 finite = arr[np.isfinite(arr)] 

226 return float(np.mean(finite)) if len(finite) > 0 else float("nan") 

227 

228 def rank_ic_std(self: _EngineProtocol, h: int = 1) -> float: 

229 """Standard deviation of the Rank IC time series, ignoring NaN values. 

230 

231 Uses ``ddof=1`` (sample standard deviation). 

232 

233 Args: 

234 h: Forward-return horizon in periods (default 1). 

235 

236 Returns: 

237 float: Sample standard deviation of all finite Rank IC values, or 

238 ``NaN`` if fewer than 2 finite values exist. 

239 """ 

240 arr = self._ic_series(use_rank=True, h=h)["rank_ic"].drop_nulls().to_numpy() 

241 finite = arr[np.isfinite(arr)] 

242 return float(np.std(finite, ddof=1)) if len(finite) > 1 else float("nan")