exchange-monitor-go/surge_detector.go

package main

import (
	"log"
	"math"
	"sort"
	"sync"
	"time"
)

// SurgeDetector detects anomalous 3-exchange max spreads using per-coin adaptive baselines.
// Theory: when a coin starts moving sharply, different exchanges update at different speeds,
// creating a temporary spike in inter-exchange spread. This detector captures that moment.
type SurgeDetector struct {
	mu    sync.Mutex
	coins map[string]*coinSurgeState

	// Config
	enabled         bool
	windowSize      int     // rolling window samples (default: 600 = ~30s at 50ms tick)
	baselineMul     float64 // baseline * N = threshold (default: 3.0)
	minAbsSpreadPct float64 // minimum absolute spread % to trigger (default: 0.05)
	cooldownSec     int     // seconds between alerts for same coin (default: 60)

	// Recent events (ring buffer)
	events   []SurgeEvent
	eventIdx int
	maxEvents int

	// DB persistence callback
	onEvent func(SurgeEvent)
}

// coinSurgeState holds per-coin adaptive baseline data.
type coinSurgeState struct {
	spreads      []float64 // rolling window of recent spread values
	lastAlertAt  time.Time
}

// SurgeEvent represents a detected surge anomaly.
type SurgeEvent struct {
	Timestamp       time.Time `json:"timestamp"`
	Coin            string    `json:"coin"`
	BnPrice         float64   `json:"bn_price"`
	OkxPrice        float64   `json:"okx_price"`
	BgPrice         float64   `json:"bg_price"`
	SpreadPct       float64   `json:"spread_pct"`       // current 3-exchange max spread
	BaselinePct     float64   `json:"baseline_pct"`     // adaptive baseline at time of event
	ThresholdPct    float64   `json:"threshold_pct"`    // trigger threshold
	Ratio           float64   `json:"ratio"`            // spread / threshold
	Direction       string    `json:"direction"`        // "up" or "down"
	LeadingExchange string    `json:"leading_exchange"` // which exchange moved first/furthest
	MidPrice        float64   `json:"mid_price"`        // median of 3 prices
}

// SurgeSnapshot holds current spread/baseline state for a coin (SSE push).
type SurgeSnapshot struct {
	Coin         string  `json:"coin"`
	SpreadPct    float64 `json:"spread_pct"`
	BaselinePct  float64 `json:"baseline_pct"`
	ThresholdPct float64 `json:"threshold_pct"`
	Direction    string  `json:"direction,omitempty"` // "up"/"down" if currently surging
	WindowSize   int     `json:"window_size"`         // current number of samples in window
}

func NewSurgeDetector() *SurgeDetector {
	return &SurgeDetector{
		coins:     make(map[string]*coinSurgeState),
		events:    make([]SurgeEvent, 200),
		maxEvents: 200,
	}
}

// Configure sets detection parameters.
func (sd *SurgeDetector) Configure(windowSize int, baselineMul, minAbsSpreadPct float64, cooldownSec int) {
	sd.enabled = true
	sd.windowSize = windowSize
	sd.baselineMul = baselineMul
	sd.minAbsSpreadPct = minAbsSpreadPct
	sd.cooldownSec = cooldownSec
}

// SetOnEvent sets the DB persistence callback.
func (sd *SurgeDetector) SetOnEvent(fn func(SurgeEvent)) {
	sd.onEvent = fn
}

