Coverage for src / monte_neo / indicators / dynamic.py: 55%

1"""Dynamic indicator implementation. 

2 

3Allows for creation of indicators from source code strings. 

4""" 

5 

6from __future__ import annotations 

7 

8from collections.abc import Callable 

9from typing import Any 

10 

11import numpy as np 

12import pandas as pd 

13 

14from monte_neo.indicators.base import BaseIndicator, IndicatorConfig 

15from monte_neo.indicators.evaluator import compile_source, evaluate_fast_signals 

16from monte_neo.utils.logger import get_logger 

17 

18logger = get_logger(__name__) 

19 

20 

21class DynamicIndicator(BaseIndicator): 

22 """Indicator matching the 'Gene' concept where logic is defined by a code string.""" 

23 

24 def __init__(self, config: IndicatorConfig | None = None) -> None: 

25 super().__init__(config) 

26 self._parameters.setdefault("source_code", "data['close']") 

27 self._compiled_code: Callable[..., Any] | None = None 

28 

29 def __getstate__(self) -> dict[str, Any]: 

30 """Prepare for pickling by removing compiled code.""" 

31 state = self.__dict__.copy() 

32 state["_compiled_code"] = None 

33 return state 

34 

35 def __setstate__(self, state: dict[str, Any]) -> None: 

36 """Restore state after unpickling.""" 

37 self.__dict__.update(state) 

38 self._compiled_code = None 

39 

40 @property 

41 def source_code(self) -> str: 

42 """Get the source code string.""" 

43 return self._parameters["source_code"] 

44 

45 def _compile_if_needed(self) -> None: 

46 """Compile the source code if not already compiled.""" 

47 if self._compiled_code is None: 

48 self._compiled_code = compile_source(self.source_code) 

49 

50 def _reset_to_safe_source(self) -> None: 

51 if self._parameters.get("source_code") != "data['close']": 

52 self._parameters["source_code"] = "data['close']" 

53 self._compiled_code = None 

54 

55 def _evaluate(self, data: pd.DataFrame | dict[str, Any]) -> Any: 

56 if self._compiled_code is None: 

57 self._compile_if_needed() 

58 

59 if self._compiled_code is not None: 

60 indicator_values = self._compiled_code(data, np, pd) 

61 else: 

62 indicator_values = np.nan 

63 

64 if callable(indicator_values) and not isinstance( 

65 indicator_values, (pd.Series, pd.DataFrame) 

66 ): 

67 try: 

68 indicator_values = indicator_values() 

69 except Exception: 

70 indicator_values = np.nan 

71 

72 if isinstance(indicator_values, pd.DataFrame): 

73 indicator_values = ( 

74 indicator_values.iloc[:, 0] if not indicator_values.empty else 0 

75 ) 

76 

77 return indicator_values 

78 

79 def _evaluate_with_fallback(self, data: pd.DataFrame | dict[str, Any]) -> Any: 

80 try: 

81 return self._evaluate(data) 

82 except Exception as e: 

83 logger.debug(f"Runtime error in dynamic indicator: {e}") 

84 self._reset_to_safe_source() 

85 try: 

86 # If data is a dict, we might need to convert it back to DataFrame for fallback 

87 # but 'data['close']' works for both. 

88 return self._evaluate(data) 

89 except Exception as safe_error: 

90 logger.debug(f"Runtime error in safe dynamic indicator: {safe_error}") 

91 return np.nan 

92 

93 def get_metal_params(self, commission_bps: float = 5.0, slippage_bps: float = 5.0) -> list[float] | None: 

94 """Return parameters for native Metal kernel if formula is supported.""" 

95 from monte_neo.indicators.metal_parser import parse_metal_params 

96 return parse_metal_params(self.source_code, commission_bps=commission_bps, slippage_bps=slippage_bps) 

97 

98 def get_formula(self) -> str: 

99 """Get the source code string used for calculation.""" 

100 return f"Dynamic: {self.source_code}" 

101 

102 def to_mlx_representation(self) -> Any | None: 

103 """Convert to MLX representation for GPU execution. 

104  

105 Tries to map common patterns to native MLX indicators for speed, 

106 otherwise falls back to MLXDynamicStrategy. 

107 """ 

108 import re 

109 

