diff --git a/benchmarks/oracle-deviation.yml b/benchmarks/oracle-deviation.yml index 07c91167..36b265c2 100644 --- a/benchmarks/oracle-deviation.yml +++ b/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: diff --git a/harnesses/oracle-deviation/cmd/script/chainlink.go b/harnesses/oracle-deviation/cmd/script/chainlink.go index e52494aa..c89d956c 100644 --- a/harnesses/oracle-deviation/cmd/script/chainlink.go +++ b/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 diff --git a/harnesses/oracle-deviation/cmd/script/loghub.go b/harnesses/oracle-deviation/cmd/script/loghub.go new file mode 100644 index 00000000..a85d663d --- /dev/null +++ b/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() {} diff --git a/harnesses/oracle-deviation/cmd/script/main.go b/harnesses/oracle-deviation/cmd/script/main.go index 05208b22..c7b96cec 100644 --- a/harnesses/oracle-deviation/cmd/script/main.go +++ b/harnesses/oracle-deviation/cmd/script/main.go @@ -21,21 +21,64 @@ 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. @@ -43,25 +86,112 @@ func recordPrice(src Source, pair Pair, v float64) { 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 @@ -69,38 +199,98 @@ func recomputeDeviations(pair Pair) { 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() diff --git a/harnesses/oracle-deviation/cmd/script/metrics.go b/harnesses/oracle-deviation/cmd/script/metrics.go index 316f8a87..49b24e30 100644 --- a/harnesses/oracle-deviation/cmd/script/metrics.go +++ b/harnesses/oracle-deviation/cmd/script/metrics.go @@ -24,7 +24,31 @@ 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"}, ) @@ -32,7 +56,7 @@ var ( 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"}, )