Coverage for src / monte_neo / core / generator_search.py: 84%

1from __future__ import annotations 

2 

3import time 

4from typing import TYPE_CHECKING 

5 

6from monte_neo.core.config import GeneratorResult 

7from monte_neo.indicators.dynamic import DynamicIndicator 

8from monte_neo.monte_carlo.engine import MCConfig, MonteCarloEngine 

9from monte_neo.monte_carlo.workers import init_worker_data 

10from monte_neo.utils.logger import get_logger 

11from monte_neo.utils.parallel import ParallelExecutor 

12 

13if TYPE_CHECKING: 

14 import pandas as pd 

15 

16 from monte_neo.core.generator import IndicatorGenerator 

17 

18logger = get_logger(__name__) 

19 

20 

21def run_search(generator: IndicatorGenerator, data: pd.DataFrame) -> GeneratorResult: 

22 """Run the search process.""" 

23 start_time = time.time() 

24 best_indicator = None 

25 best_mc_rate = 0.0 

26 best_mc_details = {} 

27 iterations_tried = 0 

28 final_metrics = {} 

29 

30 mc_cache: dict[str, float] = {} 

31 fallback_indicator = None 

32 fallback_metrics = {} 

33 best_performance_score = -float('inf') 

34 

35 # Initialize persistent executor 

36 if generator.config.use_sequential_mc: 

37 generator.executor = ParallelExecutor( 

38 n_workers=1, 

39 use_processes=False, 

40 initializer=init_worker_data, 

41 initargs=(data,) 

42 ) 

43 else: 

44 generator.executor = ParallelExecutor( 

45 initializer=init_worker_data, 

46 initargs=(data,) 

47 ) 

48 generator.executor.__enter__() 

49 

50 logger.info(f"Starting indicator generation (max {generator.config.max_iterations} iterations)") 

51 

52 # Optimization: Use larger batch size to reduce IPC and parallel overhead 

53 batch_size = max(500, generator.config.population_size * 5) 

54 total_iterations = generator.config.max_iterations 

55 

56 # Pre-generate MC scenarios 

57 mc_scenarios = _pre_generate_scenarios(generator, data, total_iterations) 

58 

59 try: 

60 if generator._progress_callback: 

61 generator._progress_callback(0, total_iterations, "Starting search...") 

62 

63 for batch_start in range(0, total_iterations, batch_size): 

64 actual_batch_size = min(batch_size, total_iterations - batch_start) 

65 batch_indicators = [generator._generate_random_indicator() for _ in range(actual_batch_size)] 

66 

67 # GPU Backtest 

68 try: 

69 gpu_results = generator.gpu_engine.backtest_batch( 

70 data, 

71 batch_indicators, 

72 executor=generator.executor, 

73 use_shared_data=True, 

74 use_sl_tp=generator.config.use_sl_tp, 

75 sl_pct=generator.config.stop_loss_pct, 

76 tp_pct=generator.config.take_profit_pct, 

77 force_parallel=True # Force parallel for large batches 

78 ) 

79 except Exception as e: 

80 logger.warning(f"Backtest failed: {e}. Skipping batch.") 

81 gpu_results = [] 

82 

83 # Process results 

84 for i, result in enumerate(gpu_results): 

85 iterations_tried += 1 

86 indicator = batch_indicators[i] 

87 metrics = result["metrics"] 

88 

89 # Fallback logic 

90 perf_score = (metrics.get("total_return", 0) * metrics.get("profit_factor", 1)) / (metrics.get("max_drawdown", 0) + 0.01) 

91 if perf_score > best_performance_score: 

92 best_performance_score = perf_score 

93 fallback_indicator = indicator 

94 fallback_metrics = metrics 

95 

96 if not generator._meets_basic_targets(metrics): 

97 continue 

98 

99 if metrics.get("trade_count", 0) < generator.config.min_trades: 

100 continue 

101 

102 # MC Validation 

103 try: 

104 ind_id = indicator.get_id() 

105 if ind_id in mc_cache: 

106 mc_pass_rate = mc_cache[ind_id] 

