Coverage for src/tinycta/_kernel.py: 100%

1"""Pure-NumPy numeric kernel for the correlation-aware position optimizer. 

2 

3Every function here operates on NumPy arrays only — no Polars, no :class:`~tinycta.config.Config` 

4— so the Polars-facing orchestration in :mod:`tinycta.engine` stays a thin adapter that 

5prepares arrays, delegates the timestamp walk to :func:`forward_walk`, and writes the result 

6back into a DataFrame. 

7""" 

8 

9from __future__ import annotations 

10 

11from collections.abc import Hashable 

12 

13import numpy as np 

14from loguru import logger 

15 

16from .linalg import inv_a_norm as _inv_a_norm 

17from .linalg import solve as _solve 

18from .signal import shrink2id as _shrink2id 

19 

20 

21def _denominator_is_degenerate(denom: float) -> bool: 

22 """Return True when the correlation-norm denominator is effectively zero. 

23 

24 ``1e-12`` is an arbitrary epsilon floor: any nearby threshold value or a 

25 ``<`` vs ``<=`` boundary behaves identically for realistic denominators, so 

26 this comparison is intentionally excluded from mutation. 

27 """ 

28 return denom <= 1e-12 # pragma: no mutate 

29 

30 

31def _risk_position(corr: np.ndarray, mu_row: np.ndarray, mask: np.ndarray, shrink: float) -> np.ndarray: 

32 """Solve the shrunk correlation system for one timestamp's tradable assets. 

33 

34 Shrinks ``corr`` towards the identity by ``shrink`` (via 

35 :func:`~tinycta.signal.shrink2id`), restricts it to the masked assets, solves 

36 for the expected returns ``mu_row`` and normalises by ``inv_a_norm`` so the raw 

37 risk position has unit norm under the correlation metric. Returns zeros when the 

38 normaliser is non-finite/degenerate or ``mu_row`` is all-zero. 

39 

40 Args: 

41 corr: Full EWMA correlation matrix for the timestamp. 

42 mu_row: Expected returns for every asset at the timestamp (NaNs tolerated). 

43 mask: Boolean mask of currently-tradable assets. 

44 shrink: Identity-shrinkage weight in ``[0, 1]``. 

45 

46 Returns: 

47 np.ndarray: The normalised risk position over the masked assets. 

48 """ 

49 matrix = _shrink2id(corr, lamb=shrink)[np.ix_(mask, mask)] 

50 expected_mu = np.nan_to_num(mu_row[mask]) 

51 denom = _inv_a_norm(expected_mu, matrix) 

52 if denom is None or not np.isfinite(denom) or _denominator_is_degenerate(denom) or np.allclose(expected_mu, 0.0): 

53 logger.debug( 

54 "Risk position zeroed for {} masked asset(s): degenerate correlation-norm " 

55 "denominator (denom={}) or all-zero expected returns.", 

56 int(mask.sum()), 

57 denom, 

58 ) 

59 return np.zeros_like(expected_mu) 

60 return _solve(matrix, expected_mu) / denom 

61 

62 

63def _update_profit_variance( 

64 profit_variance: float, 

65 cash_pos_prev: np.ndarray, 

66 returns_row: np.ndarray, 

67 ret_mask: np.ndarray, 

68 lamb: float, 

69) -> float: 

70 """EWMA-update the running profit-variance estimate with one period's P&L. 

71 

72 Realised profit is the previous cash position dotted with the current returns 

73 over the jointly-finite assets; the variance decays towards the new squared 

74 profit by ``1 - lamb``. 

75 

76 Args: 

77 profit_variance: Previous running profit-variance estimate. 

78 cash_pos_prev: Previous timestamp's cash position (NaNs tolerated). 

79 returns_row: Current timestamp's simple returns (NaNs tolerated). 

80 ret_mask: Boolean mask of assets finite in both rows. 

81 lamb: EWMA decay factor. 

82 

83 Returns: 

84 float: The updated profit-variance estimate. 

85 """ 

86 lhs = np.nan_to_num(cash_pos_prev[ret_mask], nan=0.0) 

87 rhs = np.nan_to_num(returns_row[ret_mask], nan=0.0) 

88 profit = lhs @ rhs 

89 return float(lamb * profit_variance + (1 - lamb) * profit**2) 

90 

91 

92def forward_walk( 

93 cor: dict[Hashable, np.ndarray], 

94 prices_num: np.ndarray, 

95 returns_num: np.ndarray, 

96 mu: np.ndarray, 

97 vola_np: np.ndarray, 

98 risk_pos_np: np.ndarray, 

99 cash_pos_np: np.ndarray, 

100 row_of: dict[Hashable, int], 

101 shrink: float, 

102) -> None: 

103 """Walk forward through the post-warmup timestamps, filling positions in place. 

104 

105 Mutates ``risk_pos_np`` and ``cash_pos_np`` row-by-row. At each timestamp the 

106 previous period's realised P&L EWMA-updates a running profit-variance estimate 

107 (decay ``lamb=0.99``), which scales the freshly-solved risk position before it is 

108 divided by per-asset volatility to yield the cash position. 

109 

110 Args: 

111 cor: Per-timestamp correlation matrices, keyed by ``date`` value. 

112 prices_num: Asset prices as a ``(rows, assets)`` array (NaNs tolerated). 

113 returns_num: Simple returns aligned to ``prices_num``. 

114 mu: Expected returns aligned to ``prices_num``. 

115 vola_np: Per-asset EWMA volatility aligned to ``prices_num``. 

116 risk_pos_np: Output risk-position buffer, mutated in place. 

117 cash_pos_np: Output cash-position buffer, mutated in place. 

118 row_of: Map from a ``cor`` key back to its row index. 

119 shrink: Identity-shrinkage weight in ``[0, 1]`` passed to :func:`_risk_position`. 

120 """ 

121 profit_variance = 1.0 

122 lamb = 0.99 

123 

124 prev_row: int | None = None 

125 for t in cor: 

126 row = row_of[t] 

127 mask = np.isfinite(prices_num[row]) 

128 

129 if prev_row is not None: 

130 ret_mask = np.isfinite(returns_num[row]) & mask 

131 if ret_mask.any(): 

132 cash_pos_np[prev_row] = risk_pos_np[prev_row] / vola_np[prev_row] 

133 profit_variance = _update_profit_variance( 

134 profit_variance, cash_pos_np[prev_row], returns_num[row], ret_mask, lamb 

135 ) 

136 

137 if mask.any(): 

138 pos = _risk_position(cor[t], mu[row], mask, shrink) 

139 risk_pos_np[row, mask] = pos / profit_variance 

140 cash_pos_np[row, mask] = risk_pos_np[row, mask] / vola_np[row, mask] 

141 

142 prev_row = row