Coverage for src / monte_neo / cli / menu / main.py: 20%

1"""Main interactive menu logic.""" 

2 

3from __future__ import annotations 

4 

5from pathlib import Path 

6from typing import TYPE_CHECKING 

7 

8import pandas as pd 

9import questionary 

10 

11from monte_neo.cli.progress import ProgressTracker 

12from monte_neo.cli.styles import CUSTOM_STYLE 

13from monte_neo.data.storage import ParquetStorage 

14from monte_neo.utils.console import console 

15from monte_neo.utils.logger import get_logger 

16 

17if TYPE_CHECKING: 

18 from monte_neo.core.portfolio import PortfolioManager 

19 from monte_neo.utils.config import Config 

20 

21logger = get_logger(__name__) 

22 

23class InteractiveMenu: 

24 """Interactive CLI menu with arrow navigation.""" 

25 

26 def __init__(self, config: Config) -> None: 

27 self.config = config 

28 self.storage = ParquetStorage(config.data_dir) 

29 self.progress = ProgressTracker() 

30 

31 # State 

32 self._target_metrics: dict = { 

33 "profit_factor": 2.0, 

34 "sharpe_ratio": 1.0, 

35 "max_drawdown": 0.20 

36 } 

37 self._selected_symbol: str = config.default_symbol 

38 self._selected_timeframe: str = config.default_timeframe 

39 self._mc_methods: list[str] = ["shuffling", "noise", "sensitivity", "walk_forward", "block_bootstrap"] 

40 self._pop_size, self._generations = 50, 20 

41 self._mutation_rate, self._crossover_rate = 0.3, 0.7 

42 self._cached_symbols: list[str] = [] 

43 

44 # Risk Management Settings 

45 self._stop_loss_pct: float = 1.0 

46 self._take_profit_pct: float = 2.0 

47 self._use_sl_tp: bool = True 

48 

49 # Validation Settings 

50 self._mc_pass_threshold: float = 0.80 

51 

52 # Hardware Settings 

53 self._use_gpu: bool = config.use_gpu 

54 self._gpu_precision: str = config.gpu_precision 

55 self._metal_driver: str = config.metal_driver 

56 

57 self._last_data: pd.DataFrame | None = None 

58 self._portfolio_manager: PortfolioManager | None = None 

59 

60 def run(self) -> int: 

61 """Run the interactive menu loop.""" 

62 while True: 

63 try: 

64 choice = self._show_main_menu() 

65 if choice == "exit": 

66 console.print("[green]Goodbye![/]") 

67 return 0 

68 self._handle_choice(choice) 

69 except KeyboardInterrupt: 

70 console.print("\n[yellow]Operation cancelled by user. Returning to menu...[/]") 

71 except Exception as e: 

72 console.print(f"\n[red]An error occurred: {e}[/]") 

73 logger.exception("Error in menu loop") 

74 

75 def _show_main_menu(self) -> str: 

76 choices = [ 

77 {"name": "🏆 Global Leadership Pipeline", "value": "leadership_pipeline"}, 

78 {"name": "📊 Download Market Data", "value": "download"}, 

79 {"name": "🎯 Set Target Metrics", "value": "metrics"}, 

80 {"name": "🎲 Configure Monte Carlo Methods", "value": "mc_config"}, 

81 {"name": "🚀 Generate Indicator", "value": "generate"}, 

82 {"name": "🔄 Sequential Generate Indicator", "value": "generate_sequential"}, 

83 {"name": "🧪 Test Custom Formula", "value": "test_custom"}, 

84 {"name": "💼 Smart Portfolio", "value": "portfolio"}, 

85 {"name": "📈 View Results", "value": "results"}, 

86 {"name": "⚙️ Settings", "value": "settings"}, 

87 {"name": "🔧 Hardware Configuration", "value": "hardware"}, 

88 {"name": "❌ Exit", "value": "exit"}, 

89 ] 

90 return questionary.select( 

91 "Select an option:", choices=choices, style=CUSTOM_STYLE, use_shortcuts=True 

92 ).ask() or "exit" 

93 

94 def _handle_choice(self, choice: str) -> None: 

95 from monte_neo.cli.menu.data import download_data_workflow 

96 from monte_neo.cli.menu.generator import generate_indicator_workflow 

97 from monte_neo.cli.menu.hardware import hardware_workflow 

98 from monte_neo.cli.menu.mc_config import configure_mc_workflow 

99 from monte_neo.cli.menu.metrics import set_metrics_workflow 

100 from monte_neo.cli.menu.results import view_results_workflow 

101 from monte_neo.cli.menu.settings import settings_workflow 

102 

103 if choice == "leadership_pipeline": 

104 from monte_neo.cli.menu.leadership import leadership_pipeline_workflow 

105 leadership_pipeline_workflow(self) 

106 return 

107 elif choice == "generate": 

108 generate_indicator_workflow(self, sequential=False) 

109 return 

110 elif choice == "generate_sequential": 

111 generate_indicator_workflow(self, sequential=True) 

112 return 

113 elif choice == "test_custom": 

114 from monte_neo.cli.menu.custom_test import test_custom_indicator_workflow 

115 test_custom_indicator_workflow(self) 

116 return 

117 elif choice == "portfolio": 

118 from monte_neo.cli.menu.portfolio import portfolio_workflow 

119 portfolio_workflow(self) 

120 return 

121 

122 handlers = { 

123 "download": download_data_workflow, 

124 "metrics": set_metrics_workflow, 

125 "mc_config": configure_mc_workflow, 

126 "results": view_results_workflow, 

127 "settings": settings_workflow, 

128 "hardware": hardware_workflow, 

129 } 

130 handler = handlers.get(choice) 

131 if handler: 

132 handler(self) 

133 self._sync_to_config() 

134 self._save_config() 

135 

136 def _sync_to_config(self) -> None: 

137 """Sync internal state to config object.""" 

138 self.config.default_symbol = self._selected_symbol 

139 self.config.default_timeframe = self._selected_timeframe 

140 self.config.target_profit_factor = self._target_metrics["profit_factor"] 

141 self.config.target_sharpe_ratio = self._target_metrics["sharpe_ratio"] 

142 self.config.target_max_drawdown = self._target_metrics["max_drawdown"] 

143 self.config.use_gpu = self._use_gpu 

144 self.config.gpu_precision = self._gpu_precision 

145 self.config.metal_driver = self._metal_driver 

146 

147 def _save_config(self) -> None: 

148 """Save config to default path.""" 

149 from monte_neo.utils.config import save_config 

150 config_path = Path("config.yaml") 

151 save_config(self.config, config_path)