feat(oracle): time-aligned deviation against oracle updatedAt

benchmarks/oracle-deviation.yml

@@ -69,6 +69,7 @@ methodology:
  - "Chainlink round age. The on-chain `updatedAt` timestamp from `latestRoundData()` is published separately as `ocb_oracle_last_round_age_seconds{source=\"chainlink\", pair}`. Chainlink updates only on deviation (typically 0.25-0.5% for blue chips) or heartbeat (~1 h), so a 30-minute age on a quiet ETH minute is normal; a 2-hour age on SOL during a volatile minute is the actual signal."
  - "USDT ≈ USD assumption. Binance only quotes USDT pairs (BTCUSDT, etc.). We treat USDT as ≈ USD ± 10 bps drift, which is acceptable for a bench whose alert floor is ≥ 10 bps. A real USDT depeg would surface as Binance drifting from the other three sources for *every* pair simultaneously, exactly what we want this bench to flag, not hide."
  - "MATIC → POL migration. Polygon migrated MATIC → POL 1:1 on Sep 4 2024. The Chainlink mainnet feed contract is still named MATIC/USD but on-chain `description()` confirms it tracks the POL token; Pyth renamed the feed to POL/USD; Coinbase delisted MATIC-USD and only lists POL-USD; Binance kept MATICUSDT as a frozen historical pair AND lists POLUSDT. We point Binance at POLUSDT so all four sources track the same underlying asset. Bench label is kept as `pair=\"MATIC/USD\"` for query continuity."
+ - "Time-aligned deviation (canonical headline). For every pair of sources (a, b) we anchor on the more recent of their two SourceTSs (Chainlink's on-chain `updatedAt` for Chainlink, fetch time for the continuously-updating sources) and look up the older source's price in a 30-minute per-source rolling history at the anchor moment. The result is published as `ocb_oracle_deviation_at_oracle_ts_pct` and feeds the leaderboard headline `ocb_oracle_max_deviation_pct`. This eliminates the artifact where Chainlink's heartbeat lag inflates 'deviation' by the market's drift between its updates — a researcher grading oracle quality wants to compare Chainlink's price against the market at Chainlink's own updatedAt, not at the harness's fetch instant. The legacy fetch-time gauge (`ocb_oracle_deviation_pct`, aliased as `ocb_oracle_deviation_at_fetch_ts_pct`) is preserved unchanged for backward compatibility. Alignment misses (no history sample within ±10s of the anchor) are counted as `ocb_oracle_alignment_miss_total`. Methodology recommendation from Coinpaprika data team review; matches the convention published by Chaos Labs and Risk DAO oracle risk reports."
  - "Excluded by design. Redstone (push-pull, no continuous gauge to scrape without integration contract), Uniswap V3 TWAP (per-pool integration + derivation of same CEX prints), DIA (smaller footprint than the four kept), and aggregator-of-aggregators (CoinGecko, CMC, DefiLlama), re-aggregating already-aggregated data adds latency and hides per-source disagreement."
 
 findings:

harnesses/oracle-deviation/cmd/script/chainlink.go

@@ -135,7 +135,12 @@ func runChainlink(ctx context.Context, specs []PairSpec) {
  continue
  }
  c.failureCount[s.ChainlinkFeed] = 0
