perf: optimise data formatting and implement comprehensive caching

- Add token-efficient formatters for portfolio, market, fundamentals, technical indicators, ML forecasts, and sentiment data
- Implement cached_async decorators across all MCP router methods with tailored TTL values
- Adjust cache TTL strategy: reduce market data to 1min, extend analysis results to 4hrs
- Integrate sentiment data into portfolio analysis workflow

backend/agents/portfolio_analyst.py

@@ -161,6 +161,411 @@ def format_optimisation_results_xml(opt_results: Dict[str, Any]) -> str:
     return "\n".join(lines)
 
 
+def format_fundamentals(fundamentals: Dict[str, Any]) -> str:
+    """Format company fundamentals as compact table without lengthy descriptions.
+
+    Args:
+        fundamentals: Dictionary of company profiles keyed by ticker
+
+    Returns:
+        Tab-separated table format for token efficiency
+    """
+    if not fundamentals:
+        return "No fundamental data available"
+
+    lines = ["Ticker\tSector\tIndustry\tMkt Cap\tP/E\tEmployees\tCountry"]
+    for ticker, profile in fundamentals.items():
+        # Handle both dict and object attribute access
+        if isinstance(profile, dict):
+            sector = profile.get('sector', 'N/A')
+            industry = profile.get('industry', 'N/A')
+            market_cap = profile.get('market_cap', 'N/A')
+            pe_ratio = profile.get('pe_ratio', 'N/A')
+            employees = profile.get('employees', 'N/A')
+            country = profile.get('country', 'N/A')
+        else:
+            sector = getattr(profile, 'sector', 'N/A')
+            industry = getattr(profile, 'industry', 'N/A')
+            market_cap = getattr(profile, 'market_cap', 'N/A')
+            pe_ratio = getattr(profile, 'pe_ratio', 'N/A')
+            employees = getattr(profile, 'employees', 'N/A')
+            country = getattr(profile, 'country', 'N/A')
+
+        # Format market cap and P/E for display
+        if market_cap != 'N/A':
+            market_cap = safe_float(market_cap, 0)
+            if market_cap > 1e9:
+                market_cap = f"${market_cap/1e9:.1f}B"
+            elif market_cap > 1e6:
+                market_cap = f"${market_cap/1e6:.1f}M"
+            else:
+                market_cap = f"${market_cap:.0f}"
+
+        lines.append(
+            f"{ticker}\t"
+            f"{sector}\t"
+            f"{industry}\t"
+            f"{market_cap}\t"
+            f"{safe_float(pe_ratio, 'N/A')}\t"
+            f"{employees}\t"
+            f"{country}"
+        )
+    return "\n".join(lines)
+
+
+def format_market_data(market_data: Dict[str, Any]) -> str:
+    """Format real-time market data as compact table.
+
+    Args:
+        market_data: Dictionary of quote data keyed by ticker
+
+    Returns:
+        Tab-separated table with price, change, and key metrics
+    """
+    if not market_data:
+        return "No market data available"
+
+    lines = ["Ticker\tPrice\tChange%\tVolume\tP/E\tDiv Yield\tMkt Cap"]
+    for ticker, quote in market_data.items():
+        # Handle both dict and object attribute access
+        if isinstance(quote, dict):
+            price = quote.get('price', 0)
+            prev_close = quote.get('previous_close', price)
+            volume = quote.get('volume', 'N/A')
+            pe_ratio = quote.get('pe_ratio', 'N/A')
+            dividend_yield = quote.get('dividend_yield', 'N/A')
+            market_cap = quote.get('market_cap', 'N/A')
+        else:
+            price = getattr(quote, 'price', 0)
+            prev_close = getattr(quote, 'previous_close', price)
+            volume = getattr(quote, 'volume', 'N/A')
+            pe_ratio = getattr(quote, 'pe_ratio', 'N/A')
+            dividend_yield = getattr(quote, 'dividend_yield', 'N/A')
+            market_cap = getattr(quote, 'market_cap', 'N/A')
+
+        # Calculate change percentage
+        price_float = safe_float(price, 0)
+        prev_close_float = safe_float(prev_close, price_float)
+
+        if prev_close_float > 0:
+            change_pct = ((price_float - prev_close_float) / prev_close_float * 100)
+        else:
+            change_pct = 0
+
+        # Format market cap
+        if market_cap != 'N/A':
+            market_cap_float = safe_float(market_cap, 0)
+            if market_cap_float > 1e9:
+                market_cap = f"${market_cap_float/1e9:.1f}B"
+            elif market_cap_float > 1e6:
+                market_cap = f"${market_cap_float/1e6:.1f}M"
+            else:
+                market_cap = f"${market_cap_float:.0f}"
+
+        # Format dividend yield
+        if dividend_yield != 'N/A':
+            dividend_yield = f"{safe_float(dividend_yield, 0):.2f}%"
+
+        lines.append(
+            f"{ticker}\t"
+            f"${price_float:.2f}\t"
+            f"{change_pct:+.2f}%\t"
+            f"{volume}\t"
+            f"{safe_float(pe_ratio, 'N/A')}\t"
+            f"{dividend_yield}\t"
+            f"{market_cap}"
+        )
+    return "\n".join(lines)
+
+
+def format_ml_forecasts(forecasts: Dict[str, Any]) -> str:
+    """Format ML ensemble forecasts as compact summary table.
+
+    Args:
+        forecasts: Dictionary of forecast results keyed by ticker
+
+    Returns:
+        Tab-separated table with 30-day predictions and confidence
+    """
+    if not forecasts:
+        return "No forecasts available"
+
+    lines = ["Ticker\t30d Return\tConfidence\tTrend\tModels"]
+    for ticker, forecast in forecasts.items():
+        # Handle both dict and object attribute access
+        if isinstance(forecast, dict):
+            predictions = forecast.get('predictions', [])
+            lower = forecast.get('lower_bound', [])
+            upper = forecast.get('upper_bound', [])
+            models = forecast.get('models_used', [])
+        else:
+            predictions = getattr(forecast, 'predictions', [])
+            lower = getattr(forecast, 'lower_bound', [])
+            upper = getattr(forecast, 'upper_bound', [])
+            models = getattr(forecast, 'models_used', [])
+
+        if not predictions or len(predictions) == 0:
+            continue
+
+        # Calculate return from first to last prediction
+        start_price = safe_float(predictions[0], 0)
+        end_price = safe_float(predictions[-1], start_price)
+
+        if start_price > 0:
+            return_pct = ((end_price - start_price) / start_price * 100)
+
+            # Calculate confidence range width
+            if lower and upper and len(lower) > 0 and len(upper) > 0:
+                lower_val = safe_float(lower[-1], 0)
+                upper_val = safe_float(upper[-1], 0)
+                range_width = ((upper_val - lower_val) / start_price * 100)
+            else:
+                range_width = 0
+
+            # Determine trend signal
+            if return_pct > 2:
+                trend = "Bullish"
+            elif return_pct < -2:
+                trend = "Bearish"
+            else:
+                trend = "Neutral"
+
+            # Format models list
+            models_str = f"{len(models)} models" if models else "N/A"
+
+            lines.append(
+                f"{ticker}\t"
+                f"{return_pct:+.1f}%\t"
+                f"±{abs(range_width):.1f}%\t"
+                f"{trend}\t"
+                f"{models_str}"
+            )
+
+    return "\n".join(lines)
+
+
+def format_sentiment_data(sentiment_data: Dict[str, Any]) -> str:
+    """Format news sentiment analysis as compact summary table.
+
+    Args:
+        sentiment_data: Dictionary of sentiment results keyed by ticker
+
+    Returns:
+        Tab-separated table with sentiment scores and top headlines
+    """
+    if not sentiment_data:
+        return "No sentiment data available"
+
+    lines = ["Ticker\tSentiment\tConfidence\tArticles\tTop Headlines"]
+    for ticker, data in sentiment_data.items():
+        # Handle both dict and object attribute access
+        if isinstance(data, dict):
+            sentiment = data.get('overall_sentiment', 0.0)
+            confidence = data.get('confidence', 0.0)
+            article_count = data.get('article_count', 0)
+            articles = data.get('articles', [])
+            error = data.get('error')
+        else:
+            sentiment = getattr(data, 'overall_sentiment', 0.0)
+            confidence = getattr(data, 'confidence', 0.0)
+            article_count = getattr(data, 'article_count', 0)
+            articles = getattr(data, 'articles', [])
+            error = getattr(data, 'error', None)
+
+        # Skip if there was an error
+        if error:
+            lines.append(f"{ticker}\tError\t-\t-\t{error[:50]}")
+            continue
+
+        # Determine sentiment label
+        sentiment_float = safe_float(sentiment, 0.0)
+        if sentiment_float >= 0.05:
+            label = f"Positive ({sentiment_float:+.2f})"
+        elif sentiment_float <= -0.05:
+            label = f"Negative ({sentiment_float:+.2f})"
+        else:
+            label = f"Neutral ({sentiment_float:+.2f})"
+
+        # Get top 2 headlines (truncated)
+        headlines = []
+        for article in articles[:2]:
+            if isinstance(article, dict):
+                headline = article.get('headline', '')
+            else:
+                headline = getattr(article, 'headline', '')
+
+            if headline:
+                # Truncate long headlines
+                headlines.append(headline[:50] + '...' if len(headline) > 50 else headline)
+
+        headlines_str = " | ".join(headlines) if headlines else "No recent news"
+
+        lines.append(
+            f"{ticker}\t"
+            f"{label}\t"
+            f"{safe_float(confidence, 0):.0%}\t"
+            f"{article_count}\t"
+            f"{headlines_str}"
+        )
+
+    return "\n".join(lines)
+
+
+def format_technical_indicators(technical_indicators: Dict[str, Any]) -> str:
+    """Format technical indicators as compact hybrid table.
+
+    Combines tabular layout for key metrics with inline summaries.
+    Following Anthropic best practices: flat structure with semantic signals.
+
+    Args:
+        technical_indicators: Dict of TechnicalIndicators keyed by ticker
+
+    Returns:
+        Tab-separated table with signals and key metrics (~60-80 tokens per ticker)
+    """
+    if not technical_indicators:
+        return "No technical indicators available"
+
+    lines = ["Ticker\tSignal\tRSI\tMACD\tBB Pos\tMA Trend\tVolume"]
+
+    for ticker, indicators in technical_indicators.items():
+        # Handle both dict and object attribute access
+        if isinstance(indicators, dict):
+            overall_signal = indicators.get('overall_signal', 'hold')
+            rsi = indicators.get('rsi', {})
+            macd = indicators.get('macd', {})
+            bb = indicators.get('bollinger_bands', {})
+            ma = indicators.get('moving_averages', {})
+            volume_trend = indicators.get('volume_trend', 'N/A')
+        else:
+            overall_signal = getattr(indicators, 'overall_signal', 'hold')
+            rsi = getattr(indicators, 'rsi', None)
+            macd = getattr(indicators, 'macd', None)
+            bb = getattr(indicators, 'bollinger_bands', None)
+            ma = getattr(indicators, 'moving_averages', None)
+            volume_trend = getattr(indicators, 'volume_trend', 'N/A')
+
+        # Extract RSI value and signal
+        if isinstance(rsi, dict):
+            rsi_value = safe_float(rsi.get('value'), 0)
+            rsi_signal = rsi.get('signal', 'N/A')
+        elif rsi:
+            rsi_value = safe_float(getattr(rsi, 'value', 0), 0)
+            rsi_signal = getattr(rsi, 'signal', 'N/A')
+        else:
+            rsi_value = 0
+            rsi_signal = 'N/A'
+
+        # MACD trend
+        if isinstance(macd, dict):
+            macd_trend = macd.get('trend', 'N/A')
+            macd_hist = safe_float(macd.get('histogram'), 0)
+        elif macd:
+            macd_trend = getattr(macd, 'trend', 'N/A')
+            macd_hist = safe_float(getattr(macd, 'histogram', 0), 0)
+        else:
+            macd_trend = 'N/A'
+            macd_hist = 0
+
+        # Bollinger Band position
+        if isinstance(bb, dict):
+            bb_position = bb.get('position', 'N/A')
+        elif bb:
+            bb_position = getattr(bb, 'position', 'N/A')
+        else:
+            bb_position = 'N/A'
+
+        # Moving average trend
+        if isinstance(ma, dict):
+            ma_trend = ma.get('trend', 'N/A')
+        elif ma:
+            ma_trend = getattr(ma, 'trend', 'N/A')
+        else:
+            ma_trend = 'N/A'
+
+        # Format compact RSI with signal indicator
+        rsi_compact = f"{rsi_value:.0f}"
+        if rsi_signal == 'overbought':
+            rsi_compact += "↑"
+        elif rsi_signal == 'oversold':
+            rsi_compact += "↓"
+
+        # Format MACD with trend indicator
+        macd_compact = f"{macd_hist:+.1f}"
+        if macd_trend == 'bullish':
+            macd_compact += "↑"
+        elif macd_trend == 'bearish':
+            macd_compact += "↓"
+
+        lines.append(
+            f"{ticker}\t"
+            f"{overall_signal.upper()}\t"
+            f"{rsi_compact}\t"
+            f"{macd_compact}\t"
+            f"{bb_position}\t"
+            f"{ma_trend}\t"
+            f"{volume_trend}"
+        )
+
+    return "\n".join(lines)
+
+
+def format_portfolio_data(portfolio_data: Dict[str, Any]) -> str:
+    """Format portfolio holdings as compact table.
+
+    Extracts only essential information and formats numerics for readability.
+    Omits redundant zero-value fields and internal IDs.
+
+    Args:
+        portfolio_data: Dict with holdings, portfolio_id, risk_tolerance
+
+    Returns:
+        Formatted portfolio summary with holdings table
+    """
+    if not portfolio_data:
+        return "No portfolio data available"
+
+    holdings = portfolio_data.get('holdings', [])
+    risk_tolerance = portfolio_data.get('risk_tolerance', 'moderate')
+
+    if not holdings:
+        return f"Empty portfolio (Risk Tolerance: {risk_tolerance})"
+
+    # Calculate total portfolio value
+    total_value = sum(h.get('market_value', 0) for h in holdings)
+
+    lines = [
+        f"Portfolio Summary (Risk Tolerance: {risk_tolerance}, Total Value: ${total_value:,.2f})",
+        "",
+        "Ticker\tShares\tPrice\tValue\tWeight\tCost Basis"
+    ]
+
+    for holding in holdings:
+        ticker = holding.get('ticker', 'N/A')
+        quantity = safe_float(holding.get('quantity'), 0)
+        current_price = safe_float(holding.get('current_price'), 0)
+        market_value = safe_float(holding.get('market_value'), 0)
+        weight = safe_float(holding.get('weight'), 0)
+        cost_basis = safe_float(holding.get('cost_basis'), 0)
+
+        # Format shares (hide if dollar-based entry)
+        shares_str = f"{quantity:.2f}" if quantity > 0 else "-"
+
+        # Format cost basis (hide if not set)
+        cost_basis_str = f"${cost_basis:.2f}" if cost_basis > 0 else "-"
+
+        lines.append(
+            f"{ticker}\t"
+            f"{shares_str}\t"
+            f"${current_price:.2f}\t"
+            f"${market_value:,.2f}\t"
+            f"{weight*100:.1f}%\t"
+            f"{cost_basis_str}"
+        )
+
+    return "\n".join(lines)
+
+
 class PortfolioAnalysisOutput(BaseModel):
     """Structured output from portfolio analysis."""
 
