Coverage for src / monte_neo / indicators / metal_parser.py: 83%

1 

2import re 

3 

4 

5def parse_metal_params(source_code: str, commission_bps: float = 5.0, slippage_bps: float = 5.0) -> list[float] | None: 

6 """Parse dynamic indicator source code into Metal kernel parameters.""" 

7 code = source_code.replace(" ", "") 

8 

9 # Layout: [type, p1, p2, p3, atr_period, sl_mult, tp_mult, ts_mult, commission_bps, slippage_bps] 

10 # Default SL/TP/TS params 

11 common_tail = [14.0, 1.5, 3.0, 2.0, commission_bps, slippage_bps] 

12 

13 def parse_simple_cond(cond_code: str) -> list[float] | None: 

14 # 0. Bollinger Bands (check first to avoid SMA collision) 

15 # Example: data['close']<(data['close'].rolling(20).mean()-2.0*data['close'].rolling(20).std()) 

16 bb_pattern = r"rolling\((\d+)\)\.mean\(\)[\-\+]([\d\.]+)\*.*rolling\(\1\)\.std\(\)" 

17 bb_matches = re.findall(bb_pattern, cond_code) 

18 if bb_matches: 

19 return [5.0, float(bb_matches[0][0]), float(bb_matches[0][1])] 

20 

21 # 1. SMA Crossover Pattern: SMA(f) > SMA(s) or Price > SMA(s) 

22 sma_pattern = r"rolling\((\d+)\)\.mean\(\)" 

23 matches = re.findall(sma_pattern, cond_code) 

24 

25 if len(matches) == 2: 

26 if "<" in cond_code: 

27 return [5.0, float(matches[0]), float(matches[1])] # sub_type 5 (SMA < SMA) 

28 else: 

29 # sub_type 7: SMA(f) > SMA(s) 

30 return [7.0, float(matches[0]), float(matches[1])] 

31 elif len(matches) == 1: 

32 if "data['close']>" in cond_code: 

33 return [0.0, float(matches[0]), 0.0] # sub_type 0 (Price > SMA) 

34 elif "data['close']<" in cond_code: 

35 return [4.0, float(matches[0]), 0.0] # sub_type 4 (Price < SMA) 

36 

37 # 2. Rolling Max/Min 

38 max_pattern = r"data\['high'\]\.rolling\((\d+)\)\.max\(\)" 

39 max_matches = re.findall(max_pattern, cond_code) 

40 if max_matches and "data['close']>" in cond_code: 

41 return [1.0, float(max_matches[0]), 0.0] # sub_type 1 

42 

43 min_pattern = r"data\['low'\]\.rolling\((\d+)\)\.min\(\)" 

44 min_matches = re.findall(min_pattern, cond_code) 

45 if min_matches and "data['close']<" in cond_code: 

46 return [2.0, float(min_matches[0]), 0.0] # sub_type 2 

47 

48 # 3. Momentum 

49 shift_pattern = r"data\['close'\]\.shift\((\d+)\)" 

50 shift_matches = re.findall(shift_pattern, cond_code) 

51 if shift_matches: 

52 if "data['close']>" in cond_code: 

53 return [3.0, float(shift_matches[0]), 0.0] # sub_type 3 

54 elif "data['close']<" in cond_code: 

55 return [6.0, float(shift_matches[0]), 0.0] # sub_type 6 

56 

57 # 4. RSI Pattern 

58 rsi_pattern = r"rsi\(.*?,?(\d+)\)" 

59 rsi_matches = re.findall(rsi_pattern, cond_code) 

60 if rsi_matches: 

61 period = float(rsi_matches[0]) 

62 if "<" in cond_code: 

63 thresh_match = re.findall(r"<([\d\.]+)", cond_code) 

64 if thresh_match: 

65 return [12.0, period, float(thresh_match[0])] 

66 elif ">" in cond_code: 

67 thresh_match = re.findall(r">([\d\.]+)", cond_code) 

68 if thresh_match: 

69 return [13.0, period, float(thresh_match[0])] 

70 

71 return None 

72 

73 # Check for complex logic AND/OR 

74 if "&" in code or "|" in code: 

75 op_type = 0.0 if "&" in code else 1.0 

76 parts = code.split("&" if "&" in code else "|") 

77 if len(parts) == 2: 

78 p1_params = parse_simple_cond(parts[0]) 

79 p2_params = parse_simple_cond(parts[1]) 

80 if p1_params and p2_params: 

81 # New Layout for type 4: [4, op_type, sub1, p2_1, p3_1, sub2, p2_2, p3_2, comm, slip] 

82 return [4.0, op_type, p1_params[0], p1_params[1], p1_params[2], p2_params[0], p2_params[1], p2_params[2], commission_bps, slippage_bps] 

83 

84 # Fallback to single condition parsing 

85 res = parse_simple_cond(code) 

86 if res: 

87 sub_type, p2_val, p3_val = res 

88 if sub_type == 7.0: 

89 return [0.0, p2_val, p3_val, 0.0] + common_tail 

90 if sub_type == 5.0: 

91 return [5.0, p2_val, p3_val, 0.0] + common_tail 

92 return [3.0, sub_type, p2_val, p3_val] + common_tail 

93 

94 return None