- recordPrice(SourceChainlink, s.Pair, price)
+ // Pass Chainlink's on-chain updatedAt as the SourceTS so
+ // the time-aligned deviation calc can snap market prints
+ // to this moment instead of comparing against now-fresh
+ // values (which would tag Chainlink's heartbeat lag as
+ // "deviation" — that artifact is exactly what Fix C kills).
+ recordPriceAt(SourceChainlink, s.Pair, price, time.Unix(updatedAt, 0))
  ageS := time.Since(time.Unix(updatedAt, 0)).Seconds()
  if ageS < 0 {
  ageS = 0

harnesses/oracle-deviation/cmd/script/loghub.go

@@ -0,0 +1,9 @@
+package main
+
+// Stub for the public OCB mirror. The private mobula-monorepo deploy
+// captures stdout into a ring buffer served at /logs?tail=N (X-Logs-Token
+// gated) for the openbench-monitoring admin UI. The public harness has
+// no logs endpoint — it's a transparency mirror, not an operated
+// service. Keeping the call site identical lets the two harnesses share
+// the rest of the code 1:1.
+func installLogCapture() {}

harnesses/oracle-deviation/cmd/script/main.go

@@ -21,86 +21,276 @@ import (
 // pricePoint is the in-memory record per (pair, source). The whole
 // store fits in 4 sources × 10 pairs = 40 entries, so we don't need
 // a real DB — a guarded map is more than enough.
+//
+// TS is when WE fetched the value (wall-clock at the harness).
+// SourceTS is the timestamp the source itself attached to the
+// reading: Chainlink's on-chain updatedAt for the round, time.Now()
+// for the continuously-updating sources (Pyth Hermes, Binance ticker,
+// Coinbase ticker, which all publish "current" prices at fetch time).
+// The gap (TS - SourceTS) is what oracleLastRoundAge surfaces.
 type pricePoint struct {
-	Value float64
-	TS time.Time
+	Value float64
+	TS time.Time
+	SourceTS time.Time
 }
 
 var (
 	storeMu sync.RWMutex
 	// store[pair][source] = pricePoint
 	store = make(map[Pair]map[Source]pricePoint)
+
+	// Rolling per-(pair, source) history. Used by the time-aligned
+	// deviation calculation: when Chainlink's SourceTS is 20 min
+	// older than its TS, we compare its price against the OTHER
+	// sources' historical prices at that same SourceTS, not against
+	// their current values. Eliminates the clock-skew artifact that
+	// inflates "deviation" by the market's drift during Chainlink's
+	// heartbeat window. Bounded by historyDepth; pruned on insert.
+	historyMu sync.RWMutex
+	history = make(map[Pair]map[Source][]pricePoint)
+)
+
+const (
+	// 60 entries × 30s cadence ≈ 30 min look-back. Larger than any
+	// reasonable alignment tolerance, smaller than the time scales
+	// at which a deviation > 10% off a 20 min Chainlink lag would
+	// stop being interesting.
+	historyDepth = 60
+	// Window around the anchor timestamp we search when looking up
+	// a corresponding price on another source. 10 s = ~1/3 of the
+	// 30 s polling cadence, so a healthy source should always have
+	// at least one sample inside the window. Misses are counted as
+	// oracleAlignmentMiss.
+	alignTolerance = 10 * time.Second
 )
 
-// recordPrice is called by every poller after a successful read.
-// Updates the in-memory store + the per-source gauge + the per-pair
-// deviation matrix.
+// recordPrice is called by every continuously-updating poller (Pyth,
+// Binance, Coinbase) after a successful read. SourceTS is set to
+// time.Now() because these sources publish current prices, no on-chain
+// timestamp gap. Chainlink calls recordPriceAt instead so it can pass
+// the on-chain updatedAt as SourceTS.
 func recordPrice(src Source, pair Pair, v float64) {
+	recordPriceAt(src, pair, v, time.Now())
+}
+
+// recordPriceAt is the underlying primitive. sourceTS is when the
+// source itself declared the price current (Chainlink updatedAt for
+// chainlink, fetch time for the others). The (TS, SourceTS) pair is
+// what the time-aligned deviation calculation needs to compare prices
+// at the same reference moment instead of at fetch time.
+func recordPriceAt(src Source, pair Pair, v float64, sourceTS time.Time) {
 	if v <= 0 || math.IsNaN(v) || math.IsInf(v, 0) {
  // Defensive: a 0 or NaN price would poison the deviation calc.
  // Count it as an error and bail.
  oracleScrapeErrors.WithLabelValues(string(src), string(pair)).Inc()
  return
 	}
 	now := time.Now()
+	point := pricePoint{Value: v, TS: now, SourceTS: sourceTS}
+
 	storeMu.Lock()
 	if _, ok := store[pair]; !ok {
  store[pair] = make(map[Source]pricePoint)
 	}
-	store[pair][src] = pricePoint{Value: v, TS: now}
+	store[pair][src] = point
 	storeMu.Unlock()
 
+	// Append to the rolling history under the SourceTS the source
+	// gave us, prune anything older than the depth. This is what
+	// the at_oracle_ts deviation calc reads from.
+	historyMu.Lock()
+	if _, ok := history[pair]; !ok {
+ history[pair] = make(map[Source][]pricePoint)
+	}
+	h := append(history[pair][src], point)
+	if len(h) > historyDepth {
+ h = h[len(h)-historyDepth:]
+	}
+	history[pair][src] = h
+	historyMu.Unlock()
+
 	oraclePrice.WithLabelValues(string(src), string(pair)).Set(v)
 	oracleUpdateLatencySeconds.WithLabelValues(string(src), string(pair)).Set(0)
 	recomputeDeviations(pair)
 }
 
+// lookupNearest returns the pricePoint for (pair, src) whose SourceTS
+// is nearest to target, within ±alignTolerance. Returns ok=false when
+// no such sample exists. Caller does not need to hold any lock; this
+// acquires historyMu.RLock internally.
+func lookupNearest(pair Pair, src Source, target time.Time) (pricePoint, bool) {
+	historyMu.RLock()
+	defer historyMu.RUnlock()
+	byPair := history[pair]
+	if byPair == nil {
+ return pricePoint{}, false
+	}
+	samples := byPair[src]
+	if len(samples) == 0 {
+ return pricePoint{}, false
+	}
+	var best pricePoint
+	bestDelta := time.Duration(1<<62 - 1)
+	found := false
+	for _, p := range samples {
+ delta := p.SourceTS.Sub(target)
+ if delta < 0 {
+ delta = -delta
+ }
+ if delta > alignTolerance {
+ continue
+ }
+ if delta < bestDelta {
+ best = p
+ bestDelta = delta
+ found = true
+ }
+	}
+	if !found {
+ return pricePoint{}, false
+	}
+	return best, true
+}
+
 // recomputeDeviations computes pairwise + max deviations for one
 // pair. Called inline from recordPrice so the metrics always reflect
 // the freshest data.
+//
+// Publishes TWO families of deviation gauges:
+//
+// - ocb_oracle_deviation_at_fetch_ts_pct (and the legacy alias
+// ocb_oracle_deviation_pct): pairwise deviation using whatever
+// each source last reported, regardless of when each source
+// declared its price current. This is the operational truth a
+// protocol team sees if it just polls all four sources right
+// now. Useful for showing the gap between "what Chainlink has
+// on-chain right now" and "what the live CEX/Pyth print is."
+//
+// - ocb_oracle_deviation_at_oracle_ts_pct: time-aligned deviation.
+// For every (source_a, source_b) pair, snap both prices to the
+// more recent SourceTS of the two and look up the older
+// source's price in its rolling history at THAT moment. This is
+// the deviation a researcher needs to grade oracle quality at
+// update time, free of the artifact that Chainlink's heartbeat
+// introduces when its updatedAt is 20 min behind the others.
+//
+// Headline ocb_oracle_max_deviation_pct ranks on the at_oracle_ts
+// variant because that's the canonical question: how aligned are
+// the oracles when you compare like for like. The at_fetch_ts series
+// remains queryable for backward compatibility with anyone who built
+// a dashboard on the legacy gauge.
+//
+// Methodology recommendation from Coinpaprika data team review. See
+// also Chaos Labs and Risk DAO's published methodology — both snap
+// market price to the oracle's updatedAt block timestamp, not the
+// fetch time.
 func recomputeDeviations(pair Pair) {
 	storeMu.RLock()
 	srcMap := store[pair]
-	prices := make(map[Source]float64, len(srcMap))
+	type srcSample struct {
+ value float64
+ sourceTS time.Time
+	}
+	samples := make(map[Source]srcSample, len(srcMap))
 	for s, p := range srcMap {
  // Skip stale prices (>2 polling intervals = 60s) when
  // recomputing — otherwise a dead source would keep its last
  // value frozen and quietly bias the deviation forever.
  if time.Since(p.TS) > 2*pollInterval {
  continue
  }
-prices[s] = p.Value
+samples[s] = srcSample{value: p.Value, sourceTS: p.SourceTS}
 	}
 	storeMu.RUnlock()
 
-	if len(prices) < 2 {
+	if len(samples) < 2 {
  return
 	}
 
 	// Sort sources lexicographically to keep the (source_a, source_b)
 	// label deterministic regardless of map iteration order.
-	srcs := make([]Source, 0, len(prices))
-	for s := range prices {
+	srcs := make([]Source, 0, len(samples))
+	for s := range samples {
  srcs = append(srcs, s)
 	}
 	sort.Slice(srcs, func(i, j int) bool { return srcs[i] < srcs[j] })
 
-	maxDev := 0.0
+	maxFetchDev := 0.0
+	maxAlignedDev := 0.0
+	haveAligned := false
 	for i := 0; i < len(srcs); i++ {
  for j := i + 1; j < len(srcs); j++ {
- a, b := prices[srcs[i]], prices[srcs[j]]
- mid := (a + b) / 2
+ a, b := samples[srcs[i]], samples[srcs[j]]
+ mid := (a.value + b.value) / 2
  if mid == 0 {
  continue
  }
- dev := math.Abs(a-b) / mid * 100
- oracleDeviationPct.WithLabelValues(string(pair), string(srcs[i]), string(srcs[j])).Set(dev)
- if dev > maxDev {
- maxDev = dev
+
+ // Fetch-time pairwise deviation: current behavior,
+ // preserved as the legacy gauge + a new explicit alias.
+ fetchDev := math.Abs(a.value-b.value) / mid * 100
+ oracleDeviationPct.WithLabelValues(string(pair), string(srcs[i]), string(srcs[j])).Set(fetchDev)
+ oracleDeviationAtFetchTSPct.WithLabelValues(string(pair), string(srcs[i]), string(srcs[j])).Set(fetchDev)
+ if fetchDev > maxFetchDev {
+ maxFetchDev = fetchDev
+ }
+
+ // Time-aligned pairwise deviation: anchor on the more
+ // recent SourceTS, look up the other side in history.
+ anchor := a.sourceTS
+ if b.sourceTS.After(anchor) {
+ anchor = b.sourceTS
+ }
+ alignedA, okA := alignedPriceAt(pair, srcs[i], a, anchor)
+ alignedB, okB := alignedPriceAt(pair, srcs[j], b, anchor)
+ if !okA || !okB {
+ oracleAlignmentMiss.WithLabelValues(string(pair), string(srcs[i]), string(srcs[j])).Inc()
+ continue
+ }
+ alignedMid := (alignedA + alignedB) / 2
+ if alignedMid == 0 {
+ continue
  }
+ alignedDev := math.Abs(alignedA-alignedB) / alignedMid * 100
+ oracleDeviationAtOracleTSPct.WithLabelValues(string(pair), string(srcs[i]), string(srcs[j])).Set(alignedDev)
+ if alignedDev > maxAlignedDev {
+ maxAlignedDev = alignedDev
+ }
+ haveAligned = true
  }
 	}
-	oracleMaxDeviationPct.WithLabelValues(string(pair)).Set(maxDev)
+
+	// Headline ranks on the time-aligned max when at least one
+	// aligned pair was computable, otherwise we fall back on the
+	// fetch-time max (matches old behavior so the gauge never goes
+	// dark during a buffer cold-start window).
+	if haveAligned {
+ oracleMaxDeviationPct.WithLabelValues(string(pair)).Set(maxAlignedDev)
+	} else {
+ oracleMaxDeviationPct.WithLabelValues(string(pair)).Set(maxFetchDev)
+	}
+}
+
+// alignedPriceAt returns the price the given source had at "anchor"
+// time, with fallback to the current sample when source's own SourceTS
+// is already at anchor (within tolerance, no history lookup needed).
+// Returns ok=false when no historical sample exists within tolerance.
+func alignedPriceAt(pair Pair, src Source, cur struct {
+	value float64
+	sourceTS time.Time
+}, anchor time.Time) (float64, bool) {
+	delta := cur.sourceTS.Sub(anchor)
+	if delta < 0 {
+ delta = -delta
+	}
+	if delta <= alignTolerance {
+ return cur.value, true
+	}
+	p, ok := lookupNearest(pair, src, anchor)
+	if !ok {
+ return 0, false
+	}
+	return p.Value, true
 }
 
 // runLatencyUpdater bumps the update_latency_seconds gauge once per
@@ -127,6 +317,7 @@ func runLatencyUpdater(ctx context.Context) {
 }
 
 func main() {
+	installLogCapture() // capture stdout into /logs ring buffer
 	fmt.Println("=== Oracle Deviation Harness ===")
 	fmt.Println("OpenChainBench № 025 — 4 oracles × 10 pairs, max deviation gauge.")
 	fmt.Println()

harnesses/oracle-deviation/cmd/script/metrics.go

@@ -24,15 +24,39 @@ var (
 	oracleDeviationPct = promauto.NewGaugeVec(
  prometheus.GaugeOpts{
  Name: "ocb_oracle_deviation_pct",
- Help: "Pairwise deviation between two sources for the same pair, expressed as percent of the midpoint: |a-b|/((a+b)/2)*100. One sample per (pair, source_a, source_b), source_a < source_b lexicographically to avoid double-counting.",
+ Help: "Legacy fetch-time pairwise deviation between two sources (alias of ocb_oracle_deviation_at_fetch_ts_pct, preserved for backward compat). Use ocb_oracle_deviation_at_oracle_ts_pct for the canonical time-aligned number.",
+ },
+ []string{"pair", "source_a", "source_b"},
+	)
+
+	oracleDeviationAtFetchTSPct = promauto.NewGaugeVec(
+ prometheus.GaugeOpts{
+ Name: "ocb_oracle_deviation_at_fetch_ts_pct",
+ Help: "Fetch-time pairwise deviation. Each source's last-fetched price, regardless of when the source said the price was current. Includes Chainlink's heartbeat lag as 'deviation' (operational truth, not oracle-quality).",
+ },
+ []string{"pair", "source_a", "source_b"},
+	)
+
+	oracleDeviationAtOracleTSPct = promauto.NewGaugeVec(
+ prometheus.GaugeOpts{
+ Name: "ocb_oracle_deviation_at_oracle_ts_pct",
+ Help: "Time-aligned pairwise deviation. Each source's price snapped to the more recent SourceTS of the pair (Chainlink's on-chain updatedAt for chainlink, fetch time for continuous sources), with the older source's price looked up from history. This is the canonical deviation: it answers 'do the oracles agree at the same moment in time' instead of 'do they agree at our fetch instant'.",
+ },
+ []string{"pair", "source_a", "source_b"},
+	)
+
+	oracleAlignmentMiss = promauto.NewCounterVec(
+ prometheus.CounterOpts{
+ Name: "ocb_oracle_alignment_miss_total",
+ Help: "Pairwise computations where no history sample was within ±10s of the anchor SourceTS for at least one of the two sources. Pair is skipped on the at_oracle_ts gauge for that cycle.",
  },
  []string{"pair", "source_a", "source_b"},
 	)
 
 	oracleMaxDeviationPct = promauto.NewGaugeVec(
  prometheus.GaugeOpts{
  Name: "ocb_oracle_max_deviation_pct",
- Help: "Maximum pairwise deviation observed across all available source pairs for a given asset pair. The headline metric of this bench.",
+ Help: "Maximum pairwise deviation across all available source pairs for the asset. The headline metric. Ranks on the time-aligned variant (ocb_oracle_deviation_at_oracle_ts_pct) when at least one pair is alignable, falls back to fetch-time during cold start.",
  },
  []string{"pair"},
 	)