exchange-monitor-go/cumulative.go

package main

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

// CmState represents a coin's cumulative move state.
type CmState string

const (
	CmNeutral CmState = "neutral"
	CmRising  CmState = "rising"   // strong upward consensus across exchanges
	CmFalling CmState = "falling"  // strong downward consensus across exchanges
)

// exChange holds a per-exchange price change percentage.
type exChange struct {
	name   string
	change float64
}

// CmEvent records a cumulative move state transition, persisted to DB.
type CmEvent struct {
	Coin       string  `json:"coin"`
	PrevState  string  `json:"prev_state"`
	NewState   string  `json:"new_state"`
	Direction  string  `json:"direction"`
	Score      float64 `json:"score"`       // avg_change% × ex_agree
	AvgChange  float64 `json:"avg_change"`  // average change% across all exchanges
	ExAgree    int     `json:"ex_agree"`
	ExTotal    int     `json:"ex_total"`
	BGChange1m float64 `json:"bg_1m"`
	HLChange1m float64 `json:"hl_1m"`
	BNChange1m float64 `json:"bn_1m"`
	OKXChange1m float64 `json:"okx_1m"`
	BGChange5m float64 `json:"bg_5m"`
	HLChange5m float64 `json:"hl_5m"`
	BNChange5m float64 `json:"bn_5m"`
	OKXChange5m float64 `json:"okx_5m"`
	Timestamp  int64   `json:"timestamp"`
}

// CmSnapshot is a point-in-time price snapshot for all exchanges for one coin.
type cmSnapshot struct {
	time  int64
	prices map[string]float64 // exchange → price
}

// CumulativeTracker monitors multi-exchange cumulative price changes.
// Takes 1-second snapshots, computes 1m/5m changes, detects consensus surges.
type CumulativeTracker struct {
	mu     sync.RWMutex
	coins  map[string][]cmSnapshot // coin → ring buffer of snapshots
	heads  map[string]int
	counts map[string]int

	// Per-coin state
	states    map[string]CmState
	prevState map[string]CmState

	// Ring buffer config
	maxSnapshots int // 5min worth at 1s = 300

	// Thresholds
	minExchanges int     // need at least this many exchanges with data (default: 3)
	surgePct1m  float64 // 1m change% threshold to trigger (default: 0.5%)
	surgePct5m  float64 // 5m change% threshold to trigger (default: 1.0%)

	// Event history (in-memory ring buffer)
	events     [maxTrendEvents]CmEvent
	eventsHead int
	eventsLen  int

	// Callback for DB persistence
	OnEvent func(CmEvent)
}

// NewCumulativeTracker creates a tracker with default thresholds.
func NewCumulativeTracker() *CumulativeTracker {
	return &CumulativeTracker{
		coins:        make(map[string][]cmSnapshot),
		heads:        make(map[string]int),
		counts:       make(map[string]int),
		states:       make(map[string]CmState),
		prevState:    make(map[string]CmState),
		maxSnapshots: 300, // 5min at 1s
		minExchanges: 3,
		surgePct1m:   0.5,  // 0.5% in 1min
		surgePct5m:   1.0,  // 1.0% in 5min
	}
}

// Record stores a price snapshot for a coin at the current time.
// Call this once per second with all exchange prices for each coin.
func (ct *CumulativeTracker) Record(coin string, prices map[string]float64) {
	ct.mu.Lock()
	defer ct.mu.Unlock()

	snap := cmSnapshot{
		time:   time.Now().UnixMilli(),
		prices: prices,
	}

	// Initialize buffer if needed
	if ct.coins[coin] == nil {
		ct.coins[coin] = make([]cmSnapshot, ct.maxSnapshots)
		ct.heads[coin] = 0
		ct.counts[coin] = 0
		ct.states[coin] = CmNeutral
		ct.prevState[coin] = CmNeutral
	}

	buf := ct.coins[coin]
	head := ct.heads[coin]
	buf[head] = snap
	ct.heads[coin] = (head + 1) % ct.maxSnapshots
	if ct.counts[coin] < ct.maxSnapshots {
		ct.counts[coin]++
	}
}

