gateio: add websocket resub manager and orderbook update with snapshot functionality [spot]

common/common.go

@@ -75,6 +75,7 @@ var (
 	ErrGettingField              = errors.New("error getting field")
 	ErrSettingField              = errors.New("error setting field")
 	ErrParsingWSField            = errors.New("error parsing websocket field")
+	ErrMalformedData             = errors.New("malformed data")
 )
 
 var (

currency/pair.go

@@ -7,8 +7,12 @@ import (
 	"unicode"
 )
 
+// Public errors
+var (
+	ErrCreatingPair = errors.New("error creating currency pair")
+)
+
 var (
-	errCannotCreatePair       = errors.New("cannot create currency pair")
 	errDelimiterNotFound      = errors.New("delimiter not found")
 	errDelimiterCannotBeEmpty = errors.New("delimiter cannot be empty")
 )
@@ -70,7 +74,7 @@ func NewPairWithDelimiter(base, quote, delimiter string) Pair {
 // with or without delimiter
 func NewPairFromString(currencyPair string) (Pair, error) {
 	if len(currencyPair) < 3 {
-		return EMPTYPAIR, fmt.Errorf("%w from %s string too short to be a currency pair", errCannotCreatePair, currencyPair)
+		return EMPTYPAIR, fmt.Errorf("%w from %s string too short to be a currency pair", ErrCreatingPair, currencyPair)
 	}
 
 	for x := range currencyPair {

currency/pair_methods.go

@@ -61,7 +61,7 @@ func (p *Pair) UnmarshalJSON(d []byte) error {
 	// incorrectly converted to DUS-KUSDT, ELKRW (Bithumb) which will convert
 	// converted to ELK-RW and HTUSDT (Lbank) which will be incorrectly
 	// converted to HTU-SDT.
-	return fmt.Errorf("%w from %s cannot ensure pair is in correct format, please use exchange method MatchSymbolWithAvailablePairs", errCannotCreatePair, pair)
+	return fmt.Errorf("%w from %s cannot ensure pair is in correct format, please use exchange method MatchSymbolWithAvailablePairs", ErrCreatingPair, pair)
 }
 
 // MarshalJSON conforms type to the marshaler interface

currency/pair_test.go

@@ -54,7 +54,7 @@ func TestPairUnmarshalJSON(t *testing.T) {
 	assert.Equal(t, "usd", p.Quote.String(), "Quote should be correct")
 	assert.Equal(t, "_", p.Delimiter, "Delimiter should be correct")
 
-	assert.ErrorIs(t, p.UnmarshalJSON([]byte(`"btcusd"`)), errCannotCreatePair, "UnmarshalJSON with no delimiter should error")
+	assert.ErrorIs(t, p.UnmarshalJSON([]byte(`"btcusd"`)), ErrCreatingPair, "UnmarshalJSON with no delimiter should error")
 
 	assert.NoError(t, p.UnmarshalJSON([]byte(`""`)), "UnmarshalJSON should not error on empty value")
 	assert.Equal(t, EMPTYPAIR, p, "UnmarshalJSON empty value should give EMPTYPAIR")

currency/pairs_test.go

@@ -54,7 +54,7 @@ func TestPairsFromString(t *testing.T) {
 	_, err := NewPairsFromString("", "")
 	assert.ErrorIs(t, err, errNoDelimiter)
 	_, err = NewPairsFromString("", ",")
-	assert.ErrorIs(t, err, errCannotCreatePair)
+	assert.ErrorIs(t, err, ErrCreatingPair)
 
 	pairs, err := NewPairsFromString("ALGO-AUD,BAT-AUD,BCH-AUD,BSV-AUD,BTC-AUD,COMP-AUD,ENJ-AUD,ETC-AUD,ETH-AUD,ETH-BTC,GNT-AUD,LINK-AUD,LTC-AUD,LTC-BTC,MCAU-AUD,OMG-AUD,POWR-AUD,UNI-AUD,USDT-AUD,XLM-AUD,XRP-AUD,XRP-BTC", ",")
 	require.NoError(t, err)

exchanges/deribit/deribit_types.go

@@ -42,7 +42,6 @@ var (
 	errInvalidID                           = errors.New("invalid id")
 	errInvalidMarginModel                  = errors.New("missing margin model")
 	errInvalidEmailAddress                 = errors.New("invalid email address")
-	errMalformedData                       = errors.New("malformed data")
 	errWebsocketConnectionNotAuthenticated = errors.New("websocket connection is not authenticated")
 	errResolutionNotSet                    = errors.New("resolution not set")
 	errInvalidDestinationID                = errors.New("invalid destination id")

exchanges/deribit/deribit_websocket.go

@@ -333,7 +333,7 @@ func (e *Exchange) wsSendHeartbeat(ctx context.Context) {
 
 func (e *Exchange) processUserOrders(respRaw []byte, channels []string) error {
 	if len(channels) != 4 && len(channels) != 5 {
-		return fmt.Errorf("%w, expected format 'user.orders.{instrument_name}.raw, user.orders.{instrument_name}.{interval}, user.orders.{kind}.{currency}.raw, or user.orders.{kind}.{currency}.{interval}', but found %s", errMalformedData, strings.Join(channels, "."))
+		return fmt.Errorf("%w, expected format 'user.orders.{instrument_name}.raw, user.orders.{instrument_name}.{interval}, user.orders.{kind}.{currency}.raw, or user.orders.{kind}.{currency}.{interval}', but found %s", common.ErrMalformedData, strings.Join(channels, "."))
 	}
 	var response wsResponse
 	orderData := []WsOrder{}
@@ -382,7 +382,7 @@ func (e *Exchange) processUserOrders(respRaw []byte, channels []string) error {
 
 func (e *Exchange) processUserOrderChanges(respRaw []byte, channels []string) error {
 	if len(channels) < 4 || len(channels) > 5 {
-		return fmt.Errorf("%w, expected format 'trades.{instrument_name}.{interval} or trades.{kind}.{currency}.{interval}', but found %s", errMalformedData, strings.Join(channels, "."))
+		return fmt.Errorf("%w, expected format 'trades.{instrument_name}.{interval} or trades.{kind}.{currency}.{interval}', but found %s", common.ErrMalformedData, strings.Join(channels, "."))
 	}
 	var response wsResponse
 	changeData := &wsChanges{}
@@ -492,7 +492,7 @@ func (e *Exchange) processTrades(respRaw []byte, channels []string) error {
 	}
 
 	if len(channels) < 3 || len(channels) > 5 {
-		return fmt.Errorf("%w, expected format 'trades.{instrument_name}.{interval} or trades.{kind}.{currency}.{interval}', but found %s", errMalformedData, strings.Join(channels, "."))
+		return fmt.Errorf("%w, expected format 'trades.{instrument_name}.{interval} or trades.{kind}.{currency}.{interval}', but found %s", common.ErrMalformedData, strings.Join(channels, "."))
 	}
 	var response wsResponse
 	var tradeList []wsTrade
@@ -536,7 +536,7 @@ func (e *Exchange) processTrades(respRaw []byte, channels []string) error {
 
 func (e *Exchange) processIncrementalTicker(respRaw []byte, channels []string) error {
 	if len(channels) != 2 {
-		return fmt.Errorf("%w, expected format 'incremental_ticker.{instrument_name}', but found %s", errMalformedData, strings.Join(channels, "."))
+		return fmt.Errorf("%w, expected format 'incremental_ticker.{instrument_name}', but found %s", common.ErrMalformedData, strings.Join(channels, "."))
 	}
 	a, cp, err := getAssetPairByInstrument(channels[1])
 	if err != nil {
@@ -568,7 +568,7 @@ func (e *Exchange) processIncrementalTicker(respRaw []byte, channels []string) e
 
 func (e *Exchange) processInstrumentTicker(respRaw []byte, channels []string) error {
 	if len(channels) != 3 {
-		return fmt.Errorf("%w, expected format 'ticker.{instrument_name}.{interval}', but found %s", errMalformedData, strings.Join(channels, "."))
+		return fmt.Errorf("%w, expected format 'ticker.{instrument_name}.{interval}', but found %s", common.ErrMalformedData, strings.Join(channels, "."))
 	}
 	return e.processTicker(respRaw, channels)
 }
@@ -623,7 +623,7 @@ func (e *Exchange) processData(respRaw []byte, result any) error {
 
 func (e *Exchange) processCandleChart(respRaw []byte, channels []string) error {
 	if len(channels) != 4 {
-		return fmt.Errorf("%w, expected format 'chart.trades.{instrument_name}.{resolution}', but found %s", errMalformedData, strings.Join(channels, "."))
+		return fmt.Errorf("%w, expected format 'chart.trades.{instrument_name}.{resolution}', but found %s", common.ErrInvalidResponse, strings.Join(channels, "."))
 	}
 	a, cp, err := getAssetPairByInstrument(channels[2])
 	if err != nil {
@@ -666,15 +666,15 @@ func (e *Exchange) processOrderbook(respRaw []byte, channels []string) error {
 		asks := make(orderbook.Levels, 0, len(orderbookData.Asks))
 		for x := range orderbookData.Asks {
 			if len(orderbookData.Asks[x]) != 3 {
-				return errMalformedData
+				return common.ErrMalformedData
 			}
 			price, okay := orderbookData.Asks[x][1].(float64)
 			if !okay {
-				return fmt.Errorf("%w, invalid orderbook price", errMalformedData)
+				return fmt.Errorf("%w, invalid orderbook price", common.ErrMalformedData)
 			}
 			amount, okay := orderbookData.Asks[x][2].(float64)
 			if !okay {
-				return fmt.Errorf("%w, invalid amount", errMalformedData)
+				return fmt.Errorf("%w, invalid amount", common.ErrMalformedData)
 			}
 			asks = append(asks, orderbook.Level{
 				Price:  price,
@@ -684,17 +684,17 @@ func (e *Exchange) processOrderbook(respRaw []byte, channels []string) error {
 		bids := make(orderbook.Levels, 0, len(orderbookData.Bids))
 		for x := range orderbookData.Bids {
 			if len(orderbookData.Bids[x]) != 3 {
-				return errMalformedData
+				return common.ErrMalformedData
 			}
 			price, okay := orderbookData.Bids[x][1].(float64)
 			if !okay {
-				return fmt.Errorf("%w, invalid orderbook price", errMalformedData)
+				return fmt.Errorf("%w, invalid orderbook price", common.ErrMalformedData)
 			} else if price == 0.0 {
 				continue
 			}
 			amount, okay := orderbookData.Bids[x][2].(float64)
 			if !okay {
-				return fmt.Errorf("%w, invalid amount", errMalformedData)
+				return fmt.Errorf("%w, invalid amount", common.ErrMalformedData)
 			}
 			bids = append(bids, orderbook.Level{
 				Price:  price,
@@ -735,17 +735,17 @@ func (e *Exchange) processOrderbook(respRaw []byte, channels []string) error {
 		asks := make(orderbook.Levels, 0, len(orderbookData.Asks))
 		for x := range orderbookData.Asks {
 			if len(orderbookData.Asks[x]) != 2 {
-				return errMalformedData
+				return common.ErrMalformedData
 			}
 			price, okay := orderbookData.Asks[x][0].(float64)
 			if !okay {
-				return fmt.Errorf("%w, invalid orderbook price", errMalformedData)
+				return fmt.Errorf("%w, invalid orderbook price", common.ErrMalformedData)
 			} else if price == 0 {
 				continue
 			}
 			amount, okay := orderbookData.Asks[x][1].(float64)
 			if !okay {
-				return fmt.Errorf("%w, invalid amount", errMalformedData)
+				return fmt.Errorf("%w, invalid amount", common.ErrMalformedData)
 			}
 			asks = append(asks, orderbook.Level{
 				Price:  price,
@@ -755,17 +755,17 @@ func (e *Exchange) processOrderbook(respRaw []byte, channels []string) error {
 		bids := make([]orderbook.Level, 0, len(orderbookData.Bids))
 		for x := range orderbookData.Bids {
 			if len(orderbookData.Bids[x]) != 2 {
-				return errMalformedData
+				return common.ErrMalformedData
 			}
 			price, okay := orderbookData.Bids[x][0].(float64)
 			if !okay {
-				return fmt.Errorf("%w, invalid orderbook price", errMalformedData)
+				return fmt.Errorf("%w, invalid orderbook price", common.ErrMalformedData)
 			} else if price == 0 {
 				continue
 			}
 			amount, okay := orderbookData.Bids[x][1].(float64)
 			if !okay {
-				return fmt.Errorf("%w, invalid amount", errMalformedData)
+				return fmt.Errorf("%w, invalid amount", common.ErrMalformedData)
 			}
 			bids = append(bids, orderbook.Level{
 				Price:  price,

exchanges/gateio/gateio.go

@@ -195,6 +195,7 @@ type Exchange struct {
 
 	messageIDSeq  common.Counter
 	wsOBUpdateMgr *wsOBUpdateManager
+	wsOBResubMgr  *wsOBResubManager
 }
 
 // ***************************************** SubAccounts ********************************

exchanges/gateio/gateio_types.go

@@ -2092,6 +2092,17 @@ type WsOrderbookUpdate struct {
 	Asks          orderbook.LevelsArrayPriceAmount `json:"a"`
 }
 
+// WsOrderbookUpdateWithSnapshot represents websocket orderbook update push data
+type WsOrderbookUpdateWithSnapshot struct {
+	UpdateTime    types.Time                       `json:"t"`
+	Full          bool                             `json:"full"`
+	Channel       string                           `json:"s"`
+	FirstUpdateID int64                            `json:"U"`
+	LastUpdateID  int64                            `json:"u"`
+	Bids          orderbook.LevelsArrayPriceAmount `json:"b"`
+	Asks          orderbook.LevelsArrayPriceAmount `json:"a"`
+}
+
 // WsOrderbookSnapshot represents a websocket orderbook snapshot push data
 type WsOrderbookSnapshot struct {
 	UpdateTime   types.Time                       `json:"t"`

exchanges/gateio/gateio_websocket.go

@@ -46,6 +46,7 @@ const (
 	spotOrderbookTickerChannel = "spot.book_ticker"       // Best bid or ask price
 	spotOrderbookUpdateChannel = "spot.order_book_update" // Changed order book levels
 	spotOrderbookChannel       = "spot.order_book"        // Limited-Level Full Order Book Snapshot
+	spotOrderbookV2            = "spot.obu"
 	spotOrdersChannel          = "spot.orders"
 	spotUserTradesChannel      = "spot.usertrades"
 	spotBalancesChannel        = "spot.balances"
@@ -64,6 +65,7 @@ var defaultSubscriptions = subscription.List{
 	{Enabled: true, Channel: subscription.OrderbookChannel, Asset: asset.Spot, Interval: kline.HundredMilliseconds},
 	{Enabled: false, Channel: spotOrderbookTickerChannel, Asset: asset.Spot, Interval: kline.TenMilliseconds, Levels: 1},
 	{Enabled: false, Channel: spotOrderbookChannel, Asset: asset.Spot, Interval: kline.HundredMilliseconds, Levels: 100},
+	{Enabled: false, Channel: spotOrderbookV2, Asset: asset.Spot, Levels: 50},
 	{Enabled: true, Channel: spotBalancesChannel, Asset: asset.Spot, Authenticated: true},
 	{Enabled: true, Channel: crossMarginBalanceChannel, Asset: asset.CrossMargin, Authenticated: true},
 	{Enabled: true, Channel: marginBalancesChannel, Asset: asset.Margin, Authenticated: true},
@@ -191,6 +193,8 @@ func (e *Exchange) WsHandleSpotData(ctx context.Context, conn websocket.Connecti
 		return e.processOrderbookUpdate(ctx, push.Result, push.Time)
 	case spotOrderbookChannel:
 		return e.processOrderbookSnapshot(push.Result, push.Time)
+	case spotOrderbookV2:
+		return e.processOrderbookUpdateWithSnapshot(conn, push.Result, push.Time, asset.Spot)
 	case spotOrdersChannel:
 		return e.processSpotOrders(respRaw)
 	case spotUserTradesChannel:
@@ -324,7 +328,7 @@ func (e *Exchange) processCandlestick(incoming []byte) error {
 	}
 	icp := strings.Split(data.NameOfSubscription, currency.UnderscoreDelimiter)
 	if len(icp) < 3 {
-		return errors.New("malformed candlestick websocket push data")
+		return fmt.Errorf("%w: candlestick websocket", common.ErrMalformedData)
 	}
 	currencyPair, err := currency.NewPairFromString(strings.Join(icp[1:], currency.UnderscoreDelimiter))
 	if err != nil {
@@ -409,6 +413,59 @@ func (e *Exchange) processOrderbookSnapshot(incoming []byte, lastPushed time.Tim
 	return nil
 }
 
+func (e *Exchange) processOrderbookUpdateWithSnapshot(conn websocket.Connection, incoming []byte, lastPushed time.Time, a asset.Item) error {
+	var data WsOrderbookUpdateWithSnapshot
+	if err := json.Unmarshal(incoming, &data); err != nil {
+		return err
+	}
+
+	channelParts := strings.Split(data.Channel, ".")
+	if len(channelParts) < 3 {
+		return fmt.Errorf("%w: %q", common.ErrMalformedData, data.Channel)
+	}
+
+	pair, err := currency.NewPairFromString(channelParts[1])
+	if err != nil {
+		return err
+	}
+
+	if data.Full {
+		if err := e.Websocket.Orderbook.LoadSnapshot(&orderbook.Book{
+			Exchange:     e.Name,
+			Pair:         pair,
+			Asset:        a,
+			LastUpdated:  data.UpdateTime.Time(),
+			LastPushed:   lastPushed,
+			LastUpdateID: data.LastUpdateID,
+			Bids:         data.Bids.Levels(),
+			Asks:         data.Asks.Levels(),
+		}); err != nil {
+			return err
+		}
+		e.wsOBResubMgr.CompletedResubscribe(pair, a)
+		return nil
+	}
+
+	if e.wsOBResubMgr.IsResubscribing(pair, a) {
+		return nil // Drop incremental updates; waiting for a fresh snapshot
+	}
+
+	lastUpdateID, err := e.Websocket.Orderbook.LastUpdateID(pair, a)
+	if err != nil || lastUpdateID+1 != data.FirstUpdateID {
+		return common.AppendError(err, e.wsOBResubMgr.Resubscribe(e, conn, data.Channel, pair, a))
+	}
+	return e.Websocket.Orderbook.Update(&orderbook.Update{
+		Pair:       pair,
+		Asset:      a,
+		UpdateTime: data.UpdateTime.Time(),
+		LastPushed: lastPushed,
+		UpdateID:   data.LastUpdateID,
+		Bids:       data.Bids.Levels(),
+		Asks:       data.Asks.Levels(),
+		AllowEmpty: true,
+	})
+}
+
 func (e *Exchange) processSpotOrders(data []byte) error {
 	resp := struct {
 		Time    types.Time    `json:"time"`
@@ -608,11 +665,12 @@ func (e *Exchange) GetSubscriptionTemplate(_ *subscription.Subscription) (*templ
 	return template.New("master.tmpl").
 		Funcs(sprig.FuncMap()).
 		Funcs(template.FuncMap{
-			"channelName":         channelName,
-			"singleSymbolChannel": singleSymbolChannel,
-			"orderbookInterval":   orderbookChannelInterval,
-			"candlesInterval":     candlesChannelInterval,
-			"levels":              channelLevels,
+			"channelName":             channelName,
+			"singleSymbolChannel":     singleSymbolChannel,
+			"orderbookInterval":       orderbookChannelInterval,
+			"candlesInterval":         candlesChannelInterval,
+			"levels":                  channelLevels,
+			"compactOrderbookPayload": isCompactOrderbookPayload,
 		}).Parse(subTplText)
 }
 
@@ -700,7 +758,7 @@ func channelName(s *subscription.Subscription) string {
 // singleSymbolChannel returns if the channel should be fanned out into single symbol requests
 func singleSymbolChannel(name string) bool {
 	switch name {
-	case spotCandlesticksChannel, spotOrderbookUpdateChannel, spotOrderbookChannel:
+	case spotCandlesticksChannel, spotOrderbookUpdateChannel, spotOrderbookChannel, spotOrderbookV2:
 		return true
 	}
 	return false
@@ -739,6 +797,7 @@ func isSingleOrderbookChannel(name string) bool {
 	case spotOrderbookUpdateChannel,
 		spotOrderbookChannel,
 		spotOrderbookTickerChannel,
+		spotOrderbookV2,
 		futuresOrderbookChannel,
 		futuresOrderbookTickerChannel,
 		futuresOrderbookUpdateChannel,
@@ -812,6 +871,7 @@ var channelLevelsMap = map[asset.Item]map[string][]int{
 		spotOrderbookTickerChannel: {},
 		spotOrderbookUpdateChannel: {},
 		spotOrderbookChannel:       {1, 5, 10, 20, 50, 100},
+		spotOrderbookV2:            {50, 400},
 	},
 	asset.Futures: {
 		futuresOrderbookChannel:       {1, 5, 10, 20, 50, 100},
@@ -852,16 +912,27 @@ func channelLevels(s *subscription.Subscription, a asset.Item) (string, error) {
 	return strconv.Itoa(s.Levels), nil
 }
 
+func isCompactOrderbookPayload(channel string) bool {
+	return channel == spotOrderbookV2
+}
+
 const subTplText = `
 {{- with $name := channelName $.S }}
 	{{- range $asset, $pairs := $.AssetPairs }}
 		{{- if singleSymbolChannel $name }}
 			{{- range $i, $p := $pairs -}}
-				{{- with $i := candlesInterval $.S }}{{ $i -}} , {{- end }}
-				{{- $p }}
-				{{- with $l := levels $.S $asset -}} , {{- $l }}{{ end }}
-				{{- with $i := orderbookInterval $.S $asset -}} , {{- $i }}{{- end }}
-				{{- $.PairSeparator }}
+				{{- if compactOrderbookPayload $name }}	
+					{{- with $l := levels $.S $asset -}}
+					ob.{{ $p }}.{{ $l }}
+					{{- end -}}
+					{{- $.PairSeparator }}
+				{{- else }}
+					{{- with $i := candlesInterval $.S }}{{ $i -}} , {{- end }}
+					{{- $p }}
+					{{- with $l := levels $.S $asset -}} , {{- $l }}{{ end }}
+					{{- with $i := orderbookInterval $.S $asset -}} , {{- $i }}{{- end }}
+					{{- $.PairSeparator }}
+				{{- end }}
 			{{- end }}
 			{{- $.AssetSeparator }}
 		{{- else }}