feat: integrate ensemble predictor into workflow Phase 2.5

Add ML-based price forecasting to portfolio analysis workflow:
- Add Phase 2.5 between computation and LLM synthesis
- Update MCP router with ensemble predictor integration
- Extend AgentState to include ensemble_forecasts field
- Update PortfolioAnalystAgent to consume ML forecasts
- Generate 30-day forecasts for all portfolio holdings
- Graceful degradation if forecasts fail (workflow continues)
- Pass forecast data to LLM for enhanced analysis

backend/agents/portfolio_analyst.py

@@ -280,6 +280,7 @@ class PortfolioAnalystAgent(BasePortfolioAgent[PortfolioAnalysisOutput]):
         economic_data: Dict[str, Any],
         optimization_results: Dict[str, Any],
         risk_analysis: Dict[str, Any],
+        ensemble_forecasts: Optional[Dict[str, Any]] = None,
         risk_tolerance: str = "moderate",
     ) -> "AgentResult[PortfolioAnalysisOutput]":
         """Analyze a complete portfolio with all available data.
@@ -292,6 +293,7 @@ class PortfolioAnalystAgent(BasePortfolioAgent[PortfolioAnalysisOutput]):
             economic_data: Macroeconomic indicators
             optimization_results: Portfolio optimization outputs
             risk_analysis: VaR, CVaR, and risk metrics
+            ensemble_forecasts: ML-based price forecasts (Chronos + statistical models)
             risk_tolerance: Investor's risk tolerance
 
         Returns:
@@ -302,6 +304,11 @@ class PortfolioAnalystAgent(BasePortfolioAgent[PortfolioAnalysisOutput]):
         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}"
+
         prompt = f"""Analyze this investment portfolio:
 
 PORTFOLIO:
@@ -323,7 +330,7 @@ OPTIMIZATION ANALYSIS:
 {optimisation_xml}
 
 RISK ANALYSIS:
-{risk_xml}
+{risk_xml}{forecasts_section}
 
 INVESTOR RISK TOLERANCE: {risk_tolerance}
 

backend/agents/workflow.py

@@ -1,8 +1,9 @@
 """LangGraph workflow for multi-agent portfolio analysis.
 
-This implements the three-phase architecture:
+This implements the multi-phase architecture:
 Phase 1: Data Layer MCPs (Yahoo Finance, FMP, Trading-MCP, FRED)
 Phase 2: Computation Layer MCPs (Portfolio Optimizer, Risk Analyzer)
+Phase 2.5: ML Predictions (Ensemble Predictor with Chronos)
 Phase 3: LLM Synthesis (Portfolio Analyst Agent)
 """
 
@@ -80,12 +81,14 @@ class PortfolioAnalysisWorkflow:
         # Add nodes for each phase
         workflow.add_node("phase_1_data_layer", self._phase_1_data_layer)
         workflow.add_node("phase_2_computation", self._phase_2_computation)
+        workflow.add_node("phase_2_5_ml_predictions", self._phase_2_5_ml_predictions)
         workflow.add_node("phase_3_synthesis", self._phase_3_synthesis)
 
         # Define the flow
         workflow.set_entry_point("phase_1_data_layer")
         workflow.add_edge("phase_1_data_layer", "phase_2_computation")
-        workflow.add_edge("phase_2_computation", "phase_3_synthesis")
+        workflow.add_edge("phase_2_computation", "phase_2_5_ml_predictions")
+        workflow.add_edge("phase_2_5_ml_predictions", "phase_3_synthesis")
         workflow.add_edge("phase_3_synthesis", END)
 
         return workflow.compile()
@@ -330,6 +333,79 @@ class PortfolioAnalysisWorkflow:
 
         return state
 
