# Detect cross-source price divergence in code

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](/blog/price-divergence-explained).

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:

```json
{
  "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.

```python
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.

```python
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.

```python
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](/docs/quickstart).