from typing import Annotated # Import from vendor-specific modules from .y_finance import ( get_YFin_data_online, get_stock_stats_indicators_window, get_fundamentals as get_yfinance_fundamentals, get_balance_sheet as get_yfinance_balance_sheet, get_cashflow as get_yfinance_cashflow, get_income_statement as get_yfinance_income_statement, get_insider_transactions as get_yfinance_insider_transactions, ) from .yfinance_news import get_news_yfinance, get_global_news_yfinance from .alpha_vantage import ( get_stock as get_alpha_vantage_stock, get_indicator as get_alpha_vantage_indicator, get_fundamentals as get_alpha_vantage_fundamentals, get_balance_sheet as get_alpha_vantage_balance_sheet, get_cashflow as get_alpha_vantage_cashflow, get_income_statement as get_alpha_vantage_income_statement, get_insider_transactions as get_alpha_vantage_insider_transactions, get_news as get_alpha_vantage_news, get_global_news as get_alpha_vantage_global_news, ) from .alpha_vantage_common import AlphaVantageRateLimitError # Configuration and routing logic from .config import get_config # Tools organized by category TOOLS_CATEGORIES = { "core_stock_apis": { "description": "OHLCV stock price data", "tools": [ "get_stock_data" ] }, "technical_indicators": { "description": "Technical analysis indicators", "tools": [ "get_indicators" ] }, "fundamental_data": { "description": "Company fundamentals", "tools": [ "get_fundamentals", "get_balance_sheet", "get_cashflow", "get_income_statement" ] }, "news_data": { "description": "News and insider data", "tools": [ "get_news", "get_global_news", "get_insider_transactions", ] } } VENDOR_LIST = [ "yfinance", "alpha_vantage", ] # Mapping of methods to their vendor-specific implementations VENDOR_METHODS = { # core_stock_apis "get_stock_data": { "alpha_vantage": get_alpha_vantage_stock, "yfinance": get_YFin_data_online, }, # technical_indicators "get_indicators": { "alpha_vantage": get_alpha_vantage_indicator, "yfinance": get_stock_stats_indicators_window, }, # fundamental_data "get_fundamentals": { "alpha_vantage": get_alpha_vantage_fundamentals, "yfinance": get_yfinance_fundamentals, }, "get_balance_sheet": { "alpha_vantage": get_alpha_vantage_balance_sheet, "yfinance": get_yfinance_balance_sheet, }, "get_cashflow": { "alpha_vantage": get_alpha_vantage_cashflow, "yfinance": get_yfinance_cashflow, }, "get_income_statement": { "alpha_vantage": get_alpha_vantage_income_statement, "yfinance": get_yfinance_income_statement, }, # news_data "get_news": { "alpha_vantage": get_alpha_vantage_news, "yfinance": get_news_yfinance, }, "get_global_news": { "yfinance": get_global_news_yfinance, "alpha_vantage": get_alpha_vantage_global_news, }, "get_insider_transactions": { "alpha_vantage": get_alpha_vantage_insider_transactions, "yfinance": get_yfinance_insider_transactions, }, } def get_category_for_method(method: str) -> str: """Get the category that contains the specified method.""" for category, info in TOOLS_CATEGORIES.items(): if method in info["tools"]: return category raise ValueError(f"Method '{method}' not found in any category") def get_vendor(category: str, method: str = None) -> str: """Get the configured vendor for a data category or specific tool method. Tool-level configuration takes precedence over category-level. """ config = get_config() # Check tool-level configuration first (if method provided) if method: tool_vendors = config.get("tool_vendors", {}) if method in tool_vendors: return tool_vendors[method] # Fall back to category-level configuration return config.get("data_vendors", {}).get(category, "default") def route_to_vendor(method: str, *args, **kwargs): """Route method calls to appropriate vendor implementation with fallback support.""" category = get_category_for_method(method) vendor_config = get_vendor(category, method) # Handle comma-separated vendors primary_vendors = [v.strip() for v in vendor_config.split(',')] if method not in VENDOR_METHODS: raise ValueError(f"Method '{method}' not supported") # Get all available vendors for this method for fallback all_available_vendors = list(VENDOR_METHODS[method].keys()) # Create fallback vendor list: primary vendors first, then remaining vendors as fallbacks fallback_vendors = primary_vendors.copy() for vendor in all_available_vendors: if vendor not in fallback_vendors: fallback_vendors.append(vendor) # Debug: Print fallback ordering primary_str = " → ".join(primary_vendors) fallback_str = " → ".join(fallback_vendors) print(f"DEBUG: {method} - Primary: [{primary_str}] | Full fallback order: [{fallback_str}]") # Track results and execution state results = [] vendor_attempt_count = 0 any_primary_vendor_attempted = False successful_vendor = None for vendor in fallback_vendors: if vendor not in VENDOR_METHODS[method]: if vendor in primary_vendors: print(f"INFO: Vendor '{vendor}' not supported for method '{method}', falling back to next vendor") continue vendor_impl = VENDOR_METHODS[method][vendor] is_primary_vendor = vendor in primary_vendors vendor_attempt_count += 1 # Track if we attempted any primary vendor if is_primary_vendor: any_primary_vendor_attempted = True # Debug: Print current attempt vendor_type = "PRIMARY" if is_primary_vendor else "FALLBACK" print(f"DEBUG: Attempting {vendor_type} vendor '{vendor}' for {method} (attempt #{vendor_attempt_count})") # Handle list of methods for a vendor if isinstance(vendor_impl, list): vendor_methods = [(impl, vendor) for impl in vendor_impl] print(f"DEBUG: Vendor '{vendor}' has multiple implementations: {len(vendor_methods)} functions") else: vendor_methods = [(vendor_impl, vendor)] # Run methods for this vendor vendor_results = [] for impl_func, vendor_name in vendor_methods: try: print(f"DEBUG: Calling {impl_func.__name__} from vendor '{vendor_name}'...") result = impl_func(*args, **kwargs) vendor_results.append(result) print(f"SUCCESS: {impl_func.__name__} from vendor '{vendor_name}' completed successfully") except AlphaVantageRateLimitError as e: if vendor == "alpha_vantage": print(f"RATE_LIMIT: Alpha Vantage rate limit exceeded, falling back to next available vendor") print(f"DEBUG: Rate limit details: {e}") # Continue to next vendor for fallback continue except Exception as e: # Log error but continue with other implementations print(f"FAILED: {impl_func.__name__} from vendor '{vendor_name}' failed: {e}") continue # Add this vendor's results if vendor_results: results.extend(vendor_results) successful_vendor = vendor result_summary = f"Got {len(vendor_results)} result(s)" print(f"SUCCESS: Vendor '{vendor}' succeeded - {result_summary}") # Stopping logic: Stop after first successful vendor for single-vendor configs # Multiple vendor configs (comma-separated) may want to collect from multiple sources if len(primary_vendors) == 1: print(f"DEBUG: Stopping after successful vendor '{vendor}' (single-vendor config)") break else: print(f"FAILED: Vendor '{vendor}' produced no results") # Final result summary if not results: print(f"FAILURE: All {vendor_attempt_count} vendor attempts failed for method '{method}'") raise RuntimeError(f"All vendor implementations failed for method '{method}'") else: print(f"FINAL: Method '{method}' completed with {len(results)} result(s) from {vendor_attempt_count} vendor attempt(s)") # Return single result if only one, otherwise concatenate as string if len(results) == 1: return results[0] else: # Convert all results to strings and concatenate return '\n'.join(str(result) for result in results)