// GetCurrent returns current cumulative change info for all coins, sorted by score desc.
func (ct *CumulativeTracker) GetCurrent() []map[string]interface{} {
	ct.mu.RLock()
	defer ct.mu.RUnlock()

	var results []map[string]interface{}

	for coin, buf := range ct.coins {
		count := ct.counts[coin]
		if count < 10 {
			continue // not enough data
		}
		head := ct.heads[coin]

		// Get current snapshot (most recent)
		currentIdx := (head - 1 + ct.maxSnapshots) % ct.maxSnapshots
		current := buf[currentIdx]
		if current.time == 0 {
			continue
		}
		if len(current.prices) < ct.minExchanges {
			continue
		}

		// Find snapshots from ~60s ago and ~300s ago
		now := current.time
		oneMinAgo := now - 60000
		fiveMinAgo := now - 300000
		var snap1m, snap5m *cmSnapshot
		var found1m, found5m bool

		// Walk backwards from current to find closest snapshots
		for i := 0; i < count && i < ct.maxSnapshots; i++ {
			idx := (currentIdx - i + ct.maxSnapshots) % ct.maxSnapshots
			s := &buf[idx]
			if s.time == 0 {
				continue
			}
			if !found1m && s.time <= oneMinAgo {
				snap1m = s
				found1m = true
			}
			if !found5m && s.time <= fiveMinAgo {
				snap5m = s
				found5m = true
			}
		}
		if !found1m {
			// Use oldest available as 1m approximation
			continue
		}

		// Compute 1m changes per exchange
		var changes1m, changes5m []exChange

		for ex, curP := range current.prices {
			if curP <= 0 {
				continue
			}
			if oldP, ok := snap1m.prices[ex]; ok && oldP > 0 {
				chg := (curP - oldP) / oldP * 100
				changes1m = append(changes1m, exChange{name: ex, change: chg})
			}
			if found5m && snap5m != nil {
				if oldP, ok := snap5m.prices[ex]; ok && oldP > 0 {
					chg := (curP - oldP) / oldP * 100
					changes5m = append(changes5m, exChange{name: ex, change: chg})
				}
			}
		}

		if len(changes1m) < ct.minExchanges {
			continue
		}

		// Compute averages and agreement
		var sum1m, sum5m float64
		agreeUp1m, agreeDown1m := 0, 0
		agreeUp5m, agreeDown5m := 0, 0

		for _, c := range changes1m {
			sum1m += c.change
			if c.change > 0.001 {
				agreeUp1m++
			} else if c.change < -0.001 {
				agreeDown1m++
			}
		}

		for _, c := range changes5m {
			sum5m += c.change
			if c.change > 0.005 {
				agreeUp5m++
			} else if c.change < -0.005 {
				agreeDown5m++
			}
		}

		avg1m := sum1m / float64(len(changes1m))
		var avg5m float64
		if len(changes5m) >= ct.minExchanges {
			avg5m = sum5m / float64(len(changes5m))
		}

		// Determine direction and agreement
		majorityDir := "up"
		majority := agreeUp1m
		if agreeDown1m > agreeUp1m {
			majorityDir = "down"
			majority = agreeDown1m
		}

		// Score: abs(avg1m) × agreement (weighted by magnitude)
		absAvg := math.Abs(avg1m)
		score := absAvg * float64(majority)

		entry := map[string]interface{}{
			"coin":       coin,
			"avg_1m":     math.Round(avg1m*10000) / 10000,
			"avg_5m":     math.Round(avg5m*10000) / 10000,
			"score":      math.Round(score*100) / 100,
			"direction":  majorityDir,
			"ex_agree":   majority,
			"ex_total":   len(changes1m),
		}

		// Individual exchange changes
		for _, c := range changes1m {
			entry[c.name+"_1m"] = math.Round(c.change*10000) / 10000
		}
		if len(changes5m) >= ct.minExchanges {
			for _, c := range changes5m {
				entry[c.name+"_5m"] = math.Round(c.change*10000) / 10000
			}
		}

		// Current state
		entry["state"] = string(ct.states[coin])

		results = append(results, entry)
	}

	// Sort by score descending
	sort.Slice(results, func(i, j int) bool {
		si, _ := results[i]["score"].(float64)
		sj, _ := results[j]["score"].(float64)
		return si > sj
	})

	if len(results) > 100 {
		results = results[:100]
	}
	return results
}

