(OraklNode) Improve outlier removal logic (#2165)

* feat: improve outlier removal logic

* feat: reduce copy

* feat: limit maximum outliers to be removed

* fix: skip outlier removal for small datasets

* fix: update based on feedback

* feat: update logic

node/go.mod

@@ -7,7 +7,6 @@ require (
 	github.com/elazarl/goproxy v0.0.0-20231117061959-7cc037d33fb5
 	github.com/go-playground/validator v9.31.0+incompatible
 	github.com/go-playground/validator/v10 v10.18.0
-	github.com/gofiber/contrib/websocket v1.3.2
 	github.com/gofiber/fiber/v2 v2.52.5
 	github.com/google/go-github v17.0.0+incompatible
 	github.com/google/uuid v1.6.0
@@ -29,6 +28,8 @@ require (
 	nhooyr.io/websocket v1.8.11
 )
 
+require github.com/montanaflynn/stats v0.7.1
+
 require (
 	cloud.google.com/go/compute v1.19.1 // indirect
 	cloud.google.com/go/compute/metadata v0.2.3 // indirect
@@ -70,7 +71,6 @@ require (
 	github.com/edsrzf/mmap-go v1.0.0 // indirect
 	github.com/elastic/gosigar v0.14.2 // indirect
 	github.com/ethereum/c-kzg-4844 v0.4.0 // indirect
-	github.com/fasthttp/websocket v1.5.8 // indirect
 	github.com/fatih/color v1.16.0 // indirect
 	github.com/fjl/memsize v0.0.2 // indirect
 	github.com/flynn/noise v1.0.0 // indirect
@@ -186,7 +186,6 @@ require (
 	github.com/rs/cors v1.7.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/ryanuber/go-glob v1.0.0 // indirect
-	github.com/savsgio/gotils v0.0.0-20240303185622-093b76447511 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/steakknife/bloomfilter v0.0.0-20180922174646-6819c0d2a570 // indirect

node/go.sum

@@ -178,8 +178,6 @@ github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7
 github.com/ethereum/c-kzg-4844 v0.4.0 h1:3MS1s4JtA868KpJxroZoepdV0ZKBp3u/O5HcZ7R3nlY=
 github.com/ethereum/c-kzg-4844 v0.4.0/go.mod h1:VewdlzQmpT5QSrVhbBuGoCdFJkpaJlO1aQputP83wc0=
 github.com/evanphx/json-patch v4.2.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
-github.com/fasthttp/websocket v1.5.8 h1:k5DpirKkftIF/w1R8ZzjSgARJrs54Je9YJK37DL/Ah8=
-github.com/fasthttp/websocket v1.5.8/go.mod h1:d08g8WaT6nnyvg9uMm8K9zMYyDjfKyj3170AtPRuVU0=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fatih/color v1.16.0 h1:zmkK9Ngbjj+K0yRhTVONQh1p/HknKYSlNT+vZCzyokM=
 github.com/fatih/color v1.16.0/go.mod h1:fL2Sau1YI5c0pdGEVCbKQbLXB6edEj1ZgiY4NijnWvE=
@@ -238,8 +236,6 @@ github.com/godbus/dbus/v5 v5.0.3/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5x
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/godbus/dbus/v5 v5.1.0 h1:4KLkAxT3aOY8Li4FRJe/KvhoNFFxo0m6fNuFUO8QJUk=
 github.com/godbus/dbus/v5 v5.1.0/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
-github.com/gofiber/contrib/websocket v1.3.2 h1:AUq5PYeKwK50s0nQrnluuINYeep1c4nRCJ0NWsV3cvg=
-github.com/gofiber/contrib/websocket v1.3.2/go.mod h1:07u6QGMsvX+sx7iGNCl5xhzuUVArWwLQ3tBIH24i+S8=
 github.com/gofiber/fiber/v2 v2.52.5 h1:tWoP1MJQjGEe4GB5TUGOi7P2E0ZMMRx5ZTG4rT+yGMo=
 github.com/gofiber/fiber/v2 v2.52.5/go.mod h1:KEOE+cXMhXG0zHc9d8+E38hoX+ZN7bhOtgeF2oT6jrQ=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
@@ -529,6 +525,8 @@ github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lN
 github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
+github.com/montanaflynn/stats v0.7.1 h1:etflOAAHORrCC44V+aR6Ftzort912ZU+YLiSTuV8eaE=
+github.com/montanaflynn/stats v0.7.1/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
 github.com/mr-tron/base58 v1.1.2/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
 github.com/mr-tron/base58 v1.2.0 h1:T/HDJBh4ZCPbU39/+c3rRvE0uKBQlU27+QI8LJ4t64o=
 github.com/mr-tron/base58 v1.2.0/go.mod h1:BinMc/sQntlIE1frQmRFPUoPA1Zkr8VRgBdjWI2mNwc=
@@ -665,8 +663,6 @@ github.com/ryanuber/columnize v2.1.0+incompatible/go.mod h1:sm1tb6uqfes/u+d4ooFo
 github.com/ryanuber/go-glob v1.0.0 h1:iQh3xXAumdQ+4Ufa5b25cRpC5TYKlno6hsv6Cb3pkBk=
 github.com/ryanuber/go-glob v1.0.0/go.mod h1:807d1WSdnB0XRJzKNil9Om6lcp/3a0v4qIHxIXzX/Yc=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
-github.com/savsgio/gotils v0.0.0-20240303185622-093b76447511 h1:KanIMPX0QdEdB4R3CiimCAbxFrhB3j7h0/OvpYGVQa8=
-github.com/savsgio/gotils v0.0.0-20240303185622-093b76447511/go.mod h1:sM7Mt7uEoCeFSCBM+qBrqvEo+/9vdmj19wzp3yzUhmg=
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/shurcooL/component v0.0.0-20170202220835-f88ec8f54cc4/go.mod h1:XhFIlyj5a1fBNx5aJTbKoIq0mNaPvOagO+HjB3EtxrY=
 github.com/shurcooL/events v0.0.0-20181021180414-410e4ca65f48/go.mod h1:5u70Mqkb5O5cxEA8nxTsgrgLehJeAw6Oc4Ab1c/P1HM=

node/pkg/fetcher/localaggregator.go

@@ -5,9 +5,11 @@ import (
 	"fmt"
 	"math"
 	"os"
+	"slices"
 	"time"
 
 	"bisonai.com/orakl/node/pkg/bus"
+	"github.com/montanaflynn/stats"
 	"github.com/rs/zerolog/log"
 )
 
@@ -114,21 +116,68 @@ func (c *LocalAggregator) processVolumeWeightedFeeds(ctx context.Context, feeds
 }
 
 func filterOutliers(feeds []*FeedData) ([]*FeedData, error) {
-	median, err := calculateMedian(feeds)
+	if len(feeds) < 5 {
+		return feeds, nil
+	}
+
+	data := make([]float64, len(feeds))
+	for i, feed := range feeds {
+		data[i] = feed.Value
+	}
+
+	outliers, err := stats.QuartileOutliers(data)
 	if err != nil {
 		return nil, err
 	}
 
-	var filteredFeeds []*FeedData
-	for _, feed := range feeds {
-		price := feed.Value
-		priceDifference := math.Abs((median - price) / median)
-		if priceDifference > OutlierThreshold {
-			continue
-		}
-		filteredFeeds = append(filteredFeeds, feed)
+	if outliers.Mild.Len() == 0 && outliers.Extreme.Len() == 0 {
+		return feeds, nil
+	}
+
+	median, err := stats.Median(data)
+	if err != nil {
+		return nil, err
+	}
+
+	maxOutliersToRemove := int(float64(len(feeds)) * MaxOutlierRemovalRatio)
+
+	filtered := feeds
+	var extremes stats.Float64Data
+	if outliers.Extreme.Len() > 0 {
+		slices.SortFunc(outliers.Extreme, func(a, b float64) int {
+			if math.Abs(median-a) < math.Abs(median-b) {
+				return 1
+			} else if math.Abs(median-a) > math.Abs(median-b) {
+				return -1
+			} else {
+				return 0
+			}
+		})
+
+		extremes = outliers.Extreme[:min(maxOutliersToRemove, outliers.Extreme.Len())]
+		filtered = slices.DeleteFunc(feeds, func(feed *FeedData) bool {
+			return slices.Contains(extremes, feed.Value)
+		})
 	}
-	return filteredFeeds, nil
+
+	if extremes.Len() < maxOutliersToRemove && outliers.Mild.Len() > 0 {
+		slices.SortFunc(outliers.Mild, func(a, b float64) int {
+			if math.Abs(median-a) < math.Abs(median-b) {
+				return 1
+			} else if math.Abs(median-a) > math.Abs(median-b) {
+				return -1
+			} else {
+				return 0
+			}
+		})
+
+		milds := outliers.Mild[:min(maxOutliersToRemove-extremes.Len(), outliers.Mild.Len())]
+		filtered = slices.DeleteFunc(filtered, func(feed *FeedData) bool {
+			return slices.Contains(milds, feed.Value)
+		})
+	}
+
+	return filtered, nil
 }
 
 func partitionFeeds(feeds []*FeedData) ([]*FeedData, []*FeedData) {

node/pkg/fetcher/types.go

@@ -24,7 +24,7 @@ const (
 	LocalAggregatesChannelSize            = 2_000
 	DefaultLocalAggregateInterval         = 200 * time.Millisecond
 	DefaultFeedDataDumpChannelSize        = 20000
-	OutlierThreshold                      = 0.5 // 50% price difference is considered an outlier
+	MaxOutlierRemovalRatio                = 0.25
 )
 
 type Feed = types.Feed