Coverage for src / monte_neo / indicators / numba_funcs.py: 13%

1from __future__ import annotations 

2 

3import numpy as np 

4from numba import njit 

5 

6 

7@njit 

8def sma_numba(data: np.ndarray, period: int) -> np.ndarray: 

9 """Fast SMA calculation.""" 

10 res = np.full(data.shape, np.nan) 

11 if len(data) < period: 

12 return res 

13 

14 current_sum = 0.0 

15 for i in range(period): 

16 current_sum += data[i] 

17 

18 res[period - 1] = current_sum / period 

19 

20 for i in range(period, len(data)): 

21 current_sum = current_sum - data[i - period] + data[i] 

22 res[i] = current_sum / period 

23 

24 return res 

25 

26 

27@njit 

28def ema_numba(data: np.ndarray, period: int) -> np.ndarray: 

29 """Fast EMA calculation.""" 

30 res = np.full(data.shape, np.nan) 

31 if len(data) == 0: 

32 return res 

33 

34 alpha = 2.0 / (period + 1) 

35 res[0] = data[0] 

36 

37 for i in range(1, len(data)): 

38 res[i] = (data[i] - res[i - 1]) * alpha + res[i - 1] 

39 

40 return res 

41 

42 

43@njit 

44def rsi_numba(data: np.ndarray, period: int) -> np.ndarray: 

45 """Fast RSI calculation.""" 

46 n = len(data) 

47 res = np.full(n, np.nan) 

48 if n <= period: 

49 return res 

50 

51 # First period calculation (SMA of gains/losses) 

52 gain_sum = 0.0 

53 loss_sum = 0.0 

54 

55 for i in range(1, period + 1): 

56 change = data[i] - data[i-1] 

57 if change > 0: 

58 gain_sum += change 

59 else: 

60 loss_sum -= change 

61 

62 avg_gain = gain_sum / period 

63 avg_loss = loss_sum / period 

64 

65 if avg_loss == 0: 

66 res[period] = 100.0 

67 else: 

68 rs = avg_gain / avg_loss 

69 res[period] = 100.0 - (100.0 / (1.0 + rs)) 

70 

71 # Subsequent periods (Wilder's Smoothing) 

72 for i in range(period + 1, n): 

73 change = data[i] - data[i-1] 

74 current_gain = change if change > 0 else 0.0 

75 current_loss = -change if change < 0 else 0.0 

76 

77 avg_gain = (avg_gain * (period - 1) + current_gain) / period 

78 avg_loss = (avg_loss * (period - 1) + current_loss) / period 

79 

80 if avg_loss == 0: 

81 res[i] = 100.0 

82 else: 

83 rs = avg_gain / avg_loss 

84 res[i] = 100.0 - (100.0 / (1.0 + rs)) 

85 

86 return res 

87 

88 

89@njit 

90def sma_crossover_signals_numba(data: np.ndarray, fast_period: int, slow_period: int) -> np.ndarray: 

91 """Full SMA crossover signal generation in a single Numba pass.""" 

92 n = len(data) 

93 res = np.zeros(n, dtype=np.float32) 

94 if n < slow_period or fast_period >= slow_period: 

95 return res 

96 

97 sum_fast = 0.0 

98 sum_slow = 0.0 

99 

100 # Initial sums 

101 # Initial sums for the window ending at slow_period - 1 

102 sum_fast = 0.0 

103 sum_slow = 0.0 

104 for i in range(slow_period - fast_period, slow_period): 

105 sum_fast += data[i] 

106 for i in range(slow_period): 

107 sum_slow += data[i] 

108 

109 # Initial state at slow_period - 1 

110 sma_fast = sum_fast / fast_period 

111 sma_slow = sum_slow / slow_period 

112 prev_state = 1 if sma_fast > sma_slow else -1 

113 

114 for i in range(slow_period, n): 

115 sum_fast = sum_fast - data[i - fast_period] + data[i] 

116 sum_slow = sum_slow - data[i - slow_period] + data[i] 

117 

118 sma_fast = sum_fast / fast_period 

119 sma_slow = sum_slow / slow_period 

120 

121 current_state = 1 if sma_fast > sma_slow else -1 

122 

123 if current_state != prev_state: 

124 res[i] = float(current_state) 

125 prev_state = current_state 

126 

127 return res 

128 

129 

130@njit 

131def rsi_signals_numba(data: np.ndarray, period: int, oversold: float, overbought: float) -> np.ndarray: 

132 """Fast RSI signal generation in a single Numba pass.""" 

133 n = len(data) 

134 res = np.zeros(n, dtype=np.float32) 

135 if n <= period: 

136 return res 

137 

138 # Use existing rsi_numba 

139 rsi_vals = rsi_numba(data, period) 

140 

141 for i in range(period, n): 

142 if rsi_vals[i] < oversold: 

143 res[i] = 1.0 

144 elif rsi_vals[i] > overbought: 

145 res[i] = -1.0 

146 

147 return res 

148 

149 

150@njit 

151def macd_signals_numba(data: np.ndarray, fast_p: int, slow_p: int, sig_p: int) -> np.ndarray: 

152 """Fast MACD signal generation in a single Numba pass.""" 

153 n = len(data) 

154 res = np.zeros(n, dtype=np.float32) 

155 if n < slow_p + sig_p: 

156 return res 

157 

158 fast_ema = ema_numba(data, fast_p) 

159 slow_ema = ema_numba(data, slow_p) 

160 macd_line = fast_ema - slow_ema 

161 signal_line = ema_numba(macd_line, sig_p) 

162 hist_vals = macd_line - signal_line 

163 

164 prev_sig = 0.0 

165 for i in range(slow_p + sig_p, n): 

166 curr_sig = 0.0 

167 if hist_vals[i] > 0: 

168 curr_sig = 1.0 

169 elif hist_vals[i] < 0: 

170 curr_sig = -1.0 

171 

172 if curr_sig != prev_sig: 

173 res[i] = curr_sig 

174 prev_sig = curr_sig 

175 

176 return res