"""The live view of a run: message log, agent status, report panels and timings.""" import datetime import time from collections import deque from time import monotonic from rich import box from rich.console import Console from rich.layout import Layout from rich.markdown import Markdown from rich.panel import Panel from rich.rule import Rule from rich.spinner import Spinner from rich.table import Table from rich.text import Text from tradingagents.graph.analyst_execution import ( ANALYST_NODE_SPECS, AnalystExecutionPlan, ) console = Console() class MessageBuffer: # Fixed teams that always run (not user-selectable) FIXED_AGENTS = { "Research Team": ["Bull Researcher", "Bear Researcher", "Research Manager"], "Trading Team": ["Trader"], "Risk Management": ["Aggressive Analyst", "Neutral Analyst", "Conservative Analyst"], "Portfolio Management": ["Portfolio Manager"], } # Analyst name mapping ANALYST_MAPPING = { "market": "Market Analyst", "social": "Sentiment Analyst", "news": "News Analyst", "fundamentals": "Fundamentals Analyst", } # Report section mapping: section -> (analyst_key for filtering, finalizing_agent) # analyst_key: which analyst selection controls this section (None = always included) # finalizing_agent: which agent must be "completed" for this report to count as done REPORT_SECTIONS = { "market_report": ("market", "Market Analyst"), "sentiment_report": ("social", "Sentiment Analyst"), "news_report": ("news", "News Analyst"), "fundamentals_report": ("fundamentals", "Fundamentals Analyst"), "investment_plan": (None, "Research Manager"), "trader_investment_plan": (None, "Trader"), "final_trade_decision": (None, "Portfolio Manager"), } def __init__(self, max_length=100): self.messages = deque(maxlen=max_length) self.tool_calls = deque(maxlen=max_length) self.current_report = None self.agent_status = {} self.report_sections = {} self.selected_analysts = [] self._processed_message_ids = set() def init_for_analysis(self, selected_analysts): """Initialize agent status and report sections based on selected analysts. Args: selected_analysts: List of analyst type strings (e.g., ["market", "news"]) """ self.selected_analysts = [a.lower() for a in selected_analysts] self.agent_status = {} for analyst_key in self.selected_analysts: if analyst_key in self.ANALYST_MAPPING: self.agent_status[self.ANALYST_MAPPING[analyst_key]] = "pending" for team_agents in self.FIXED_AGENTS.values(): for agent in team_agents: self.agent_status[agent] = "pending" self.report_sections = {} for section, (analyst_key, _) in self.REPORT_SECTIONS.items(): if analyst_key is None or analyst_key in self.selected_analysts: self.report_sections[section] = None # Reset other state self.current_report = None self.messages.clear() self.tool_calls.clear() self._processed_message_ids.clear() def get_completed_reports_count(self): """Count reports that are finalized (their finalizing agent is completed). A report is considered complete when: 1. The report section has content (not None), AND 2. The agent responsible for finalizing that report has status "completed" This prevents interim updates (like debate rounds) from counting as completed. """ count = 0 for section in self.report_sections: if section not in self.REPORT_SECTIONS: continue _, finalizing_agent = self.REPORT_SECTIONS[section] # Report is complete if it has content AND its finalizing agent is done has_content = self.report_sections.get(section) is not None agent_done = self.agent_status.get(finalizing_agent) == "completed" if has_content and agent_done: count += 1 return count def add_message(self, message_type, content): timestamp = datetime.datetime.now().strftime("%H:%M:%S") self.messages.append((timestamp, message_type, content)) def add_tool_call(self, tool_name, args): timestamp = datetime.datetime.now().strftime("%H:%M:%S") self.tool_calls.append((timestamp, tool_name, args)) def update_agent_status(self, agent, status): if agent in self.agent_status: self.agent_status[agent] = status def update_report_section(self, section_name, content): if section_name in self.report_sections: self.report_sections[section_name] = content self._update_current_report() def _update_current_report(self): # For the panel display, only show the most recently updated section latest_section = None latest_content = None for section, content in self.report_sections.items(): if content is not None: latest_section = section latest_content = content if latest_section and latest_content: section_titles = { "market_report": "Market Analysis", "sentiment_report": "Social Sentiment", "news_report": "News Analysis", "fundamentals_report": "Fundamentals Analysis", "investment_plan": "Research Team Decision", "trader_investment_plan": "Trading Team Plan", "final_trade_decision": "Portfolio Management Decision", } self.current_report = ( f"### {section_titles[latest_section]}\n{latest_content}" ) message_buffer = MessageBuffer() def create_layout(): layout = Layout() layout.split_column( Layout(name="header", size=3), Layout(name="main"), Layout(name="footer", size=3), ) layout["main"].split_column( Layout(name="upper", ratio=3), Layout(name="analysis", ratio=5) ) layout["upper"].split_row( Layout(name="progress", ratio=2), Layout(name="messages", ratio=3) ) return layout def format_tokens(n): """Format token count for display.""" if n >= 1000: return f"{n/1000:.1f}k" return str(n) def update_display(layout, spinner_text=None, stats_handler=None, start_time=None): # Header with welcome message layout["header"].update( Panel( "[bold green]Welcome to TradingAgents CLI[/bold green]\n" "[dim]© [Tauric Research](https://github.com/TauricResearch)[/dim]", title="Welcome to TradingAgents", border_style="green", padding=(1, 2), expand=True, ) ) # Progress panel showing agent status progress_table = Table( show_header=True, header_style="bold magenta", show_footer=False, box=box.SIMPLE_HEAD, # Use simple header with horizontal lines title=None, # Remove the redundant Progress title padding=(0, 2), # Add horizontal padding expand=True, # Make table expand to fill available space ) progress_table.add_column("Team", style="cyan", justify="center", width=20) progress_table.add_column("Agent", style="green", justify="center", width=20) progress_table.add_column("Status", style="yellow", justify="center", width=20) # Group agents by team - filter to only include agents in agent_status all_teams = { "Analyst Team": [ "Market Analyst", "Sentiment Analyst", "News Analyst", "Fundamentals Analyst", ], "Research Team": ["Bull Researcher", "Bear Researcher", "Research Manager"], "Trading Team": ["Trader"], "Risk Management": ["Aggressive Analyst", "Neutral Analyst", "Conservative Analyst"], "Portfolio Management": ["Portfolio Manager"], } teams = {} for team, agents in all_teams.items(): active_agents = [a for a in agents if a in message_buffer.agent_status] if active_agents: teams[team] = active_agents for team, agents in teams.items(): first_agent = agents[0] status = message_buffer.agent_status.get(first_agent, "pending") if status == "in_progress": spinner = Spinner( "dots", text="[blue]in_progress[/blue]", style="bold cyan" ) status_cell = spinner else: status_color = { "pending": "yellow", "completed": "green", "error": "red", }.get(status, "white") status_cell = f"[{status_color}]{status}[/{status_color}]" progress_table.add_row(team, first_agent, status_cell) for agent in agents[1:]: status = message_buffer.agent_status.get(agent, "pending") if status == "in_progress": spinner = Spinner( "dots", text="[blue]in_progress[/blue]", style="bold cyan" ) status_cell = spinner else: status_color = { "pending": "yellow", "completed": "green", "error": "red", }.get(status, "white") status_cell = f"[{status_color}]{status}[/{status_color}]" progress_table.add_row("", agent, status_cell) progress_table.add_row("─" * 20, "─" * 20, "─" * 20, style="dim") layout["progress"].update( Panel(progress_table, title="Progress", border_style="cyan", padding=(1, 2)) ) # Messages panel showing recent messages and tool calls messages_table = Table( show_header=True, header_style="bold magenta", show_footer=False, expand=True, # Make table expand to fill available space box=box.MINIMAL, # Use minimal box style for a lighter look show_lines=True, # Keep horizontal lines padding=(0, 1), # Add some padding between columns ) messages_table.add_column("Time", style="cyan", width=8, justify="center") messages_table.add_column("Type", style="green", width=10, justify="center") messages_table.add_column( "Content", style="white", no_wrap=False, ratio=1 ) # Make content column expand # Combine tool calls and messages all_messages = [] for timestamp, tool_name, args in message_buffer.tool_calls: formatted_args = format_tool_args(args) all_messages.append((timestamp, "Tool", f"{tool_name}: {formatted_args}")) for timestamp, msg_type, content in message_buffer.messages: content_str = str(content) if content else "" if len(content_str) > 200: content_str = content_str[:197] + "..." all_messages.append((timestamp, msg_type, content_str)) # Sort by timestamp descending (newest first) all_messages.sort(key=lambda x: x[0], reverse=True) max_messages = 12 recent_messages = all_messages[:max_messages] for timestamp, msg_type, content in recent_messages: wrapped_content = Text(content, overflow="fold") messages_table.add_row(timestamp, msg_type, wrapped_content) layout["messages"].update( Panel( messages_table, title="Messages & Tools", border_style="blue", padding=(1, 2), ) ) # Analysis panel showing current report if message_buffer.current_report: layout["analysis"].update( Panel( Markdown(message_buffer.current_report), title="Current Report", border_style="green", padding=(1, 2), ) ) else: layout["analysis"].update( Panel( "[italic]Waiting for analysis report...[/italic]", title="Current Report", border_style="green", padding=(1, 2), ) ) # Footer with statistics # Agent progress - derived from agent_status dict agents_completed = sum( 1 for status in message_buffer.agent_status.values() if status == "completed" ) agents_total = len(message_buffer.agent_status) # Report progress - based on agent completion (not just content existence) reports_completed = message_buffer.get_completed_reports_count() reports_total = len(message_buffer.report_sections) stats_parts = [f"Agents: {agents_completed}/{agents_total}"] # LLM and tool stats from callback handler if stats_handler: stats = stats_handler.get_stats() stats_parts.append(f"LLM: {stats['llm_calls']}") stats_parts.append(f"Tools: {stats['tool_calls']}") # Token display with graceful fallback if stats["tokens_in"] > 0 or stats["tokens_out"] > 0: tokens_str = f"Tokens: {format_tokens(stats['tokens_in'])}\u2191 {format_tokens(stats['tokens_out'])}\u2193" else: tokens_str = "Tokens: --" stats_parts.append(tokens_str) stats_parts.append(f"Reports: {reports_completed}/{reports_total}") # Elapsed time if start_time: elapsed = time.time() - start_time elapsed_str = f"\u23f1 {int(elapsed // 60):02d}:{int(elapsed % 60):02d}" stats_parts.append(elapsed_str) stats_table = Table(show_header=False, box=None, padding=(0, 2), expand=True) stats_table.add_column("Stats", justify="center") stats_table.add_row(" | ".join(stats_parts)) layout["footer"].update(Panel(stats_table, border_style="grey50")) def display_complete_report(final_state): """Display the complete analysis report sequentially (avoids truncation).""" console.print() console.print(Rule("Complete Analysis Report", style="bold green")) # I. Analyst Team Reports analysts = [] if final_state.get("market_report"): analysts.append(("Market Analyst", final_state["market_report"])) if final_state.get("sentiment_report"): analysts.append(("Sentiment Analyst", final_state["sentiment_report"])) if final_state.get("news_report"): analysts.append(("News Analyst", final_state["news_report"])) if final_state.get("fundamentals_report"): analysts.append(("Fundamentals Analyst", final_state["fundamentals_report"])) if analysts: console.print(Panel("[bold]I. Analyst Team Reports[/bold]", border_style="cyan")) for title, content in analysts: console.print(Panel(Markdown(content), title=title, border_style="blue", padding=(1, 2))) # II. Research Team Reports if final_state.get("investment_debate_state"): debate = final_state["investment_debate_state"] research = [] if debate.get("bull_history"): research.append(("Bull Researcher", debate["bull_history"])) if debate.get("bear_history"): research.append(("Bear Researcher", debate["bear_history"])) if final_state.get("investment_plan"): research.append(("Research Manager", final_state["investment_plan"])) if research: console.print(Panel("[bold]II. Research Team Decision[/bold]", border_style="magenta")) for title, content in research: console.print(Panel(Markdown(content), title=title, border_style="blue", padding=(1, 2))) # III. Trading Team if final_state.get("trader_investment_plan"): console.print(Panel("[bold]III. Trading Team Plan[/bold]", border_style="yellow")) console.print(Panel(Markdown(final_state["trader_investment_plan"]), title="Trader", border_style="blue", padding=(1, 2))) # IV. Risk Management Team if final_state.get("risk_debate_state"): risk = final_state["risk_debate_state"] risk_reports = [] if risk.get("aggressive_history"): risk_reports.append(("Aggressive Analyst", risk["aggressive_history"])) if risk.get("conservative_history"): risk_reports.append(("Conservative Analyst", risk["conservative_history"])) if risk.get("neutral_history"): risk_reports.append(("Neutral Analyst", risk["neutral_history"])) if risk_reports: console.print(Panel("[bold]IV. Risk Management Team Decision[/bold]", border_style="red")) for title, content in risk_reports: console.print(Panel(Markdown(content), title=title, border_style="blue", padding=(1, 2))) # V. Portfolio Manager Decision if final_state.get("final_trade_decision"): console.print(Panel("[bold]V. Portfolio Manager Decision[/bold]", border_style="green")) console.print(Panel(Markdown(final_state["final_trade_decision"]), title="Portfolio Manager", border_style="blue", padding=(1, 2))) def update_research_team_status(status): """Update status for research team members (not Trader).""" research_team = ["Bull Researcher", "Bear Researcher", "Research Manager"] for agent in research_team: message_buffer.update_agent_status(agent, status) # Ordered list of analysts for status transitions ANALYST_ORDER = ["market", "social", "news", "fundamentals"] ANALYST_AGENT_NAMES = { "market": "Market Analyst", "social": "Sentiment Analyst", "news": "News Analyst", "fundamentals": "Fundamentals Analyst", } ANALYST_REPORT_MAP = { "market": "market_report", "social": "sentiment_report", "news": "news_report", "fundamentals": "fundamentals_report", } def update_analyst_statuses(message_buffer, chunk, wall_time_tracker=None): """Update analyst statuses based on accumulated report state. Logic: - Store new report content from the current chunk if present - Check accumulated report_sections (not just current chunk) for status - Analysts with reports = completed - First analyst without report = in_progress - Remaining analysts without reports = pending - When all analysts done, set Bull Researcher to in_progress """ selected = message_buffer.selected_analysts found_active = False if wall_time_tracker is not None: sync_analyst_tracker_from_chunk(wall_time_tracker, chunk) for analyst_key in ANALYST_ORDER: if analyst_key not in selected: continue agent_name = ANALYST_AGENT_NAMES[analyst_key] report_key = ANALYST_REPORT_MAP[analyst_key] # Capture new report content from current chunk if chunk.get(report_key): message_buffer.update_report_section(report_key, chunk[report_key]) # Determine status from accumulated sections, not just current chunk has_report = bool(message_buffer.report_sections.get(report_key)) if has_report: message_buffer.update_agent_status(agent_name, "completed") elif not found_active: message_buffer.update_agent_status(agent_name, "in_progress") found_active = True else: message_buffer.update_agent_status(agent_name, "pending") # When all analysts complete, transition research team to in_progress if ( not found_active and selected and message_buffer.agent_status.get("Bull Researcher") == "pending" ): message_buffer.update_agent_status("Bull Researcher", "in_progress") def extract_content_string(content): """Extract string content from various message formats. Returns None if no meaningful text content is found. """ def is_empty(val): """Whether a value carries nothing to show. Text is judged by whether anything was written, not by what it would mean as Python: a report saying "0" or "None" is a message the run produced, and reading it as a falsy literal dropped it from the display. """ if isinstance(val, str): return not val.strip() return val is None or not bool(val) if is_empty(content): return None if isinstance(content, str): return content.strip() if isinstance(content, dict): text = content.get('text', '') return text.strip() if not is_empty(text) else None if isinstance(content, list): text_parts = [ item.get('text', '').strip() if isinstance(item, dict) and item.get('type') == 'text' else (item.strip() if isinstance(item, str) else '') for item in content ] result = ' '.join(t for t in text_parts if t and not is_empty(t)) return result if result else None return str(content).strip() if not is_empty(content) else None def classify_message_type(message) -> tuple[str, str | None]: """Classify LangChain message into display type and extract content. Returns: (type, content) - type is one of: User, Agent, Data, Control - content is extracted string or None """ from langchain_core.messages import AIMessage, HumanMessage, ToolMessage content = extract_content_string(getattr(message, 'content', None)) if isinstance(message, HumanMessage): if content and content.strip() == "Continue": return ("Control", content) return ("User", content) if isinstance(message, ToolMessage): return ("Data", content) if isinstance(message, AIMessage): return ("Agent", content) # Fallback for unknown types return ("System", content) def format_tool_args(args, max_length=80) -> str: """Format tool arguments for terminal display.""" result = str(args) if len(result) > max_length: return result[:max_length - 3] + "..." return result class AnalystWallTimeTracker: def __init__(self, plan: AnalystExecutionPlan): self.plan = plan self._started_at: dict[str, float] = {} self._wall_times: dict[str, float] = {} def mark_started(self, analyst_key: str, started_at: float | None = None) -> None: if analyst_key not in ANALYST_NODE_SPECS: raise ValueError(f"unknown analyst key: {analyst_key}") self._started_at.setdefault(analyst_key, monotonic() if started_at is None else started_at) def mark_completed( self, analyst_key: str, completed_at: float | None = None, ) -> None: if analyst_key not in ANALYST_NODE_SPECS: raise ValueError(f"unknown analyst key: {analyst_key}") if analyst_key in self._wall_times: return started_at = self._started_at.get(analyst_key) if started_at is None: return finished_at = monotonic() if completed_at is None else completed_at self._wall_times[analyst_key] = max(0.0, finished_at - started_at) def format_summary(self) -> str: parts = [] for spec in self.plan.specs: duration = self._wall_times.get(spec.key) if duration is not None: label = spec.agent_node.removesuffix(" Analyst") parts.append(f"{label} {duration:.2f}s") if not parts: return "Analyst wall time: pending" return "Analyst wall time: " + " | ".join(parts) def sync_analyst_tracker_from_chunk( tracker: AnalystWallTimeTracker, chunk: dict[str, str], now: float | None = None, ) -> None: current_time = monotonic() if now is None else now active_found = False for spec in tracker.plan.specs: has_report = bool(chunk.get(spec.report_key)) if has_report: tracker.mark_started(spec.key, started_at=current_time) tracker.mark_completed(spec.key, completed_at=current_time) continue if not active_found: tracker.mark_started(spec.key, started_at=current_time) active_found = True