Coverage for src / monte_neo / indicators / evaluator.py: 82%

1"""Evaluator for dynamic indicators. 

2""" 

3 

4from __future__ import annotations 

5 

6from collections.abc import Callable 

7from typing import Any 

8 

9import numpy as np 

10import pandas as pd 

11 

12from monte_neo.utils.logger import get_logger 

13 

14logger = get_logger(__name__) 

15 

16def rsi(data, period=14): 

17 if isinstance(data, dict): 

18 close = pd.Series(data['close']) 

19 else: 

20 close = data['close'] if hasattr(data, 'close') else data 

21 delta = close.diff() 

22 gain = (delta.where(delta > 0, 0)).rolling(window=period).mean() 

23 loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean() 

24 rs = gain / loss 

25 return 100 - (100 / (1 + rs)) 

26 

27def sma(data, period=20): 

28 if isinstance(data, dict): 

29 close = pd.Series(data['close']) 

30 else: 

31 close = data['close'] if hasattr(data, 'close') else data 

32 return close.rolling(window=period).mean() 

33 

34def compile_source(source_code: str) -> Callable[..., Any]: 

35 """Compile source code into a function.""" 

36 try: 

37 exec_globals = { 

38 "np": np, 

39 "pd": pd, 

40 "_np": np, 

41 "_pd": pd, 

42 "rsi": rsi, 

43 "sma": sma, 

44 "__builtins__": __builtins__, 

45 } 

46 

47 func_code = ( 

48 f"def _dynamic_calc(data, np, pd):\n return {source_code}" 

49 ) 

50 

51 local_scope: dict[str, Any] = {} 

52 exec(func_code, exec_globals, local_scope) 

53 return local_scope["_dynamic_calc"] 

54 except Exception as e: 

55 logger.debug(f"Failed to compile dynamic indicator: {e}") 

56 # Fallback to safe source 

57 safe_source = "data['close']" 

58 func_code = ( 

59 f"def _dynamic_calc(data, np, pd):\n return {safe_source}" 

60 ) 

61 safe_scope: dict[str, Any] = {} 

62 exec(func_code, exec_globals, safe_scope) 

63 return safe_scope["_dynamic_calc"] 

64 

65def evaluate_fast_signals(compiled_code: Callable, data: pd.DataFrame | np.ndarray) -> np.ndarray: 

66 """Fast version of signal generation.""" 

67 if isinstance(data, pd.DataFrame): 

68 fast_data = { 

69 "open": data["open"], 

70 "high": data["high"], 

71 "low": data["low"], 

72 "close": data["close"], 

73 "volume": data["volume"], 

74 } 

75 n_rows = len(data) 

76 else: 

77 # If it's a numpy array, check dimensions 

78 if data.ndim == 1: 

79 # Only close prices provided (typical for MC scenarios) 

80 fast_data = { 

81 "open": pd.Series(data), 

82 "high": pd.Series(data), 

83 "low": pd.Series(data), 

84 "close": pd.Series(data), 

85 "volume": pd.Series(np.ones_like(data)), 

86 } 

87 else: 

88 # OHLCV provided 

89 fast_data = { 

90 "open": pd.Series(data[:, 0]), 

91 "high": pd.Series(data[:, 1]), 

92 "low": pd.Series(data[:, 2]), 

93 "close": pd.Series(data[:, 3]), 

94 "volume": pd.Series(data[:, 4]), 

95 } 

96 n_rows = len(data) 

97 

98 try: 

99 vals = compiled_code(fast_data, np, pd) 

100 

101 if isinstance(vals, pd.Series): 

102 vals = vals.values 

103 

104 if not isinstance(vals, np.ndarray): 

105 vals = np.full(n_rows, vals) 

106 

107 sig_vals = np.zeros(n_rows, dtype=np.float32) 

108 

109 # Standardize signals: >0 is 1, <0 is -1, 0 is 0 

110 sig_vals[np.greater(vals, 0)] = 1.0 

111 sig_vals[np.less(vals, 0)] = -1.0 

112 return sig_vals 

113 except Exception as e: 

114 logger.debug(f"Error in fast evaluation: {e}") 

115 return np.zeros(n_rows, dtype=np.float32)