perf: asynchronize and parallelize analysis tools to prevent main loop blocking

backend/core/pipeline.py

@@ -72,12 +72,10 @@ def _resolve_evo_agent_ids() -> set[str]:
 # Team infrastructure imports (graceful import - may not exist yet)
 try:
     from backend.agents.team.team_coordinator import TeamCoordinator
-    from backend.agents.team.msg_hub import MsgHub as TeamMsgHub
     TEAM_COORD_AVAILABLE = True
 except ImportError:
     TEAM_COORD_AVAILABLE = False
     TeamCoordinator = None
-    TeamMsgHub = None
 
 
 logger = logging.getLogger(__name__)

backend/tools/analysis_tools.py

@@ -103,18 +103,28 @@ def _safe_float(value, default=0.0) -> float:
 
 
 def safe(func):
-    """Decorator to catch exceptions in tool functions."""
+    """Decorator to catch exceptions in both sync and async tool functions."""
 
+    if asyncio.iscoroutinefunction(func):
         @wraps(func)
-    def wrapper(*args, **kwargs):
+        async def async_wrapper(*args, **kwargs):
+            try:
+                return await func(*args, **kwargs)
+            except Exception as e:
+                error_msg = f"Error in {func.__name__}: {str(e)}"
+                logger.error(f"{error_msg}\n{traceback.format_exc()}")
+                return _to_text_response(f"[ERROR] {error_msg}")
+        return async_wrapper
+    else:
+        @wraps(func)
+        def sync_wrapper(*args, **kwargs):
             try:
                 return func(*args, **kwargs)
             except Exception as e:
                 error_msg = f"Error in {func.__name__}: {str(e)}"
                 logger.error(f"{error_msg}\n{traceback.format_exc()}")
                 return _to_text_response(f"[ERROR] {error_msg}")
-
+        return sync_wrapper
-    return wrapper
 
 
 def _fmt(val, fmt=".2f", suffix="") -> str:
@@ -141,7 +151,7 @@ def _resolved_date(current_date: Optional[str]) -> str:
 
 
 @safe
