exchange-monitor-go/trend.go

package main

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

// TrendState represents the state of a coin's trend detection lifecycle.
type TrendState string

const (
	TrendIdle       TrendState = "idle"
	TrendAlert      TrendState = "alert"      // anomaly detected, awaiting confirmation
	TrendConfirmed  TrendState = "confirmed"  // trend confirmed by 3+ exchanges
	TrendExhausting TrendState = "exhausting" // momentum fading
)

// TrendDirection indicates the direction of a detected trend.
type TrendDirection string

const (
	TrendUp   TrendDirection = "up"
	TrendDown TrendDirection = "down"
)

// TrendEvent records a state transition for one coin, persisted in a ring buffer for UI display.
type TrendEvent struct {
	Coin       string  `json:"coin"`
	PrevState  string  `json:"prev_state"`
	NewState   string  `json:"new_state"`
	Direction  string  `json:"direction"`
	ZScore     float64 `json:"z_score"`
	Volatility float64 `json:"volatility"`
	BGChange   float64 `json:"bg_change"`
	HLChange   float64 `json:"hl_change"`
	BNChange   float64 `json:"bn_change"`
	OKXChange  float64 `json:"okx_change"`
	ExAgree    int     `json:"ex_agree"`
	ExTotal    int     `json:"ex_total"`
	Timestamp  int64   `json:"timestamp"`
}

const maxTrendEvents = 500

// TrendEntry is one coin's trend data sent via SSE.
type TrendEntry struct {
	Coin         string        `json:"coin"`
	State        TrendState    `json:"state"`
	Direction    TrendDirection `json:"direction"`
	AnomalyScore float64       `json:"anomaly_score"` // max z-score across all exchanges
	Volatility   float64       `json:"volatility"`    // current EMA volatility baseline
	BGChange     float64       `json:"bg_change"`     // 60s change %
	HLChange     float64       `json:"hl_change"`
	BNChange     float64       `json:"bn_change"`
	OKXChange    float64       `json:"okx_change"`
	AlertedAt    int64         `json:"alerted_at,omitempty"`   // unix millis
	ConfirmedAt  int64         `json:"confirmed_at,omitempty"` // unix millis
	Duration     string        `json:"duration,omitempty"`     // how long in current state
	ExChanges    int           `json:"ex_changes"`             // how many exchanges agree on direction
}

// exchangeChange holds the 15s change % for one exchange.
type exchangeChange struct {
	name   string
	change float64
}

// trendCoinState tracks the state machine for one coin.
type trendCoinState struct {
	state       TrendState
	direction   TrendDirection
	anomalyScore float64
	peakScore   float64 // highest z-score seen during current cycle
	volatility  float64

	alertedAt   time.Time
	confirmedAt time.Time
	stateSince  time.Time

	// For confirmation: track how many consecutive ticks agree
	confirmCount int
	misalignCount int
}

// TrendDetector detects price anomalies and confirms trends across exchanges.
type TrendDetector struct {
	mu     sync.RWMutex
	coins  map[string]*trendCoinState
	momentum *MomentumTracker

	// Configuration
	baselineWindow int     // ticks for EMA baseline (default: 600 = 30s at 50ms)
	anomalyMul     float64 // z-score multiplier for alert threshold (default: 3.0)
	confirmTicks   int     // ticks needed for confirmation (default: 3)
	alertCooldown  int64   // ms cooldown between alerts for same coin (default: 60000)

	// Event history ring buffer (for UI display)
	events     [maxTrendEvents]TrendEvent
	eventsHead int
	eventsLen  int

	// OnEvent is called whenever a state transition is recorded.
	// Set this to persist events to database.
	OnEvent func(TrendEvent)
}

// NewTrendDetector creates a trend detector that reads from MomentumTracker.
func NewTrendDetector(mt *MomentumTracker) *TrendDetector {
	return &TrendDetector{
		coins:          make(map[string]*trendCoinState),
		momentum:       mt,
		baselineWindow: 600,  // ~30s at 50ms tick
		anomalyMul:     3.0,  // 3 sigma
		confirmTicks:   3,    // 3 consecutive ticks
		alertCooldown:  60000, // 1 min
	}
}