110 from monte_neo.core.acceleration.indicators import ( 

111 MLXSMA, 

112 MLXCrossStrategy, 

113 MLXDynamicStrategy, 

114 MLXSMACrossStrategy, 

115 ) 

116 

117 code = self.source_code.replace(" ", "") 

118 

119 # 1. Price > SMA(P) 

120 sma_pattern = r"data\['close'\]>data\['close'\]\.rolling\((\d+)\)\.mean\(\)" 

121 match = re.search(sma_pattern, code) 

122 if match: 

123 return MLXCrossStrategy(MLXSMA(int(match.group(1))), mode="greater") 

124 

125 # 2. Price < SMA(P) 

126 sma_pattern_lt = r"data\['close'\]<data\['close'\]\.rolling\((\d+)\)\.mean\(\)" 

127 match = re.search(sma_pattern_lt, code) 

128 if match: 

129 return MLXCrossStrategy(MLXSMA(int(match.group(1))), mode="less") 

130 

131 # 3. SMA(F) > SMA(S) 

132 sma_cross_pattern = r"data\['close'\]\.rolling\((\d+)\)\.mean\(\)>data\['close'\]\.rolling\((\d+)\)\.mean\(\)" 

133 match = re.search(sma_cross_pattern, code) 

134 if match: 

135 return MLXSMACrossStrategy(int(match.group(1)), int(match.group(2))) 

136 

137 # 4. RSI < Threshold 

138 rsi_pattern_lt = r"rsi\(.*?,?(\d+)\)<([\d\.]+)" 

139 match = re.search(rsi_pattern_lt, code) 

140 if match: 

141 # Re-use MLXCrossStrategy but with RSI as indicator 

142 # Wait, MLXCrossStrategy compares close vs indicator. 

143 # For RSI < 30, we need a different strategy or a constant indicator. 

144 # For now, let's just use the fallback for RSI to be safe. 

145 pass 

146 

147 # Fallback to general strategy 

148 return MLXDynamicStrategy(self) 

149 

150 def calculate(self, data: pd.DataFrame) -> pd.DataFrame: 

151 """Calculate indicator values using the generated code. 

152 

153 Args: 

154 data: OHLCV DataFrame. 

155 

156 Returns: 

157 DataFrame with 'value' column (for now) or dynamic columns. 

158 """ 

159 result = data.copy() 

160 indicator_values = self._evaluate_with_fallback(data) 

161 

162 if isinstance(indicator_values, (pd.Series, np.ndarray)): 

163 result["dynamic"] = indicator_values 

164 else: 

165 result["dynamic"] = indicator_values 

166 

167 return result 

168 

169 def generate_signals(self, data: pd.DataFrame) -> pd.DataFrame: 

170 """Generate signals. 

171 

172 For dynamic indicators, the 'source_code' might calculate a boolean signal directly, 

173 or a continuous value. 

174 """ 

175 vals = self._evaluate_with_fallback(data) 

176 

177 signals = pd.DataFrame(index=data.index) 

178 signals["signal"] = 0 

179 

180 vals = pd.to_numeric(vals, errors="coerce") 

181 

182 if isinstance(vals, pd.Series): 

183 aligned = vals.reindex(data.index).fillna(0) 

184 elif isinstance(vals, np.ndarray): 

185 aligned = pd.Series(vals, index=data.index).fillna(0) 

186 else: 

187 aligned = pd.Series([vals] * len(data), index=data.index).fillna(0) 

188 

189 if aligned.dtype == bool: 

190 signals.loc[aligned, "signal"] = 1 

191 else: 

192 signals.loc[aligned > 0, "signal"] = 1 

193 signals.loc[aligned < 0, "signal"] = -1 

194 

195 return signals 

196 

197 def generate_signals_fast(self, data: pd.DataFrame | np.ndarray) -> np.ndarray: 

198 """Fast version of signal generation for dynamic indicators.""" 

199 if self._compiled_code is None: 

200 self._compile_if_needed() 

201 

202 if self._compiled_code is None: 

203 return np.zeros(len(data), dtype=np.float32) 

204 

205 return evaluate_fast_signals(self._compiled_code, data) 

206 

207 def get_min_periods(self) -> int: 

208 # Difficult to know statically. Default to something safe or 0. 

209 return 50