@@ -281,6 +686,7 @@ class PortfolioAnalystAgent(BasePortfolioAgent[PortfolioAnalysisOutput]):
         optimization_results: Dict[str, Any],
         risk_analysis: Dict[str, Any],
         ensemble_forecasts: Optional[Dict[str, Any]] = None,
+        sentiment_data: Optional[Dict[str, Any]] = None,
         risk_tolerance: str = "moderate",
     ) -> "AgentResult[PortfolioAnalysisOutput]":
         """Analyze a complete portfolio with all available data.
@@ -294,34 +700,36 @@ class PortfolioAnalystAgent(BasePortfolioAgent[PortfolioAnalysisOutput]):
             optimization_results: Portfolio optimization outputs
             risk_analysis: VaR, CVaR, and risk metrics
             ensemble_forecasts: ML-based price forecasts (Chronos + statistical models)
+            sentiment_data: News sentiment analysis for each ticker
             risk_tolerance: Investor's risk tolerance
 
         Returns:
             Structured portfolio analysis
         """
-        # Construct comprehensive analysis prompt
+        # Format all data sections for token efficiency
+        portfolio_table = format_portfolio_data(portfolio_data)
+        market_table = format_market_data(market_data)
+        fundamentals_table = format_fundamentals(fundamentals)
+        technical_table = format_technical_indicators(technical_indicators)
         economic_summary = format_economic_summary(economic_data)
