Coverage for src / monte_neo / metrics / numba_funcs.py: 4%
209 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-28 16:27 +0200
1from __future__ import annotations
3import numpy as np
4from numba import njit, prange
7@njit
8def extract_trades_fast(
9 prices: np.ndarray,
10 highs: np.ndarray,
11 lows: np.ndarray,
12 signals: np.ndarray,
13 use_sl_tp: bool = False,
14 sl_pct: float = 0.0,
15 tp_pct: float = 0.0,
16 commission_pct: float = 0.0,
17 slippage_pct: float = 0.0,
18) -> list[tuple[int, int, float, float, int, float, float]]:
19 """Fast trade extraction using Numba JIT."""
20 results = []
22 position = 0
23 entry_idx = 0
24 entry_price = 0.0
26 sl_price = 0.0
27 tp_price = 0.0
29 blocked_signal = 0
31 for i in range(len(signals)):
32 signal = int(signals[i])
33 price = prices[i]
34 high = highs[i]
35 low = lows[i]
37 if position == 0:
38 if signal == 0:
39 blocked_signal = 0
40 elif signal != blocked_signal:
41 # Open position
42 position = signal
43 entry_idx = i
44 entry_price = price * (1.0 + float(position) * slippage_pct)
46 if use_sl_tp:
47 if position == 1: # Long
48 sl_price = entry_price * (1.0 - sl_pct / 100.0)
49 tp_price = entry_price * (1.0 + tp_pct / 100.0)
50 else: # Short
51 sl_price = entry_price * (1.0 + sl_pct / 100.0)
52 tp_price = entry_price * (1.0 - tp_pct / 100.0)
53 else:
54 # Check for SL/TP first
55 hit_exit = False
56 exit_price = price
58 if use_sl_tp:
59 if position == 1: # Long
60 if low <= sl_price:
61 exit_price = sl_price
62 hit_exit = True
63 elif high >= tp_price:
64 exit_price = tp_price
65 hit_exit = True
66 else: # Short
67 if high >= sl_price:
68 exit_price = sl_price
69 hit_exit = True
70 elif low <= tp_price:
71 exit_price = tp_price
72 hit_exit = True
74 # Check for signal exit if SL/TP not hit
75 if not hit_exit and signal == -position:
76 exit_price = price
77 hit_exit = True
79 if hit_exit:
80 # Close position
81 exit_price_adj = exit_price * (1.0 - float(position) * slippage_pct)
82 pnl = (exit_price_adj - entry_price) * position
83 pnl_pct = (pnl / entry_price) - commission_pct * 2.0 # Round trip commission
85 results.append(
86 (entry_idx, i, entry_price, exit_price, position, pnl, pnl_pct)
87 )
89 # Reset position and block current signal
90 blocked_signal = signal
91 position = 0
93 return results
96@njit(parallel=True)
97def calculate_batch_fast(
98 prices: np.ndarray,
99 highs: np.ndarray,
100 lows: np.ndarray,
101 signal_matrix: np.ndarray,
102 use_sl_tp: bool,
103 sl_pct: float,
104 tp_pct: float,
105 commission_pct: float = 0.0,
106 slippage_pct: float = 0.0,
107) -> np.ndarray:
108 """Calculate basic metrics for a batch of signal sets in parallel."""
109 n_indicators = signal_matrix.shape[0]
110 results = np.zeros((n_indicators, 4), dtype=np.float64) # total_return, max_dd, pf, n_trades
112 for i in prange(n_indicators):
113 signals = signal_matrix[i]
115 # Simplified extraction for speed
116 position: int = 0
117 entry_price: float = 0.0
118 sl_price: float = 0.0
119 tp_price: float = 0.0
120 blocked_signal: int = 0
122 total_pnl_pct = 0.0
123 gross_profit = 0.0
124 gross_loss = 0.0
125 n_trades = 0
127 # Equity curve for drawdown
128 equity = 1.0
129 max_equity = 1.0
130 max_dd = 0.0
132 for j in range(len(signals)):
133 signal = int(signals[j])
134 price = prices[j]
135 high = highs[j]
136 low = lows[j]
138 if position == 0:
139 if signal == 0:
140 blocked_signal = 0
141 elif signal != blocked_signal:
142 position = signal
143 entry_price = price
144 if use_sl_tp:
145 if position == 1:
146 sl_price = entry_price * (1.0 - sl_pct / 100.0)
147 tp_price = entry_price * (1.0 + tp_pct / 100.0)
148 else:
149 sl_price = entry_price * (1.0 + sl_pct / 100.0)
150 tp_price = entry_price * (1.0 - tp_pct / 100.0)
151 else:
152 hit_exit = False
153 exit_price = price
155 if use_sl_tp:
156 if position == 1:
157 if low <= sl_price:
158 exit_price = sl_price
159 hit_exit = True
160 elif high >= tp_price:
161 exit_price = tp_price
162 hit_exit = True
163 else:
164 if high >= sl_price:
165 exit_price = sl_price
166 hit_exit = True
167 elif low <= tp_price:
168 exit_price = tp_price
169 hit_exit = True
171 if not hit_exit and signal == -position:
172 exit_price = price
173 hit_exit = True
175 if hit_exit:
176 pnl = (exit_price - entry_price) * position
177 pnl_pct = pnl / entry_price
178 total_pnl_pct += pnl_pct
180 if pnl > 0: gross_profit += pnl
181 else: gross_loss += abs(pnl)
183 # Update equity and drawdown
184 equity *= (1.0 + pnl_pct)
185 if equity > max_equity: max_equity = equity
186 dd = (max_equity - equity) / max_equity
187 if dd > max_dd: max_dd = dd
189 blocked_signal = signal
190 position = 0
191 n_trades += 1
193 pf = gross_profit / gross_loss if gross_loss > 0 else 100.0
194 results[i, 0] = total_pnl_pct
195 results[i, 1] = max_dd
196 results[i, 2] = pf
197 results[i, 3] = n_trades
199 return results
202@njit(parallel=True)
203def calculate_batch_multi_price_fast(
204 price_matrix: np.ndarray,
205 high_matrix: np.ndarray,
206 low_matrix: np.ndarray,
207 signal_matrix: np.ndarray,
208 use_sl_tp: bool,
209 sl_pct: float,
210 tp_pct: float,
211 commission_pct: float = 0.0,
212 slippage_pct: float = 0.0,
213) -> np.ndarray:
214 """Calculate basic metrics for a batch where each row has its own prices."""
215 n_rows = signal_matrix.shape[0]
216 results = np.zeros((n_rows, 4), dtype=np.float64) # total_return, max_dd, pf, n_trades
218 for i in prange(n_rows):
219 signals = signal_matrix[i]
220 prices = price_matrix[i]
221 highs = high_matrix[i]
222 lows = low_matrix[i]
224 # Simplified extraction for speed
225 position: int = 0
226 entry_price: float = 0.0
227 sl_price: float = 0.0
228 tp_price: float = 0.0
229 blocked_signal: int = 0
231 total_pnl_pct = 0.0
232 gross_profit = 0.0
233 gross_loss = 0.0
234 n_trades = 0
236 # Equity curve for drawdown
237 equity = 1.0
238 max_equity = 1.0
239 max_dd = 0.0
241 # We need to know the actual length of this row (ignoring padding)
242 # We assume non-zero prices mean actual data
243 row_len = signals.shape[0]
244 while row_len > 0 and prices[row_len-1] == 0:
245 row_len -= 1
247 for j in range(row_len):
248 signal = int(signals[j])
249 price = prices[j]
250 high = highs[j]
251 low = lows[j]
253 if position == 0:
254 if signal == 0:
255 blocked_signal = 0
256 elif signal != blocked_signal:
257 position = signal
258 entry_price = price
259 if use_sl_tp:
260 if position == 1:
261 sl_price = entry_price * (1.0 - sl_pct / 100.0)
262 tp_price = entry_price * (1.0 + tp_pct / 100.0)
263 else:
264 sl_price = entry_price * (1.0 + sl_pct / 100.0)
265 tp_price = entry_price * (1.0 - tp_pct / 100.0)
266 else:
267 hit_exit = False
268 exit_price = price
270 if use_sl_tp:
271 if position == 1:
272 if low <= sl_price:
273 exit_price = sl_price
274 hit_exit = True
275 elif high >= tp_price:
276 exit_price = tp_price
277 hit_exit = True
278 else:
279 if high >= sl_price:
280 exit_price = sl_price
281 hit_exit = True
282 elif low <= tp_price:
283 exit_price = tp_price
284 hit_exit = True
286 if not hit_exit and signal == -position:
287 exit_price = price
288 hit_exit = True
290 if hit_exit:
291 pnl = (exit_price - entry_price) * position
292 pnl_pct = pnl / entry_price
293 total_pnl_pct += pnl_pct
295 if pnl > 0: gross_profit += pnl
296 else: gross_loss += abs(pnl)
298 # Update equity and drawdown
299 equity *= (1.0 + pnl_pct)
300 if equity > max_equity: max_equity = equity
301 dd = (max_equity - equity) / max_equity
302 if dd > max_dd: max_dd = dd
304 blocked_signal = signal
305 position = 0
306 n_trades += 1
308 pf = gross_profit / gross_loss if gross_loss > 0 else 100.0
309 results[i, 0] = total_pnl_pct
310 results[i, 1] = max_dd
311 results[i, 2] = pf
312 results[i, 3] = n_trades
314 return results