// Configure sets trend detection parameters.
func (td *TrendDetector) Configure(baselineWindow int, anomalyMul float64, confirmTicks int, alertCooldownMs int64) {
	td.mu.Lock()
	defer td.mu.Unlock()
	if baselineWindow > 0 {
		td.baselineWindow = baselineWindow
	}
	if anomalyMul > 0 {
		td.anomalyMul = anomalyMul
	}
	if confirmTicks > 0 {
		td.confirmTicks = confirmTicks
	}
	if alertCooldownMs > 0 {
		td.alertCooldown = alertCooldownMs
	}
}

// recordEvent stores a state transition in the ring buffer.
func (td *TrendDetector) recordEvent(coin, prevState, newState, direction string, zScore, vola float64, bgC, hlC, bnC, okxC float64, exAgree, exTotal int) {
	ev := TrendEvent{
		Coin:       coin,
		PrevState:  prevState,
		NewState:   newState,
		Direction:  direction,
		ZScore:     math.Round(zScore*100) / 100,
		Volatility: math.Round(vola*10000) / 10000,
		BGChange:   bgC,
		HLChange:   hlC,
		BNChange:   bnC,
		OKXChange:  okxC,
		ExAgree:    exAgree,
		ExTotal:    exTotal,
		Timestamp:  time.Now().UnixMilli(),
	}
	td.events[td.eventsHead] = ev
	td.eventsHead = (td.eventsHead + 1) % maxTrendEvents
	if td.eventsLen < maxTrendEvents {
		td.eventsLen++
	}

	// Fire callback for DB persistence
	if td.OnEvent != nil {
		td.OnEvent(ev)
	}
}

// GetEvents returns trend event history, newest first.
func (td *TrendDetector) GetEvents(limit int) []TrendEvent {
	td.mu.RLock()
	defer td.mu.RUnlock()

	n := td.eventsLen
	if limit > 0 && limit < n {
		n = limit
	}
	result := make([]TrendEvent, 0, n)
	for i := 0; i < n; i++ {
		idx := (td.eventsHead - 1 - i + maxTrendEvents) % maxTrendEvents
		if td.events[idx].Timestamp == 0 {
			continue
		}
		result = append(result, td.events[idx])
	}
	return result
}

