"""Comprehensive error handling for the trading analysis platform.""" from typing import Optional class TradingAnalysisError(Exception): """Base exception for all trading analysis errors.""" def __init__( self, message: str, details: Optional[str] = None, suggestion: Optional[str] = None, ): """ Initialize error with comprehensive information. Args: message: The main error message details: Additional technical details about the error suggestion: User-friendly suggestion on how to resolve the issue """ self.message = message self.details = details self.suggestion = suggestion super().__init__(self.format_error()) def format_error(self) -> str: """Format error message with all available information.""" parts = [f"āŒ {self.message}"] if self.details: parts.append(f"\nDetails: {self.details}") if self.suggestion: parts.append(f"\nšŸ’” Suggestion: {self.suggestion}") return "\n".join(parts) # Data Provider Errors class DataProviderError(TradingAnalysisError): """Base class for data provider errors.""" pass class TickerNotFoundError(DataProviderError): """Raised when a ticker symbol is not found.""" def __init__(self, ticker: str, provider: str = "data provider"): super().__init__( message=f"Ticker '{ticker}' not found", details=f"The ticker symbol '{ticker}' could not be found in {provider}", suggestion=( f"Please verify the ticker symbol is correct. " f"Examples: AAPL (stock), BTC-USD (crypto), GC=F (commodity), ^GSPC (index)" ), ) class NoDataReturnedError(DataProviderError): """Raised when no data is returned for a valid ticker.""" def __init__(self, ticker: str, timeframe: str, provider: str = "data provider"): super().__init__( message=f"No data available for {ticker}", details=f"No OHLC data returned for ticker '{ticker}' with timeframe '{timeframe}' from {provider}", suggestion=( f"This could mean: " f"1) The market is closed and no recent data is available, " f"2) The ticker doesn't support this timeframe, " f"3) Try a different timeframe (e.g., '1d' instead of '{timeframe}')" ), ) class RateLimitError(DataProviderError): """Raised when API rate limits are exceeded.""" def __init__(self, provider: str, wait_time: Optional[int] = None): wait_msg = f" Please wait {wait_time} seconds." if wait_time else "" super().__init__( message=f"Rate limit exceeded for {provider}", details=f"Too many requests to {provider} API.{wait_msg}", suggestion=( f"1) Wait a few minutes before trying again, " f"2) Consider upgrading to a premium API key, " f"3) Use a different data provider in Settings" ), ) class InsufficientDataError(DataProviderError): """Raised when there's insufficient data for analysis.""" def __init__(self, ticker: str, required: int, actual: int): super().__init__( message=f"Insufficient data for analysis of {ticker}", details=f"Need at least {required} data points for analysis, but only {actual} available", suggestion=( f"1) Try a longer time period (increase date range), " f"2) Use a higher timeframe (e.g., '1d' instead of '1m'), " f"3) Check if the asset has been trading long enough" ), ) class DataValidationError(DataProviderError): """Raised when data validation fails.""" def __init__(self, issue: str): super().__init__( message="Data validation failed", details=issue, suggestion="This might be a temporary data provider issue. Try again in a few moments.", ) # API and Configuration Errors class APIKeyError(TradingAnalysisError): """Raised when API keys are missing or invalid.""" def __init__(self, provider: str): super().__init__( message=f"Missing or invalid API key for {provider}", details=f"The {provider} API key is either not configured or invalid", suggestion=( f"1) Set your {provider} API key in the .env file or API Keys tab, " f"2) Verify the API key is valid and active, " f"3) Restart the application after setting the key" ), ) class ConfigurationError(TradingAnalysisError): """Raised when configuration is invalid.""" def __init__(self, parameter: str, issue: str): super().__init__( message=f"Invalid configuration for {parameter}", details=issue, suggestion=f"Check the Settings tab and verify {parameter} is configured correctly", ) # LLM Errors class LLMError(TradingAnalysisError): """Base class for LLM-related errors.""" pass class LLMProviderError(LLMError): """Raised when LLM provider encounters an error.""" def __init__(self, provider: str, model: str, error: str): super().__init__( message=f"{provider} API error", details=f"Model '{model}' returned error: {error}", suggestion=( f"1) Check your {provider} API key is valid and has credits, " f"2) Verify you have access to the '{model}' model, " f"3) Try a different model in Settings, " f"4) Check your internet connection" ), ) class LLMTimeoutError(LLMError): """Raised when LLM request times out.""" def __init__(self, agent: str, timeout: int): super().__init__( message=f"Agent '{agent}' timed out", details=f"LLM request exceeded timeout of {timeout} seconds", suggestion=( f"1) Try again - this might be temporary network congestion, " f"2) Check your internet connection, " f"3) Consider using a faster model in Settings" ), ) class LLMContextLimitError(LLMError): """Raised when context length is exceeded.""" def __init__(self, agent: str, tokens: int, limit: int): super().__init__( message=f"Context length exceeded for agent '{agent}'", details=f"Input requires {tokens} tokens but model limit is {limit}", suggestion=( f"1) Try analyzing a shorter time period, " f"2) Use a higher timeframe (fewer data points), " f"3) Use a model with larger context window" ), ) # Analysis Errors class AnalysisError(TradingAnalysisError): """Base class for analysis errors.""" pass class IndicatorCalculationError(AnalysisError): """Raised when indicator calculation fails.""" def __init__(self, indicator: str, reason: str): super().__init__( message=f"Failed to calculate {indicator}", details=reason, suggestion=( f"1) Ensure there's enough data for {indicator} calculation, " f"2) Try adjusting indicator parameters in Settings, " f"3) Check if the data quality is sufficient" ), ) class PatternRecognitionError(AnalysisError): """Raised when pattern recognition fails.""" def __init__(self, reason: str): super().__init__( message="Pattern recognition failed", details=reason, suggestion="This might be due to unusual market data or chart generation issues. Try again.", ) class ChartGenerationError(AnalysisError): """Raised when chart generation fails.""" def __init__(self, reason: str): super().__init__( message="Failed to generate chart", details=reason, suggestion=( "1) Ensure matplotlib and mplfinance are installed correctly, " "2) Check that the data directory is writable, " "3) Verify sufficient disk space is available" ), ) # Workflow Errors class WorkflowError(TradingAnalysisError): """Base class for workflow execution errors.""" pass class AgentExecutionError(WorkflowError): """Raised when an agent fails to execute.""" def __init__(self, agent: str, step: str, error: str): super().__init__( message=f"Agent '{agent}' failed", details=f"Error during {step}: {error}", suggestion=( f"1) Check the logs for more details, " f"2) Verify API keys are configured correctly, " f"3) Try running the analysis again" ), ) class WorkflowTimeoutError(WorkflowError): """Raised when workflow execution times out.""" def __init__(self, workflow: str, timeout: int): super().__init__( message=f"Workflow '{workflow}' timed out", details=f"Analysis exceeded maximum time of {timeout} seconds", suggestion=( f"1) Try using Technical Analysis instead of Comprehensive for faster results, " f"2) Check your internet connection, " f"3) The analysis was taking too long - try again" ), ) # Helper Functions def format_exception_for_user(e: Exception) -> str: """ Format any exception into a user-friendly error message. Args: e: The exception to format Returns: Formatted error message with helpful information """ if isinstance(e, TradingAnalysisError): return e.format_error() # Handle common Python exceptions error_type = type(e).__name__ error_msg = str(e) # Map common exceptions to user-friendly messages if isinstance(e, KeyError): return ( f"āŒ Missing required data field: {error_msg}\n" f"šŸ’” Suggestion: This might be due to incomplete data from the provider. Try again." ) elif isinstance(e, ValueError): return ( f"āŒ Invalid value: {error_msg}\n" f"šŸ’” Suggestion: Check your input parameters and try again." ) elif isinstance(e, ConnectionError): return ( f"āŒ Connection error: {error_msg}\n" f"šŸ’” Suggestion: Check your internet connection and try again." ) elif isinstance(e, TimeoutError): return ( f"āŒ Request timed out: {error_msg}\n" f"šŸ’” Suggestion: The service is taking too long to respond. Please try again." ) else: return ( f"āŒ Unexpected error ({error_type}): {error_msg}\n" f"šŸ’” Suggestion: Please try again. If the problem persists, check the logs for details." ) def wrap_provider_error( provider: str, ticker: str, operation: str, error: Exception ) -> TradingAnalysisError: """ Wrap provider exceptions into appropriate TradingAnalysisError subclasses. Args: provider: Name of the data provider ticker: Ticker symbol being accessed operation: Operation being performed (e.g., "fetch_ohlc", "fetch_fundamentals") error: The original exception Returns: Appropriate TradingAnalysisError subclass """ error_msg = str(error).lower() # Check for specific error patterns if "no data" in error_msg or "empty" in error_msg: return NoDataReturnedError(ticker, "unknown", provider) elif "rate limit" in error_msg or "too many requests" in error_msg: return RateLimitError(provider) elif ( "not found" in error_msg or "invalid ticker" in error_msg or "invalid symbol" in error_msg ): return TickerNotFoundError(ticker, provider) elif ( "api key" in error_msg or "unauthorized" in error_msg or "forbidden" in error_msg ): return APIKeyError(provider) else: return DataProviderError( message=f"{provider} error during {operation}", details=str(error), suggestion=( f"1) Verify the ticker '{ticker}' is correct, " f"2) Check your internet connection, " f"3) Try using a different data provider in Settings" ), )