Stop Querying Google Every Time. Here's a SERP Caching Strategy That Cut My API Cost by 73%

#devops #python #tutorial #webdev

If you're running any kind of rank tracker, SEO tool, or search-data pipeline, there's a good chance you're making redundant API calls you don't need to.

I was querying the same keywords daily — sometimes multiple times a day — and paying for it every single time. Same query. Same result. Full price.

After some embarrassingly basic analysis, I found 73% of my SERP queries returned results that hadn't meaningfully changed since the last check.

Here's the caching layer I built, and the logic for deciding what's worth re-querying vs. what's fine to serve from cache.

The Core Problem: Not All Keywords Change at the Same Rate

Brand keywords for stable companies: changes maybe once a month. "Breaking news" queries: can change every 5 minutes. Long-tail informational keywords: changes maybe every 2-4 weeks. Transactional/price-sensitive keywords: changes every 1-7 days.

If you're checking everything on the same schedule, you're massively over-querying stable keywords and potentially under-querying volatile ones.

Step 1: Classify your keywords by volatility

import hashlib
import json
import redis
from datetime import timedelta
from enum import Enum

class KeywordVolatility(Enum):
    REALTIME = "realtime"      # News, breaking events: TTL 15 min
    HIGH = "high"              # Trending topics, prices: TTL 4 hours  
    MEDIUM = "medium"          # Standard commercial: TTL 24 hours
    LOW = "low"                # Brand/evergreen: TTL 72 hours

VOLATILITY_TTL = {
    KeywordVolatility.REALTIME: timedelta(minutes=15),
    KeywordVolatility.HIGH: timedelta(hours=4),
    KeywordVolatility.MEDIUM: timedelta(hours=24),
    KeywordVolatility.LOW: timedelta(hours=72),
}

def classify_keyword_volatility(keyword: str) -> KeywordVolatility:
    """
    Classify keyword by expected SERP change rate.
    Crude heuristic — refine with your own data over time.
    """
    keyword_lower = keyword.lower()

    realtime_signals = ["news", "today", "now", "live", "breaking", "latest"]
    high_signals = ["price", "cost", "deal", "vs", "2026", "sale", "discount"]
    low_signals = ["what is", "how to", "guide", "tutorial", "definition"]

    if any(s in keyword_lower for s in realtime_signals):
        return KeywordVolatility.REALTIME
    if any(s in keyword_lower for s in high_signals):
        return KeywordVolatility.HIGH
    if any(s in keyword_lower for s in low_signals):
        return KeywordVolatility.LOW

    return KeywordVolatility.MEDIUM  # Default

Step 2: Build the caching wrapper

import requests

class CachedSERPClient:
    def __init__(self, api_key: str, redis_client: redis.Redis):
        self.api_key = api_key
        self.redis = redis_client
        self.base_url = "https://apiserpent.com/api/search"

    def _cache_key(self, query: str, engine: str) -> str:
        raw = f"{query}:{engine}".lower().strip()
        return f"serp:{hashlib.md5(raw.encode()).hexdigest()}"

    def search(
        self,
        query: str,
        engine: str = "google",
        volatility: KeywordVolatility = None,
        force_refresh: bool = False
    ) -> dict:
        # Auto-classify if not provided
        if volatility is None:
            volatility = classify_keyword_volatility(query)

        cache_key = self._cache_key(query, engine)
        ttl = VOLATILITY_TTL[volatility]

        # Check cache first (unless force refresh)
        if not force_refresh:
            cached = self.redis.get(cache_key)
            if cached:
                result = json.loads(cached)
                result["_cache_hit"] = True
                result["_volatility"] = volatility.value
                return result

        # Cache miss — query the API
        response = requests.get(
            self.base_url,
            params={"q": query, "engine": engine, "num": 10},
            headers={"X-API-Key": self.api_key},
            timeout=10
        )
        response.raise_for_status()
        data = response.json()

        # Store in cache
        data["_cache_hit"] = False
        data["_volatility"] = volatility.value
        self.redis.setex(cache_key, int(ttl.total_seconds()), json.dumps(data))

        return data

Step 3: Track your cache hit rate to measure savings

class SERPMetrics:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def record_query(self, cache_hit: bool, volatility: str):
        today = datetime.now().strftime("%Y-%m-%d")
        key = f"serp_metrics:{today}"

        self.redis.hincrby(key, "total_queries", 1)
        self.redis.expire(key, 60 * 60 * 24 * 30)  # Keep 30 days

        if cache_hit:
            self.redis.hincrby(key, "cache_hits", 1)
        else:
            self.redis.hincrby(key, "api_calls", 1)

        self.redis.hincrby(key, f"volatility_{volatility}", 1)

    def get_daily_stats(self, date: str = None) -> dict:
        date = date or datetime.now().strftime("%Y-%m-%d")
        raw = self.redis.hgetall(f"serp_metrics:{date}")

        stats = {k.decode(): int(v) for k, v in raw.items()}
        total = stats.get("total_queries", 0)
        hits = stats.get("cache_hits", 0)

        return {
            **stats,
            "cache_hit_rate": hits / total if total > 0 else 0,
            "estimated_savings_pct": (hits / total * 100) if total > 0 else 0
        }

Real results from my pipeline

Before caching: 14,200 API calls/month → $0.85 at Scale tier pricing After caching: 3,820 API calls/month → $0.23 at Scale tier pricing

73% reduction. $0.62/month saved.

I know — the absolute number is tiny because Serpent API is cheap. But the pattern matters at scale. If you're on a more expensive provider or running higher volumes, the same logic cuts proportionally:

SerpAPI at $100/10K pages: 14,200 calls → $142/month → with caching → $38/month
The architectural pattern is the same regardless of provider

When to force-refresh (bypass cache)

# Force refresh on explicit user request
result = client.search("apiserpent.com competitors", force_refresh=True)

# Force refresh if SERP change events detected
# (e.g., Google core update announcements in your monitoring)
if google_core_update_detected():
    client.search(keyword, force_refresh=True)