""" Report viewer component for displaying analysis results. """ from typing import Any, Dict, List, Optional, Union import gradio as gr from data.providers.base import DataProvider from graph.state.trading_state import TechnicalWorkflowState from graph.workflows.comprehensive_workflow import ComprehensiveState from utils.cost_tracker import format_cost def format_config_info(config: Dict[str, Any]) -> str: """ Format configuration information for display. Args: config: Configuration dictionary Returns: Markdown formatted configuration info """ if not config: return "" lines = [] lines.append("### âš™ī¸ Analysis Configuration") lines.append("") # Indicator parameters if "indicator_parameters" in config: params = config["indicator_parameters"] lines.append("**Indicator Parameters:**") lines.append(f"- RSI Period: {params.get('rsi_period', 14)}") lines.append( f"- MACD: Fast={params.get('macd_fast', 12)}, Slow={params.get('macd_slow', 26)}, Signal={params.get('macd_signal', 9)}" ) lines.append( f"- Stochastic: K={params.get('stoch_k_period', 14)}, D={params.get('stoch_d_period', 3)}" ) lines.append("") # Data providers if "data_providers" in config: providers = config["data_providers"] lines.append("**Data Sources:**") lines.append(f"- OHLC: {providers.get('ohlc_primary', 'yfinance')}") lines.append( f"- Fundamentals: {providers.get('fundamentals_primary', 'alpha_vantage')}" ) lines.append("") return "\n".join(lines) def format_cost_summary(cost_summary: Optional[Dict[str, Any]]) -> str: """ Format cost tracking summary for display. Args: cost_summary: Cost summary dictionary from CostTracker Returns: Markdown formatted cost summary """ if not cost_summary: return "" lines = [] lines.append("### 💰 Analysis Cost") lines.append("") total_cost = cost_summary.get("total_cost", 0.0) total_tokens = cost_summary.get("total_tokens", 0) call_count = cost_summary.get("call_count", 0) lines.append(f"**Total Cost:** {format_cost(total_cost)}") lines.append(f"**Total Tokens:** {total_tokens:,}") lines.append(f"**API Calls:** {call_count}") if call_count > 0: avg_cost = cost_summary.get("average_cost_per_call", 0.0) lines.append(f"**Average per Call:** {format_cost(avg_cost)}") # Show per-agent breakdown if available agent_costs = cost_summary.get("agent_costs", {}) if agent_costs: lines.append("") lines.append("
") lines.append("Cost by Agent (click to expand)") lines.append("") for agent_name, cost in sorted( agent_costs.items(), key=lambda x: x[1], reverse=True ): lines.append(f"- **{agent_name}:** {format_cost(cost)}") lines.append("") lines.append("
") lines.append("") lines.append("---") lines.append("") return "\n".join(lines) def get_asset_guidance(ticker: str) -> str: """ Get asset-specific guidance and warnings based on asset type. Args: ticker: Asset ticker symbol Returns: Markdown formatted guidance/warning string """ asset_type = DataProvider.detect_asset_type(ticker) asset_characteristics = DataProvider.get_asset_characteristics(asset_type) guidance_parts = [] if asset_type == "crypto": guidance_parts.append("## đŸĒ™ Cryptocurrency Asset") guidance_parts.append("") guidance_parts.append("> **âš ī¸ HIGH VOLATILITY WARNING**") guidance_parts.append("> ") guidance_parts.append( "> - Cryptocurrencies are **extremely volatile** and can experience 10-20%+ swings in minutes" ) guidance_parts.append( "> - Market operates **24/7 (365 days)** - no market close for risk management" ) guidance_parts.append( "> - **Highly sentiment-driven** - news and social media have immediate impact" ) guidance_parts.append( "> - **Regulatory risk** - government actions can cause instant severe moves" ) guidance_parts.append( "> - **Liquidity varies** - wide spreads possible during volatility" ) guidance_parts.append("> - Traditional fundamental analysis **does not apply**") guidance_parts.append("> ") guidance_parts.append( "> **Focus on:** Technical patterns, market sentiment, news, trading volume" ) guidance_parts.append("") elif asset_type == "commodity": guidance_parts.append("## 🌾 Commodity Futures") guidance_parts.append("") guidance_parts.append("> **âš ī¸ FUTURES CONTRACT NOTICE**") guidance_parts.append("> ") guidance_parts.append( "> - Most commodity tickers represent **futures contracts** with expiration dates" ) guidance_parts.append( "> - **Contract rollover** required - positions may need to be closed before expiry" ) guidance_parts.append( "> - **Contango/backwardation** affects long-term holding costs" ) guidance_parts.append( "> - **High leverage** inherent in futures - margin requirements apply" ) guidance_parts.append( "> - **Gap risk** from overnight geopolitical/weather events" ) guidance_parts.append("> - Traditional financial metrics **do not apply**") guidance_parts.append("> ") guidance_parts.append( "> **Focus on:** Supply/demand dynamics, inventory levels, weather, geopolitical events" ) guidance_parts.append("") elif asset_type == "index": guidance_parts.append("## 📈 Market Index") guidance_parts.append("") guidance_parts.append("> **â„šī¸ INDEX TRACKING NOTICE**") guidance_parts.append("> ") guidance_parts.append( "> - Indices **cannot be traded directly** - use ETFs, futures, or options" ) guidance_parts.append( "> - Represents **weighted basket** of underlying securities" ) guidance_parts.append( "> - **Sector composition** affects behavior - tech-heavy indices more volatile" ) guidance_parts.append( "> - **Diversification** reduces single-stock risk but limits upside" ) guidance_parts.append("> - Traditional company fundamentals **do not apply**") guidance_parts.append("> ") guidance_parts.append( "> **Focus on:** Macro trends, sector rotation, interest rates, earnings season" ) guidance_parts.append("") elif asset_type == "forex": guidance_parts.append("## 💱 Foreign Exchange") guidance_parts.append("") guidance_parts.append("> **âš ī¸ CURRENCY PAIR NOTICE**") guidance_parts.append("> ") guidance_parts.append( "> - Forex markets are **highly leveraged** (50:1 to 500:1 typical)" ) guidance_parts.append( "> - Market operates **24/5** (Sunday evening to Friday evening)" ) guidance_parts.append( "> - **Macro-driven** - central bank policy, interest rates, geopolitical events" ) guidance_parts.append( "> - **Bid-ask spreads** vary by broker and market conditions" ) guidance_parts.append("> - Traditional equity fundamentals **do not apply**") guidance_parts.append("> ") guidance_parts.append( "> **Focus on:** Interest rate differentials, economic data, central bank policy" ) guidance_parts.append("") elif asset_type == "stock": # For stocks, just add a brief note about traditional analysis guidance_parts.append("## 📊 Equity Stock") guidance_parts.append("") guidance_parts.append("> **â„šī¸ Traditional equity analysis applies**") guidance_parts.append("> ") guidance_parts.append("> This is a publicly traded stock. Analysis includes:") guidance_parts.append("> - Financial fundamentals (earnings, revenue, margins)") guidance_parts.append("> - Technical indicators and chart patterns") guidance_parts.append("> - News and market sentiment") guidance_parts.append("> - Company-specific events and announcements") guidance_parts.append("") return "\n".join(guidance_parts) def create_timeframe_alignment_indicator( alignment_score: float, trend_direction: str = "unknown" ) -> str: """ Create a visual indicator for timeframe alignment. Args: alignment_score: Alignment score (0-1) trend_direction: Overall trend direction (bullish, bearish, mixed) Returns: Visual indicator string with emoji and description """ # Determine alignment strength if alignment_score >= 0.75: strength = "Strong" emoji = "đŸŸĸđŸŸĸđŸŸĸ" elif alignment_score >= 0.5: strength = "Moderate" emoji = "đŸŸĸđŸŸĸ" elif alignment_score >= 0.25: strength = "Weak" emoji = "🟡" else: strength = "Conflicting" emoji = "🔴" # Add trend direction if trend_direction.lower() == "bullish": direction_emoji = "📈" elif trend_direction.lower() == "bearish": direction_emoji = "📉" else: direction_emoji = "â†”ī¸" return f"{emoji} {direction_emoji} {strength} Alignment ({alignment_score:.0%})" def format_timeframe_context(timeframe_context: Dict[str, Any]) -> str: """ Format timeframe context information for display. Args: timeframe_context: Timeframe context dict Returns: Formatted string """ if not timeframe_context: return "" label = timeframe_context.get("label", "unknown") scope = timeframe_context.get("scope", "trading") hold_duration = timeframe_context.get("hold_duration", "varies") weight = timeframe_context.get("weight", 0.5) return f"**Timeframe:** {label} | **Scope:** {scope.title()} | **Expected Hold:** {hold_duration} | **Weight:** {weight:.1f}" def create_report_viewer() -> gr.Markdown: """ Create report viewer component. Returns: Gradio Markdown component for displaying reports """ return gr.Markdown( label="Analysis Report", value="Analysis report will appear here...", ) def format_error_report(error_message: str) -> str: """ Format error message as report. Args: error_message: Error message Returns: Markdown formatted error report """ return f"""# âš ī¸ Analysis Error An error occurred during analysis: ``` {error_message} ``` Please check: - Ticker symbol is valid - Internet connection is working - API keys are configured (if using Alpha Vantage) - Try a different timeframe """ def format_progress_message(current_agent: str) -> str: """ Format progress message during analysis. Args: current_agent: Currently executing agent Returns: Progress message """ agent_display = current_agent.replace("_", " ").title() return f"""# 🔄 Analysis in Progress Currently running: **{agent_display}** Please wait while the multi-agent system analyzes the market data... **Workflow Steps:** 1. ✅ Fetch Market Data 2. 🔄 Indicator Agent - Calculate technical indicators 3. âŗ Pattern Agent - Identify chart patterns 4. âŗ Trend Agent - Analyze trends 5. âŗ Decision Agent - Generate recommendation 6. âŗ Generate Charts """