"""Data models for phase-organized analysis reports.""" from dataclasses import dataclass, field from datetime import datetime from typing import Any, Dict, List, Optional from config.models import AnalysisPhase, InvestmentStyle, PhaseConfiguration @dataclass class AgentOutput: """Output from a single agent within a phase.""" agent_name: str """Name of the agent (e.g., 'IndicatorAgent', 'PatternAgent').""" report: str """Markdown-formatted agent analysis report.""" charts: List[str] = field(default_factory=list) """List of base64-encoded chart images (PNG format, no data URI prefix).""" educational_notes: Optional[str] = None """Plain-English explanation of agent findings (if educational_mode enabled).""" execution_time_seconds: float = 0.0 """Time taken to execute this agent.""" metadata: Dict[str, Any] = field(default_factory=dict) """Additional agent-specific metadata (e.g., indicator values, pattern names).""" @dataclass class PhaseOutput: """Output from a complete analysis phase.""" phase: AnalysisPhase """Which phase this output represents.""" agents: List[AgentOutput] = field(default_factory=list) """Outputs from all agents in this phase.""" phase_summary: str = "" """High-level summary of this phase's findings.""" execution_time_seconds: float = 0.0 """Total time to execute all agents in this phase.""" score: Optional[float] = None """Numeric score (0-10) representing phase analysis outcome. 0=extremely bearish, 5=neutral, 10=extremely bullish.""" @dataclass class AnalysisReport: """Complete analysis report with phase-organized results.""" ticker: str """Ticker symbol analyzed.""" timeframe: str """Timeframe used for analysis (1d, 1w, 1mo, etc.).""" investment_style: InvestmentStyle """Investment style profile used.""" timestamp: datetime = field(default_factory=datetime.now) """When analysis was generated.""" phases: List[PhaseOutput] = field(default_factory=list) """Phase-by-phase analysis results.""" executive_summary: str = "" """High-level summary of entire analysis (generated by final agent).""" final_decision: Optional[str] = None """Final trading decision (BUY/SELL/HOLD) if DECISION phase was enabled.""" total_execution_time_seconds: float = 0.0 """End-to-end analysis execution time.""" config: Optional[PhaseConfiguration] = None """Configuration used for this analysis.""" cost_summary: Optional[Dict[str, Any]] = None """Cost tracking information from analysis (if available). Includes total_cost, tokens, provider breakdown, and free tier usage.""" def get_phase_output(self, phase: AnalysisPhase) -> Optional[PhaseOutput]: """Retrieve output for a specific phase. Args: phase: The phase to retrieve Returns: PhaseOutput if found, None otherwise """ for p in self.phases: if p.phase == phase: return p return None def to_markdown(self) -> str: """Convert report to markdown format for display. Returns: Markdown-formatted report string """ # Header section style_name = self.investment_style.value.replace("_", " ").title() md = f"""# Analysis Report: {self.ticker} **Timeframe**: {self.timeframe} **Investment Style**: {style_name} **Generated**: {self.timestamp.strftime("%Y-%m-%d %H:%M:%S")} **Execution Time**: {self.total_execution_time_seconds:.1f}s --- ## Executive Summary {self.executive_summary} """ # Final decision (if available) if self.final_decision: md += f"\n**Final Decision**: {self.final_decision}\n" # Cost information (if available) if self.cost_summary: md += "\n## Analysis Cost Summary\n\n" md += f"**Total Cost**: ${self.cost_summary['total_cost']:.4f}\n" md += f"**Total Tokens**: {self.cost_summary['total_tokens']:,} " md += f"({self.cost_summary['total_input_tokens']:,} in + {self.cost_summary['total_output_tokens']:,} out)\n" # Provider breakdown if self.cost_summary.get("provider_costs"): md += "\n**Cost by Provider:**\n\n" md += "| Provider | Cost | Tokens | Free Tier Calls |\n" md += "|----------|------|--------|----------------|\n" for provider, cost in sorted( self.cost_summary["provider_costs"].items(), key=lambda x: x[1], reverse=True, ): tokens = self.cost_summary.get("provider_tokens", {}).get( provider, 0 ) free_calls = self.cost_summary.get("free_tier_calls", {}).get( provider, 0 ) if free_calls > 0: cost_str = ( f"${cost:.4f} (free)" if cost == 0.0 else f"${cost:.4f}" ) md += ( f"| {provider} | {cost_str} | {tokens:,} | {free_calls} |\n" ) else: md += f"| {provider} | ${cost:.4f} | {tokens:,} | 0 |\n" md += "\n*Costs are estimates based on current pricing. Free tier usage is tracked automatically.*\n" md += "\n---\n\n" # Phase-by-phase breakdown for phase_output in self.phases: phase_name = phase_output.phase.value.replace("_", " ").title() md += f"## {phase_name} Phase\n\n" if phase_output.phase_summary: md += f"*{phase_output.phase_summary}*\n\n" md += f"*Execution time: {phase_output.execution_time_seconds:.1f}s*\n\n" # Agent outputs within this phase for agent_output in phase_output.agents: md += f"### {agent_output.agent_name}\n\n" md += f"{agent_output.report}\n\n" # Educational notes if agent_output.educational_notes: md += f"💡 **Plain English**: {agent_output.educational_notes}\n\n" # Embedded charts for chart_b64 in agent_output.charts: md += f'\n\n' md += "\n---\n\n" return md def to_dict(self) -> Dict[str, Any]: """Convert report to dictionary for JSON serialization. Returns: Dictionary representation of the report """ return { "ticker": self.ticker, "timeframe": self.timeframe, "investment_style": self.investment_style.value, "timestamp": self.timestamp.isoformat(), "executive_summary": self.executive_summary, "final_decision": self.final_decision, "total_execution_time_seconds": self.total_execution_time_seconds, "cost_summary": self.cost_summary if self.cost_summary else None, "config": { "investment_style": self.config.investment_style.value if self.config else None, "enabled_phases": [p.value for p in self.config.enabled_phases] if self.config else [], "timeframe": self.config.timeframe if self.config else None, "chart_period_days": self.config.chart_period_days if self.config else None, "educational_mode": self.config.educational_mode if self.config else None, } if self.config else None, "phases": [ { "phase": phase.phase.value, "phase_summary": phase.phase_summary, "execution_time_seconds": phase.execution_time_seconds, "agents": [ { "agent_name": agent.agent_name, "report": agent.report, "charts": agent.charts, "educational_notes": agent.educational_notes, "execution_time_seconds": agent.execution_time_seconds, "metadata": agent.metadata, } for agent in phase.agents ], } for phase in self.phases ], } def to_html(self) -> str: """Convert report to HTML format with enhanced styling. Returns: HTML-formatted report string with CSS styling """ style_name = self.investment_style.value.replace("_", " ").title() # CSS styling html = """ Analysis Report: {ticker}

