Coverage for src / monte_neo / core / optimization / production_exporter.py: 100%

1import json 

2import os 

3from typing import Any 

4 

5from monte_neo.indicators.base import BaseIndicator 

6 

7 

8class ProductionExporter: 

9 """Exports validated indicators for production use (JSON and C++).""" 

10 

11 def __init__(self, export_dir: str = "exports/production"): 

12 self.export_dir = export_dir 

13 os.makedirs(self.export_dir, exist_ok=True) 

14 

15 def export(self, indicator: BaseIndicator, validation_results: dict[str, Any], metadata: dict[str, Any] | None = None) -> str: 

16 """ 

17 Exports indicator config, validation certificate, and C++ source. 

18  

19 Returns: 

20 Path to the export directory. 

21 """ 

22 indicator_id = id(indicator) 

23 export_subdir = os.path.join(self.export_dir, f"indicator_{indicator.__class__.__name__}_{indicator_id}") 

24 os.makedirs(export_subdir, exist_ok=True) 

25 

26 # 1. Export JSON Metadata 

27 export_data = { 

28 "indicator_type": indicator.__class__.__name__, 

29 "parameters": indicator.get_parameters(), 

30 "formula": indicator.get_formula(), 

31 "validation": validation_results, 

32 "metadata": metadata or {}, 

33 "version": "1.0.0" 

34 } 

35 

36 json_path = os.path.join(export_subdir, "config.json") 

37 with open(json_path, 'w') as f: 

38 json.dump(export_data, f, indent=4) 

39 

40 # 2. Export C++ Production Logic 

41 cpp_path = self._export_cpp(indicator, export_subdir) 

42 

43 # 3. Create a basic README for the export 

44 with open(os.path.join(export_subdir, "README.md"), 'w') as f: 

45 f.write(f"# Production Export: {indicator.__class__.__name__}\n\n") 

46 f.write(f"Robustness Score: {validation_results.get('robustness_score', 'N/A')}\n") 

47 f.write(f"Is Production Ready: {validation_results.get('is_production_ready', False)}\n\n") 

48 f.write("## Usage\n") 

49 f.write("Compile the C++ code for standalone execution:\n") 

50 f.write("```bash\n./compile.sh\n```\n") 

51 

52 return export_subdir 

53 

54 def _export_cpp(self, indicator: BaseIndicator, target_dir: str) -> str: 

55 """Generates standalone C++ code for the indicator.""" 

56 params = indicator.get_parameters() 

57 formula = indicator.get_formula() 

58 

59 cpp_content = self._generate_cpp_source(indicator.__class__.__name__, params, formula) 

60 cpp_path = os.path.join(target_dir, "production_indicator.cpp") 

61 

62 with open(cpp_path, 'w') as f: 

63 f.write(cpp_content) 

64 

65 # Add a simple compile script 

66 compile_sh = os.path.join(target_dir, "compile.sh") 

67 with open(compile_sh, 'w') as f: 

68 f.write("#!/bin/bash\n") 

69 f.write("echo 'Compiling Production C++ Indicator...'\n") 

70 f.write("g++ -O3 production_indicator.cpp -o production_indicator\n") 

71 f.write("if [ $? -eq 0 ]; then\n") 

72 f.write(" echo 'Successfully compiled to ./production_indicator'\n") 

73 f.write("else\n") 

74 f.write(" echo 'Compilation failed!'\n") 

75 f.write(" exit 1\n") 

76 f.write("fi\n") 

77 os.chmod(compile_sh, 0o755) 

78 

79 # Export Metal shader for GPU execution 

80 self._export_metal(indicator, target_dir) 

81 

82 return cpp_path 

83 

84 def _export_metal(self, indicator: BaseIndicator, target_dir: str) -> str: 

85 """Generates standalone Metal shader for the indicator.""" 

86 params = indicator.get_parameters() 

87 formula = indicator.get_formula() 

88 

89 metal_content = self._generate_metal_source(indicator.__class__.__name__, params, formula) 

90 metal_path = os.path.join(target_dir, "production_indicator.metal") 

91 

92 with open(metal_path, 'w') as f: 

93 f.write(metal_content) 

94 

95 return metal_path 

96 

97 def _generate_cpp_source(self, name: str, params: dict[str, Any], formula: str) -> str: 

98 """Template for C++ production source.""" 

99 param_init = "\n ".join([f"float {k} = {v};" for k, v in params.items()]) 

100 

101 return f""" 

102#include <iostream> 

103#include <vector> 

104#include <string> 

105#include <cmath> 

106#include <algorithm> 

107 

108struct Candle {{ 

109 double open, high, low, close, volume; 

110}}; 

111 

112class {name} {{ 

113private: 

114 // Parameters 

115 {param_init} 

116  

117 // State (for indicators like EMA/RSI) 

118 double last_ema = 0; 

119 bool initialized = false; 

120 

121public: 

122 {name}() {{}} 

123 

124 /** 

125 * @brief Get signal for the current candle. 

126 * @return 1 for Buy, -1 for Sell, 0 for Neutral. 

127 */ 

128 int get_signal(const std::vector<Candle>& data, int index) {{ 

129 if (index < 1) return 0; 

130  

131 // Logic for formula: {formula} 

132 // AUTO-GENERATED LOGIC START 

133 const Candle& current = data[index]; 

134 const Candle& prev = data[index-1]; 

135  

136 // Example: Simple Trend Follower 

137 if (current.close > prev.close) return 1; 

138 if (current.close < prev.close) return -1; 

139  

140 return 0; 

141 // AUTO-GENERATED LOGIC END 

142 }} 

143}}; 

144 

145int main() {{ 

146 std::cout << "--- Monte-Neo Production Node ---" << std::endl; 

147 std::cout << "Indicator: {name}" << std::endl; 

148 std::cout << "Formula: {formula}" << std::endl; 

149 std::cout << "Status: Ready for zero-latency execution" << std::endl; 

150  

151 // Example usage 

152 {name} strategy; 

153 std::vector<Candle> mock_data = {{{{100, 105, 95, 102, 1000}}, {{102, 108, 101, 106, 1100}}}}; 

154 int signal = strategy.get_signal(mock_data, 1); 

155  

156 std::cout << "Mock Signal (last candle): " << signal << std::endl; 

157  

158 return 0; 

159}} 

160""" 

161 

162 def _generate_metal_source(self, name: str, params: dict[str, Any], formula: str) -> str: 

163 """Template for Metal shader production source.""" 

164 return f""" 

165#include <metal_stdlib> 

166using namespace metal; 

167 

168struct Candle {{ 

169 float open; 

170 float high; 

171 float low; 

172 float close; 

173 float volume; 

174}}; 

175 

176// Formula: {formula} 

177kernel void {name}_kernel( 

178 const device Candle* data [[buffer(0)]], 

179 device int* signals [[buffer(1)]], 

180 uint id [[thread_position_in_grid]] 

181) {{ 

182 if (id < 1) {{ 

183 signals[id] = 0; 

184 return; 

185 }} 

186  

187 const device Candle& current = data[id]; 

188 const device Candle& prev = data[id-1]; 

189  

190 // Simplified logic translation 

191 int signal = 0; 

192 if (current.close > prev.close) signal = 1; 

193 else if (current.close < prev.close) signal = -1; 

194  

195 signals[id] = signal; 

196}} 

197"""