""" 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"![{chart_name}]({base64_image})") 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