// Tick processes one snapshot tick, detecting surges for all coins.
// Returns newly detected events for immediate SSE broadcast.
func (sd *SurgeDetector) Tick(snap map[string]map[string]float64) []SurgeEvent {
	if !sd.enabled {
		return nil
	}

	sd.mu.Lock()
	defer sd.mu.Unlock()

	var newEvents []SurgeEvent
	now := time.Now()

	for _, coin := range TrackedCoins {
		exMap := snap[coin.Name]
		if exMap == nil {
			continue
		}

		bnP := exMap[ExBinance]
		okxP := exMap[ExOKX]
		bgP := exMap[ExBitget]

		// Need at least 2 exchanges
		prices := []float64{}
		if bnP > 0 { prices = append(prices, bnP) }
		if okxP > 0 { prices = append(prices, okxP) }
		if bgP > 0 { prices = append(prices, bgP) }
		if len(prices) < 2 {
			continue
		}

		// Compute 3-exchange max spread
		minP, maxP := prices[0], prices[0]
		for _, p := range prices[1:] {
			if p < minP { minP = p }
			if p > maxP { maxP = p }
		}
		spread := (maxP - minP) / minP * 100

		// Get or create coin state
		state, exists := sd.coins[coin.Name]
		if !exists {
			state = &coinSurgeState{
				spreads: make([]float64, 0, sd.windowSize),
			}
			sd.coins[coin.Name] = state
		}

		// Add spread to rolling window
		state.spreads = append(state.spreads, spread)
		if len(state.spreads) > sd.windowSize {
			state.spreads = state.spreads[len(state.spreads)-sd.windowSize:]
		}

		// Need minimum samples for baseline (at least 10)
		if len(state.spreads) < 10 {
			continue
		}

		// Compute baseline = median of recent spreads
		baseline := median(state.spreads)

		// Threshold = baseline * multiplier, but at least minAbsSpreadPct
		threshold := baseline * sd.baselineMul
		if threshold < sd.minAbsSpreadPct {
			threshold = sd.minAbsSpreadPct
		}

		// Check if spread exceeds threshold AND cooldown has passed
		if spread < threshold {
			continue
		}
		if !state.lastAlertAt.IsZero() && now.Sub(state.lastAlertAt).Seconds() < float64(sd.cooldownSec) {
			continue
		}

		// Surge detected — determine direction
		midPrice := median(prices)
		var direction, leadingEx string

		// Find max and min exchanges for reporting
		exPrices := map[string]float64{}
		if bnP > 0 { exPrices[ExBinance] = bnP }
		if okxP > 0 { exPrices[ExOKX] = okxP }
		if bgP > 0 { exPrices[ExBitget] = bgP }

		var maxEx, minEx string
		var maxVal, minVal float64 = -1, math.MaxFloat64
		for ex, p := range exPrices {
			if p > maxVal { maxVal = p; maxEx = ex }
			if p < minVal { minVal = p; minEx = ex }
		}

		// Direction: if highest is further from median than lowest → up, else down
		if (maxVal - midPrice) > (midPrice - minVal) {
			direction = "up"
			leadingEx = maxEx
		} else {
			direction = "down"
			leadingEx = minEx
		}

		ratio := spread / threshold

		event := SurgeEvent{
			Timestamp:       now,
			Coin:            coin.Name,
			BnPrice:         bnP,
			OkxPrice:        okxP,
			BgPrice:         bgP,
			SpreadPct:       math.Round(spread*10000) / 10000,
			BaselinePct:     math.Round(baseline*10000) / 10000,
			ThresholdPct:    math.Round(threshold*10000) / 10000,
			Ratio:           math.Round(ratio*100) / 100,
			Direction:       direction,
			LeadingExchange: leadingEx,
			MidPrice:        math.Round(midPrice*10000) / 10000,
		}

		state.lastAlertAt = now
		newEvents = append(newEvents, event)

		// Store in ring buffer
		sd.events[sd.eventIdx%sd.maxEvents] = event
		sd.eventIdx++

		log.Printf("[Surge] %s %s surge detected: spread=%.4f%% baseline=%.4f%% threshold=%.4f%% ratio=%.1fx leading=%s",
			coin.Name, direction, event.SpreadPct, event.BaselinePct, event.ThresholdPct, event.Ratio, leadingEx)

		// Persist to DB if callback set
		if sd.onEvent != nil {
			sd.onEvent(event)
		}
	}

	return newEvents
}

// GetRecentEvents returns the most recent N surge events.
func (sd *SurgeDetector) GetRecentEvents(n int) []SurgeEvent {
	sd.mu.Lock()
	defer sd.mu.Unlock()

	if n <= 0 || n > sd.maxEvents {
		n = sd.maxEvents
	}

	total := sd.eventIdx
	if total > sd.maxEvents {
		total = sd.maxEvents
	}

	result := make([]SurgeEvent, 0, total)
	for i := 0; i < total; i++ {
		idx := (sd.eventIdx - total + i) % sd.maxEvents
		if idx < 0 {
			idx += sd.maxEvents
		}
		ev := sd.events[idx]
		if ev.Coin != "" {
			result = append(result, ev)
		}
	}

	// Return at most n, most recent first
	if len(result) <= n {
		// Reverse to get newest first
		for i, j := 0, len(result)-1; i < j; i, j = i+1, j-1 {
			result[i], result[j] = result[j], result[i]
		}
		return result
	}

	// Take last n and reverse
	out := make([]SurgeEvent, n)
	for i := 0; i < n; i++ {
		out[i] = result[len(result)-1-i]
	}
	return out
}

// Snapshot returns current spread/baseline state for all coins (SSE push).
func (sd *SurgeDetector) Snapshot() []SurgeSnapshot {
	sd.mu.Lock()
	defer sd.mu.Unlock()

	var result []SurgeSnapshot
	now := time.Now()

	for _, coin := range TrackedCoins {
		state, exists := sd.coins[coin.Name]
		if !exists || len(state.spreads) < 10 {
			continue
		}

		currentSpread := state.spreads[len(state.spreads)-1]
		baseline := median(state.spreads)
		threshold := baseline * sd.baselineMul
		if threshold < sd.minAbsSpreadPct {
			threshold = sd.minAbsSpreadPct
		}

		snap := SurgeSnapshot{
			Coin:         coin.Name,
			SpreadPct:    math.Round(currentSpread*10000) / 10000,
			BaselinePct:  math.Round(baseline*10000) / 10000,
			ThresholdPct: math.Round(threshold*10000) / 10000,
			WindowSize:   len(state.spreads),
		}

		// Check if currently surging (within cooldown)
		if currentSpread >= threshold && !state.lastAlertAt.IsZero() && now.Sub(state.lastAlertAt).Seconds() < float64(sd.cooldownSec) {
			if state.spreads[len(state.spreads)-1] >= threshold {
				snap.Direction = "up" // placeholder, real direction calculated in Tick
			}
		}

		result = append(result, snap)
	}

	// Sort by spread descending
	sort.Slice(result, func(i, j int) bool {
		return result[i].SpreadPct > result[j].SpreadPct
	})

	// Limit to top 50
	if len(result) > 50 {
		result = result[:50]
	}

	return result
}

// median computes the median of a slice of float64s.
// The input slice is NOT modified.
func median(vals []float64) float64 {
	if len(vals) == 0 {
		return 0
	}
	sorted := make([]float64, len(vals))
	copy(sorted, vals)
	sort.Float64s(sorted)
	n := len(sorted)
	if n%2 == 1 {
		return sorted[n/2]
	}
	return (sorted[n/2-1] + sorted[n/2]) / 2
}