Analysis Report: {ticker}

Timeframe {timeframe}
Investment Style {style_name}
Generated {timestamp}
Execution Time {execution_time:.1f}s

Executive Summary

{executive_summary}
""".format( ticker=self.ticker, timeframe=self.timeframe, style_name=style_name, timestamp=self.timestamp.strftime("%Y-%m-%d %H:%M:%S"), execution_time=self.total_execution_time_seconds, executive_summary=self._markdown_to_html(self.executive_summary), ) # Final decision if self.final_decision: html += f"""
Final Decision: {self.final_decision}
""" # Cost summary if self.cost_summary: html += """

Analysis Cost Summary

Total Cost ${total_cost:.4f}
Total Tokens {total_tokens:,}
Input Tokens {input_tokens:,}
Output Tokens {output_tokens:,}
""".format( total_cost=self.cost_summary["total_cost"], total_tokens=self.cost_summary["total_tokens"], input_tokens=self.cost_summary["total_input_tokens"], output_tokens=self.cost_summary["total_output_tokens"], ) # Provider breakdown table if self.cost_summary.get("provider_costs"): html += """

Cost by Provider

""" for provider, cost in sorted( self.cost_summary["provider_costs"].items(), key=lambda x: x[1], reverse=True, ): tokens = self.cost_summary.get("provider_tokens", {}).get( provider, 0 ) free_calls = self.cost_summary.get("free_tier_calls", {}).get( provider, 0 ) if free_calls > 0: cost_str = ( f"${cost:.4f} (free)" if cost == 0.0 else f"${cost:.4f}" ) style = "background-color: #e8f5e9;" if cost == 0.0 else "" else: cost_str = f"${cost:.4f}" style = "" html += f""" """ html += """
Provider Cost Tokens Free Tier Calls
{provider} {cost_str} {tokens:,} {free_calls}

Costs are estimates based on current pricing. Free tier usage is tracked automatically.

""" html += "\n
\n" # Phase-by-phase breakdown for phase_output in self.phases: phase_name = phase_output.phase.value.replace("_", " ").title() html += f"""
{phase_name} Phase
""" if phase_output.phase_summary: html += f"""
{self._markdown_to_html(phase_output.phase_summary)}
""" html += f"""

Execution time: {phase_output.execution_time_seconds:.1f}s

""" # Agent outputs within this phase for agent_output in phase_output.agents: html += f"""
{agent_output.agent_name}
{self._markdown_to_html(agent_output.report)}
""" # Educational notes if agent_output.educational_notes: html += f"""
Plain English: {self._markdown_to_html(agent_output.educational_notes)}
""" # Embedded charts if agent_output.charts: for chart_b64 in agent_output.charts: html += f"""
Chart
""" html += """
""" html += """
""" html += """
""" return html def _markdown_to_html(self, markdown_text: str) -> str: """Convert basic markdown to HTML (simplified converter). Args: markdown_text: Markdown-formatted text Returns: HTML-formatted text with basic markdown converted """ if not markdown_text: return "" import re html = markdown_text # Bold text html = re.sub(r"\*\*(.+?)\*\*", r"\1", html) # Italic text html = re.sub(r"\*(.+?)\*", r"\1", html) # Links html = re.sub(r"\[(.+?)\]\((.+?)\)", r'\1', html) # Code blocks html = re.sub(r"`(.+?)`", r"\1", html) # Line breaks html = html.replace("\n\n", "

") html = html.replace("\n", "
") # Wrap in paragraph if not already if not html.startswith("<"): html = f"

{html}

" return html