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

1"""Per-step position solvers for the incremental (streaming) optimiser. 

2 

3Two entry points — `solve_sliding_window_position` and `solve_ewma_position` — 

4each take the resolved step inputs plus the current `_StreamState` and return a 

5``(cash_position, status)`` pair. They hold no reference to the 

6`BasanosStream` façade, so the two covariance modes can be read and tested 

7independently of the streaming loop that drives them. 

8""" 

9 

10from __future__ import annotations 

11 

12import logging 

13from typing import Any, cast 

14 

15import numpy as np 

16import polars as pl 

17from cvx.linalg import SingularMatrixError, cov_to_corr 

18from cvx.linalg.covariance.ewm_cov import ewm_covariance 

19 

20from ._config import BasanosConfig, SlidingWindowConfig 

21from ._engine_solve import MatrixBundle, SolveStatus, _SolveMixin 

22from ._factor_model import FactorModel 

23from ._signal import shrink2id 

24from ._stream_state import _StreamState 

25 

26_logger = logging.getLogger(__name__) 

27 

28 

29def _fit_sliding_factor_model( 

30 cfg: BasanosConfig, 

31 state: _StreamState, 

32 mask: np.ndarray, 

33 date: Any, 

34) -> FactorModel | None: 

35 """Fit a truncated factor model on the masked rolling window; ``None`` on SVD failure.""" 

36 sw_config = cast(SlidingWindowConfig, cfg.covariance_config) 

37 sw_ret_buf = cast(np.ndarray, state.sw_ret_buf) 

38 window_ret = np.where( 

39 np.isfinite(sw_ret_buf[:, mask]), 

40 sw_ret_buf[:, mask], 

41 0.0, 

42 ) 

43 n_sub = int(mask.sum()) 

44 k_eff = min(sw_config.n_factors, sw_config.window, n_sub) 

45 if sw_config.max_components is not None: 

46 k_eff = min(k_eff, sw_config.max_components) 

47 try: 

48 return FactorModel.from_returns(window_ret, k=k_eff) 

49 except (np.linalg.LinAlgError, ValueError) as exc: 

50 _logger.debug("Sliding window SVD failed at date=%s: %s", date, exc) 

51 return None 

52 

53 

54def _woodbury_normalised( 

55 fm: FactorModel, 

56 expected_mu: np.ndarray, 

57 cfg: BasanosConfig, 

58 date: Any, 

59) -> np.ndarray | None: 

60 """Woodbury-solve and normalise; ``None`` on solve failure or degenerate denominator.""" 

61 try: 

62 x = fm.solve(expected_mu) 

63 denom_val = float(np.sqrt(max(0.0, float(np.dot(expected_mu, x))))) 

64 except (SingularMatrixError, np.linalg.LinAlgError) as exc: 

65 _logger.warning("Woodbury solve failed at date=%s: %s", date, exc) 

66 return None 

67 

68 if not np.isfinite(denom_val) or denom_val <= cfg.denom_tol: 

69 _logger.warning( 

70 "Positions zeroed at date=%s (sliding_window): normalisation " 

71 "denominator degenerate (denom=%s, denom_tol=%s).", 

72 date, 

73 denom_val, 

74 cfg.denom_tol, 

75 ) 

76 return None 

77 

78 return cast("np.ndarray", x / denom_val) 

79 

80 

81def solve_sliding_window_position( 

82 *, 

83 cfg: BasanosConfig, 

84 state: _StreamState, 

85 mask: np.ndarray, 

86 new_m: np.ndarray, 

87 vola_vec: np.ndarray, 

88 n_assets: int, 

89 date: Any, 

90) -> tuple[np.ndarray, SolveStatus]: 

91 """Solve one step in SlidingWindow mode and return cash position + status.""" 

92 new_cash_pos = np.full(n_assets, np.nan, dtype=float) 

93 status = SolveStatus.DEGENERATE 

94 if not mask.any(): 

95 return new_cash_pos, status 

96 

97 fm = _fit_sliding_factor_model(cfg, state, mask, date) 

98 if fm is None: 

99 new_cash_pos[mask] = 0.0 

100 return new_cash_pos, status 

101 

102 expected_mu = np.nan_to_num(new_m[mask]) 

103 if np.allclose(expected_mu, 0.0): 

104 new_cash_pos[mask] = 0.0 

105 return new_cash_pos, SolveStatus.ZERO_SIGNAL 

106 

107 risk_pos = _woodbury_normalised(fm, expected_mu, cfg, date) 

108 if risk_pos is None: 

109 new_cash_pos[mask] = 0.0 

110 return new_cash_pos, status 

111 

112 with np.errstate(invalid="ignore"): 

113 new_cash_pos[mask] = risk_pos / vola_vec[mask] 

114 return new_cash_pos, SolveStatus.VALID 

115 

116 

117def solve_ewma_position( 

118 *, 

119 cfg: BasanosConfig, 

120 state: _StreamState, 

121 corr_ret_buf: np.ndarray, 

122 mask: np.ndarray, 

123 new_m: np.ndarray, 

124 vola_vec: np.ndarray, 

125 assets: list[str], 

126 n_assets: int, 

127 date: Any, 

128) -> tuple[np.ndarray, SolveStatus]: 

129 """Solve one step in EWMA mode and return cash position + status.""" 

130 new_cash_pos = np.full(n_assets, np.nan, dtype=float) 

131 buf = corr_ret_buf # (T, N) — already includes the new row 

132 span = 2 * cfg.corr + 1 

133 t = buf.shape[0] 

134 cols = [pl.Series(a, buf[:, i]).fill_nan(None) for i, a in enumerate(assets)] 

135 pl_df = pl.DataFrame([pl.Series("t", list(range(t))), *cols]) 

136 cov_dict = ewm_covariance(pl_df, assets=assets, index_col="t", window=span, warmup=cfg.corr) 

137 if not cov_dict: 

138 corr = np.full((n_assets, n_assets), np.nan) 

139 else: 

140 # keys are the integer ``t`` index values built from ``range(t)`` above 

141 latest = max(cov_dict, key=lambda k: cast("int", k)) 

142 corr = cov_to_corr(cov_dict[latest], cfg.min_corr_denom) 

143 matrix = shrink2id(corr, lamb=cfg.shrink) 

144 expected_mu, early = _SolveMixin._row_early_check(state.step_count, date, mask, new_m) 

145 if early is not None: 

146 _, _, _, pos, status = early 

147 new_cash_pos[mask] = pos 

148 return new_cash_pos, status 

149 

150 corr_sub = matrix[np.ix_(mask, mask)] 

151 _, _, _, pos, status = _SolveMixin._compute_position( 

152 state.step_count, date, mask, expected_mu, MatrixBundle(matrix=corr_sub), cfg.denom_tol 

153 ) 

154 if status == SolveStatus.VALID: 

155 new_cash_pos[mask] = _SolveMixin._scale_to_cash(cast(np.ndarray, pos), vola_vec[mask]) 

156 else: 

157 new_cash_pos[mask] = pos 

158 return new_cash_pos, status