107 else: 

108 mc_result = generator._run_mc_validation(data, indicator, scenarios=mc_scenarios) 

109 mc_pass_rate = mc_result.pass_rate 

110 mc_cache[ind_id] = mc_pass_rate 

111 

112 if mc_pass_rate > best_mc_rate: 

113 best_mc_rate = mc_pass_rate 

114 best_indicator = indicator 

115 final_metrics = metrics 

116 best_mc_details = { 

117 "step_results": [ 

118 {"method": r.method_name, "passed": r.passed, "rate": r.pass_rate, "advice": r.advice} 

119 for r in mc_result.step_results 

120 ], 

121 "timing_stats": getattr(mc_result, "timing_stats", {}) 

122 } 

123 logger.info(f"New best: {indicator.name} MC rate={mc_pass_rate:.2%}") 

124 

125 if mc_pass_rate > 0.0: 

126 generator._candidates.append((indicator, mc_pass_rate)) 

127 

128 except Exception as e: 

129 logger.warning(f"Error validating indicator: {e}") 

130 continue 

131 

132 # Progress callback 

133 _update_progress(generator, start_time, batch_start, actual_batch_size, total_iterations, best_mc_rate) 

134 

135 # Early stopping 

136 if generator.config.early_stopping and best_mc_rate >= 0.95: 

137 logger.info(f"Early stopping: found solution at iteration {batch_start + actual_batch_size}") 

138 break 

139 

140 # Check for shutdown requested 

141 if generator.executor and getattr(generator.executor, "_shutdown_requested", False): 

142 logger.info("Shutdown requested. Stopping search.") 

143 break 

144 

145 except KeyboardInterrupt: 

146 logger.info("KeyboardInterrupt caught in generator. Cleaning up...") 

147 if generator._progress_callback: 

148 generator._progress_callback(iterations_tried, total_iterations, "Interrupted by user") 

149 if generator.executor: 

150 generator.executor.__exit__(None, None, None) 

151 generator.executor = None 

152 # Return what we found so far instead of crashing 

153 return _create_result(generator, best_indicator, fallback_indicator, best_mc_rate, best_mc_details, final_metrics, fallback_metrics, iterations_tried, start_time, data) 

154 

155 # Evolution Phase 

156 best_indicator, best_mc_rate, best_mc_details = _run_evolution_phase( 

157 generator, data, best_indicator, best_mc_rate, best_mc_details 

158 ) 

159 

160 elapsed = time.time() - start_time 

161 if generator._progress_callback: 

162 generator._progress_callback(total_iterations, total_iterations, f"Best MC rate: {best_mc_rate:.1%} [Done]") 

163 

164 if generator.executor: 

165 generator.executor.__exit__(None, None, None) 

166 generator.executor = None 

167 

168 return _create_result( 

169 generator, best_indicator, fallback_indicator, best_mc_rate, 

170 best_mc_details, final_metrics, fallback_metrics, iterations_tried, start_time, data 

171 ) 

172 

173 

174def _create_result( 

175 generator: IndicatorGenerator, 

176 best_indicator, 

177 fallback_indicator, 

178 best_mc_rate: float, 

179 best_mc_details: dict, 

180 final_metrics: dict, 

181 fallback_metrics: dict, 

182 iterations_tried: int, 

183 start_time: float, 

184 data: pd.DataFrame 

185) -> GeneratorResult: 

186 """Helper to create GeneratorResult.""" 

187 if not best_indicator and fallback_indicator: 

188 best_indicator = fallback_indicator 

189 final_metrics = fallback_metrics 

190 

191 if best_indicator and not final_metrics: 

192 signals = best_indicator.generate_signals(data) 

193 final_metrics = generator.metrics_calc.calculate_all( 

194 data, signals, use_sl_tp=generator.config.use_sl_tp, 

195 sl_pct=generator.config.stop_loss_pct, tp_pct=generator.config.take_profit_pct 

196 ) 

197 

198 elapsed = time.time() - start_time 

199 

