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

1"""Portfolio management menu workflow.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7import questionary 

8from rich.console import Console 

9from rich.table import Table 

10 

11from monte_neo.cli.styles import CUSTOM_STYLE, press_any_key 

12from monte_neo.core.portfolio import PortfolioManager 

13 

14if TYPE_CHECKING: 

15 from monte_neo.cli.menu.main import InteractiveMenu 

16 

17console = Console() 

18 

19def portfolio_workflow(menu: InteractiveMenu) -> None: 

20 """Portfolio management sub-menu.""" 

21 # Initialize manager if not exists (in a real app, we'd load it from config/storage) 

22 if menu._portfolio_manager is None: 

23 menu._portfolio_manager = PortfolioManager() 

24 

25 manager = menu._portfolio_manager 

26 while True: 

27 choices = [ 

28 {"name": "📋 View Portfolio Summary", "value": "summary"}, 

29 {"name": "➕ Add Indicator to Portfolio", "value": "add_asset"}, 

30 {"name": "🎲 Portfolio Monte Carlo", "value": "portfolio_mc"}, 

31 {"name": "📊 Analyze Clusters (Correlation)", "value": "clusters"}, 

32 {"name": "⚖️ Auto-Rebalance (Risk Parity/Kelly)", "value": "optimize"}, 

33 {"name": "🔙 Back to Main Menu", "value": "back"}, 

34 ] 

35 

36 choice = questionary.select( 

37 "Portfolio Management:", 

38 choices=choices, 

39 style=CUSTOM_STYLE 

40 ).ask() 

41 

42 if choice == "back" or choice is None: 

43 break 

44 

45 if choice == "summary": 

46 _show_summary(manager) 

47 elif choice == "add_asset": 

48 _add_asset_workflow(menu) 

49 elif choice == "portfolio_mc": 

50 _portfolio_mc_workflow(manager) 

51 elif choice == "clusters": 

52 _cluster_analysis_workflow(manager) 

53 elif choice == "optimize": 

54 _optimize_workflow(manager) 

55 

56def _portfolio_mc_workflow(manager: PortfolioManager) -> None: 

57 """Runs Monte Carlo on the combined portfolio.""" 

58 with console.status("[bold blue]Running Portfolio Monte Carlo..."): 

59 results = manager.run_portfolio_monte_carlo() 

60 

61 if not results: 

62 console.print("[red]Portfolio is empty or has no equity data.[/]") 

63 return 

64 

65 table = Table(title="Portfolio Robustness (Monte Carlo)") 

66 table.add_column("Metric", style="cyan") 

67 table.add_column("Value", style="magenta") 

68 

69 table.add_row("Avg Expected Return", f"{results['avg_return']:.2%}") 

70 table.add_row("Max Drawdown (95% CI)", f"{results['max_drawdown_95th']:.2%}") 

71 table.add_row("VaR (95%)", f"{results['var_95']:.2%}") 

72 

73 console.print(table) 

74 press_any_key() 

75 

76def _cluster_analysis_workflow(manager: PortfolioManager) -> None: 

77 """Displays asset clusters based on correlation.""" 

78 summary = manager.get_portfolio_summary() 

79 clusters = summary.get("clusters", {}) 

80 

81 if not clusters: 

82 console.print("[yellow]Not enough data for cluster analysis. Need at least 2 assets with equity curves.[/]") 

83 return 

84 

85 table = Table(title="Correlation Clusters (Strategy Redundancy)") 

86 table.add_column("Cluster ID", style="dim") 

87 table.add_column("Assets", style="cyan") 

88 

89 for cid, assets in clusters.items(): 

90 table.add_row(str(cid), ", ".join(assets)) 

91 

92 console.print(table) 

93 console.print("[dim]Assets in the same cluster are highly correlated and might be redundant.[/]") 

94 press_any_key() 

95 

96def _optimize_workflow(manager: PortfolioManager) -> None: 

97 """Runs portfolio optimization.""" 

98 method = questionary.select( 

99 "Select Optimization Method:", 

100 choices=[ 

101 {"name": "⚖️ Risk Parity (Equal Risk Contribution)", "value": "risk_parity"}, 

102 {"name": "💰 Kelly Criterion (Optimal Growth)", "value": "kelly"}, 

103 {"name": "📏 Equal Weights", "value": "equal"}, 

104 ], 

105 style=CUSTOM_STYLE 

106 ).ask() 

107 

108 if method: 

109 with console.status(f"[bold green]Optimizing using {method}..."): 

110 weights = manager.auto_rebalance(method=method) 

111 

112 console.print("[green]Portfolio rebalanced successfully![/]") 

113 _show_summary(manager) 

114 

115def _show_summary(manager: PortfolioManager) -> None: 

116 summary = manager.get_portfolio_summary() 

117 

118 table = Table(title="Portfolio Summary") 

119 table.add_column("Property", style="cyan") 

120 table.add_column("Value", style="magenta") 

121 

122 table.add_row("Total Assets", str(summary["total_assets"])) 

123 table.add_row("Active Assets", str(summary["active_assets"])) 

124 table.add_row("Initial Capital", f"${summary['initial_capital']:.2f}") 

125 

126 console.print(table) 

127 

128 if summary["weights"]: 

129 weight_table = Table(title="Asset Weights") 

130 weight_table.add_column("Asset ID", style="cyan") 

131 weight_table.add_column("Weight", style="green") 

132 

133 for asset_id, weight in summary["weights"].items(): 

134 weight_table.add_row(asset_id, f"{weight:.2%}") 

135 

136 console.print(weight_table) 

137 

138def _add_asset_workflow(menu: InteractiveMenu) -> None: 

139 # In a real app, this would list generated indicators from exports/production 

140 from pathlib import Path 

141 

142 production_dir = Path("exports/production") 

143 if not production_dir.exists(): 

144 console.print("[red]No production indicators found in exports/production[/]") 

145 return 

146 

147 indicators = list(production_dir.glob("*.json")) 

148 if not indicators: 

149 console.print("[red]No .json indicators found[/]") 

150 return 

151 

152 indicator_choices = [{"name": idx.name, "value": str(idx)} for idx in indicators] 

153 selected_idx = questionary.select( 

154 "Select indicator to add:", 

155 choices=indicator_choices, 

156 style=CUSTOM_STYLE 

157 ).ask() 

158 

159 if selected_idx: 

160 from monte_neo.core.portfolio import PortfolioAsset 

161 

162 if menu._portfolio_manager is None: 

163 menu._portfolio_manager = PortfolioManager() 

164 

165 asset_id = Path(selected_idx).stem 

166 asset = PortfolioAsset( 

167 id=asset_id, 

168 indicator_path=selected_idx, 

169 symbol=menu._selected_symbol 

170 ) 

171 menu._portfolio_manager.add_asset(asset) 

172 console.print(f"[green]Successfully added {asset_id} to portfolio![/]")