Spaces:
Sleeping
Sleeping
| """ | |
| Phase-Organized Report Formatter | |
| Formats analysis reports organized by phases for the web interface. | |
| """ | |
| import base64 | |
| from pathlib import Path | |
| from typing import Any, Dict, List | |
| from config.models import AnalysisPhase, InvestmentStyle | |
| def _image_to_base64(image_path: str) -> str: | |
| """ | |
| Convert image file to base64 string for embedding in markdown. | |
| Args: | |
| image_path: Path to image file | |
| Returns: | |
| Base64-encoded image string with data URI prefix | |
| """ | |
| try: | |
| image_file = Path(image_path) | |
| if not image_file.exists(): | |
| return "" | |
| with open(image_file, "rb") as f: | |
| image_data = f.read() | |
| base64_image = base64.b64encode(image_data).decode("utf-8") | |
| return f"data:image/png;base64,{base64_image}" | |
| except Exception as e: | |
| return "" | |
| def format_phase_organized_report(final_state: Dict[str, Any]) -> str: | |
| """ | |
| Format a phase-organized analysis report for display. | |
| Args: | |
| final_state: Final workflow state from ConditionalComprehensiveWorkflow | |
| Returns: | |
| Markdown-formatted report string | |
| """ | |
| ticker = final_state.get("ticker", "UNKNOWN") | |
| timeframe = final_state.get("timeframe", "UNKNOWN") | |
| # Extract configuration | |
| config_dict = final_state.get("config", {}) | |
| investment_style_str = config_dict.get("investment_style", "long_term") | |
| investment_style = ( | |
| InvestmentStyle(investment_style_str) | |
| if investment_style_str | |
| else InvestmentStyle.LONG_TERM | |
| ) | |
| completed_phases = final_state.get("completed_phases", []) | |
| phase_outputs = final_state.get("phase_outputs", {}) | |
| # Build report with prominent timeframe indicator (US3) | |
| lines = [ | |
| f"# 📊 Phase-Organized Analysis: {ticker}", | |
| f"## ⏱️ Timeframe: {timeframe.upper()}", | |
| "", | |
| f"**Investment Style**: {investment_style.value.replace('_', ' ').title()}", | |
| f"**Phases Completed**: {len(completed_phases)}", | |
| "", | |
| "---", | |
| "", | |
| ] | |
| # Executive Summary | |
| executive_summary = final_state.get("executive_summary") | |
| if executive_summary: | |
| lines.extend( | |
| [ | |
| "## 📋 Executive Summary", | |
| "", | |
| executive_summary, | |
| "", | |
| "---", | |
| "", | |
| ] | |
| ) | |
| # Final Decision (if available) | |
| final_decision = final_state.get("final_decision") | |
| if final_decision: | |
| lines.extend( | |
| [ | |
| "## 🎯 Final Recommendation", | |
| "", | |
| f"**Decision**: {final_decision}", | |
| "", | |
| ] | |
| ) | |
| # Add portfolio decision details if available | |
| portfolio_decision = final_state.get("portfolio_decision", {}) | |
| if portfolio_decision: | |
| decision_dict = portfolio_decision.get("decision", {}) | |
| # Display phase rationale with confidence (US8) | |
| phase_rationale = portfolio_decision.get("phase_rationale") | |
| if phase_rationale: | |
| lines.extend( | |
| [ | |
| "### 📊 Decision Rationale", | |
| "", | |
| phase_rationale, | |
| "", | |
| ] | |
| ) | |
| else: | |
| # Fallback to old confidence display | |
| confidence_pct = decision_dict.get("confidence_pct") | |
| if confidence_pct is not None: | |
| lines.append(f"**Confidence**: {confidence_pct:.0f}%") | |
| else: | |
| lines.append( | |
| f"**Confidence**: {decision_dict.get('confidence', 'N/A')}" | |
| ) | |
| lines.append("") | |
| lines.extend( | |
| [ | |
| f"**Risk Level**: {decision_dict.get('risk_level', 'N/A').upper()}", | |
| "", | |
| ] | |
| ) | |
| if decision_dict.get("entry_price"): | |
| lines.extend( | |
| [ | |
| "### 💰 Price Levels", | |
| "", | |
| f"- **Entry**: ${decision_dict.get('entry_price', 0):.2f}", | |
| f"- **Target**: ${decision_dict.get('target_price', 0):.2f}", | |
| f"- **Stop Loss**: ${decision_dict.get('stop_loss', 0):.2f}", | |
| "", | |
| ] | |
| ) | |
| lines.extend(["---", ""]) | |
| # Phase-by-Phase Results | |
| lines.extend( | |
| [ | |
| "## 📑 Phase-by-Phase Analysis", | |
| "", | |
| ] | |
| ) | |
| phase_names = { | |
| "technical": "🔧 Technical Analysis", | |
| "fundamental": "💼 Fundamental Analysis", | |
| "sentiment": "📰 Sentiment Analysis", | |
| "research_synthesis": "🔬 Research Synthesis", | |
| "risk": "⚠️ Risk Assessment", | |
| "decision": "🎯 Final Decision", | |
| } | |
| for phase_name in completed_phases: | |
| phase_output = phase_outputs.get(phase_name) | |
| if not phase_output: | |
| continue | |
| display_name = phase_names.get(phase_name, phase_name.replace("_", " ").title()) | |
| lines.extend( | |
| [ | |
| f"### {display_name}", | |
| "", | |
| ] | |
| ) | |
| # Phase summary - handle both dict and PhaseOutput object | |
| if isinstance(phase_output, dict): | |
| phase_summary = phase_output.get("phase_summary", "Analysis completed") | |
| execution_time = phase_output.get("execution_time_seconds", 0) | |
| agent_outputs = phase_output.get("agents", []) | |
| phase_score = phase_output.get("score") | |
| else: | |
| # It's a PhaseOutput object | |
| phase_summary = getattr(phase_output, "phase_summary", "Analysis completed") | |
| execution_time = getattr(phase_output, "execution_time_seconds", 0) | |
| agent_outputs = getattr(phase_output, "agents", []) | |
| phase_score = getattr(phase_output, "score", None) | |
| # Display phase score if available (US8) | |
| if phase_score is not None: | |
| # Add visual indicator based on score | |
| if phase_score >= 7: | |
| score_indicator = "🟢" # Bullish | |
| elif phase_score <= 3: | |
| score_indicator = "🔴" # Bearish | |
| else: | |
| score_indicator = "🟡" # Neutral | |
| lines.append(f"**Phase Score**: {score_indicator} {phase_score:.1f}/10") | |
| lines.extend( | |
| [ | |
| f"**Summary**: {phase_summary}", | |
| f"**Execution Time**: {execution_time:.1f}s", | |
| "", | |
| ] | |
| ) | |
| # Agent outputs | |
| if agent_outputs: | |
| lines.append("**Agent Outputs**:") | |
| lines.append("") | |
| for agent_output in agent_outputs: | |
| if isinstance(agent_output, dict): | |
| agent_name = agent_output.get("agent_name", "Unknown") | |
| report = agent_output.get("report", "") | |
| educational_notes = agent_output.get("educational_notes") | |
| charts = agent_output.get("charts", []) | |
| else: | |
| # Handle AgentOutput dataclass object | |
| agent_name = getattr(agent_output, "agent_name", "Unknown") | |
| report = getattr(agent_output, "report", "") | |
| educational_notes = getattr(agent_output, "educational_notes", None) | |
| charts = getattr(agent_output, "charts", []) | |
| lines.extend( | |
| [ | |
| f"**{agent_name.replace('_', ' ').title()}**:", | |
| "", | |
| ] | |
| ) | |
| # Show full report (removed truncation for complete analysis visibility) | |
| lines.append(report) | |
| lines.append("") | |
| # Add charts if available (User Story 5: Technical Indicator Charts) | |
| if charts: | |
| lines.append("**📊 Charts**:") | |
| lines.append("") | |
| for chart_path in charts: | |
| if chart_path: | |
| chart_file = Path(chart_path) | |
| chart_name = chart_file.stem.replace("_", " ").title() | |
| base64_image = _image_to_base64(chart_path) | |
| if base64_image: | |
| lines.append(f"*{chart_name}*") | |
| lines.append(f"") | |
| lines.append("") | |
| lines.append("") | |
| # Add educational notes if available | |
| if educational_notes: | |
| lines.extend( | |
| [ | |
| "**💡 Educational Notes**:", | |
| "", | |
| educational_notes, | |
| "", | |
| ] | |
| ) | |
| lines.append("") | |
| lines.extend(["---", ""]) | |
| # Add detailed analysis sections if available | |
| if final_state.get("technical_analyst"): | |
| lines.extend( | |
| [ | |
| "## 🔍 Technical Synthesis", | |
| "", | |
| str( | |
| final_state.get("technical_analyst", {}).get( | |
| "assessment", "No synthesis available" | |
| ) | |
| ), | |
| "", | |
| "---", | |
| "", | |
| ] | |
| ) | |
| if final_state.get("risk_assessment"): | |
| risk = final_state.get("risk_assessment", {}) | |
| lines.extend( | |
| [ | |
| "## ⚠️ Risk Assessment", | |
| "", | |
| f"**Risk Score**: {risk.get('risk_score', 0):.1f}/100", | |
| "", | |
| str(risk.get("assessment", "No risk assessment available")), | |
| "", | |
| "---", | |
| "", | |
| ] | |
| ) | |
| # Footer | |
| lines.extend( | |
| [ | |
| "", | |
| "---", | |
| "", | |
| "*This analysis is for educational purposes only and does not constitute financial advice.*", | |
| ] | |
| ) | |
| return "\n".join(lines) | |
| def format_phase_report_details(final_state: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| Extract key details from phase-organized report for JSON display. | |
| Args: | |
| final_state: Final workflow state | |
| Returns: | |
| Dictionary with key report details | |
| """ | |
| completed_phases = final_state.get("completed_phases", []) | |
| phase_outputs = final_state.get("phase_outputs", {}) | |
| # Calculate total execution time | |
| total_time = 0 | |
| for phase in completed_phases: | |
| phase_output = phase_outputs.get(phase) | |
| if phase_output: | |
| # Handle both dict and PhaseOutput object | |
| if hasattr(phase_output, "execution_time_seconds"): | |
| total_time += phase_output.execution_time_seconds | |
| elif isinstance(phase_output, dict): | |
| total_time += phase_output.get("execution_time_seconds", 0) | |
| details = { | |
| "ticker": final_state.get("ticker"), | |
| "timeframe": final_state.get("timeframe"), | |
| "investment_style": final_state.get("config", {}).get("investment_style"), | |
| "completed_phases": completed_phases, | |
| "total_phases": len(completed_phases), | |
| "total_execution_time_seconds": total_time, | |
| "final_decision": final_state.get("final_decision"), | |
| "workflow_status": "completed" if completed_phases else "incomplete", | |
| } | |
| # Add decision details if available | |
| portfolio_decision = final_state.get("portfolio_decision", {}) | |
| if portfolio_decision: | |
| decision_dict = portfolio_decision.get("decision", {}) | |
| details.update( | |
| { | |
| "recommendation": decision_dict.get("recommendation"), | |
| "confidence": decision_dict.get("confidence"), | |
| "risk_level": decision_dict.get("risk_level"), | |
| "entry_price": decision_dict.get("entry_price"), | |
| "target_price": decision_dict.get("target_price"), | |
| "stop_loss": decision_dict.get("stop_loss"), | |
| } | |
| ) | |
| # Add risk assessment if available | |
| risk_assessment = final_state.get("risk_assessment", {}) | |
| if risk_assessment: | |
| details["risk_score"] = risk_assessment.get("risk_score") | |
| return details | |