When building automated backend systems like trading bots, algorithmic platforms, or financial scrapers, connection resilience is critical. A simple remote network hiccup or temporary API gateway timeout shouldn't completely crash your entire runtime script execution pipeline.
In this tutorial, we will build a production-grade Python command-line utility that extracts real-time stock, market index, and cryptocurrency metrics via SerpApi's Google Finance engine. We will explicitly implement custom exception routing and an exponential backoff retry loop to handle unstable network connections safely.
Prerequisites & Setup
To follow along with this implementation, ensure you have Python 3.x installed on your local operating system.
First, open your terminal environment and install the required external HTTP network connection module:
pip install requests
Next, ensure you register for a free developer access key by visiting https://serpapi.com/dashboard. To protect your private authentication credentials from being hardcoded inside your public script files, inject your key directly as a local session environment variable in your active terminal terminal panel:
Windows Command Prompt (cmd):
set SERPAPI_KEY=your_private_api_key_here
Mac / Linux Terminal:
export SERPAPI_KEY="your_private_api_key_here"
The Resilient Code Engine
Create a new file in your local workspace using Notepad++ or your preferred code editor, name it exactly serpapi_market_data.py, and paste the following fault-tolerant implementation structure:
import os
import sys
import time
from typing import Optional
import requests
SERPAPI_ENDPOINT = "https://serpapi.com"
DEFAULT_TIMEOUT = 10 # seconds
MAX_RETRIES = 3
RETRY_BACKOFF = 2 # seconds
class SerpApiError(Exception):
"""Custom exception routing configurations for SerpApi specific runtime errors."""
def get_api_key() -> str:
api_key = os.environ.get("SERPAPI_KEY")
if not api_key:
raise SerpApiError("No API key found. Ensure the SERPAPI_KEY environment variable is configured.")
return api_key
def fetch_quote(ticker: str, api_key: str) -> dict:
params = {
"engine": "google_finance",
"q": ticker,
"api_key": api_key,
}
last_error: Optional[Exception] = None
for attempt in range(1, MAX_RETRIES + 1):
try:
response = requests.get(params=params, url=SERPAPI_ENDPOINT, timeout=DEFAULT_TIMEOUT)
response.raise_for_status()
data = response.json()
if "error" in data:
raise SerpApiError(f"SerpApi dashboard error for {ticker}: {data['error']}")
return data
except (requests.RequestException, SerpApiError) as exc:
last_error = exc
if attempt < MAX_RETRIES:
sleep_time = RETRY_BACKOFF * attempt
print(f"[warn] {ticker}: Attempt {attempt} failed. Retrying execution in {sleep_time}s...")
time.sleep(sleep_time)
raise SerpApiError(f"Failed to fetch tracking data for {ticker} after {MAX_RETRIES} attempts. Engine error: {last_error}")
def parse_quote(data: dict, ticker: str) -> dict:
summary = data.get("summary", {})
if not summary:
raise SerpApiError(f"No summary data structures returned for {ticker}. Verify parameter definitions.")
return {
"ticker": ticker,
"title": summary.get("title"),
"price": summary.get("extracted_price") or summary.get("price"),
"currency": summary.get("currency"),
"change": summary.get("price_movement", {}).get("value"),
"change_percent": summary.get("price_movement", {}).get("percentage"),
"movement": summary.get("price_movement", {}).get("movement"),
}
def main():
tickers = sys.argv[1:] or ["AAPL:NASDAQ", "GOOGL:NASDAQ", "BTC-USD"]
try:
api_key = get_api_key()
except SerpApiError as fatal_err:
print(f"[fatal] {fatal_err}")
sys.exit(1)
print(f"\n{'Ticker':<15}{'Price':<12}{'Change':<12}{'% Change':<12}{'Movement':<10}")
print("-" * 61)
for ticker in tickers:
try:
raw = fetch_quote(ticker, api_key)
r = parse_quote(raw, ticker)
print(f"{r['ticker']:<15}{str(r['price']):<12}{str(r['change']):<12}{str(r['change_percent']):<12}{str(r['movement']):<10}")
except SerpApiError as exc:
print(f"{ticker:<15}ERROR BLOCK: {exc}")
if __name__ == "__main__":
main()
Execution & Grid Verification
Execute the operational script engine directly within your terminal window while supplying any target asset trackers as arguments:
python serpapi_market_data.py AAPL:NASDAQ GOOGL:NASDAQ BTC-USD
The application layer will automatically process the commands, query the server endpoints, run the mathematical exponential backoff algorithm if an unexpected network timeout happens, and print out a clean, structured asset data grid directly into your console pipeline.
Check out my working software tools and open-source backend layouts on my public portfolio link: (https://github.com/ssebinacharles/quant-bot-command-center).
Top comments (2)
The retry gate currently groups SerpApiError with transport failures, so an invalid key or invalid request is retried exactly like a timeout. Split retryable transport, 429, and 5xx failures from permanent 4xx and payload validation errors, then add a negative control where a 401 must stop after one attempt while a timeout reaches the cap.
thanks for following me