Coverage for src / monte_neo / core / generator_worker.py: 86%

1from __future__ import annotations 

2 

3from typing import TYPE_CHECKING 

4 

5from monte_neo.indicators.base import BaseIndicator 

6 

7if TYPE_CHECKING: 

8 pass 

9 

10 

11def _search_worker(args: tuple) -> tuple[BaseIndicator | None, float]: 

12 """Worker for parallel indicator search.""" 

13 ( 

14 indicator, 

15 data, 

16 metrics_calc, 

17 target_metrics, 

18 mc_iterations, 

19 mc_shuffling, 

20 mc_noise, 

21 mc_sensitivity, 

22 mc_walk_forward, 

23 min_trades, 

24 use_sl_tp, 

25 sl_pct, 

26 tp_pct, 

27 ) = args 

28 

29 # Quick pre-check 

30 signals = indicator.generate_signals_fast(data) 

31 basic_metrics = metrics_calc.calculate_all( 

32 data, signals, use_sl_tp=use_sl_tp, sl_pct=sl_pct, tp_pct=tp_pct 

33 ) 

34 

35 # Skip if too few trades 

36 if basic_metrics.get("trade_count", 0) < min_trades: 

37 return None, 0.0 

38 

39 # Skip if basic metrics don't meet targets 

40 # Inline check for performance 

41 for name, target in target_metrics.items(): 

42 if name not in basic_metrics: 

43 continue 

44 actual = basic_metrics[name] 

45 if name in ["max_drawdown", "consecutive_losses"]: 

46 if actual > target: 

47 return None, 0.0 

48 else: 

49 if actual < target: 

50 return None, 0.0 

51 

52 # Run Monte Carlo validation 

53 # Note: We need a static version of MC validation or use engine directly 

54 from monte_neo.monte_carlo.engine import MCConfig, MonteCarloEngine 

55 

56 mc_config = MCConfig( 

57 iterations=mc_iterations, 

58 use_shuffling=mc_shuffling, 

59 use_noise=mc_noise, 

60 use_sensitivity=mc_sensitivity, 

61 use_walk_forward=mc_walk_forward, 

62 use_sl_tp=use_sl_tp, 

63 sl_pct=sl_pct, 

64 tp_pct=tp_pct, 

65 ) 

66 mc_engine = MonteCarloEngine(mc_config) 

67 result = mc_engine.run(data, indicator, metrics_calc, target_metrics) 

68 

69 return indicator, result.pass_rate