Error Handling Mechanisms in Daily Stock Analysis: A 5-Layer Defense Strategy

The ZhuLinsen/daily_stock_analysis repository implements a layered error handling strategy combining FastAPI middleware, Pydantic schemas, custom exception hierarchies, tenacity-based retries, and structured logging to ensure resilient API operations.

This comprehensive guide examines the error handling mechanisms implemented across the daily_stock_analysis codebase. The repository employs a sophisticated multi-tiered approach that catches exceptions at the API boundary, enforces typed error responses, and implements domain-specific retry logic for network and database operations.

Centralized Exception Middleware

The foundation of the error handling strategy rests in api/middlewares/error_handler.py, where a global middleware intercepts all unhandled exceptions before they reach the client. The ErrorHandlerMiddleware class extends BaseHTTPMiddleware to wrap every request in a try-catch block, ensuring consistent JSON error responses regardless of where an exception originates.

# api/middlewares/error_handler.py

class ErrorHandlerMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        try:
            return await call_next(request)
        except Exception as e:
            logger.error(
                f"未处理的异常: {e}\n"
                f"请求路径: {request.url.path}\n"
                f"请求方法: {request.method}\n"
                f"堆栈: {traceback.format_exc()}"
            )
            return JSONResponse(
                status_code=500,
                content={
                    "error": "internal_error",
                    "message": "服务器内部错误，请稍后重试",
                    "detail": str(e) if logger.isEnabledFor(logging.DEBUG) else None,
                },
            )

The middleware registers via add_error_handlers(app), which adds specific handlers for HTTPException and validation errors. This ensures that every error—whether a validation failure or a catastrophic server crash—follows the same JSON format, preventing information leakage while maintaining debuggability.

Structured Error Response Schemas

All API responses conform to a strict Pydantic model defined in api/v1/schemas/common.py. The ErrorResponse base class enforces a machine-readable structure that clients can reliably parse, distinguishing between user-facing messages and internal debug details.

# api/v1/schemas/common.py

class ErrorResponse(BaseModel):
    error: str          # machine-readable code

    message: str        # short description

    detail: Optional[str] = None   # extra debug info (only when DEBUG)

Subclasses like SystemConfigValidationErrorResponse extend this base for specific domains while maintaining the core contract. The detail field is conditionally populated based on the logging level, ensuring that sensitive stack traces never expose internal implementation details in production environments.

Domain-Specific Exception Hierarchy

Rather than raising generic Exception objects, the codebase implements a rich hierarchy of custom exceptions that communicate specific failure modes. These domain exceptions live in their respective service modules and enable precise error handling logic.

Configuration Exceptions

Located in src/services/system_config_service.py, these handle system setup failures:

• ConfigValidationError: Raised when configuration files fail schema validation
• ConfigConflictError: Triggered when configuration updates create state conflicts
• ConfigImportError: Handles failures during external configuration imports

Data Provider Exceptions

Defined in src/data_provider/base.py, these manage external API interactions:

• DataFetchError: General wrapper for failed data retrieval operations
• RateLimitError: Specific handling for API throttling scenarios
• DataSourceUnavailableError: Indicates complete upstream service failures

Business Logic Exceptions

Service-layer exceptions enforce domain rules:

• DuplicateTaskError (src/services/task_queue.py): Prevents redundant task processing
• PortfolioConflictError (src/services/portfolio_service.py): Handles concurrent modification conflicts
• PortfolioOversellError: Enforces trading constraints when sell orders exceed holdings
• DuplicateTradeUidError (src/repositories/portfolio_repo.py): Ensures database-level trade uniqueness
• PortfolioBusyError: Signals when the portfolio is locked by another operation

Retry Strategies with Tenacity

Transient failures from network I/O or rate limiting are handled using the tenacity library. The @retry decorator implements exponential back-off with jitter, automatically retrying operations on whitelisted exception types.

In src/services/social_sentiment_service.py, external API calls are wrapped with configurable retry logic:

# src/services/social_sentiment_service.py

@retry(
    retry=retry_if_exception_type(_TRANSIENT_EXCEPTIONS),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
)
def _get_with_retry(url: str, *, headers: Dict[str, str], params: Optional[Dict[str, Any]] = None):
    """GET with retry on transient network errors."""
    resp = requests.get(url, headers=headers, params=params, timeout=10)
    resp.raise_for_status()
    return resp

