Coverage for src / monte_neo / cli / app.py: 22%

1"""Main CLI application. 

2 

3Entry point for the Monte-Neo CLI. 

4""" 

5 

6from __future__ import annotations 

7 

8import argparse 

9import sys 

10from typing import TYPE_CHECKING 

11 

12from rich.console import Console 

13 

14from monte_neo._version import __version__ 

15from monte_neo.cli.menu import InteractiveMenu 

16from monte_neo.cli.styles import print_banner, print_error 

17from monte_neo.utils.config import load_config 

18from monte_neo.utils.logger import get_logger, setup_logging 

19 

20if TYPE_CHECKING: 

21 pass 

22 

23logger = get_logger(__name__) 

24console = Console() 

25 

26 

27class MonteNeoCLI: 

28 """Main CLI application class.""" 

29 

30 def __init__(self) -> None: 

31 """Initialize CLI.""" 

32 self.config = load_config() 

33 self.menu = InteractiveMenu(self.config) 

34 

35 def run(self, interactive: bool = True) -> int: 

36 """Run the CLI application. 

37 

38 Args: 

39 interactive: Run in interactive mode. 

40 

41 Returns: 

42 Exit code. 

43 """ 

44 print_banner() 

45 

46 if interactive: 

47 return self.menu.run() 

48 else: 

49 console.print("[yellow]Non-interactive mode not yet implemented[/]") 

50 return 1 

51 

52 def run_export(self, json_path: str) -> int: 

53 """Export an indicator to C++.""" 

54 import json 

55 

56 from monte_neo.core.optimization.production_exporter import ProductionExporter 

57 

58 try: 

59 with open(json_path) as f: 

60 data = json.load(f) 

61 

62 # Mock indicator for export 

63 from monte_neo.indicators.dynamic import DynamicIndicator 

64 indicator = DynamicIndicator() 

65 indicator.set_parameter("source_code", data.get("formula", "")) 

66 

67 exporter = ProductionExporter() 

68 export_path = exporter.export(indicator, data.get("validation", {})) 

69 

70 console.print(f"[green]Successfully exported to: {export_path}[/]") 

71 return 0 

72 except Exception as e: 

73 console.print(f"[red]Export failed: {e}[/]") 

74 return 1 

75 

76 def run_evolve(self, symbol: str) -> int: 

77 """Run AI-driven evolution for a symbol.""" 

78 from monte_neo.core.evolution_ai import AIEvolutionEngine 

79 from monte_neo.data.storage import ParquetStorage 

80 from monte_neo.utils.config import Config 

81 

82 try: 

83 console.print(f"[bold cyan]Starting AI Evolution for {symbol}...[/]") 

84 config = Config() 

85 storage = ParquetStorage(config.data_dir) 

86 data = storage.load(symbol, timeframe=config.default_timeframe) 

87 

88 if data is None or data.empty: 

89 console.print(f"[red]No data found for {symbol}[/]") 

90 return 1 

91 

92 engine = AIEvolutionEngine() 

93 best_indicator = engine.evolve(data, {"profit_factor": 1.5, "sharpe_ratio": 1.0}, generations=5) 

94 

95 console.print("[green]Evolution complete! Best formula:[/]") 

96 console.print(f"[bold white]{best_indicator.get_formula()}[/]") 

97 return 0 

98 except Exception as e: 

99 console.print(f"[red]Evolution failed: {e}[/]") 

100 return 1 

101 

102 def run_headless( 

103 self, 

104 config_file: str, 

105 ) -> int: 

106 """Run in headless mode with config file. 

107 

108 Args: 

109 config_file: Path to YAML config. 

110 

111 Returns: 

112 Exit code. 

113 """ 

114 from monte_neo.cli.headless import run_headless_generation 

115 

116 return run_headless_generation(config_file) 

117 

118 

119def parse_args() -> argparse.Namespace: 

120 """Parse command line arguments.""" 

121 parser = argparse.ArgumentParser( 

122 prog="monte-neo", 

123 description="Monte Carlo Indicator Generator Framework", 

124 ) 

125 

126 parser.add_argument( 

127 "--interactive", 

128 "-i", 

129 action="store_true", 

130 default=True, 

131 help="Run in interactive mode (default)", 

132 ) 

133 

134 parser.add_argument( 

135 "--config", 

136 "-c", 

137 type=str, 

138 help="Path to YAML config file (for headless mode)", 

139 ) 

140 

141 parser.add_argument( 

142 "--headless", 

143 action="store_true", 

144 help="Run in headless mode with config file", 

145 ) 

146 

147 parser.add_argument( 

148 "--export", 

149 type=str, 

150 help="Path to indicator JSON to export to C++", 

151 ) 

152 

153 parser.add_argument( 

154 "--evolve", 

155 type=str, 

156 help="Symbol to run AI evolution for (e.g. BTCUSDT)", 

157 ) 

158 

159 parser.add_argument( 

160 "--log-level", 

161 type=str, 

162 default="INFO", 

163 choices=["DEBUG", "INFO", "WARNING", "ERROR"], 

164 help="Log level", 

165 ) 

166 

167 parser.add_argument( 

168 "--version", 

169 "-v", 

170 action="version", 

171 version=f"monte-neo {__version__}", 

172 ) 

173 

174 return parser.parse_args() 

175 

176 

177def main() -> int: 

178 """Main entry point.""" 

179 args = parse_args() 

180 

181 # Setup logging 

182 setup_logging(args.log_level) 

183 

184 try: 

185 app = MonteNeoCLI() 

186 

187 if args.export: 

188 return app.run_export(args.export) 

189 elif args.evolve: 

190 return app.run_evolve(args.evolve) 

191 elif args.headless and args.config: 

192 return app.run_headless(args.config) 

193 else: 

194 return app.run(interactive=args.interactive) 

195 

196 except KeyboardInterrupt: 

197 console.print("\n[yellow]Interrupted by user[/]") 

198 sys.exit(0) 

199 

200 except Exception as e: 

201 print_error(f"Error: {e}") 

202 logger.exception("Unhandled exception") 

203 return 1 

204 

205 

206if __name__ == "__main__": 

207 sys.exit(main())