From db562b1889d7fa9cc01c649a9573070196c73eab Mon Sep 17 00:00:00 2001 From: Florent Tapponnier Date: Tue, 9 Jun 2026 14:46:26 +0200 Subject: [PATCH] feat(peg): multi-venue consensus outlier rule Replaces the flat 20% drop / 10% cap with a consensus-based filter: - >50% off peg: dropped as parser-bug sanity floor - >2% off peg: kept ONLY if another venue corroborates within 30s in the same direction (real multi-venue depeg) - single-venue >2% with no corroborator: dropped_isolated (glitch) The old rule would have clipped USDC at $0.87 during March 2023 SVB to $0.90, erasing the event in the percentile metric. The new rule preserves multi-venue depeg signals while filtering single-venue glitches and parser bugs. New peg_source_call_total{result} labels: dropped_sanity, dropped_isolated, kept_corroborated. Methodology recommendation from Coinpaprika data team review. Mirror of mobula PR (private monorepo). --- benchmarks/stablecoin-peg.yml | 1 + .../stablecoin-peg/cmd/script/aggregator.go | 151 +++++++++++++++--- .../stablecoin-peg/cmd/script/metrics.go | 2 +- 3 files changed, 132 insertions(+), 22 deletions(-) diff --git a/benchmarks/stablecoin-peg.yml b/benchmarks/stablecoin-peg.yml index 66f27b57..2680e5a8 100644 --- a/benchmarks/stablecoin-peg.yml +++ b/benchmarks/stablecoin-peg.yml @@ -63,6 +63,7 @@ methodology: - "USDT-anchored secondary metric: Binance USDC/USDT, FDUSD/USDT and USDE/USDT are exposed on `peg_deviation_usdt_anchored_bps{venue}` separately so the USD-anchored primary leaderboard is not contaminated by USDT's own peg deviation." - "Time outside band: total seconds in the trailing 24 h during which the per-minute aggregated price fell outside [0.995, 1.005] (±50 bps). Split into `peg_time_below_peg_24h_seconds` (< 0.995) and `peg_time_above_peg_24h_seconds` (> 1.005) because Circle redemption only clears above-peg, so the direction tells you which failure mode is active." - "Depeg event flag: binary `peg_depeg_event_flag` set to 1 when the per-minute aggregated price has been outside [0.97, 1.03] for ≥5 consecutive minutes; cleared after 30 minutes back inside. Conservative so it does not flap during normal stress events." + - "Outlier rule (multi-venue consensus): a single sample more than 2% off peg is kept only when at least one OTHER venue has also been outside the same band in the same direction within the last 30 seconds. Single-venue glitches (one CEX returns a stale or fat-finger print while every other venue is at $1.00) are dropped as `dropped_isolated`. Multi-venue confirmation (Kraken AND Bitstamp both at $0.87 = real depeg) is kept as `kept_corroborated` so the percentile metric surfaces the event. Replaces the previous flat 20% drop / 10% cap which would have clipped USDC at $0.87 during the March 2023 SVB depeg to $0.90, erasing the event in the percentile metric. Sanity floor: anything more than 50% off peg is treated as a parser bug and dropped regardless. Methodology recommendation from Coinpaprika data team review." - "Excluded by design: aggregator-only prices (CoinGecko, Coinmarketcap, DefiLlama) because they are themselves liquidity-weighted medians of the venues we already poll directly. Algo-stables that have already failed (UST, USDR) are out of scope; the bench tracks live, currently-redeemable stables." findings: diff --git a/harnesses/stablecoin-peg/cmd/script/aggregator.go b/harnesses/stablecoin-peg/cmd/script/aggregator.go index 88416a06..f9b9fada 100644 --- a/harnesses/stablecoin-peg/cmd/script/aggregator.go +++ b/harnesses/stablecoin-peg/cmd/script/aggregator.go @@ -19,6 +19,28 @@ type sample struct { minuteBucket time.Time } +// recentSample is one (price, time) observation kept in a rolling +// 30-second buffer per (stable, venue), used exclusively by the +// consensus outlier rule. We don't keep the full sample struct because +// nothing else in the consensus check needs the bucket / quote / liquidity. +type recentSample struct { + price float64 + receivedAt time.Time +} + +// Consensus outlier rule constants. A sample more than +// consensusThreshold off peg is kept only when at least one OTHER venue +// has also been outside the band in the same direction within the last +// consensusWindow. Single-venue glitches get filtered. Multi-venue +// confirmation = real depeg signal preserved. Sanity floor at +// sanityCeiling because anything wider than that on a USD-quoted +// stable is a parser bug, not a market event. +const ( + consensusWindow = 30 * time.Second + consensusThreshold = 0.02 + sanityCeiling = 0.50 +) + // Aggregator collects per-venue samples, closes minute buckets, // computes liquidity-weighted median per stable per minute, and // updates the deviation metrics. Single-threaded internal logic @@ -42,38 +64,128 @@ type Aggregator struct { band ringBuffer // Current minute timestamp. currentMinute time.Time + // Rolling 30s buffer of (price, time) per (stable, venue) used + // by the consensus outlier rule to decide whether a >2% sample + // is a single-venue glitch or a corroborated depeg event. + // Every sample is recorded here regardless of accept/reject so + // two venues that depeg simultaneously can validate each other + // even when the first arrives before the second. + recentByStableVenue map[string]map[string][]recentSample } func NewAggregator() *Aggregator { return &Aggregator{ - primaryBucket: make(map[string]map[string][]sample), - usdtBucket: make(map[string]map[string][]sample), - lastClosedPrice: make(map[string]float64), - depegStreak: make(map[string]int), - recoveryStreak: make(map[string]int), - depegActive: make(map[string]bool), - band: newRingBuffer(percentileWindow), + primaryBucket: make(map[string]map[string][]sample), + usdtBucket: make(map[string]map[string][]sample), + lastClosedPrice: make(map[string]float64), + depegStreak: make(map[string]int), + recoveryStreak: make(map[string]int), + depegActive: make(map[string]bool), + band: newRingBuffer(percentileWindow), + recentByStableVenue: make(map[string]map[string][]recentSample), + } +} + +// corroboratedLocked answers: at time `at`, with this `stable` showing +// `price` on `exceptVenue`, has at least one OTHER venue published a +// reading in the last consensusWindow that is ALSO outside the +// consensusThreshold band in the SAME direction (both above peg or both +// below peg)? If yes, the print is a corroborated depeg signal and +// should be kept. If no, it's a single-venue glitch and should be +// dropped. Caller holds a.mu. +func (a *Aggregator) corroboratedLocked(stable, exceptVenue string, price float64, at time.Time) bool { + byVenue := a.recentByStableVenue[stable] + if byVenue == nil { + return false + } + cutoff := at.Add(-consensusWindow) + subjectAbove := price > 1.0 + for venue, samples := range byVenue { + if venue == exceptVenue { + continue + } + for _, ss := range samples { + if ss.receivedAt.Before(cutoff) { + continue + } + otherDev := math.Abs(ss.price - 1.0) + if otherDev <= consensusThreshold { + continue + } + if (ss.price > 1.0) == subjectAbove { + return true + } + } + } + return false +} + +// recordRecentLocked appends the sample to the rolling 30s buffer for +// (stable, venue) and prunes entries that fell out of the window. We +// record every observed sample (including ones we drop via the +// consensus rule) because the rule cares about "what has the source +// recently published," not "what we agreed to use." Caller holds a.mu. +func (a *Aggregator) recordRecentLocked(s sample) { + if a.recentByStableVenue[s.stable] == nil { + a.recentByStableVenue[s.stable] = make(map[string][]recentSample) + } + cutoff := s.receivedAt.Add(-consensusWindow) + prev := a.recentByStableVenue[s.stable][s.venue] + pruned := prev[:0] + for _, ss := range prev { + if !ss.receivedAt.Before(cutoff) { + pruned = append(pruned, ss) + } } + pruned = append(pruned, recentSample{price: s.price, receivedAt: s.receivedAt}) + a.recentByStableVenue[s.stable][s.venue] = pruned } // Ingest accepts a fresh sample from a source goroutine. -// Applies outlier rules synchronously; rejected samples are -// counted via the source counter but never reach the buckets. +// Applies the consensus outlier rule synchronously; rejected samples +// are counted via peg_source_call_total and never reach the buckets. +// +// Consensus outlier rule (replaces the old flat 20% drop / 10% cap): +// +// - >sanityCeiling (50%) off peg → drop. This is a parser bug, +// not a market event, on any USD-quoted stable. +// - >consensusThreshold (2%) off peg → keep only if at least one +// OTHER venue has also been outside the threshold in the SAME +// direction within the last consensusWindow (30s). This +// distinguishes a single-venue glitch (Kraken returns $0.97 +// while every other venue is at $1.00 = stale print, drop) from +// a corroborated depeg (Kraken AND Bitstamp both at $0.87 = real +// event, keep all samples so the percentile metric surfaces it). +// - ≤2% off peg → always keep, this is normal noise. +// +// Every sample (even rejected ones) is also recorded in the rolling +// 30s buffer so that two venues which depeg within the same 30s +// window can validate each other regardless of arrival order. +// +// This is the rule the Coinpaprika data team review identified as the +// single most important fix to the bench: the old 20% drop / 10% cap +// would clip USDC at $0.87 (March 2023 SVB) to $0.90, erasing the +// event in the percentile metric. func (a *Aggregator) Ingest(s sample) { dev := math.Abs(s.price - 1.0) - // Outlier rule 1: hard drop > 20%. - if dev > 0.20 { - pegSourceCallTotal.WithLabelValues(s.stable, s.venue, "dropped").Inc() + a.mu.Lock() + defer a.mu.Unlock() + + // Record in the rolling buffer before applying the rule so the + // next venue to observe a corroborating print can find this one. + a.recordRecentLocked(s) + + if dev > sanityCeiling { + pegSourceCallTotal.WithLabelValues(s.stable, s.venue, "dropped_sanity").Inc() return } - // Outlier rule 2: cap-and-tag 10-20% → cap at 10% so percentile - // stays meaningful, still in the bucket. - if dev > 0.10 { - if s.price > 1.0 { - s.price = 1.10 + if dev > consensusThreshold { + if a.corroboratedLocked(s.stable, s.venue, s.price, s.receivedAt) { + pegSourceCallTotal.WithLabelValues(s.stable, s.venue, "kept_corroborated").Inc() } else { - s.price = 0.90 + pegSourceCallTotal.WithLabelValues(s.stable, s.venue, "dropped_isolated").Inc() + return } } @@ -81,9 +193,6 @@ func (a *Aggregator) Ingest(s sample) { pegSourceCallTotal.WithLabelValues(s.stable, s.venue, "ok").Inc() pegSourceHealth.WithLabelValues(s.stable, s.venue).Set(1) - a.mu.Lock() - defer a.mu.Unlock() - // Roll the minute bucket if needed. now := s.receivedAt bucket := now.Truncate(minuteBucketSize) diff --git a/harnesses/stablecoin-peg/cmd/script/metrics.go b/harnesses/stablecoin-peg/cmd/script/metrics.go index 96aeb71a..846ea9a7 100644 --- a/harnesses/stablecoin-peg/cmd/script/metrics.go +++ b/harnesses/stablecoin-peg/cmd/script/metrics.go @@ -183,7 +183,7 @@ var ( pegSourceCallTotal = promauto.NewCounterVec( prometheus.CounterOpts{ Name: "peg_source_call_total", - Help: "Number of source polls broken down by result: ok, http_err, parse_err, stale, dropped (sample dropped by outlier rule).", + Help: "Number of source polls broken down by result. ok = sample accepted. http_err / parse_err / stale = source failure. dropped_sanity = sample >50% off peg (parser bug). dropped_isolated = sample >2% off peg with no corroborating venue in last 30s (single-venue glitch). kept_corroborated = sample >2% off peg confirmed by ≥1 other venue in last 30s (real depeg signal, sample kept).", }, []string{"stable", "venue", "result"}, )