// Tick runs one detection cycle: updates state machines, fires events.
func (ct *CumulativeTracker) Tick() {
	ct.mu.Lock()
	defer ct.mu.Unlock()

	for coin, buf := range ct.coins {
		count := ct.counts[coin]
		if count < 60 {
			continue // need at least 1min of data
		}
		head := ct.heads[coin]
		currentIdx := (head - 1 + ct.maxSnapshots) % ct.maxSnapshots
		current := buf[currentIdx]
		if current.time == 0 || len(current.prices) < ct.minExchanges {
			continue
		}

		// Find 1min ago snapshot
		oneMinAgo := current.time - 60000
		var snap1m *cmSnapshot
		for i := 0; i < count && i < ct.maxSnapshots; i++ {
			idx := (currentIdx - i + ct.maxSnapshots) % ct.maxSnapshots
			s := &buf[idx]
			if s.time > 0 && s.time <= oneMinAgo {
				snap1m = s
				break
			}
		}
		if snap1m == nil {
			continue
		}

		// Compute 1m changes
		var changes []exChange
		for ex, curP := range current.prices {
			if curP <= 0 {
				continue
			}
			if oldP, ok := snap1m.prices[ex]; ok && oldP > 0 {
				chg := (curP - oldP) / oldP * 100
				changes = append(changes, exChange{name: ex, change: chg})
			}
		}
		if len(changes) < ct.minExchanges {
			continue
		}

		var sum float64
		agreeUp, agreeDown := 0, 0
		for _, c := range changes {
			sum += c.change
			if c.change > 0.001 {
				agreeUp++
			} else if c.change < -0.001 {
				agreeDown++
			}
		}
		avg := sum / float64(len(changes))
		majority := agreeUp
		majorityDir := "up"
		if agreeDown > agreeUp {
			majority = agreeDown
			majorityDir = "down"
		}

		// Determine new state
		absAvg := math.Abs(avg)
		newState := ct.states[coin]

		// Map exchange changes for individual values
		exMap := make(map[string]float64)
		for _, c := range changes {
			exMap[c.name] = c.change
		}

		if absAvg >= ct.surgePct1m && majority >= ct.minExchanges {
			if majorityDir == "up" {
				if ct.states[coin] == CmNeutral || ct.states[coin] == CmFalling {
					ct.prevState[coin] = ct.states[coin]
					ct.states[coin] = CmRising
					newState = CmRising
					// Fire event
					ev := ct.makeEvent(coin, string(ct.prevState[coin]), "rising", majorityDir,
						absAvg*float64(majority), avg, majority, len(changes), exMap)
					ct.storeEvent(ev)
				}
			} else {
				if ct.states[coin] == CmNeutral || ct.states[coin] == CmRising {
					ct.prevState[coin] = ct.states[coin]
					ct.states[coin] = CmFalling
					newState = CmFalling
					ev := ct.makeEvent(coin, string(ct.prevState[coin]), "falling", majorityDir,
						absAvg*float64(majority), avg, majority, len(changes), exMap)
					ct.storeEvent(ev)
				}
			}
		} else if absAvg < ct.surgePct1m*0.3 || majority < 2 {
			if ct.states[coin] != CmNeutral {
				ct.prevState[coin] = ct.states[coin]
				ct.states[coin] = CmNeutral
				ev := ct.makeEvent(coin, string(ct.prevState[coin]), "neutral", majorityDir,
					absAvg*float64(majority), avg, majority, len(changes), exMap)
				ct.storeEvent(ev)
			}
		}
		_ = newState
	}
}

// makeEvent builds a CmEvent struct with 1m and 5m data.
func (ct *CumulativeTracker) makeEvent(coin, prevState, newState, direction string, score, avgChange float64, exAgree, exTotal int, exChanges map[string]float64) CmEvent {
	return CmEvent{
		Coin:       coin,
		PrevState:  prevState,
		NewState:   newState,
		Direction:  direction,
		Score:      math.Round(score*100) / 100,
		AvgChange:  math.Round(avgChange*10000) / 10000,
		ExAgree:    exAgree,
		ExTotal:    exTotal,
		BGChange1m: exChanges[ExBitget],
		HLChange1m: exChanges[ExHyperLiquid],
		BNChange1m: exChanges[ExBinance],
		OKXChange1m: exChanges[ExOKX],
		Timestamp:  time.Now().UnixMilli(),
	}
}

// storeEvent adds to ring buffer and fires callback.
func (ct *CumulativeTracker) storeEvent(ev CmEvent) {
	ct.events[ct.eventsHead] = ev
	ct.eventsHead = (ct.eventsHead + 1) % maxTrendEvents
	if ct.eventsLen < maxTrendEvents {
		ct.eventsLen++
	}
	if ct.OnEvent != nil {
		ct.OnEvent(ev)
	}
}

// GetEvents returns stored events, newest first.
func (ct *CumulativeTracker) GetEvents(limit int) []CmEvent {
	ct.mu.RLock()
	defer ct.mu.RUnlock()

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

// GetTopCoins returns top surging coins by score.
func (ct *CumulativeTracker) GetTopCoins(limit int) []map[string]interface{} {
	all := ct.GetCurrent()
	if limit > 0 && limit < len(all) {
		return all[:limit]
	}
	return all
}