Coverage for src / monte_neo / indicators / sma.py: 67%

1from __future__ import annotations 

2 

3import numpy as np 

4import pandas as pd 

5 

6from monte_neo.indicators.base import BaseIndicator, IndicatorConfig 

7from monte_neo.indicators.numba_funcs import sma_crossover_signals_numba 

8from monte_neo.indicators.technical_lib import TechnicalIndicators 

9 

10 

11class SMAIndicator(BaseIndicator): 

12 """SMA Crossover indicator.""" 

13 

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

15 super().__init__(config) 

16 self._parameters.setdefault("fast_period", 10) 

17 self._parameters.setdefault("slow_period", 20) 

18 

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

20 result = data.copy() 

21 result["sma_fast"] = TechnicalIndicators.sma( 

22 data["close"], self._parameters["fast_period"] 

23 ) 

24 result["sma_slow"] = TechnicalIndicators.sma( 

25 data["close"], self._parameters["slow_period"] 

26 ) 

27 return result 

28 

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

30 # High-performance single-pass Numba calculation 

31 close = data["close"].to_numpy() 

32 fast_period = self._parameters["fast_period"] 

33 slow_period = self._parameters["slow_period"] 

34 

35 # Ensure fast < slow for crossover logic 

36 # Cast to int for Numba compatibility 

37 p1 = int(round(min(fast_period, slow_period))) 

38 p2 = int(round(max(fast_period, slow_period))) 

39 

40 # Minimum period is 2 

41 p1 = max(2, p1) 

42 p2 = max(p1 + 1, p2) 

43 

44 final_signals = sma_crossover_signals_numba(close, p1, p2) 

45 return pd.DataFrame({"signal": final_signals}, index=data.index) 

46 

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

48 if isinstance(data, pd.DataFrame): 

49 close = data["close"].to_numpy() 

50 else: 

51 # Assume 1D close prices or OHLC matrix (close is col 3) 

52 close = data[:, 3] if data.ndim > 1 else data 

53 

54 p1 = int(round(min(self._parameters["fast_period"], self._parameters["slow_period"]))) 

55 p2 = int(round(max(self._parameters["fast_period"], self._parameters["slow_period"]))) 

56 

57 # Minimum period is 2 

58 p1 = max(2, p1) 

59 p2 = max(p1 + 1, p2) 

60 

61 return sma_crossover_signals_numba(close, p1, p2) 

62 

63 def get_formula(self) -> str: 

64 p1 = min(self._parameters["fast_period"], self._parameters["slow_period"]) 

65 p2 = max(self._parameters["fast_period"], self._parameters["slow_period"]) 

66 return f"SMA({p1}) Cross SMA({p2})" 

67 

68 def get_min_periods(self) -> int: 

69 return self._parameters["slow_period"] 

70 

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

72 """Return parameters for native Metal kernel.""" 

73 # Layout: [type, p1, p2, p3, atr_period, sl_mult, tp_mult, ts_mult, commission, slippage] 

74 # type 0: SMA 

75 return [ 

76 0.0, # type 

77 float(self._parameters.get("fast_period", 10)), 

78 float(self._parameters.get("slow_period", 20)), 

79 0.0, # p3 

80 14.0, # ATR 

81 1.5, # SL 

82 3.0, # TP 

83 2.0, # TS 

84 commission_bps, 

85 slippage_bps 

86 ] 

87 

88 def to_mlx_representation(self): 

89 try: 

90 from monte_neo.core.acceleration.indicators import MLXSMACrossStrategy 

91 fast = int(self._parameters["fast_period"]) 

92 slow = int(self._parameters["slow_period"]) 

93 return MLXSMACrossStrategy(fast, slow) 

94 except ImportError: 

95 return None