Similar tenacity-based patterns appear in src/search_service.py and various data provider fetchers, ensuring that temporary network blips do not cascade into system failures.

Database Write Resilience

SQLite operations in src/storage.py implement custom retry logic specifically for sqlite3.OperationalError, which typically indicates database locks or busy states. The storage layer reads configuration values sqlite_write_retry_max and sqlite_write_retry_base_delay to determine retry behavior.

# src/storage.py (excerpt)

max_retries = self._sqlite_write_retry_max if self._is_sqlite_engine else 0
for attempt in range(max_retries + 1):
    try:
        # perform DB write

        break
    except sqlite3.OperationalError as exc:
        if attempt == max_retries:
            raise
        delay = self._sqlite_write_retry_base_delay * (2 ** attempt)
        time.sleep(delay)

This exponential back-off strategy prevents thundering herd problems when multiple processes contend for the SQLite database file.

Notification Sender Reliability

The Telegram notification system in src/notification_sender/telegram_sender.py implements custom retry logic respecting the Retry-After header for rate-limited requests. This approach combines fixed-delay and exponential back-off strategies.

# src/notification_sender/telegram_sender.py

def _send_message(...):
    for attempt in range(5):
        response = requests.post(...)
        if response.status_code == 429:          # rate limited

            retry_after = int(response.headers.get('Retry-After', 2 ** attempt))
            logger.warning(f"Telegram rate limited, retrying in {retry_after}s ...")
            time.sleep(retry_after)
            continue
        response.raise_for_status()
        return response

By parsing the upstream API's rate-limit headers, the system optimizes retry timing rather than blindly applying exponential back-off.

Logging and Observability

The error handling mechanisms integrate with a centralized logging configuration. The global middleware captures full tracebacks and request metadata (path, method, headers) for every unhandled exception. Service-layer code logs contextual information at appropriate severity levels before re-raising or handling exceptions.

The conditional inclusion of the detail field in error responses—only when logger.isEnabledFor(logging.DEBUG) returns true—ensures that production deployments remain secure while development environments retain full debugging capabilities.

Summary

The daily_stock_analysis repository demonstrates production-grade error handling through these key mechanisms:

• Global middleware in api/middlewares/error_handler.py ensures uniform error responses and prevents information leakage
• Typed Pydantic schemas enforce consistent API contracts for error responses
• Custom exception hierarchies across services enable precise, domain-specific error handling
• Tenacity decorators provide resilient retry logic for external network calls
• SQLite-specific retry loops handle database contention without external dependencies
• Rate-limit-aware retries for notification services respect upstream API constraints
• Structured logging captures full context for debugging while respecting production security

Frequently Asked Questions

What is the purpose of the ErrorHandlerMiddleware in the FastAPI application?

The ErrorHandlerMiddleware serves as the last line of defense for unhandled exceptions in the daily_stock_analysis API. It catches any exception that bubbles up through the request stack and converts it into a standardized JSON response containing error, message, and conditional detail fields. This ensures clients always receive predictable error payloads even when unexpected failures occur, while logging full tracebacks server-side for debugging.

How does the repository handle transient network failures when fetching stock data?

The codebase uses the tenacity library to wrap external HTTP calls with exponential back-off retry logic. Functions like _get_with_retry in src/services/social_sentiment_service.py are decorated with @retry configurations that specify stop_after_attempt(3) and wait_exponential parameters. This mechanism specifically targets transient exceptions—such as connection timeouts or temporary 5xx errors—automatically retrying requests before surfacing persistent failures to the calling code.

Why does the SQLite storage implementation use custom retry logic instead of tenacity?

The SQLite retry logic in src/storage.py handles sqlite3.OperationalError specifically, which occurs when the database file is locked by another process. Unlike network requests, SQLite contention requires tighter integration with the storage class's internal state. The custom loop respects configuration parameters sqlite_write_retry_max and sqlite_write_retry_base_delay, applying exponential delays directly through time.sleep() without the overhead of a decorator chain. This approach provides granular control over retry counts based on whether the engine is SQLite or another database backend.

What information is included in production error responses versus development environments?

Production error responses include only error (machine-readable code) and message (user-friendly description) fields to prevent information leakage. The detail field, which may contain stack traces or internal error specifics, is only populated when logger.isEnabledFor(logging.DEBUG) returns true. This distinction ensures that production deployments expose minimal attack surface while development instances provide verbose debugging information for troubleshooting.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}