Blog/Developer

Detect cross-source price divergence in code

Read per-source prices from the API response, compute the spread in basis points, and flag when sources disagree beyond your tolerance threshold.

Developer.md

A single price number hides information. When the API aggregates quotes from multiple sources, those quotes are not always identical. Sometimes they diverge — by a small amount, normally, and by a larger amount when something is off: a feed is delayed, a source has a stale quote, or a genuine dislocation is forming. For the market explanation of why sources diverge, see What it means when gold sources disagree.

Reading that divergence and deciding what to do with it is something you can code in an afternoon.

What the response looks like

The GET /v1/spot/XAU-USD-SPOT response includes a top-level price (the aggregated value) and a sources array with the per-source breakdown:

{
  "symbol": "XAU-USD-SPOT",
  "price": 2341.80,
  "computed_at": "2026-06-19T08:14:22.411Z",
  "sources": [
    { "source": "feed_a", "price": 2341.20, "fetched_at": "2026-06-19T08:14:20.100Z" },
    { "source": "feed_b", "price": 2342.40, "fetched_at": "2026-06-19T08:14:21.800Z" },
    { "source": "feed_c", "price": 2341.80, "fetched_at": "2026-06-19T08:14:22.000Z" }
  ]
}

The aggregated price is a weighted or median value across those sources. The sources array tells you what went into it.

Measuring spread in basis points

Basis points (bps) are the right unit here. One basis point is 0.01% of price. At $2341, 1 bps = $0.234. Using bps instead of dollars makes your threshold code stable as the price level changes over months and years.

import os
import requests

API_KEY = os.environ["GOLDPRICE_API_KEY"]

def fetch_spot(symbol: str) -> dict:
    resp = requests.get(
        f"https://api.goldprice.dev/v1/spot/{symbol}",
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()

def compute_spread_bps(sources: list[dict]) -> dict:
    prices = [s["price"] for s in sources]
    if len(prices) < 2:
        return {"spread_bps": 0.0, "min": prices[0], "max": prices[0]}

    lo = min(prices)
    hi = max(prices)
    midpoint = (lo + hi) / 2
    spread_bps = ((hi - lo) / midpoint) * 10_000

    return {
        "spread_bps": round(spread_bps, 2),
        "min": lo,
        "max": hi,
        "source_count": len(prices),
    }

data = fetch_spot("XAU-USD-SPOT")
spread = compute_spread_bps(data.get("sources", []))
print(spread)
# {'spread_bps': 5.13, 'min': 2341.2, 'max': 2342.4, 'source_count': 3}

Flagging divergence

Define your threshold based on what you care about. Five basis points ($1.17 at $2341) is a normal intraday spread. Anything above 20–30 bps is worth a second look; above 50 bps in a liquid market, something unusual is happening.

WARN_BPS = 20.0   # log a warning
ALERT_BPS = 50.0  # page someone / halt downstream

def evaluate_divergence(spread: dict) -> str:
    bps = spread["spread_bps"]
    if bps >= ALERT_BPS:
        return "ALERT"
    if bps >= WARN_BPS:
        return "WARN"
    return "OK"

data = fetch_spot("XAU-USD-SPOT")
spread = compute_spread_bps(data.get("sources", []))
status = evaluate_divergence(spread)

print(f"Spread: {spread['spread_bps']} bps — {status}")

In a production system you would route ALERT to PagerDuty or your on-call webhook, and WARN to a Slack channel. The point is that the decision logic lives in your code, not in the API.

Staleness as a divergence signal

A source might be current on price but stale on timestamp. That is its own signal — a feed that has not updated in 10 minutes is effectively diverged, even if its last price happened to match.

from datetime import datetime, timezone, timedelta

STALE_THRESHOLD = timedelta(minutes=5)

def flag_stale_sources(sources: list[dict]) -> list[dict]:
    now = datetime.now(timezone.utc)
    stale = []
    for s in sources:
        fetched = datetime.fromisoformat(s["fetched_at"].replace("Z", "+00:00"))
        age = now - fetched
        if age > STALE_THRESHOLD:
            stale.append({
                "source": s["source"],
                "age_seconds": int(age.total_seconds()),
                "last_price": s["price"],
            })
    return stale

data = fetch_spot("XAU-USD-SPOT")
stale = flag_stale_sources(data.get("sources", []))
if stale:
    print("Stale sources:", stale)

Combine this with the spread check and you have two independent signals. A wide spread with all sources fresh suggests genuine market dislocation. A normal spread with one stale source suggests a feed problem. Different causes, different responses.

Why this matters

If your application displays the price to end users or makes decisions based on it, you want to know when the inputs are degraded. Showing a stale or unreliable price is worse than showing nothing; acting on a dislocated price in an automated system can be expensive.

Most applications never check this. They take the price field and move on. The sources array exists precisely so you can do better than that.

Get your API key via the quickstart guide.

// related guides

// goldprice.dev

Live gold prices, historical OHLC, and multi-source aggregation — available via REST and SSE.