Spaces:
Sleeping
Sleeping
| """Phase configuration component for Gradio UI.""" | |
| from typing import List, Tuple | |
| import gradio as gr | |
| from config.default_config import DEFAULT_CONFIG | |
| from config.models import AnalysisPhase, PhaseConfiguration | |
| # Phase display information | |
| PHASE_INFO = { | |
| AnalysisPhase.TECHNICAL.value: { | |
| "label": "Technical Analysis", | |
| "description": "Indicators, patterns, trends (~15-20s)", | |
| "agents": "IndicatorAgent, PatternAgent, TrendAgent", | |
| }, | |
| AnalysisPhase.FUNDAMENTAL.value: { | |
| "label": "Fundamental Analysis", | |
| "description": "Financial metrics, valuation (~30-40s)", | |
| "agents": "FundamentalsAgent", | |
| }, | |
| AnalysisPhase.SENTIMENT.value: { | |
| "label": "Sentiment Analysis", | |
| "description": "Social & news sentiment (~20-30s)", | |
| "agents": "SentimentAgent, NewsAgent", | |
| }, | |
| AnalysisPhase.RESEARCH_SYNTHESIS.value: { | |
| "label": "Research Synthesis", | |
| "description": "Integrated perspective (~15-20s)", | |
| "agents": "TechnicalAnalystAgent", | |
| }, | |
| AnalysisPhase.RISK.value: { | |
| "label": "Risk Assessment", | |
| "description": "Risk scores, volatility (~15-20s)", | |
| "agents": "RiskManagerAgent", | |
| }, | |
| AnalysisPhase.DECISION.value: { | |
| "label": "Portfolio Decision", | |
| "description": "Final BUY/SELL/HOLD (~15-20s)", | |
| "agents": "PortfolioManagerAgent, DecisionAgent", | |
| }, | |
| } | |
| def create_phase_configuration(): | |
| """Create phase configuration UI component. | |
| Returns: | |
| tuple: (checkboxgroup, preset dropdown, validation message, estimated time) | |
| """ | |
| with gr.Group(): | |
| gr.Markdown("### Workflow Phases") | |
| gr.Markdown( | |
| "Select which analysis phases to execute. You can use presets or customize:" | |
| ) | |
| # Preset selector | |
| preset_choices = [ | |
| ("Quick Technical Check (~15-20s)", "quick_technical"), | |
| ("Investment Research (~90s)", "investment_research"), | |
| ("Full Analysis (~2min)", "full_analysis"), | |
| ("Custom", "custom"), | |
| ] | |
| preset_dropdown = gr.Dropdown( | |
| choices=preset_choices, | |
| value="full_analysis", | |
| label="Analysis Preset", | |
| info="Choose a preset or select 'Custom' to manually choose phases", | |
| ) | |
| # Phase checkboxes | |
| phase_choices = [ | |
| (PHASE_INFO[phase.value]["label"], phase.value) for phase in AnalysisPhase | |
| ] | |
| phase_checkboxes = gr.CheckboxGroup( | |
| choices=phase_choices, | |
| value=[ | |
| phase.value for phase in AnalysisPhase | |
| ], # All phases selected by default | |
| label="Enabled Phases", | |
| info="Select phases to execute (order is automatic)", | |
| ) | |
| # Validation message | |
| validation_msg = gr.Markdown(value="", visible=False) | |
| # Educational mode checkbox | |
| educational_mode_checkbox = gr.Checkbox( | |
| value=True, | |
| label="Educational Mode", | |
| info="Include explanations and learning notes in the report", | |
| ) | |
| # Estimated execution time | |
| estimated_time = gr.Markdown( | |
| value="⏱️ Estimated execution time: ~2 minutes", visible=True | |
| ) | |
| # Phase details (collapsible) | |
| with gr.Accordion("Phase Details", open=False): | |
| phase_details_md = "\n\n".join( | |
| [ | |
| f"**{info['label']}**: {info['description']}\n- Agents: {info['agents']}" | |
| for phase, info in PHASE_INFO.items() | |
| ] | |
| ) | |
| gr.Markdown(phase_details_md) | |
| def apply_preset(preset_name: str) -> List[str]: | |
| """Apply preset phase configuration. | |
| Args: | |
| preset_name: Name of preset (quick_technical, investment_research, full_analysis, custom) | |
| Returns: | |
| List of enabled phase values | |
| """ | |
| if preset_name == "custom": | |
| # Don't change current selection | |
| return gr.update() | |
| preset_config = DEFAULT_CONFIG["phase_presets"].get(preset_name, {}) | |
| enabled_phases = preset_config.get("phases", []) | |
| return enabled_phases | |
| def validate_phases(selected_phases: List[str]) -> Tuple[str, str]: | |
| """Validate phase selection and estimate execution time. | |
| Args: | |
| selected_phases: List of selected phase values | |
| Returns: | |
| tuple: (validation_message, estimated_time_message) | |
| """ | |
| if not selected_phases: | |
| return ( | |
| "⚠️ **Error**: At least one phase must be selected", | |
| "⏱️ Estimated execution time: N/A", | |
| ) | |
| # Create temporary config for validation | |
| try: | |
| phases_enum = [AnalysisPhase(p) for p in selected_phases] | |
| except ValueError as e: | |
| return ( | |
| f"⚠️ **Error**: Invalid phase selection: {e}", | |
| "⏱️ Estimated execution time: N/A", | |
| ) | |
| config = PhaseConfiguration(enabled_phases=phases_enum) | |
| errors = config.validate() | |
| # Calculate estimated time | |
| time_estimates = { | |
| AnalysisPhase.TECHNICAL.value: 20, | |
| AnalysisPhase.FUNDAMENTAL.value: 35, | |
| AnalysisPhase.SENTIMENT.value: 25, | |
| AnalysisPhase.RESEARCH_SYNTHESIS.value: 18, | |
| AnalysisPhase.RISK.value: 18, | |
| AnalysisPhase.DECISION.value: 18, | |
| } | |
| total_seconds = sum(time_estimates.get(p, 20) for p in selected_phases) | |
| if total_seconds < 60: | |
| time_str = f"~{total_seconds}s" | |
| else: | |
| minutes = total_seconds // 60 | |
| seconds = total_seconds % 60 | |
| if seconds > 0: | |
| time_str = f"~{minutes}min {seconds}s" | |
| else: | |
| time_str = f"~{minutes}min" | |
| estimated_msg = f"⏱️ Estimated execution time: {time_str}" | |
| if errors: | |
| error_msg = "⚠️ **Validation Errors**:\n\n" + "\n".join( | |
| f"- {err}" for err in errors | |
| ) | |
| return (error_msg, estimated_msg) | |
| else: | |
| return ("", estimated_msg) | |
| # Wire up preset dropdown to update checkboxes | |
| preset_dropdown.change( | |
| fn=apply_preset, inputs=[preset_dropdown], outputs=[phase_checkboxes] | |
| ) | |
| # Wire up phase validation | |
| phase_checkboxes.change( | |
| fn=validate_phases, | |
| inputs=[phase_checkboxes], | |
| outputs=[validation_msg, estimated_time], | |
| ) | |
| return ( | |
| preset_dropdown, | |
| phase_checkboxes, | |
| educational_mode_checkbox, | |
| validation_msg, | |
| estimated_time, | |
| ) | |
| def get_phase_configuration_from_ui( | |
| selected_phases: List[str], | |
| investment_style: str, | |
| timeframe: str, | |
| educational_mode: bool = True, | |
| ) -> PhaseConfiguration: | |
| """Create PhaseConfiguration from UI inputs. | |
| Args: | |
| selected_phases: List of selected phase values from UI | |
| investment_style: Investment style value (long_term or swing_trading) | |
| timeframe: Selected timeframe (1d, 1w, etc.) | |
| educational_mode: Whether to include educational content | |
| Returns: | |
| PhaseConfiguration object | |
| """ | |
| from config.models import InvestmentStyle | |
| # Convert strings to enums | |
| style_enum = InvestmentStyle(investment_style) | |
| phases_enum = [AnalysisPhase(p) for p in selected_phases] | |
| # Get chart period from investment style config | |
| style_config = DEFAULT_CONFIG["investment_styles"].get(investment_style, {}) | |
| chart_period_days = style_config.get("chart_period_days", 180) | |
| return PhaseConfiguration( | |
| investment_style=style_enum, | |
| enabled_phases=phases_enum, | |
| timeframe=timeframe, | |
| chart_period_days=chart_period_days, | |
| educational_mode=educational_mode, | |
| ) | |