""" 基础错误处理和异常类 定义系统中使用的所有自定义异常 """ from typing import Optional, Dict, Any class StockAnalysisError(Exception): """基础异常类""" def __init__(self, message: str, details: Optional[Dict[str, Any]] = None): self.message = message self.details = details or {} super().__init__(self.message) class DataSourceError(StockAnalysisError): """数据源错误""" def __init__(self, message: str, data_source: Optional[str] = None, details: Optional[Dict[str, Any]] = None): self.data_source = data_source super().__init__(message, details) class AIAnalysisError(StockAnalysisError): """AI分析错误""" def __init__(self, message: str, model: Optional[str] = None, details: Optional[Dict[str, Any]] = None): self.model = model super().__init__(message, details) class ConfigurationError(StockAnalysisError): """配置错误""" def __init__(self, message: str, config_key: Optional[str] = None, details: Optional[Dict[str, Any]] = None): self.config_key = config_key super().__init__(message, details) class DatabaseError(StockAnalysisError): """数据库错误""" def __init__(self, message: str, operation: Optional[str] = None, details: Optional[Dict[str, Any]] = None): self.operation = operation super().__init__(message, details) class ValidationError(StockAnalysisError): """数据验证错误""" def __init__(self, message: str, field: Optional[str] = None, details: Optional[Dict[str, Any]] = None): self.field = field super().__init__(message, details) class APIError(StockAnalysisError): """API调用错误""" def __init__(self, message: str, status_code: Optional[int] = None, api_name: Optional[str] = None, details: Optional[Dict[str, Any]] = None): self.status_code = status_code self.api_name = api_name super().__init__(message, details) class ReportGenerationError(StockAnalysisError): """报告生成错误""" def __init__(self, message: str, module_type: Optional[str] = None, details: Optional[Dict[str, Any]] = None): self.module_type = module_type super().__init__(message, details) class SymbolNotFoundError(DataSourceError): """证券代码未找到错误""" def __init__(self, symbol: str, market: str, data_source: Optional[str] = None): message = f"证券代码 {symbol} 在市场 {market} 中未找到" details = {"symbol": symbol, "market": market} super().__init__(message, data_source, details) class RateLimitError(APIError): """API调用频率限制错误""" def __init__(self, api_name: str, retry_after: Optional[int] = None): message = f"API {api_name} 调用频率超限" details = {"retry_after": retry_after} if retry_after else {} super().__init__(message, 429, api_name, details) class AuthenticationError(APIError): """API认证错误""" def __init__(self, api_name: str, details: Optional[Dict[str, Any]] = None): message = f"API {api_name} 认证失败" super().__init__(message, 401, api_name, details)