// Tick runs one iteration of trend detection.
// Reads exchange changes from MomentumTracker buffers, computes volatility baselines,
// and advances the state machine for each coin.
func (td *TrendDetector) Tick() {
	// Get all momentum entries to access exchange changes
	entries := td.momentum.Snapshot(0)
	if len(entries) == 0 {
		return
	}

	td.mu.Lock()
	defer td.mu.Unlock()

	for _, entry := range entries {
		// Collect 60s changes from all 3 exchanges
		var changes []exchangeChange
		if entry.BG60s != 0 {
			changes = append(changes, exchangeChange{name: ExBitget, change: entry.BG60s})
		}
		if entry.BN60s != 0 {
			changes = append(changes, exchangeChange{name: ExBinance, change: entry.BN60s})
		}
		if entry.OKX60s != 0 {
			changes = append(changes, exchangeChange{name: ExOKX, change: entry.OKX60s})
		}

		if len(changes) < 2 {
			continue // need at least 2 exchanges for reliable detection
		}

		// Compute aggregate stats
		_, std := meanStdDev(changes)
		maxAbs := 0.0
		agreeUp := 0
		agreeDown := 0
		for _, c := range changes {
			abs := math.Abs(c.change)
			if abs > maxAbs {
				maxAbs = abs
			}
			if c.change > 0.001 {
				agreeUp++
			} else if c.change < -0.001 {
				agreeDown++
			}
		}

		// Z-score: how anomalous is the max movement?
		var zScore float64
		if std > 0.0001 {
			zScore = maxAbs / std
		}

		// Update or create coin state
		cs, exists := td.coins[entry.Coin]
		if !exists {
			cs = &trendCoinState{
				state:      TrendIdle,
				stateSince: time.Now(),
			}
			td.coins[entry.Coin] = cs
		}

		// Update volatility baseline (EMA of maxAbs)
		if cs.volatility == 0 {
			cs.volatility = maxAbs
		} else {
			alpha := 2.0 / float64(td.baselineWindow+1)
			cs.volatility = cs.volatility*(1-alpha) + maxAbs*alpha
		}

		// Update anomaly score — track the peak during the current cycle
		cs.anomalyScore = zScore
		if zScore > cs.peakScore {
			cs.peakScore = zScore
		}

		// Determine majority direction
		majorityDir := TrendUp
		majorityCount := agreeUp
		if agreeDown > agreeUp {
			majorityDir = TrendDown
			majorityCount = agreeDown
		}

		// State machine transitions
		now := time.Now()
		switch cs.state {
		case TrendIdle:
			// Alert if z-score exceeds threshold AND majority exchanges agree
			if zScore >= td.anomalyMul && majorityCount >= 3 {
				cs.state = TrendAlert
				cs.direction = majorityDir
				cs.alertedAt = now
				cs.stateSince = now
				cs.peakScore = zScore
				cs.confirmCount = 1
				cs.misalignCount = 0
				td.recordEvent(entry.Coin, "idle", "alert", string(majorityDir),
					zScore, cs.volatility, entry.BG60s, 0, entry.BN60s, entry.OKX60s,
					majorityCount, len(changes))
			}

		case TrendAlert:
			// Check if majority still agrees
			if majorityCount >= 3 && majorityDir == cs.direction {
				cs.confirmCount++
				cs.misalignCount = 0
				if cs.confirmCount >= td.confirmTicks {
					cs.state = TrendConfirmed
					cs.confirmedAt = now
					cs.stateSince = now
					td.recordEvent(entry.Coin, "alert", "confirmed", string(cs.direction),
						zScore, cs.volatility, entry.BG60s, 0, entry.BN60s, entry.OKX60s,
						majorityCount, len(changes))
				}
			} else {
				cs.misalignCount++
				if cs.misalignCount >= td.confirmTicks {
					// Failed to confirm — back to idle
					cs.state = TrendIdle
					cs.stateSince = now
					cs.confirmCount = 0
					cs.misalignCount = 0
					td.recordEvent(entry.Coin, "alert", "idle", string(cs.direction),
						zScore, cs.volatility, entry.BG60s, 0, entry.BN60s, entry.OKX60s,
						majorityCount, len(changes))
				}
			}

		case TrendConfirmed:
			// Check if momentum is exhausting (fewer than 3 exchanges agree)
			// Also track if z-score drops below threshold
			if majorityCount < 2 || zScore < td.anomalyMul*0.5 {
				cs.state = TrendExhausting
				cs.stateSince = now
				td.recordEvent(entry.Coin, "confirmed", "exhausting", string(cs.direction),
					zScore, cs.volatility, entry.BG60s, 0, entry.BN60s, entry.OKX60s,
					majorityCount, len(changes))
			}

		case TrendExhausting:
			// After exhausting, go back to idle
			if time.Since(cs.stateSince) > 5*time.Second {
				cs.state = TrendIdle
				cs.stateSince = now
				cs.confirmCount = 0
				cs.misalignCount = 0
				td.recordEvent(entry.Coin, "exhausting", "idle", string(cs.direction),
					zScore, cs.volatility, entry.BG60s, 0, entry.BN60s, entry.OKX60s,
					majorityCount, len(changes))
			}
			// Also immediately go to idle if below threshold
			if zScore < td.anomalyMul*0.3 || majorityCount < 1 {
				cs.state = TrendIdle
				cs.stateSince = now
				cs.confirmCount = 0
				cs.misalignCount = 0
				td.recordEvent(entry.Coin, "exhausting", "idle", string(cs.direction),
					zScore, cs.volatility, entry.BG60s, 0, entry.BN60s, entry.OKX60s,
					majorityCount, len(changes))
			}
		}
	}

	// Cleanup stale entries that never even reached alert state (> 120s)
	cutoff := time.Now().Add(-120 * time.Second)
	for coin, cs := range td.coins {
		if cs.state == TrendIdle && cs.alertedAt.IsZero() && cs.stateSince.Before(cutoff) {
			delete(td.coins, coin)
		}
	}
}

