Coverage for src / monte_neo / indicators / macd.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 macd_signals_numba 

8from monte_neo.indicators.technical_lib import TechnicalIndicators 

9 

10 

11class MACDIndicator(BaseIndicator): 

12 """MACD crossover indicator.""" 

13 

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

15 super().__init__(config) 

16 self._parameters.setdefault("fast", 12) 

17 self._parameters.setdefault("slow", 26) 

18 self._parameters.setdefault("signal", 9) 

19 

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

21 result = data.copy() 

22 macd, signal, hist = TechnicalIndicators.macd( 

23 data["close"], 

24 self._parameters["fast"], 

25 self._parameters["slow"], 

26 self._parameters["signal"], 

27 ) 

28 result["macd"] = macd 

29 result["macd_signal"] = signal 

30 result["macd_histogram"] = hist 

31 return result 

32 

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

34 # High-performance Numba-based MACD signals 

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

36 fast_p = int(round(self._parameters["fast"])) 

37 slow_p = int(round(self._parameters["slow"])) 

38 sig_p = int(round(self._parameters["signal"])) 

39 

40 # Minimum period is 2 

41 fast_p = max(2, fast_p) 

42 slow_p = max(fast_p + 1, slow_p) 

43 sig_p = max(2, sig_p) 

44 

45 sig_vals = macd_signals_numba(close, fast_p, slow_p, sig_p) 

46 return pd.DataFrame({"signal": sig_vals}, index=data.index) 

47 

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

49 if isinstance(data, pd.DataFrame): 

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

51 else: 

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

53 

54 fast_p = int(round(self._parameters["fast"])) 

55 slow_p = int(round(self._parameters["slow"])) 

56 sig_p = int(round(self._parameters["signal"])) 

57 

58 # Minimum period is 2 

59 fast_p = max(2, fast_p) 

60 slow_p = max(fast_p + 1, slow_p) 

61 sig_p = max(2, sig_p) 

62 

63 return macd_signals_numba(close, fast_p, slow_p, sig_p) 

64 

65 def get_formula(self) -> str: 

66 f = self._parameters["fast"] 

67 s = self._parameters["slow"] 

68 sig = self._parameters["signal"] 

69 return f"MACD({f}, {s}, {sig}) Histogram Cross 0" 

70 

71 def get_min_periods(self) -> int: 

72 return self._parameters["slow"] + self._parameters["signal"] 

73 

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

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

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

77 # type 2: MACD 

78 return [ 

79 2.0, # type 

80 float(self._parameters.get("fast", 12)), 

81 float(self._parameters.get("slow", 26)), 

82 float(self._parameters.get("signal", 9)), 

83 14.0, # ATR 

84 1.5, # SL 

85 3.0, # TP 

86 2.0, # TS 

87 commission_bps, 

88 slippage_bps 

89 ]