Coverage for src / monte_neo / monte_carlo / workers.py: 100%

1"""Monte Carlo workers. 

2 

3Independent worker functions for parallel execution. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import TYPE_CHECKING, Any 

9 

10import numpy as np 

11import pandas as pd 

12 

13if TYPE_CHECKING: 

14 from monte_neo.indicators.base import BaseIndicator 

15 from monte_neo.metrics.calculator import MetricsCalculator 

16 

17# Global variable for shared memory in workers 

18SHARED_SCENARIOS: list[pd.DataFrame] | None = None 

19SHARED_DATA: pd.DataFrame | None = None 

20 

21 

22def _generate_signals_wrapper(args): 

23 """Helper for parallel signal generation.""" 

24 indicator, df = args 

25 # Use shared data if available and df is None 

26 if df is None: 

27 if SHARED_DATA is None: 

28 # Fallback or error? For safety, return zeros or raise 

29 # But in parallel execution exceptions might be swallowed or messy. 

30 # We assume SHARED_DATA is set if df is None. 

31 return np.zeros(0, dtype=np.float32) 

32 df = SHARED_DATA 

33 

34 sigs = indicator.generate_signals_fast(df) 

35 # Return numpy array to reduce IPC 

36 return np.asarray(sigs, dtype=np.float32) 

37 

38 

39def run_indicator_batch(args: tuple[list[BaseIndicator], pd.DataFrame | None]) -> list[np.ndarray]: 

40 """Run a batch of indicators on shared data.""" 

41 indicators, df = args 

42 if df is None: 

43 if SHARED_DATA is None: 

44 return [] 

45 df = SHARED_DATA 

46 

47 results = [] 

48 for ind in indicators: 

49 try: 

50 sigs = ind.generate_signals_fast(df) 

51 results.append(np.asarray(sigs, dtype=np.float32)) 

52 except Exception: 

53 # Return zeros on failure to keep alignment 

54 results.append(np.zeros(len(df), dtype=np.float32)) 

55 return results 

56 

57 

58def _generate_lazy_scenario_wrapper(args): 

59 """Helper for lazy scenario generation.""" 

60 return run_block_bootstrap_scenario(*args) 

61 

62 

63def init_worker(scenarios: list[pd.DataFrame]) -> None: 

64 """Initialize worker with shared scenarios.""" 

65 global SHARED_SCENARIOS 

66 SHARED_SCENARIOS = scenarios 

67 

68 

69def init_worker_data(data: pd.DataFrame) -> None: 

70 """Initialize worker with shared data.""" 

71 global SHARED_DATA 

72 SHARED_DATA = data 

73 

74 

75def run_scenario_batch( 

76 scenarios: list[pd.DataFrame] | None, 

77 indicator: BaseIndicator, 

78 metrics_calc: MetricsCalculator, 

79 target_metrics: dict[str, float], 

80 indices: list[int] | None = None, 

81) -> list[tuple[bool, dict[str, float]]]: 

82 """Run a batch of scenarios in a single worker task. 

83 

84 This reduces IPC overhead and allows reusing compiled indicator code. 

85 Can use explicitly passed scenarios OR shared scenarios via indices. 

86 """ 

87 results = [] 

88 required_metrics = list(target_metrics.keys()) 

89 

90 # Determine data source 

91 batch_data: list[pd.DataFrame] 

92 if scenarios is not None: 

93 batch_data = scenarios 

94 elif SHARED_SCENARIOS is not None and indices is not None: 

95 batch_data = [SHARED_SCENARIOS[i] for i in indices] 

96 else: 

97 # Fallback or error 

98 return [] 

99 

100 # Compile once per batch if needed 

101 if hasattr(indicator, "_compile_if_needed"): 

102 try: 

103 indicator._compile_if_needed() 

104 except Exception: 

105 pass 

106 

107 for data in batch_data: 

108 try: 

109 signals = indicator.generate_signals_fast(data) 

110 metrics = metrics_calc.calculate_all(data, signals, required_metrics=required_metrics) 

111 

112 # Inline check 

113 passed = True 

114 for metric_name, target_value in target_metrics.items(): 

115 if metric_name not in metrics: 

116 continue 

117 actual = metrics[metric_name] 

118 if metric_name in ["max_drawdown", "consecutive_losses"]: 

119 if actual > target_value: 

120 passed = False 

121 break 

122 else: 

123 if actual < target_value: 

124 passed = False 

125 break 

126 results.append((passed, metrics)) 

127 except Exception: 

128 results.append((False, {})) 

129 

130 return results 

131 

132 

133def run_single_scenario( 

134 scenario_data: pd.DataFrame, 

135 indicator: BaseIndicator, 

136 metrics_calc: MetricsCalculator, 

137 target_metrics: dict[str, float], 

138) -> tuple[bool, dict[str, float]]: 

139 """Helper for parallel execution (legacy/single mode).""" 

140 signals = indicator.generate_signals_fast(scenario_data) 

141 metrics = metrics_calc.calculate_all( 

142 scenario_data, signals, required_metrics=list(target_metrics.keys()) 

143 ) 

144 

145 # Inline check_targets to avoid dependency on self 

146 passed = True 

147 for metric_name, target_value in target_metrics.items(): 

148 if metric_name not in metrics: 

149 continue 

150 actual = metrics[metric_name] 

151 if metric_name in ["max_drawdown", "consecutive_losses"]: 

152 if actual > target_value: 

153 passed = False 

154 break 

155 else: 

156 if actual < target_value: 

157 passed = False 

158 break 

159 return passed, metrics 

160 

161 

162def run_block_bootstrap_scenario( 

163 indicator: BaseIndicator, 

164 seed: int, 

165 block_size: int | None = None 

166) -> Any | None: 

167 """Generate block bootstrap scenario on fly and run signal generation.""" 

168 if SHARED_DATA is None: 

169 return None 

170 

171 # Replicate block bootstrap logic for SINGLE scenario 

172 rng = np.random.default_rng(seed) 

173 n = len(SHARED_DATA) 

174 if block_size is None: 

175 block_size = max(1, int(np.sqrt(n))) 

176 

177 n_blocks = n // block_size 

178 indices_range = np.arange(block_size) 

179 

180 block_starts = rng.choice(n - block_size + 1, size=n_blocks, replace=True) 

181 full_indices = (block_starts[:, None] + indices_range).ravel() 

182 

183 # Use iloc for speed (returns copy by default for fancy indexing) 

184 scenario_data = SHARED_DATA.iloc[full_indices] 

185 

186 # Run indicator 

187 try: 

188 if hasattr(indicator, "_compile_if_needed"): 

189 indicator._compile_if_needed() 

190 

191 signals = indicator.generate_signals_fast(scenario_data) 

192 

193 # Extract signal array to reduce IPC 

194 signal_arr = np.asarray(signals, dtype=np.float32) 

195 

196 # Calculate returns for GPU engine 

197 close_prices = scenario_data["close"].values 

198 returns = (close_prices[1:] / close_prices[:-1]) - 1 

199 

200 # Extract OHLC for SL/TP calculations 

201 ohlc = scenario_data[["open", "high", "low", "close"]].values.astype(np.float32) 

202 

203 return signal_arr, returns.astype(np.float32), ohlc 

204 except Exception: 

205 return None