// Snapshot returns current trend state for all coins.
func (td *TrendDetector) Snapshot() []TrendEntry {
	td.mu.RLock()
	defer td.mu.RUnlock()

	entries := td.momentum.Snapshot(0)
	entryMap := make(map[string]MomentumEntry, len(entries))
	for _, e := range entries {
		entryMap[e.Coin] = e
	}

	var result []TrendEntry
	for coin, cs := range td.coins {
		// Skip coins that never even reached alert state
		if cs.state == TrendIdle && cs.alertedAt.IsZero() {
			continue
		}

		entry := TrendEntry{
			Coin:         coin,
			State:        cs.state,
			Direction:    cs.direction,
			AnomalyScore: math.Round(cs.peakScore*100) / 100,
			Volatility:   math.Round(cs.volatility*10000) / 10000,
			ExChanges:    0,
		}

		if me, ok := entryMap[coin]; ok {
			entry.BGChange = me.BG15s
			entry.HLChange = 0
			entry.BNChange = me.BN15s
			entry.OKXChange = me.OKX15s

			// Count how many exchanges agree with the trend direction
			agree := 0
			changes := []float64{entry.BGChange, entry.BNChange, entry.OKXChange}
			for _, c := range changes {
				if cs.direction == TrendUp && c > 0.001 {
					agree++
				} else if cs.direction == TrendDown && c < -0.001 {
					agree++
				}
			}
			entry.ExChanges = agree
		}

		if !cs.alertedAt.IsZero() {
			entry.AlertedAt = cs.alertedAt.UnixMilli()
		}
		if !cs.confirmedAt.IsZero() {
			entry.ConfirmedAt = cs.confirmedAt.UnixMilli()
		}

		// Duration in current state
		dur := time.Since(cs.stateSince).Round(time.Second)
		entry.Duration = dur.String()

		result = append(result, entry)
	}

	// Sort: confirmed first, then alert, then exhausting, then idle (completed)
	sort.Slice(result, func(i, j int) bool {
		order := map[TrendState]int{
			TrendConfirmed:  0,
			TrendAlert:      1,
			TrendExhausting: 2,
			TrendIdle:       3,
		}
		oi := order[result[i].State]
		oj := order[result[j].State]
		if oi != oj {
			return oi < oj
		}
		return result[i].AnomalyScore > result[j].AnomalyScore
	})

	return result
}

// meanStdDev computes mean and standard deviation of exchange change values.
func meanStdDev(changes []exchangeChange) (mean, stdDev float64) {
	if len(changes) == 0 {
		return 0, 0
	}
	var sum float64
	for _, c := range changes {
		sum += c.change
	}
	mean = sum / float64(len(changes))

	var varianceSum float64
	for _, c := range changes {
		diff := c.change - mean
		varianceSum += diff * diff
	}
	variance := varianceSum / float64(len(changes))
	stdDev = math.Sqrt(variance)

	return mean, stdDev
}

// IsTrending returns true if the given coin is in confirmed trend state.
func (td *TrendDetector) IsTrending(coin string) bool {
	td.mu.RLock()
	defer td.mu.RUnlock()
	cs, ok := td.coins[coin]
	return ok && cs.state == TrendConfirmed
}

// IsAnomalous returns true if the coin is in alert or confirmed trend state.
func (td *TrendDetector) IsAnomalous(coin string) bool {
	td.mu.RLock()
	defer td.mu.RUnlock()
	cs, ok := td.coins[coin]
	return ok && (cs.state == TrendAlert || cs.state == TrendConfirmed)
}

// State returns the human-readable trend state for a coin, or empty string if unknown.
func (td *TrendDetector) State(coin string) string {
	td.mu.RLock()
	defer td.mu.RUnlock()
	cs, ok := td.coins[coin]
	if !ok {
		return ""
	}
	return string(cs.state)
}

// GetTrendingCoins returns all coins currently in confirmed trend.
func (td *TrendDetector) GetTrendingCoins() map[string]TrendDirection {
	td.mu.RLock()
	defer td.mu.RUnlock()
	result := make(map[string]TrendDirection)
	for coin, cs := range td.coins {
		if cs.state == TrendConfirmed {
			result[coin] = cs.direction
		}
	}
	return result
}