feat(peg): multi-venue consensus outlier rule

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:

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,48 +64,135 @@ 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
  }
 	}
 
 	pegRawPrice.WithLabelValues(s.stable, s.venue, string(s.quote)).Set(s.price)
 	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)

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"},
 	)