Price Monitoring Competitor Intelligence Workflows

Caching Exchange Rates to Avoid Rate Limits

A price pipeline that normalizes ten marketplaces into one base currency will call a foreign-exchange provider thousands of times per crawl, and most providers cap the free tier at a few hundred requests an hour. Hammering the API is not only a way to earn a 429; it also makes your conversions non-deterministic, because two rows scraped ninety seconds apart can pick up two different rates and disagree on which competitor is cheaper. This guide is a focused recipe for a rate cache that eliminates redundant calls and makes every conversion in a run reproducible. It sits under the parent guide on currency conversion and exchange rate sync, and pairs directly with converting multi-currency prices to base currency, which consumes the cached rates this recipe produces.

Prerequisites & Input Contract

The cache sits between your normalization stage and any FX provider (exchangerate.host, Open Exchange Rates, the ECB reference feed, or an internal treasury service). It only needs a callable that fetches one day’s rates for a base currency and returns a mapping of quote codes to decimal rates. Everything else — retries, currency rounding, promo flags — is handled elsewhere in the data normalization and promo parsing pipelines.

# The provider contract the cache depends on. Your real fetcher wraps an
# HTTP client; the signature is all the cache cares about.
from decimal import Decimal
from datetime import date

def fetch_rates(base: str, on: date) -> dict[str, Decimal]:
    """Return {quote_code: rate} for `base` on day `on`. Raises on HTTP error."""
    # e.g. GET https://api.example.com/{on}?base={base}
    ...

Library versions: Python 3.11+, and optionally redis>=5.0 for a shared cache across workers. Rates are stored and compared as Decimal, never float, so that 100.00 EUR converts identically on every worker. The cache key is the triple (base, quote, date) — pinning the date is what makes a crawl reproducible.

Step-by-Step Implementation

The design walks four rungs: an in-process dict for the hot path, a persistent backing store for cross-run and cross-worker reuse, a network fetch only on a true miss, and a last-known-good fallback when the provider is down. The flow is shown below.

Four-tier lookup path for the exchange-rate cacheA conversion request for a base, quote and date triple enters the cache. Tier one checks the in-process dictionary; a hit returns immediately. A miss falls through to tier two, the persistent store such as Redis; a hit backfills the dictionary and returns. A miss there triggers tier three, a single network fetch of the whole day's rate table for that base, which is written to both stores and returned. If the fetch fails, tier four serves the most recent last-known-good rate for the pair and flags the result as stale. Every tier writes to an audit counter recording hits, misses and fallbacks.missmissmissRequest(base,quote,date)Tier 1in-process dictTier 2Redis / diskTier 3provider fetchfetch failsTier 4 · last-known-goodserved stale, flaggedhits backfill the tier above; every tier increments an audit counter

Step 1 — Define the key and an immutable rate record

Normalize currency codes to upper case and pin the date so the same pair on the same day always resolves to the same key. Storing the fetch timestamp lets you detect staleness later.

from dataclasses import dataclass
from decimal import Decimal
from datetime import date, datetime, timezone

def rate_key(base: str, quote: str, on: date) -> str:
    # Canonical, provider-agnostic key. Upper-casing avoids eur/EUR splits.
    return f"{base.upper()}:{quote.upper()}:{on.isoformat()}"

@dataclass(frozen=True)
class RateRecord:
    rate: Decimal          # units of quote per 1 unit of base
    as_of: date            # the market day this rate applies to
    fetched_at: datetime   # when we retrieved it (for staleness checks)
    stale: bool = False    # True if served as last-known-good fallback

print(rate_key("eur", "usd", date(2026, 7, 15)))  # -> 'EUR:USD:2026-07-15'

Step 2 — Build the cache with a two-tier store and batch prefetch

The core class checks the in-process dict, then the persistent store, then fetches. Crucially, on a miss it fetches the whole day’s table for that base and caches every pair at once, so the second currency in the same run is free.

