Coverage for src / monte_neo / core / generator.py: 100%

1"""Indicator generator module. 

2 

3Core engine for generating robust trading indicators. 

4""" 

5 

6from __future__ import annotations 

7 

8from collections.abc import Callable 

9from typing import TYPE_CHECKING 

10 

11import numpy as np 

12import pandas as pd 

13 

14from monte_neo.core.config import GeneratorConfig, GeneratorResult 

15from monte_neo.core.evolution import EvolutionEngine 

16from monte_neo.core.generator_search import run_search 

17from monte_neo.core.generator_utils import PARAM_SPACES, estimate_time 

18from monte_neo.core.gpu_engine import MLXBacktestEngine 

19from monte_neo.indicators.base import BaseIndicator 

20from monte_neo.indicators.code_gen import CodeGenerator 

21from monte_neo.indicators.dynamic import DynamicIndicator 

22from monte_neo.indicators.technical import MACDIndicator, RSIIndicator, SMAIndicator 

23from monte_neo.metrics.calculator import MetricsCalculator 

24from monte_neo.monte_carlo.engine import MCConfig, MonteCarloEngine 

25from monte_neo.utils.logger import get_logger 

26from monte_neo.utils.parallel import ParallelExecutor 

27 

28if TYPE_CHECKING: 

29 from monte_neo.monte_carlo.engine import MCResult 

30 

31logger = get_logger(__name__) 

32 

33 

34class IndicatorGenerator: 

35 """Generate robust trading indicators.""" 

36 

37 def __init__(self, config: GeneratorConfig | None = None) -> None: 

38 """Initialize generator. 

39 

40 Args: 

41 config: Generator configuration. 

42 """ 

43 self.config = config or GeneratorConfig() 

44 self.rng = np.random.default_rng() 

45 self.metrics_calc = MetricsCalculator( 

46 initial_capital=self.config.initial_capital, 

47 leverage=self.config.leverage 

48 ) 

49 self.gpu_engine = MLXBacktestEngine( 

50 precision=self.config.gpu_precision, 

51 metal_driver=self.config.metal_driver, 

52 initial_capital=self.config.initial_capital, 

53 leverage=self.config.leverage 

54 ) 

55 self.executor: ParallelExecutor | None = None 

56 

57 # State 

58 self.population: list[BaseIndicator] = [] 

59 self._progress_callback: Callable[[int, int, str], None] | None = None 

60 self._candidates: list[tuple[BaseIndicator, float]] = [] 

61 

62 def set_progress_callback( 

63 self, 

64 callback: Callable[[int, int, str], None], 

65 ) -> None: 

66 """Set progress callback.""" 

67 self._progress_callback = callback 

68 

69 def _run_mc_validation( 

70 self, 

71 data: pd.DataFrame, 

72 indicator: BaseIndicator, 

73 scenarios: list[pd.DataFrame] | None = None 

74 ) -> MCResult: 

75 """Run Monte Carlo validation for a single indicator.""" 

76 mc_config = MCConfig( 

77 iterations=self.config.mc_iterations, 

78 use_shuffling=self.config.use_mc_shuffling, 

79 use_noise=self.config.use_mc_noise, 

80 use_sensitivity=self.config.use_mc_sensitivity, 

81 use_walk_forward=self.config.use_mc_walk_forward, 

82 use_block_bootstrap=self.config.use_mc_block_bootstrap, 

83 use_sequential=self.config.use_sequential_mc, 

84 pass_threshold=self.config.mc_pass_threshold, 

85 use_sl_tp=self.config.use_sl_tp, 

86 sl_pct=self.config.stop_loss_pct, 

87 tp_pct=self.config.take_profit_pct, 

88 use_gpu=self.config.use_gpu, 

89 gpu_precision=self.config.gpu_precision, 

90 metal_driver=self.config.metal_driver, 

91 initial_capital=self.config.initial_capital, 

92 leverage=self.config.leverage, 

93 ) 

94 mc_engine = MonteCarloEngine(mc_config, executor=self.executor) 

95 

96 return mc_engine.run( 

97 data, 

98 indicator, 

99 self.metrics_calc, 

100 self.config.target_metrics, 

101 existing_scenarios=scenarios, 

102 interactive=False 

103 ) 

104 

105 def generate(self, data: pd.DataFrame) -> GeneratorResult: 

106 """Generate a robust indicator.""" 

107 return run_search(self, data) 

108 

109 def _mutate_indicator(self, indicator: BaseIndicator) -> BaseIndicator: 

110 """Mutate an indicator (wrapper for EvolutionEngine).""" 

111 evo = EvolutionEngine(self.config) 

112 return evo._mutate_indicator(indicator) 

113 

114 def _generate_random_indicator(self) -> BaseIndicator: 

115 """Generate a random indicator with random parameters.""" 

116 ind_type = self.rng.choice(self.config.indicator_types) 

117 indicator: BaseIndicator 

118 

119 if ind_type == "sma": 

120 indicator = SMAIndicator() 

121 elif ind_type == "rsi": 

122 indicator = RSIIndicator() 

123 elif ind_type == "macd": 

124 indicator = MACDIndicator() 

125 elif ind_type == "dynamic": 

126 indicator = DynamicIndicator() 

127 code_gen = CodeGenerator(self.rng) 

128 code = code_gen.generate_code() 

129 indicator.set_parameter("source_code", code) 

130 return indicator 

131 else: 

132 indicator = SMAIndicator() 

133 

134 # Set random parameters 

135 param_space = PARAM_SPACES.get(ind_type, {}) 

136 for param_name, (min_val, max_val) in param_space.items(): 

137 value = int(self.rng.integers(min_val, max_val + 1)) 

138 indicator.set_parameter(param_name, value) 

139 

140 return indicator 

141 

142 def _meets_basic_targets(self, metrics: dict) -> bool: 

143 """Check if metrics meet basic targets.""" 

144 for name, target in self.config.target_metrics.items(): 

145 if name not in metrics: 

146 continue 

147 actual = metrics[name] 

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

149 if actual > target: 

150 return False 

151 else: 

152 if actual < target: 

153 return False 

154 return True 

155 

156 def _run_evolution( 

157 self, 

158 data: pd.DataFrame, 

159 initial_population: list[BaseIndicator] | None = None 

160 ) -> BaseIndicator | None: 

161 """Run evolutionary optimization on candidates.""" 

162 population = initial_population if initial_population is not None else [c[0] for c in self._candidates] 

163 evolution = EvolutionEngine( 

164 self.config, 

165 metrics_calc=self.metrics_calc, 

166 progress_callback=self._progress_callback 

167 ) 

168 return evolution.run(data, population, executor=self.executor) 

169 

170 def estimate_time(self, data: pd.DataFrame) -> float: 

171 """Estimate generation time in minutes.""" 

172 return estimate_time(self, data)