-        risk_xml = format_risk_analysis_xml(risk_analysis)
         optimisation_xml = format_optimisation_results_xml(optimization_results)
-
-        # Format ensemble forecasts if available
-        forecasts_section = ""
-        if ensemble_forecasts:
-            forecasts_section = f"\n\nML FORECASTS (30-day predictions):\n{ensemble_forecasts}"
+        risk_xml = format_risk_analysis_xml(risk_analysis)
+        forecasts_table = format_ml_forecasts(ensemble_forecasts) if ensemble_forecasts else "No forecasts available"
+        sentiment_table = format_sentiment_data(sentiment_data) if sentiment_data else "No sentiment data available"
 
         prompt = f"""Analyze this investment portfolio:
 
 PORTFOLIO:
-{portfolio_data}
+{portfolio_table}
 
-MARKET DATA:
-{market_data}
+MARKET DATA (Real-time Quotes):
+{market_table}
 
 FUNDAMENTALS:
-{fundamentals}
+{fundamentals_table}
 
 TECHNICAL INDICATORS:
-{technical_indicators}
+{technical_table}
 
 ECONOMIC CONTEXT:
 {economic_summary}
@@ -330,7 +738,13 @@ OPTIMIZATION ANALYSIS:
 {optimisation_xml}
 
 RISK ANALYSIS:
-{risk_xml}{forecasts_section}
+{risk_xml}
+
+ML FORECASTS (30-day predictions):
+{forecasts_table}
+
+NEWS SENTIMENT (7-day analysis):
+{sentiment_table}
 
 INVESTOR RISK TOLERANCE: {risk_tolerance}
 

backend/agents/workflow.py

@@ -506,6 +506,7 @@ class PortfolioAnalysisWorkflow:
                 optimization_results=state.get("optimisation_results", {}),
                 risk_analysis=state.get("risk_analysis", {}),
                 ensemble_forecasts=state.get("ensemble_forecasts", {}),
+                sentiment_data=state.get("sentiment_data", {}),
                 risk_tolerance=state["risk_tolerance"],
             )
 

backend/caching/redis_cache.py

@@ -91,11 +91,11 @@ class TTLStrategy:
 
     # Base TTL values in seconds
     TTL_CONFIG: Dict[CacheDataType, int] = {
-        CacheDataType.MARKET_DATA: 300,  # 5 minutes
+        CacheDataType.MARKET_DATA: 60,  # 1 minute (reduced for real-time data)
         CacheDataType.PORTFOLIO_METRICS: 1800,  # 30 minutes
-        CacheDataType.ANALYSIS_RESULTS: 3600,  # 1 hour
-        CacheDataType.HISTORICAL_DATA: 86400,  # 24 hours
-        CacheDataType.MCP_RESPONSE: 600,  # 10 minutes
+        CacheDataType.ANALYSIS_RESULTS: 14400,  # 4 hours (extended from 1hr for cost savings)
+        CacheDataType.HISTORICAL_DATA: 43200,  # 12 hours (extended for older data)
+        CacheDataType.MCP_RESPONSE: 300,  # 5 minutes (extended from 10min)
         CacheDataType.USER_DATA: 7200,  # 2 hours
     }
 

backend/mcp_router.py

@@ -16,6 +16,10 @@ from backend.mcp_servers import yahoo_finance_mcp, fmp_mcp, trading_mcp, fred_mc
 from backend.mcp_servers import portfolio_optimizer_mcp, risk_analyzer_mcp, ensemble_predictor_mcp
 from backend.mcp_servers import news_sentiment_mcp
 
+# Import caching decorator
+from backend.caching.decorators import cached_async
+from backend.caching.redis_cache import CacheDataType
+
 logger = logging.getLogger(__name__)
 
 
@@ -51,6 +55,10 @@ class MCPRouter:
         logger.info(f"Initialised {len(self.servers)} MCP servers")
 
     # Yahoo Finance MCP methods
+    @cached_async(
+        namespace="yahoo_finance",
+        data_type=CacheDataType.MARKET_DATA,  # 60s TTL for real-time quotes
+    )
     async def call_yahoo_finance_mcp(self, tool: str, params: Dict[str, Any]) -> Dict[str, Any]:
         """Call Yahoo Finance MCP tool.
 
