Coverage for src / monte_neo / monte_carlo / engine.py: 100%
123 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-28 16:27 +0200
1"""Monte Carlo simulation engine.
3Main engine for running Monte Carlo simulations with various methods.
4"""
6from __future__ import annotations
8import time
9from collections.abc import Callable
10from typing import TYPE_CHECKING
12import numpy as np
13import pandas as pd
15from monte_neo.monte_carlo.scenarios import ScenarioBuilder
16from monte_neo.monte_carlo.types import MCConfig, MCResult
17from monte_neo.monte_carlo.utils import summarize_metrics
18from monte_neo.monte_carlo.workers import init_worker_data, run_scenario_batch, run_single_scenario
19from monte_neo.utils.logger import get_logger
20from monte_neo.utils.parallel import ParallelExecutor
22if TYPE_CHECKING:
23 from monte_neo.indicators.base import BaseIndicator
24 from monte_neo.metrics.calculator import MetricsCalculator
26logger = get_logger(__name__)
28class MonteCarloEngine:
29 """Monte Carlo simulation engine."""
31 def __init__(
32 self,
33 config: MCConfig | None = None,
34 executor: ParallelExecutor | None = None,
35 ) -> None:
36 """Initialize Monte Carlo engine.
38 Args:
39 config: Monte Carlo configuration.
40 executor: Optional shared parallel executor.
41 """
42 self.config = config or MCConfig()
43 self.executor = executor
44 self.rng = np.random.default_rng(self.config.random_seed)
46 # Initialize sub-modules
47 self.scenario_builder = ScenarioBuilder(self.config)
49 from monte_neo.core.gpu_engine import MLXBacktestEngine
50 self.gpu_engine = MLXBacktestEngine(
51 precision=self.config.gpu_precision,
52 metal_driver=self.config.metal_driver,
53 initial_capital=self.config.initial_capital,
54 leverage=self.config.leverage
55 )
57 self._progress_callback: Callable[[int, int], None] | None = None
59 def set_progress_callback(self, callback: Callable[[int, int], None]) -> None:
60 """Set progress callback function.
62 Args:
63 callback: Function(current, total) for progress updates.
64 """
65 self._progress_callback = callback
67 def run(
68 self,
69 data: pd.DataFrame,
70 indicator: BaseIndicator,
71 metrics_calc: MetricsCalculator,
72 target_metrics: dict[str, float],
73 existing_scenarios: list[pd.DataFrame] | None = None,
74 interactive: bool = True,
75 ) -> MCResult:
76 """Run Monte Carlo simulation.
78 Args:
79 data: OHLCV DataFrame.
80 indicator: Indicator to test.
81 metrics_calc: Metrics calculator.
82 target_metrics: Target metrics to achieve.
83 existing_scenarios: Optional list of pre-generated scenarios.
84 interactive: Whether to ask for confirmation in sequential mode.
86 Returns:
87 MCResult with simulation results.
88 """
89 start_time = time.time()
90 passed_count = 0
91 all_results = []
93 def _meets_targets(metrics: dict[str, float]) -> bool:
94 for metric_name, target_value in target_metrics.items():
95 if metric_name not in metrics:
96 continue
97 actual = metrics[metric_name]
98 if metric_name in ["max_drawdown", "consecutive_losses"]:
99 if actual > target_value:
100 return False
101 else:
102 if actual < target_value:
103 return False
104 return True
106 if self.config.use_sequential:
107 return self.run_sequential(data, indicator, metrics_calc, target_metrics, interactive=interactive)
109 # Check for Pure GPU Acceleration (End-to-End on GPU)
110 # Only for Shuffling method currently, and if indicator supports it.
111 # This bypasses CPU scenario generation and data transfer overhead.
112 has_mlx = indicator.to_mlx_representation() is not None
113 has_metal = indicator.get_metal_params() is not None
115 only_shuffling = (
116 (self.config.use_shuffling or not any([
117 self.config.use_noise,
118 self.config.use_sensitivity,
119 self.config.use_walk_forward,
120 self.config.use_block_bootstrap
121 ]))
122 and not self.config.use_noise
123 and not self.config.use_sensitivity
124 and not self.config.use_walk_forward
125 and not self.config.use_block_bootstrap
126 )
128 if (has_mlx or has_metal) and only_shuffling and existing_scenarios is None and self.config.iterations > 100:
129 try:
130 engine_type = "Native Metal" if has_metal else "MLX"
131 logger.info(f"🚀 Using High-Performance {engine_type} Engine for {self.config.iterations} iterations")
132 results, timing_stats = self.gpu_engine.run_full_simulation(
133 data=data,
134 indicator=indicator,
135 n_scenarios=self.config.iterations,
136 method="shuffling",
137 seed=self.config.random_seed or 42,
138 use_sl_tp=self.config.use_sl_tp,
139 sl_pct=self.config.sl_pct,
140 tp_pct=self.config.tp_pct,
141 )
143 # Transform results to match MCResult format
144 passed_count = sum(1 for r in results if _meets_targets(r.get("metrics", {})))
145 total = len(results)
147 all_results = []
148 for i, res in enumerate(results):
149 metrics = res.get("metrics", {})
150 passed = _meets_targets(metrics)
151 all_results.append({
152 "scenario_idx": i,
153 "passed": passed,
154 "metrics": metrics
155 })
157 if self._progress_callback:
158 self._progress_callback(total, total)
160 finalize_res = self._finalize_results(passed_count, total, all_results, start_time)
161 finalize_res.timing_stats = timing_stats
162 return finalize_res
164 except Exception as e:
165 logger.warning(f"Pure GPU execution failed, falling back: {e}")
166 # Fall through to standard methods
168 # Generate test scenarios
169 other_methods_enabled = (
170 self.config.use_shuffling
171 or self.config.use_noise
172 or self.config.use_sensitivity
173 or self.config.use_walk_forward
174 )
175 if (
176 self.config.use_block_bootstrap
177 and not other_methods_enabled
178 and existing_scenarios is None
179 ):
180 # Lazy generation for Block Bootstrap to avoid memory overhead
181 logger.debug(f"Running lazy Block Bootstrap with {self.config.iterations} iterations")
183 # Ensure we have an executor with initialized data
184 executor = self.executor
185 should_shutdown = False
187 if executor is None:
188 # Create a local executor with data initialization
189 executor = ParallelExecutor(
190 n_workers=self.config.n_workers,
191 initializer=init_worker_data,
192 initargs=(data,)
193 )
194 should_shutdown = True
195 executor.__enter__()
197 try:
198 # Use GPU engine's lazy method
199 results = self.gpu_engine.backtest_lazy_scenarios(
200 indicator,
201 self.config.iterations,
202 executor=executor,
203 block_size=None, # Auto-calculated
204 base_seed=self.config.random_seed or 42,
205 use_sl_tp=self.config.use_sl_tp,
206 sl_pct=self.config.sl_pct,
207 tp_pct=self.config.tp_pct,
208 )
210 # Transform results
211 passed_count = sum(1 for r in results if _meets_targets(r.get("metrics", {})))
212 total = len(results)
214 all_results = []
215 for i, res in enumerate(results):
216 metrics = res.get("metrics", {})
217 passed = _meets_targets(metrics)
218 all_results.append({
219 "scenario_idx": i,
220 "passed": passed,
221 "metrics": metrics
222 })
224 if self._progress_callback:
225 self._progress_callback(total, total)
227 return self._finalize_results(passed_count, total, all_results, start_time)
229 finally:
230 if should_shutdown and executor:
231 executor.__exit__(None, None, None)
233 if existing_scenarios is not None:
234 scenarios = existing_scenarios
235 else:
236 scenarios = self.scenario_builder.generate(data)
237 total = len(scenarios)
239 logger.debug(f"Running {total} Monte Carlo scenarios in parallel")
241 # Prepare arguments for parallel execution
242 # We use a helper function to avoid pickling issues with 'self' if possible,
243 # but ProcessPoolExecutor usually handles methods if they are defined at module level.
244 # Alternatively, we can use a standalone function.
246 # Try GPU acceleration if many scenarios
247 if len(scenarios) > 10:
248 try:
249 logger.debug(f"Offloading {total} scenarios to GPU (MLX)...")
250 gpu_results = self.gpu_engine.backtest_scenarios(
251 indicator,
252 scenarios,
253 executor=self.executor,
254 use_sl_tp=self.config.use_sl_tp,
255 sl_pct=self.config.sl_pct,
256 tp_pct=self.config.tp_pct,
257 )
259 for i, res in enumerate(gpu_results):
260 # The GPU engine returns a dict with 'passed' and 'metrics'
261 metrics = res.get("metrics", {})
262 passed = _meets_targets(metrics)
263 if passed:
264 passed_count += 1
265 all_results.append(
266 {
267 "scenario_idx": i,
268 "passed": passed,
269 "metrics": metrics,
270 }
271 )
273 if self._progress_callback:
274 self._progress_callback(total, total)
276 return self._finalize_results(
277 passed_count, total, all_results, start_time
278 )
280 except Exception as e:
281 logger.warning(f"GPU acceleration failed, falling back to CPU: {e}")
283 # Fallback to CPU parallel execution
284 cpu_results = self._run_cpu_parallel(scenarios, indicator, metrics_calc, target_metrics)
285 passed_count += cpu_results["passed_count"]
286 all_results.extend(cpu_results["all_results"])
288 if self._progress_callback:
289 self._progress_callback(total, total)
291 return self._finalize_results(passed_count, total, all_results, start_time)
293 def _run_cpu_parallel(
294 self,
295 scenarios: list[pd.DataFrame],
296 indicator: BaseIndicator,
297 metrics_calc: MetricsCalculator,
298 target_metrics: dict[str, float],
299 ) -> dict:
300 """Run scenarios on CPU in parallel."""
301 from functools import partial
302 passed_count = 0
303 all_results = []
304 total = len(scenarios)
306 executor = self.executor
307 if executor is None:
308 executor = ParallelExecutor(n_workers=self.config.n_workers)
310 n_workers = executor.n_workers
312 if total < 50 or n_workers == 1:
313 worker_func = partial(
314 run_single_scenario,
315 indicator=indicator,
316 metrics_calc=metrics_calc,
317 target_metrics=target_metrics,
318 )
319 results = executor.map(worker_func, scenarios)
321 for i, result in enumerate(results):
322 if result is None: continue
323 meets_targets, metrics = result
324 if meets_targets: passed_count += 1
325 all_results.append({"scenario_idx": i, "passed": meets_targets, "metrics": metrics})
326 else:
327 batch_size = max(10, total // (n_workers * 4))
328 scenario_batches = [scenarios[i : i + batch_size] for i in range(0, total, batch_size)]
329 batch_worker = partial(run_scenario_batch, indicator=indicator, metrics_calc=metrics_calc, target_metrics=target_metrics)
330 batch_results_list = executor.map(batch_worker, scenario_batches)
332 current_idx = 0
333 for batch_res in batch_results_list:
334 if not batch_res: continue
335 for meets_targets, metrics in batch_res:
336 if meets_targets: passed_count += 1
337 all_results.append({"scenario_idx": current_idx, "passed": meets_targets, "metrics": metrics})
338 current_idx += 1
340 return {"passed_count": passed_count, "all_results": all_results}
342 def _finalize_results(
343 self, passed_count: int, total: int, all_results: list, start_time: float
344 ) -> MCResult:
345 """Finalize and summarize results."""
346 elapsed = time.time() - start_time
347 pass_rate = passed_count / total if total > 0 else 0
348 passed = pass_rate >= self.config.pass_threshold
349 metrics_summary = summarize_metrics(all_results)
351 logger.info(f"✨ Monte Carlo simulation completed in {elapsed:.2f}s ({total} iterations)")
352 logger.info(f"📊 Pass Rate: {pass_rate:.1%} ({'PASSED' if passed else 'FAILED'})")
354 return MCResult(
355 passed=passed,
356 pass_rate=pass_rate,
357 iterations_run=total,
358 elapsed_time=elapsed,
359 metrics_summary=metrics_summary,
360 detailed_results=all_results,
361 )
363 def run_sequential(
364 self,
365 data: pd.DataFrame,
366 indicator: BaseIndicator,
367 metrics_calc: MetricsCalculator,
368 target_metrics: dict[str, float],
369 interactive: bool = True,
370 ) -> MCResult:
371 """Run Monte Carlo simulation sequentially.
373 Args:
374 data: OHLCV DataFrame.
375 indicator: Indicator to test.
376 metrics_calc: Metrics calculator.
377 target_metrics: Target metrics.
378 interactive: Whether to ask for confirmation before each step.
380 Returns:
381 MCResult with sequential simulation results.
382 """
383 from monte_neo.monte_carlo.sequential import SequentialMCRunner
384 runner = SequentialMCRunner(self)
385 return runner.run(data, indicator, metrics_calc, target_metrics, interactive=interactive)