200 return GeneratorResult( 

201 success=best_mc_rate >= generator.config.mc_pass_threshold, 

202 indicator=best_indicator, 

203 parameters=best_indicator.get_parameters() if best_indicator else {}, 

204 final_metrics=final_metrics, 

205 mc_pass_rate=best_mc_rate, 

206 mc_details=best_mc_details, 

207 iterations_tried=iterations_tried, 

208 elapsed_time=elapsed, 

209 candidates_found=len(generator._candidates), 

210 ) 

211 

212 

213def _pre_generate_scenarios(generator: IndicatorGenerator, data: pd.DataFrame, total_iterations: int) -> list[pd.DataFrame] | None: 

214 # If GPU is enabled and we are using shuffling, skip pre-generation to use end-to-end GPU path 

215 if generator.config.use_gpu and generator.config.use_mc_shuffling and not any([ 

216 generator.config.use_mc_noise, 

217 generator.config.use_mc_sensitivity, 

218 generator.config.use_mc_walk_forward, 

219 generator.config.use_mc_block_bootstrap 

220 ]): 

221 logger.info("Skipping pre-generation to use end-to-end GPU path") 

222 return None 

223 

224 if not generator.config.use_mc_block_bootstrap: 

225 try: 

226 temp_mc_config = MCConfig( 

227 iterations=generator.config.mc_iterations, 

228 use_shuffling=generator.config.use_mc_shuffling, 

229 use_noise=generator.config.use_mc_noise, 

230 use_sensitivity=generator.config.use_mc_sensitivity, 

231 use_walk_forward=generator.config.use_mc_walk_forward, 

232 use_block_bootstrap=False, 

233 ) 

234 if generator._progress_callback: 

235 generator._progress_callback(0, total_iterations, "Generating shared MC scenarios...") 

236 

237 temp_engine = MonteCarloEngine(temp_mc_config) 

238 return temp_engine.scenario_builder.generate(data) 

239 except Exception as e: 

240 logger.warning(f"Failed to pre-generate scenarios: {e}. Will generate per candidate.") 

241 return None 

242 

243 

244def _update_progress(generator: IndicatorGenerator, start_time: float, batch_start: int, batch_size: int, total: int, best_rate: float): 

245 if generator._progress_callback: 

246 current_iter = min(batch_start + batch_size, total) 

247 elapsed = time.time() - start_time 

248 ops_sec = current_iter / elapsed if elapsed > 0 else 0.0 

249 status = f"Best MC rate: {best_rate:.1%} | Speed: {ops_sec:.1f} op/s" 

250 generator._progress_callback(current_iter, total, status) 

251 

252 

253def _run_evolution_phase(generator: IndicatorGenerator, data: pd.DataFrame, best_ind, best_rate, best_details): 

254 if "dynamic" in generator.config.indicator_types and len(generator._candidates) >= 2: 

255 logger.info(f"Starting evolutionary optimization on {len(generator._candidates)} candidates...") 

256 try: 

257 dynamic_candidates = [c[0] for c in generator._candidates if isinstance(c[0], DynamicIndicator)] 

258 if len(dynamic_candidates) >= 2: 

259 evolved_best = generator._run_evolution(data, initial_population=dynamic_candidates) 

260 if evolved_best: 

261 mc_result = generator._run_mc_validation(data, evolved_best) 

262 mc_rate = mc_result.pass_rate 

263 

264 if mc_rate > 0: 

265 generator._candidates.append((evolved_best, mc_rate)) 

266 

267 if mc_rate > best_rate: 

268 best_rate = mc_rate 

269 best_ind = evolved_best 

270 best_details = { 

271 "step_results": [ 

272 {"method": r.method_name, "passed": r.passed, "rate": r.pass_rate, "advice": r.advice} 

273 for r in mc_result.step_results 

274 ], 

275 "timing_stats": getattr(mc_result, "timing_stats", {}) 

276 } 

277 logger.info(f"Evolution found better indicator: {evolved_best.name} MC rate={mc_rate:.2%}") 

278 except Exception as e: 

279 logger.error(f"Evolutionary optimization failed: {e}") 

280 return best_ind, best_rate, best_details