@@ -89,6 +97,11 @@ class MCPRouter:
         return result
 
     # FMP MCP methods
+    @cached_async(
+        namespace="fmp",
+        data_type=CacheDataType.HISTORICAL_DATA,  # 12 hours TTL
+        ttl=21600,  # Override to 6 hours for company fundamentals
+    )
     async def call_fmp_mcp(self, tool: str, params: Dict[str, Any]) -> Dict[str, Any]:
         """Call Financial Modeling Prep MCP tool.
 
@@ -126,6 +139,10 @@ class MCPRouter:
         return result
 
     # Trading MCP methods
+    @cached_async(
+        namespace="trading",
+        data_type=CacheDataType.HISTORICAL_DATA,  # 12 hours TTL for technical indicators
+    )
     async def call_trading_mcp(self, tool: str, params: Dict[str, Any]) -> Dict[str, Any]:
         """Call Trading MCP tool.
 
@@ -150,6 +167,11 @@ class MCPRouter:
         return result
 
     # FRED MCP methods
+    @cached_async(
+        namespace="fred",
+        data_type=CacheDataType.HISTORICAL_DATA,  # 12 hours default
+        ttl=86400,  # Override to 24 hours for economic data (changes infrequently)
+    )
     async def call_fred_mcp(self, tool: str, params: Dict[str, Any]) -> Dict[str, Any]:
         """Call FRED MCP tool.
 
@@ -174,6 +196,11 @@ class MCPRouter:
         return result
 
     # Portfolio Optimizer MCP methods
+    @cached_async(
+        namespace="portfolio_optimizer",
+        data_type=CacheDataType.PORTFOLIO_METRICS,  # 30 min default
+        ttl=14400,  # Override to 4 hours for optimization results (computational, deterministic)
+    )
     async def call_portfolio_optimizer_mcp(self, tool: str, params: Dict[str, Any]) -> Dict[str, Any]:
         """Call Portfolio Optimizer MCP tool.
 
