Coverage for src / monte_neo / cli / menu / leadership.py: 0%

1"""Leadership Pipeline Workflow. 

2 

3End-to-end automated indicator discovery and production deployment. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import TYPE_CHECKING 

9 

10import numpy as np 

11from rich.panel import Panel 

12from rich.table import Table 

13 

14from monte_neo.cli.styles import press_any_key 

15from monte_neo.core.evolution_ai import AIEvolutionEngine 

16from monte_neo.core.optimization.production_gate import ProductionGate 

17from monte_neo.monte_carlo.engine import MonteCarloEngine 

18from monte_neo.utils.console import console 

19from monte_neo.utils.logger import get_logger 

20 

21if TYPE_CHECKING: 

22 from monte_neo.cli.menu.main import InteractiveMenu 

23 

24logger = get_logger(__name__) 

25 

26def leadership_pipeline_workflow(menu: InteractiveMenu) -> None: 

27 """Runs the full end-to-end leadership pipeline.""" 

28 console.print(Panel.fit( 

29 "[bold gold1]🏆 Global Leadership Pipeline[/]\n" 

30 "[white]Automated Discovery -> Evolution -> Validation -> Certification -> Production Export[/]", 

31 border_style="gold1" 

32 )) 

33 

34 # 1. Select Symbol 

35 from monte_neo.cli.menu.symbol_selector import select_symbol 

36 symbol = select_symbol(menu) 

37 if not symbol: 

38 return 

39 

40 # 2. Setup Evolution 

41 console.print(f"\n[cyan]Initializing AI Evolution Engine for {symbol}...[/]") 

42 data = menu.storage.load(symbol, timeframe=menu._selected_timeframe) 

43 if data is None or data.empty: 

44 console.print(f"[red]No data found for {symbol}. Please download it first.[/]") 

45 return 

46 

47 # 3. Evolution Phase 

48 with console.status("[bold green]Evolving high-performance indicator formulas..."): 

49 engine = AIEvolutionEngine( 

50 population_size=menu._pop_size, 

51 initial_capital=menu.config.initial_capital, 

52 leverage=menu.config.leverage 

53 ) 

54 best_indicator = engine.evolve(data, menu._target_metrics, generations=menu._generations) 

55 

56 console.print("\n[green]✅ Best formula discovered:[/]") 

57 console.print(Panel(f"[bold white]{best_indicator.get_formula()}[/]", border_style="green")) 

58 

59 # 4. Robustness Validation Phase 

60 console.print("\n[cyan]Running intensive robustness validation suite...[/]") 

61 from monte_neo.core.validator import OverfitValidator 

62 from monte_neo.metrics.calculator import MetricsCalculator 

63 

64 metrics_calc = MetricsCalculator( 

65 initial_capital=menu.config.initial_capital, 

66 leverage=menu.config.leverage 

67 ) 

68 validator = OverfitValidator() 

69 

70 with console.status("[bold blue]Validating robustness across multiple folds and methods..."): 

71 validation_res = validator.validate(best_indicator, data, metrics_calc, menu._target_metrics) 

72 

73 # Additional Monte Carlo validation 

74 mc_config = MonteCarloEngine().config 

75 mc_config.initial_capital = menu.config.initial_capital 

76 mc_config.leverage = menu.config.leverage 

77 mc_engine = MonteCarloEngine(config=mc_config) 

78 mc_res = mc_engine.run(data, best_indicator, metrics_calc, menu._target_metrics) 

79 

80 # CSCV Analysis for PBO 

81 from monte_neo.monte_carlo.cscv import CSCVAnalyzer 

82 cscv_analyzer = CSCVAnalyzer() 

83 cscv_res = cscv_analyzer.analyze(best_indicator, data, metrics_calc) 

84 

85 # Prepare unified results 

86 validation_results = { 

87 "robustness_score": validation_res.overall_score * 100, 

88 "is_production_ready": validation_res.passed and mc_res.passed and cscv_res.get("is_robust", False), 

89 "wfe": validation_res.out_sample_metrics.get("sharpe_ratio", 0) / max(0.001, validation_res.in_sample_metrics.get("sharpe_ratio", 0)), 

90 "mc_robustness": mc_res.pass_rate, 

91 "pbo": cscv_res.get("pbo", 1.0), 

92 "consistency": np.mean(validation_res.cross_val_scores) if validation_res.cross_val_scores else 0, 

93 "recommendation": "Highly robust. Ready for production." if validation_res.passed else "Warning: Potential overfitting detected.", 

94 "stress_test_score": 0.0 # Will be filled by production gate 

95 } 

96 

97 if validation_res.warnings: 

98 validation_results["recommendation"] += "\nWarnings: " + "; ".join(validation_res.warnings) 

99 

100 # 5. Production Gate Phase 

101 console.print("\n[cyan]Finalizing through Production Gate...[/]") 

102 gate = ProductionGate() 

103 

104 with console.status("[bold magenta]Stress testing and generating production assets..."): 

105 gate_results = gate.process(best_indicator, data, validation_results) 

106 

107 # 6. Final Results 

108 _display_pipeline_results(gate_results) 

109 

110def _display_pipeline_results(results: dict) -> None: 

111 """Display the final outcome of the pipeline.""" 

112 table = Table(title="Pipeline Outcome", show_header=True, header_style="bold magenta") 

113 table.add_column("Stage", style="dim") 

114 table.add_column("Result") 

115 

116 status_str = "[bold green]CERTIFIED & EXPORTED[/]" if results["is_certified"] else "[bold red]REJECTED[/]" 

117 table.add_row("Final Status", status_str) 

118 table.add_row("Final Score", f"{results['final_score']:.2f}/100") 

119 table.add_row("Certificate", results["certificate_path"]) 

120 table.add_row("Export Path", results["export_path"] or "N/A") 

121 

122 console.print("\n", table) 

123 

124 if results["is_certified"]: 

125 console.print(f"\n[bold green]Indicator is ready for zero-latency deployment in {results['export_path']}![/]") 

126 else: 

127 console.print("\n[bold yellow]Indicator did not meet leadership standards. Try adjusting target metrics or increasing evolution generations.[/]") 

128 

129 press_any_key()