class RateCache:
    def __init__(self, fetcher, store=None, ttl_seconds: int = 86_400):
        self._fetch = fetcher              # fetch_rates(base, on) -> {quote: Decimal}
        self._local: dict[str, RateRecord] = {}
        self._store = store                # optional Redis-like KV, or None
        self._ttl = ttl_seconds
        self.stats = {"hit": 0, "miss": 0, "fallback": 0}

    def _from_store(self, key):
        if not self._store:
            return None
        raw = self._store.get(key)
        return _decode(raw) if raw else None

    def _to_store(self, key, rec: RateRecord):
        self._local[key] = rec
        if self._store:
            self._store.set(key, _encode(rec), ex=self._ttl)

    def get(self, base: str, quote: str, on: date) -> RateRecord:
        base, quote = base.upper(), quote.upper()
        if base == quote:                          # identity, never hits the wire
            return RateRecord(Decimal("1"), on, _now())
        key = rate_key(base, quote, on)
        rec = self._local.get(key) or self._from_store(key)
        if rec:
            self.stats["hit"] += 1
            self._local[key] = rec                 # backfill hot path from store
            return rec
        return self._fetch_day(base, quote, on)    # true miss

    def _fetch_day(self, base, quote, on) -> RateRecord:
        self.stats["miss"] += 1
        try:
            table = self._fetch(base, on)          # one call, whole table
        except Exception:
            return self._last_known_good(base, quote, on)
        fetched = _now()
        wanted = None
        for q, r in table.items():                 # cache every pair at once
            rec = RateRecord(Decimal(str(r)), on, fetched)
            self._to_store(rate_key(base, q.upper(), on), rec)
            if q.upper() == quote:
                wanted = rec
        return wanted or self._last_known_good(base, quote, on)

Step 3 — Add the last-known-good fallback

When the provider errors, do not raise into the pipeline and stall the whole crawl. Walk backwards up to a bounded number of days and serve the most recent cached rate, flagged stale=True so downstream code can decide whether to trust it.

from datetime import timedelta

    def _last_known_good(self, base, quote, on, max_back: int = 7) -> RateRecord:
        for delta in range(1, max_back + 1):
            prev = on - timedelta(days=delta)
            key = rate_key(base, quote, prev)
            rec = self._local.get(key) or self._from_store(key)
            if rec:
                self.stats["fallback"] += 1
                # Re-stamp with the requested day but keep the true as_of + stale flag
                return RateRecord(rec.rate, rec.as_of, rec.fetched_at, stale=True)
        raise LookupError(f"no rate for {base}->{quote} within {max_back}d of {on}")

Step 4 — Pin a per-run snapshot for reproducibility

For a single crawl you usually want every row converted at one consistent set of rates. Prefetch the day’s tables for all bases once, then freeze the cache: any pair not already loaded raises instead of silently drifting to a live rate mid-run.

class SnapshotCache(RateCache):
    def __init__(self, *a, **kw):
        super().__init__(*a, **kw)
        self._frozen = False

    def prefetch(self, bases: list[str], on: date):
        for base in bases:                         # warm one table per base
            self._fetch_day(base, base, on)        # quote==base is harmless
        self._frozen = True

    def get(self, base, quote, on):
        rec = self._local.get(rate_key(base.upper(), quote.upper(), on))
        if self._frozen and rec is None and base.upper() != quote.upper():
            raise LookupError(f"snapshot missing {base}->{quote} on {on}")
        return super().get(base, quote, on)

Running a quick end-to-end check with a stub fetcher shows the second lookup never touches the wire:

calls = {"n": 0}
def stub(base, on):
    calls["n"] += 1
    return {"USD": Decimal("1.08"), "GBP": Decimal("0.85")}

cache = RateCache(stub)
d = date(2026, 7, 15)
print(cache.get("EUR", "USD", d).rate)   # -> 1.08   (miss, fetches table)
print(cache.get("EUR", "GBP", d).rate)   # -> 0.85   (hit, same table)
print(calls["n"], cache.stats)           # -> 1 {'hit': 1, 'miss': 1, 'fallback': 0}

Verification & Testing