@@ -209,6 +236,11 @@ class MCPRouter:
         return result
 
     # Risk Analyzer MCP methods
+    @cached_async(
+        namespace="risk_analyzer",
+        data_type=CacheDataType.PORTFOLIO_METRICS,  # 30 min default
+        ttl=14400,  # Override to 4 hours for risk analysis (computational, deterministic)
+    )
     async def call_risk_analyzer_mcp(self, tool: str, params: Dict[str, Any]) -> Dict[str, Any]:
         """Call Risk Analyzer MCP tool.
 
@@ -233,6 +265,11 @@ class MCPRouter:
         return result
 
     # Ensemble Predictor MCP methods
+    @cached_async(
+        namespace="ensemble_predictor",
+        data_type=CacheDataType.HISTORICAL_DATA,  # 12 hours default
+        ttl=21600,  # Override to 6 hours for ML forecasts (expensive computation)
+    )
     async def call_ensemble_predictor_mcp(self, tool: str, params: Dict[str, Any]) -> Dict[str, Any]:
         """Call Ensemble Predictor MCP tool.
 
@@ -256,6 +293,11 @@ class MCPRouter:
             return result.model_dump()
         return result
 
+    @cached_async(
+        namespace="news_sentiment",
+        data_type=CacheDataType.USER_DATA,  # 2 hours default
+        ttl=7200,  # 2 hours for news sentiment (balance freshness vs API costs)
+    )
     async def call_news_sentiment_mcp(self, tool: str, params: Dict[str, Any]) -> Dict[str, Any]:
         """Call News Sentiment MCP tool.