-def analyze_efficiency_ratios(
+async def analyze_efficiency_ratios(
     tickers: Optional[List[str]] = None,
     current_date: Optional[str] = None,
 ) -> ToolResponse:
@@ -163,21 +173,26 @@ def analyze_efficiency_ratios(
     tickers = _parse_tickers(tickers)
     lines = [f"=== Efficiency Ratios Analysis ({current_date}) ===\n"]
 
-    for ticker in tickers:
+    async def _fetch_one(ticker):
-        metrics = get_financial_metrics(ticker=ticker, end_date=current_date)
+        try:
+            metrics = await asyncio.to_thread(get_financial_metrics, ticker=ticker, end_date=current_date)
             if not metrics:
-            lines.append(f"{ticker}: No data available\n")
+                return f"{ticker}: No data available\n"
-            continue
 
             m = metrics[0]
-        lines.append(f"{ticker}:")
+            ticker_lines = [
-        lines.append(f"  Asset Turnover: {_fmt(m.asset_turnover)}")
+                f"{ticker}:",
-        lines.append(f"  Inventory Turnover: {_fmt(m.inventory_turnover)}")
+                f"  Asset Turnover: {_fmt(m.asset_turnover)}",
-        lines.append(f"  Receivables Turnover: {_fmt(m.receivables_turnover)}")
+                f"  Inventory Turnover: {_fmt(m.inventory_turnover)}",
-        lines.append(
+                f"  Receivables Turnover: {_fmt(m.receivables_turnover)}",
-            f"  Working Capital Turnover: {_fmt(m.working_capital_turnover)}",
+                f"  Working Capital Turnover: {_fmt(m.working_capital_turnover)}\n",
-        )
+            ]
-        lines.append("")
+            return "\n".join(ticker_lines)
+        except Exception as e:
+            return f"{ticker}: Error - {str(e)}\n"
+
+    results = await asyncio.gather(*[_fetch_one(t) for t in tickers])
+    lines.extend(results)
 
     return _to_text_response("\n".join(lines))
 
@@ -310,7 +325,7 @@ def analyze_financial_health(
 
 
 @safe
-def analyze_valuation_ratios(
+async def analyze_valuation_ratios(
     tickers: Optional[List[str]] = None,
     current_date: Optional[str] = None,
 ) -> ToolResponse:
@@ -332,24 +347,31 @@ def analyze_valuation_ratios(
     tickers = _parse_tickers(tickers)
     lines = [f"=== Valuation Ratios Analysis ({current_date}) ===\n"]
 
-    for ticker in tickers:
+    async def _fetch_one(ticker):
-        metrics = get_financial_metrics(ticker=ticker, end_date=current_date)
+        try:
+            metrics = await asyncio.to_thread(get_financial_metrics, ticker=ticker, end_date=current_date)
             if not metrics:
-            lines.append(f"{ticker}: No data available\n")
+                return f"{ticker}: No data available\n"
-            continue
 
             m = metrics[0]
-        lines.append(f"{ticker}:")
+            ticker_lines = [
-        lines.append(f"  P/E Ratio: {_fmt(m.price_to_earnings_ratio)}")
+                f"{ticker}:",
-        lines.append(f"  P/B Ratio: {_fmt(m.price_to_book_ratio)}")
+                f"  P/E Ratio: {_fmt(m.price_to_earnings_ratio)}",
-        lines.append(f"  P/S Ratio: {_fmt(m.price_to_sales_ratio)}")
+                f"  P/B Ratio: {_fmt(m.price_to_book_ratio)}",
-        lines.append("")
+                f"  P/S Ratio: {_fmt(m.price_to_sales_ratio)}\n",
+            ]
+            return "\n".join(ticker_lines)
+        except Exception as e:
+            return f"{ticker}: Error - {str(e)}\n"
+
+    results = await asyncio.gather(*[_fetch_one(t) for t in tickers])
+    lines.extend(results)
 
     return _to_text_response("\n".join(lines))
 
 
 @safe
-def get_financial_metrics_tool(
+async def get_financial_metrics_tool(
     tickers: Optional[List[str]] = None,
     current_date: Optional[str] = None,
     period: str = "ttm",
@@ -374,35 +396,35 @@ def get_financial_metrics_tool(
         f"=== Comprehensive Financial Metrics ({current_date}, {period}) ===\n",
     ]
 
-    for ticker in tickers:
+    async def _fetch_one(ticker):
-        metrics = get_financial_metrics(
+        try:
+            # Offload synchronous data fetching to thread to keep loop snappy
+            metrics = await asyncio.to_thread(
+                get_financial_metrics,
                 ticker=ticker,
                 end_date=current_date,
                 period=period,
             )
             if not metrics:
-            lines.append(f"{ticker}: No data available\n")
+                return f"{ticker}: No data available\n"
-            continue
 
             m = metrics[0]
-        lines.append(f"{ticker}:")
+            ticker_lines = [
-        lines.append(f"  Market Cap: ${_fmt(m.market_cap, ',.0f')}")
+                f"{ticker}:",
-        lines.append(
+                f"  Market Cap: ${_fmt(m.market_cap, ',.0f')}",
                 f"  P/E: {_fmt(m.price_to_earnings_ratio)} | P/B: {_fmt(m.price_to_book_ratio)} | P/S: {_fmt(m.price_to_sales_ratio)}",
-        )
-        lines.append(
                 f"  ROE: {_fmt(m.return_on_equity, '.1%')} | Net Margin: {_fmt(m.net_margin, '.1%')}",
-        )
-        lines.append(
                 f"  Revenue Growth: {_fmt(m.revenue_growth, '.1%')} | Earnings Growth: {_fmt(m.earnings_growth, '.1%')}",
-        )
-        lines.append(
                 f"  Current Ratio: {_fmt(m.current_ratio)} | D/E: {_fmt(m.debt_to_equity)}",
-        )
+                f"  EPS: ${_fmt(m.earnings_per_share)} | FCF/Share: ${_fmt(m.free_cash_flow_per_share)}\n",
-        lines.append(
+            ]
-            f"  EPS: ${_fmt(m.earnings_per_share)} | FCF/Share: ${_fmt(m.free_cash_flow_per_share)}",
+            return "\n".join(ticker_lines)
-        )
+        except Exception as e:
-        lines.append("")
+            return f"{ticker}: Error fetching data - {str(e)}\n"
+
+    # Parallelize data retrieval for all tickers
+    results = await asyncio.gather(*[_fetch_one(t) for t in tickers])
+    lines.extend(results)
 
     return _to_text_response("\n".join(lines))