+    async def _phase_2_5_ml_predictions(self, state: AgentState) -> AgentState:
+        """Phase 2.5: Generate ML-based price forecasts using Ensemble Predictor.
+
+        MCP called:
+        - Ensemble Predictor: Chronos + statistical models for price forecasting
+        """
+        logger.info("PHASE 2.5: Generating ML predictions")
+        phase_start = time.perf_counter()
+
+        try:
+            # Generate forecasts for each holding
+            logger.debug("Running ensemble forecasts for portfolio holdings")
+            ensemble_forecasts = {}
+
+            for holding in state["holdings"]:
+                ticker = holding["ticker"]
+
+                # Get historical prices from Phase 1 data
+                hist_data = state["historical_prices"].get(ticker, {})
+                prices = hist_data.get("close_prices", [])
+
+                if not prices or len(prices) < 10:
+                    logger.warning(f"Insufficient price data for {ticker}, skipping forecast")
+                    continue
+
+                try:
+                    # Call ensemble predictor
+                    forecast_result = await self.mcp_router.call_ensemble_predictor_mcp(
+                        "forecast_ensemble",
+                        {
+                            "ticker": ticker,
+                            "prices": prices,
+                            "forecast_horizon": 30,  # 30-day forecast
+                            "confidence_level": 0.95,
+                            "use_returns": True,  # Forecast returns for stability
+                            "ensemble_method": "mean",  # Simple averaging
+                        }
+                    )
+
+                    ensemble_forecasts[ticker] = forecast_result
+                    logger.debug(f"Generated forecast for {ticker} using {len(forecast_result.get('models_used', []))} models")
+
+                except Exception as e:
+                    logger.warning(f"Forecast failed for {ticker}: {e}")
+                    continue
+
+            # Update state
+            state["ensemble_forecasts"] = ensemble_forecasts
+            state["current_step"] = "phase_2_5_complete"
+
+            # Log MCP calls
+            state["mcp_calls"].extend([
+                MCPCall.model_validate({
+                    "mcp": "ensemble_predictor",
+                    "tool": "forecast_ensemble"
+                }).model_dump(),
+            ])
+
+            # Track phase duration
+            phase_duration_ms = int((time.perf_counter() - phase_start) * 1000)
+
+            logger.info(
+                f"PHASE 2.5 COMPLETE: Generated forecasts for {len(ensemble_forecasts)} assets ({phase_duration_ms}ms)"
+            )
+
+        except Exception as e:
+            logger.error(f"Error in Phase 2.5: {e}")
+            state["errors"].append(f"Phase 2.5 error: {str(e)}")
+            # Set empty forecasts to allow workflow to continue
+            state["ensemble_forecasts"] = {}
+
+        return state
+
     async def _phase_3_synthesis(self, state: AgentState) -> AgentState:
         """Phase 3: LLM synthesis of all data into actionable insights."""
         logger.info("PHASE 3: LLM Synthesis")
@@ -352,6 +428,7 @@ class PortfolioAnalysisWorkflow:
                 economic_data=state.get("economic_data", {}),
                 optimization_results=state.get("optimisation_results", {}),
                 risk_analysis=state.get("risk_analysis", {}),
+                ensemble_forecasts=state.get("ensemble_forecasts", {}),
                 risk_tolerance=state["risk_tolerance"],
             )
 

backend/mcp_router.py

@@ -13,7 +13,7 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
 
 # Import all MCP servers
 from backend.mcp_servers import yahoo_finance_mcp, fmp_mcp, trading_mcp, fred_mcp
-from backend.mcp_servers import portfolio_optimizer_mcp, risk_analyzer_mcp
+from backend.mcp_servers import portfolio_optimizer_mcp, risk_analyzer_mcp, ensemble_predictor_mcp
 
 logger = logging.getLogger(__name__)
 
@@ -23,7 +23,7 @@ class MCPRouter:
 
     Manages connections to:
     - P0 (Week 1): Yahoo Finance, FMP, Trading-MCP, FRED, Portfolio Optimizer, Risk Analyzer
-    - P1 (Week 2): Ensemble Predictor (if time permits)
+    - P1 (Week 2): Ensemble Predictor (Chronos + statistical models)
     """
 
     def __init__(self):
@@ -43,6 +43,7 @@ class MCPRouter:
             "fred": fred_mcp,
             "portfolio_optimizer": portfolio_optimizer_mcp,
             "risk_analyzer": risk_analyzer_mcp,
+            "ensemble_predictor": ensemble_predictor_mcp,
         }
 
         logger.info(f"Initialised {len(self.servers)} MCP servers")
@@ -229,6 +230,30 @@ class MCPRouter:
             return result.model_dump()
         return result
 
+    # Ensemble Predictor MCP methods
+    async def call_ensemble_predictor_mcp(self, tool: str, params: Dict[str, Any]) -> Dict[str, Any]:
+        """Call Ensemble Predictor MCP tool.
+
+        Args:
+            tool: Tool name
+            params: Tool parameters
+
+        Returns:
+            Tool result
+        """
+        logger.debug(f"Calling Ensemble Predictor MCP: {tool}")
+
+        if tool == "forecast_ensemble":
+            from backend.mcp_servers.ensemble_predictor_mcp import forecast_ensemble, ForecastRequest
+            request = ForecastRequest(**params)
+            result = await forecast_ensemble.fn(request)
+        else:
+            raise ValueError(f"Unknown Ensemble Predictor tool: {tool}")
+
+        if hasattr(result, 'model_dump'):
+            return result.model_dump()
+        return result
+
     # High-level helper methods
     async def fetch_market_data(self, tickers: List[str]) -> Dict[str, Any]:
         """Fetch market data for given tickers.

backend/models/agent_state.py

@@ -51,6 +51,9 @@ class AgentState(TypedDict):
     optimisation_results: Annotated[Dict[str, Any], merge_dicts]
     risk_analysis: Annotated[Dict[str, Any], merge_dicts]
 
+    # Phase 2.5: ML Predictions (P1)
+    ensemble_forecasts: Annotated[Dict[str, Any], merge_dicts]
+
     # Phase 3: LLM Synthesis
     ai_synthesis: str
     recommendations: List[str]