Assert the three behaviours that matter: one fetch serves a whole base’s pairs, a provider outage degrades to stale rather than crashing, and a frozen snapshot refuses unknown pairs.

def test_cache_behaviour():
    d = date(2026, 7, 15)
    n = {"c": 0}
    def fetch(base, on):
        n["c"] += 1
        return {"USD": Decimal("1.08"), "GBP": Decimal("0.85")}

    c = RateCache(fetch)
    assert c.get("EUR", "USD", d).rate == Decimal("1.08")
    assert c.get("EUR", "GBP", d).rate == Decimal("0.85")
    assert n["c"] == 1                     # batch prefetch: one call, two pairs

    # Outage after warm-up -> last-known-good, flagged stale
    def boom(base, on):
        raise ConnectionError("429 rate limited")
    c2 = RateCache(boom)
    c2._to_store(rate_key("EUR", "USD", d - timedelta(days=1)),
                 RateRecord(Decimal("1.07"), d - timedelta(days=1), _now()))
    r = c2.get("EUR", "USD", d)
    assert r.stale and r.rate == Decimal("1.07")
    print("cache behaviour verified")

A second layer of defence is a daily reconciliation job: compare each cached daily snapshot against the provider’s published close and alert if any pair drifts more than a few basis points, which usually means a mid-day schema change on the provider side.

Edge Cases & Gotchas

  • Weekend and holiday gaps. FX markets close, so many providers return Friday’s rate for Saturday and Sunday, and some return nothing at all. Decide one policy — carry the last trading day forward — and encode it by having fetch_rates roll the request date back to the previous business day, so EUR:USD:2026-07-18 (a Saturday) deliberately resolves to the Friday key rather than firing a fallback that looks like an outage.
  • Stale rate served silently. The stale flag is worthless if downstream ignores it. In converting multi-currency prices to base currency, refuse to emit an automated repricing signal from a stale conversion and route the row to review, the same way a suspect discount is held in statistical outlier detection for price data.
  • Missing pair / exotic currency. Not every provider quotes every cross. If the day’s table lacks the quote you need, triangulate through the base’s own reference (e.g. TRY->EUR->USD) rather than falling back seven days, since thinly traded pairs move fast and a week-old rate is worse than a same-day triangulation.
  • TTL vs daily-snapshot confusion. A TTL cache (ex=3600) is right for an intraday live rate; a daily snapshot keyed by date should effectively never expire, because 15 July’s rate does not change after the fact. Mixing the two — a one-hour TTL on a date-keyed record — silently forces refetches and reintroduces the rate-limit problem you set out to solve.

Performance Notes

The hot path is an O(1) dict lookup, so a warmed cache converts millions of rows without a single network call. Cost concentrates entirely in cold misses, and the batch-per-base fetch is what keeps that bounded: a crawl spanning eight source currencies needs eight provider calls per day regardless of how many million rows or how many quote currencies you touch. With a shared Redis tier, the first worker to see a base pays the fetch and every other worker reads the warmed table, cutting a fleet’s provider traffic to the single-worker figure. Keep Decimal throughout — the conversion arithmetic is cheap relative to the network, and the determinism it buys is the entire point of caching by (base, quote, date) in the first place.

Frequently Asked Questions

Should I use a TTL or a dated snapshot for exchange rates? Use a dated snapshot keyed by (base, quote, date) for reproducible batch conversion, and set an effectively infinite TTL because a historical day’s rate is immutable. Reserve short TTLs for live intraday rates where you deliberately want freshness over reproducibility.

How do I keep a whole crawl converting at one consistent rate? Prefetch the day’s rate table for every source currency once at the start of the run and freeze the cache. Every subsequent conversion reads the frozen snapshot, so two rows scraped minutes apart cannot disagree, and any pair you forgot to warm raises loudly instead of drifting to a live rate.

What should happen when the FX provider is rate-limited or down? Serve the most recent last-known-good rate, walking back a bounded number of days, and flag the result as stale rather than raising and stalling the pipeline. Downstream repricing must treat a stale conversion as review-only and never auto-act on it.