Kraken: Subscription improvements (#1587)

* Convert: Fix TimeFromUnixTimestampDecimal using local

All parsed times should be in UTC

* Subscriptions: Add IgnoringAssetsKey

* Tests: Pass tb to curried WS handlers

* Websocket: Make ErrNoMessageListener a public error

* Kraken: Fix URLMap ignored for websocket URLs

* Kraken: Move SeedAssets from Setup to Bootstrap

Having SeedAssets in Setup is cruel and unusual because it calls the
API. Most other interactive data seeding happens in Bootstrap.

This made it so that fixing and creating unit tests for Kraken was
painfully slow, particularly on flaky internet.

* Kraken: Remove convert test

Duplicate of convert_test.go TestTimeFromUnixTimestampDecimal

* Kraken: Test config upgrades

* Kraken: Sub Channel improvements

* Use Websocket subscriptionChannels instead of local slice
* Remove ChannelID - Deprecated in docs
* Simplify ping handlers and hardcodes message
* Add Depth as configurable orderbook channel param
* Simplify auth/non-auth channel updates
* Add configurable Book depth
* Add configurable Candle timeframes

Kraken: Simplify all WS handlers with reqId

* Kraken: Subscription templating

* Generate N+ subs for pairs
If we generate one sub for all pairs, but then fan it out in the
responses, we end up with a mis-match between the sub store and
GenerateSubs, and when we do FlushChannels it will try to resub
everything again.

* Kraken: Rename channelName var throughout

Avoid shadowing func of same name

* Kraken: Add TestEnforceStandardChannelNames

* Websocket: Fix Resubscribe erroring Duplicate

exchanges/binance/binance_test.go

@@ -1982,10 +1982,11 @@ func TestSubscribe(t *testing.T) {
 	require.NoError(t, err, "generateSubscriptions must not error")
 	if mockTests {
 		exp := []string{"btcusdt@depth@100ms", "btcusdt@kline_1m", "btcusdt@ticker", "btcusdt@trade", "dogeusdt@depth@100ms", "dogeusdt@kline_1m", "dogeusdt@ticker", "dogeusdt@trade"}
-		mock := func(msg []byte, w *websocket.Conn) error {
+		mock := func(tb testing.TB, msg []byte, w *websocket.Conn) error {
+			tb.Helper()
 			var req WsPayload
-			require.NoError(t, json.Unmarshal(msg, &req), "Unmarshal should not error")
-			require.ElementsMatch(t, req.Params, exp, "Params should have correct channels")
+			require.NoError(tb, json.Unmarshal(msg, &req), "Unmarshal should not error")
+			require.ElementsMatch(tb, req.Params, exp, "Params should have correct channels")
 			return w.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"result":null,"id":%d}`, req.ID)))
 		}
 		b = testexch.MockWsInstance[Binance](t, testexch.CurryWsMockUpgrader(t, mock))
@@ -2003,10 +2004,11 @@ func TestSubscribeBadResp(t *testing.T) {
 	channels := subscription.List{
 		{Channel: "moons@ticker"},
 	}
-	mock := func(msg []byte, w *websocket.Conn) error {
+	mock := func(tb testing.TB, msg []byte, w *websocket.Conn) error {
+		tb.Helper()
 		var req WsPayload
 		err := json.Unmarshal(msg, &req)
-		require.NoError(t, err, "Unmarshal should not error")
+		require.NoError(tb, err, "Unmarshal should not error")
 		return w.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"result":{"error":"carrots"},"id":%d}`, req.ID)))
 	}
 	b := testexch.MockWsInstance[Binance](t, testexch.CurryWsMockUpgrader(t, mock)) //nolint:govet // Intentional shadow to avoid future copy/paste mistakes

exchanges/bitfinex/bitfinex_websocket.go

@@ -438,16 +438,16 @@ func (b *Bitfinex) handleWSEvent(respRaw []byte) error {
 			return fmt.Errorf("%w 'chanId': %w from message: %s", errParsingWSField, err, respRaw)
 		}
 		if !b.Websocket.Match.IncomingWithData("unsubscribe:"+chanID, respRaw) {
-			return fmt.Errorf("%v channel unsubscribe listener not found", chanID)
+			return fmt.Errorf("%w: unsubscribe:%v", stream.ErrNoMessageListener, chanID)
 		}
 	case wsEventError:
 		if subID, err := jsonparser.GetUnsafeString(respRaw, "subId"); err == nil {
 			if !b.Websocket.Match.IncomingWithData("subscribe:"+subID, respRaw) {
-				return fmt.Errorf("%v channel subscribe listener not found", subID)
+				return fmt.Errorf("%w: subscribe:%v", stream.ErrNoMessageListener, subID)
 			}
 		} else if chanID, err := jsonparser.GetUnsafeString(respRaw, "chanId"); err == nil {
 			if !b.Websocket.Match.IncomingWithData("unsubscribe:"+chanID, respRaw) {
-				return fmt.Errorf("%v channel unsubscribe listener not found", chanID)
+				return fmt.Errorf("%w: unsubscribe:%v", stream.ErrNoMessageListener, chanID)
 			}
 		} else {
 			return fmt.Errorf("unknown channel error; Message: %s", respRaw)
@@ -520,7 +520,7 @@ func (b *Bitfinex) handleWSSubscribed(respRaw []byte) error {
 		log.Debugf(log.ExchangeSys, "%s Subscribed to Channel: %s Pair: %s ChannelID: %d\n", b.Name, c.Channel, c.Pairs, chanID)
 	}
 	if !b.Websocket.Match.IncomingWithData("subscribe:"+subID, respRaw) {
-		return fmt.Errorf("%v channel subscribe listener not found", subID)
+		return fmt.Errorf("%w: subscribe:%v", stream.ErrNoMessageListener, subID)
 	}
 	return nil
 }

exchanges/kraken/kraken.go

@@ -9,7 +9,6 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/thrasher-corp/gocryptotrader/common"
@@ -38,7 +37,6 @@ const (
 // Kraken is the overarching type across the kraken package
 type Kraken struct {
 	exchange.Base
-	wsRequestMtx sync.Mutex
 }
 
 // GetCurrentServerTime returns current server time

exchanges/kraken/kraken_types.go

@@ -10,7 +10,6 @@ import (
 	"github.com/thrasher-corp/gocryptotrader/common"
 	"github.com/thrasher-corp/gocryptotrader/currency"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/order"
-	"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
 	"github.com/thrasher-corp/gocryptotrader/types"
 )
 
@@ -77,10 +76,18 @@ const (
 	statusOpen = "open"
 
 	krakenFormat = "2006-01-02T15:04:05.000Z"
+
+	// ChannelOrderbookDepthKey configures the orderbook depth in stream.ChannelSubscription.Params
+	ChannelOrderbookDepthKey = "_depth"
+	// ChannelCandlesTimeframeKey configures the candle bar timeframe in stream.ChannelSubscription.Params
+	ChannelCandlesTimeframeKey = "_timeframe"
 )
 
 var (
 	assetTranslator assetTranslatorStore
+
+	errNoWebsocketOrderbookData = errors.New("no websocket orderbook data")
+	errBadChannelSuffix         = errors.New("bad websocket channel suffix")
 )
 
 // GenericResponse stores general response data for functions that only return success
@@ -492,43 +499,29 @@ type WithdrawStatusResponse struct {
 	Status string  `json:"status"`
 }
 
-// WebsocketSubscriptionEventRequest handles WS subscription events
-type WebsocketSubscriptionEventRequest struct {
-	Event        string                    `json:"event"`           // subscribe
-	RequestID    int64                     `json:"reqid,omitempty"` // Optional, client originated ID reflected in response message.
-	Pairs        []string                  `json:"pair,omitempty"`  // Array of currency pairs (pair1,pair2,pair3).
+// WebsocketSubRequest contains request data for Subscribe/Unsubscribe to channels
+type WebsocketSubRequest struct {
+	Event        string                    `json:"event"`
+	RequestID    int64                     `json:"reqid,omitempty"`
+	Pairs        []string                  `json:"pair,omitempty"`
 	Subscription WebsocketSubscriptionData `json:"subscription,omitempty"`
-	Channels     subscription.List         `json:"-"` // Keeps track of associated subscriptions in batched outgoings
-}
-
-// WebsocketBaseEventRequest Just has an "event" property
-type WebsocketBaseEventRequest struct {
-	Event string `json:"event"` // eg "unsubscribe"
-}
-
-// WebsocketUnsubscribeByChannelIDEventRequest  handles WS unsubscribe events
-type WebsocketUnsubscribeByChannelIDEventRequest struct {
-	WebsocketBaseEventRequest
-	RequestID int64    `json:"reqid,omitempty"` // Optional, client originated ID reflected in response message.
-	Pairs     []string `json:"pair,omitempty"`  // Array of currency pairs (pair1,pair2,pair3).
-	ChannelID int64    `json:"channelID,omitempty"`
 }
 
 // WebsocketSubscriptionData contains details on WS channel
 type WebsocketSubscriptionData struct {
 	Name     string `json:"name,omitempty"`     // ticker|ohlc|trade|book|spread|*, * for all (ohlc interval value is 1 if all channels subscribed)
-	Interval int64  `json:"interval,omitempty"` // Optional - Time interval associated with ohlc subscription in minutes. Default 1. Valid Interval values: 1|5|15|30|60|240|1440|10080|21600
-	Depth    int64  `json:"depth,omitempty"`    // Optional - depth associated with book subscription in number of levels each side, default 10. Valid Options are: 10, 25, 100, 500, 1000
-	Token    string `json:"token,omitempty"`    // Optional used for authenticated requests
+	Interval int    `json:"interval,omitempty"` // Optional - Timeframe for candles subscription in minutes; default 1. Valid: 1|5|15|30|60|240|1440|10080|21600
+	Depth    int    `json:"depth,omitempty"`    // Optional - Depth associated with orderbook; default 10. Valid: 10|25|100|500|1000
+	Token    string `json:"token,omitempty"`    // Optional - Token for authenticated channels
 
 }
 
 // WebsocketEventResponse holds all data response types
 type WebsocketEventResponse struct {
-	WebsocketBaseEventRequest
+	Event        string                            `json:"event"`
 	Status       string                            `json:"status"`
 	Pair         currency.Pair                     `json:"pair,omitempty"`
-	RequestID    int64                             `json:"reqid,omitempty"` // Optional, client originated ID reflected in response message.
+	RequestID    int64                             `json:"reqid,omitempty"`
 	Subscription WebsocketSubscriptionResponseData `json:"subscription,omitempty"`
 	ChannelName  string                            `json:"channelName,omitempty"`
 	WebsocketSubscriptionEventResponse
@@ -545,23 +538,11 @@ type WebsocketSubscriptionResponseData struct {
 	Name string `json:"name"`
 }
 
-// WebsocketDataResponse defines a websocket data type
-type WebsocketDataResponse []interface{}
-
 // WebsocketErrorResponse defines a websocket error response
 type WebsocketErrorResponse struct {
 	ErrorMessage string `json:"errorMessage"`
 }
 
-// WebsocketChannelData Holds relevant data for channels to identify what we're
-// doing
-type WebsocketChannelData struct {
-	Subscription string
-	Pair         currency.Pair
-	ChannelID    *int64
-	MaxDepth     int
-}
-
 // WsTokenResponse holds the WS auth token
 type WsTokenResponse struct {
 	Expires int64  `json:"expires"`
@@ -575,21 +556,6 @@ type wsSystemStatus struct {
 	Version      string  `json:"version"`
 }
 
-type wsSubscription struct {
-	ChannelID    *int64 `json:"channelID"`
-	ChannelName  string `json:"channelName"`
-	ErrorMessage string `json:"errorMessage"`
-	Event        string `json:"event"`
-	Pair         string `json:"pair"`
-	RequestID    int64  `json:"reqid"`
-	Status       string `json:"status"`
-	Subscription struct {
-		Depth    int    `json:"depth"`
-		Interval int    `json:"interval"`
-		Name     string `json:"name"`
-	} `json:"subscription"`
-}
-
 // WsOpenOrder contains all open order data from ws feed
 type WsOpenOrder struct {
 	UserReferenceID int64   `json:"userref"`

exchanges/kraken/kraken_wrapper.go

@@ -168,6 +168,7 @@ func (k *Kraken) SetDefaults() {
 				GlobalResultLimit: 720,
 			},
 		},
+		Subscriptions: defaultSubscriptions.Clone(),
 	}
 
 	k.Requester, err = request.New(k.Name,
@@ -178,10 +179,11 @@ func (k *Kraken) SetDefaults() {
 	}
 	k.API.Endpoints = k.NewEndpoints()
 	err = k.API.Endpoints.SetDefaultEndpoints(map[exchange.URL]string{
-		exchange.RestSpot:                 krakenAPIURL,
-		exchange.RestFutures:              krakenFuturesURL,
-		exchange.WebsocketSpot:            krakenWSURL,
-		exchange.RestFuturesSupplementary: krakenFuturesSupplementaryURL,
+		exchange.RestSpot:                   krakenAPIURL,
+		exchange.RestFutures:                krakenFuturesURL,
+		exchange.WebsocketSpot:              krakenWSURL,
+		exchange.WebsocketSpotSupplementary: krakenAuthWSURL,
+		exchange.RestFuturesSupplementary:   krakenFuturesSupplementaryURL,
 	})
 	if err != nil {
 		log.Errorln(log.ExchangeSys, err)
@@ -207,11 +209,6 @@ func (k *Kraken) Setup(exch *config.Exchange) error {
 		return err
 	}
 
-	err = k.SeedAssets(context.TODO())
-	if err != nil {
-		return err
-	}
-
 	wsRunningURL, err := k.API.Endpoints.GetURL(exchange.WebsocketSpot)
 	if err != nil {
 		return err
@@ -223,7 +220,7 @@ func (k *Kraken) Setup(exch *config.Exchange) error {
 		Connector:             k.WsConnect,
 		Subscriber:            k.Subscribe,
 		Unsubscriber:          k.Unsubscribe,
-		GenerateSubscriptions: k.GenerateDefaultSubscriptions,
+		GenerateSubscriptions: k.generateSubscriptions,
 		Features:              &k.Features.Supports.WebsocketCapabilities,
 		OrderbookBufferConfig: buffer.Config{SortBuffer: true},
 	})
@@ -235,27 +232,47 @@ func (k *Kraken) Setup(exch *config.Exchange) error {
 		RateLimit:            request.NewWeightedRateLimitByDuration(50 * time.Millisecond),
 		ResponseCheckTimeout: exch.WebsocketResponseCheckTimeout,
 		ResponseMaxLimit:     exch.WebsocketResponseMaxLimit,
-		URL:                  krakenWSURL,
 	})
 	if err != nil {
 		return err
 	}
 
+	wsRunningAuthURL, err := k.API.Endpoints.GetURL(exchange.WebsocketSpotSupplementary)
+	if err != nil {
+		return err
+	}
 	return k.Websocket.SetupNewConnection(stream.ConnectionSetup{
 		RateLimit:            request.NewWeightedRateLimitByDuration(50 * time.Millisecond),
 		ResponseCheckTimeout: exch.WebsocketResponseCheckTimeout,
 		ResponseMaxLimit:     exch.WebsocketResponseMaxLimit,
-		URL:                  krakenAuthWSURL,
 		Authenticated:        true,
+		URL:                  wsRunningAuthURL,
 	})
 }
 
+// Bootstrap provides initialisation for an exchange
+func (k *Kraken) Bootstrap(_ context.Context) (continueBootstrap bool, err error) {
+	continueBootstrap = true
+
+	if err = k.SeedAssets(context.TODO()); err != nil {
+		err = fmt.Errorf("failed to Seed Assets: %w", err)
+	}
+
+	return
+}
+
 // UpdateOrderExecutionLimits sets exchange execution order limits for an asset type
 func (k *Kraken) UpdateOrderExecutionLimits(ctx context.Context, a asset.Item) error {
 	if a != asset.Spot {
 		return common.ErrNotYetImplemented
 	}
 
+	if !assetTranslator.Seeded() {
+		if err := k.SeedAssets(ctx); err != nil {
+			return err
+		}
+	}
+
 	pairInfo, err := k.fetchSpotPairInfo(ctx)
 	if err != nil {
 		return fmt.Errorf("%s failed to load %s pair execution limits. Err: %s", k.Name, a, err)
@@ -547,6 +564,11 @@ func (k *Kraken) UpdateAccountInfo(ctx context.Context, assetType asset.Item) (a
 	var info account.Holdings
 	var balances []account.Balance
 	info.Exchange = k.Name
+	if !assetTranslator.Seeded() {
+		if err := k.SeedAssets(ctx); err != nil {
+			return info, err
+		}
+	}
 	switch assetType {
 	case asset.Spot:
 		bal, err := k.GetBalance(ctx)
@@ -798,8 +820,7 @@ func (k *Kraken) SubmitOrder(ctx context.Context, s *order.Submit) (*order.Submi
 	return resp, nil
 }
 
-// ModifyOrder will allow of changing orderbook placement and limit to
-// market conversion
+// ModifyOrder will allow of changing orderbook placement and limit to market conversion
 func (k *Kraken) ModifyOrder(_ context.Context, _ *order.Modify) (*order.ModifyResponse, error) {
 	return nil, common.ErrFunctionNotSupported
 }

exchanges/kraken/mock_ws_test.go

@@ -0,0 +1,76 @@
+package kraken
+
+import (
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/buger/jsonparser"
+	"github.com/gorilla/websocket"
+	"github.com/stretchr/testify/assert"
+)
+
+func mockWsServer(tb testing.TB, msg []byte, w *websocket.Conn) error {
+	tb.Helper()
+	event, err := jsonparser.GetUnsafeString(msg, "event")
+	if err != nil {
+		return err
+	}
+	switch event {
+	case krakenWsCancelOrder:
+		return mockWsCancelOrders(tb, msg, w)
+	case krakenWsAddOrder:
+		return mockWsAddOrder(tb, msg, w)
+	}
+	return nil
+}
+
+func mockWsCancelOrders(tb testing.TB, msg []byte, w *websocket.Conn) error {
+	tb.Helper()
+	var req WsCancelOrderRequest
+	if err := json.Unmarshal(msg, &req); err != nil {
+		return err
+	}
+	resp := WsCancelOrderResponse{
+		Event:     krakenWsCancelOrderStatus,
+		Status:    "ok",
+		RequestID: req.RequestID,
+		Count:     int64(len(req.TransactionIDs)),
+	}
+	if len(req.TransactionIDs) == 0 || strings.Contains(req.TransactionIDs[0], "FISH") { // Reject anything that smells suspicious
+		resp.Status = "error"
+		resp.ErrorMessage = "[EOrder:Unknown order]"
+	}
+	msg, err := json.Marshal(resp)
+	if err != nil {
+		return err
+	}
+	return w.WriteMessage(websocket.TextMessage, msg)
+}
+
+func mockWsAddOrder(tb testing.TB, msg []byte, w *websocket.Conn) error {
+	tb.Helper()
+	var req WsAddOrderRequest
+	if err := json.Unmarshal(msg, &req); err != nil {
+		return err
+	}
+
+	assert.Equal(tb, "buy", req.OrderSide, "OrderSide should be correct")
+	assert.Equal(tb, "limit", req.OrderType, "OrderType should be correct")
+	assert.Equal(tb, "XBT/USD", req.Pair, "Pair should be correct")
+	assert.Equal(tb, 80000.0, req.Price, "Pair should be correct")
+
+	resp := WsAddOrderResponse{
+		Event:         krakenWsAddOrderStatus,
+		Status:        "ok",
+		RequestID:     req.RequestID,
+		TransactionID: "ONPNXH-KMKMU-F4MR5V",
+		Description:   fmt.Sprintf("%s %.f %s @ %s %.f", req.OrderSide, req.Volume, req.Pair, req.OrderSide, req.Price),
+	}
+	msg, err := json.Marshal(resp)
+	if err != nil {
+		return err
+	}
+	return w.WriteMessage(websocket.TextMessage, msg)
+}

exchanges/kraken/testdata/wsHandleData.json

@@ -0,0 +1,10 @@
+{"event":"pong"}
+{"connectionID":8628615390848610000,"event":"systemStatus","status":"online","version":"1.0.0"}
+[1337,{"a":["5525.40000",1,"1.000"],"b":["5525.10000",1,"1.000"],"c":["5525.10000","0.00398963"],"h":["5783.00000","5783.00000"],"l":["5505.00000","5505.00000"],"o":["5760.70000","5763.40000"],"p":["5631.44067","5653.78939"],"t":[11493,16267],"v":["2634.11501494","3591.17907851"]},"ticker","XBT/USD"]
+[13337,["1542057314.748456","1542057360.435743","3586.70000","3586.70000","3586.60000","3586.60000","3586.68894","0.03373000",2],"ohlc-5","XBT/USD"]
+[133337,[["5541.20000","0.15850568","1534614057.321597","s","l",""],["6060.00000","0.02455000","1534614057.324998","b","l",""]],"trade","XBT/USD"]
+[1333337,["5698.40000","5700.00000","1542057299.545897","1.01234567","0.98765432"],"spread","XBT/USD"]
+[13333337,{"as":[["5541.30000","2.50700000","1534614248.123678"],["5541.80000","0.33000000","1534614098.345543"],["5542.70000","0.64700000","1534614244.654432"],["5544.30000","2.50700000","1534614248.123678"],["5545.80000","0.33000000","1534614098.345543"],["5546.70000","0.64700000","1534614244.654432"],["5547.70000","0.64700000","1534614244.654432"],["5548.30000","2.50700000","1534614248.123678"],["5549.80000","0.33000000","1534614098.345543"],["5550.70000","0.64700000","1534614244.654432"]],"bs":[["5541.20000","1.52900000","1534614248.765567"],["5539.90000","0.30000000","1534614241.769870"],["5539.50000","5.00000000","1534613831.243486"],["5538.20000","1.52900000","1534614248.765567"],["5537.90000","0.30000000","1534614241.769870"],["5536.50000","5.00000000","1534613831.243486"],["5535.20000","1.52900000","1534614248.765567"],["5534.90000","0.30000000","1534614241.769870"],["5533.50000","5.00000000","1534613831.243486"],["5532.50000","5.00000000","1534613831.243486"]]},"book-100","XBT/USD"]
+[13333337,{"a":[["5541.30000","2.50700000","1534614248.456738"],["5542.50000","0.40100000","1534614248.456738"]],"c":"4187525586"},"book-10","XBT/USD"]
+[13333337,{"b":[["5541.30000","0.00000000","1534614335.345903"]],"c":"4187525586"},"book-10","XBT/USD"]
+[[{"TDLH43-DVQXD-2KHVYY":{"cost":"1000000.00000","fee":"1600.00000","margin":"0.00000","ordertxid":"TDLH43-DVQXD-2KHVYY","ordertype":"limit","pair":"XBT/USD","postxid":"OGTT3Y-C6I3P-XRI6HX","price":"100000.00000","time":"1560516023.070651","type":"sell","vol":"1000000000.00000000"}},{"TDLH43-DVQXD-2KHVYY":{"cost":"1000000.00000","fee":"600.00000","margin":"0.00000","ordertxid":"TDLH43-DVQXD-2KHVYY","ordertype":"limit","pair":"XBT/USD","postxid":"OGTT3Y-C6I3P-XRI6HX","price":"100000.00000","time":"1560516023.070658","type":"buy","vol":"1000000000.00000000"}},{"TDLH43-DVQXD-2KHVYY":{"cost":"1000000.00000","fee":"1600.00000","margin":"0.00000","ordertxid":"TDLH43-DVQXD-2KHVYY","ordertype":"limit","pair":"XBT/USD","postxid":"OGTT3Y-C6I3P-XRI6HX","price":"100000.00000","time":"1560520332.914657","type":"sell","vol":"1000000000.00000000"}},{"TDLH43-DVQXD-2KHVYY":{"cost":"1000000.00000","fee":"600.00000","margin":"0.00000","ordertxid":"TDLH43-DVQXD-2KHVYY","ordertype":"limit","pair":"XBT/USD","postxid":"OGTT3Y-C6I3P-XRI6HX","price":"100000.00000","time":"1560520332.914664","type":"buy","vol":"1000000000.00000000"}}],"ownTrades",{"sequence":1}]

exchanges/kraken/testdata/wsOpenTrades.json

@@ -1,4 +1,4 @@
-[[{"OGTT3Y-C6I3P-XRI6HR":{"cost":"0.00000","descr":{"close":"","leverage":"0.1","order":"sell 10.00345345 XBT/USD @ limit 34.50000 with 0:1 leverage","ordertype":"limit","pair":"XBT/USD","price":"34.50000","price2":"0.00000","type":"sell"},"expiretm":"0.000000","fee":"0.00000","limitprice":"34.50000","misc":"","oflags":"fcib","opentm":"0.000000","avg_price":"0.00000","refid":"LIMIT-OPEN","starttm":"0.000000","status":"open","stopprice":"0.000000","userref":0,"vol":"10.00345345","vol_exec":"0.00000000"}},{"OKB55A-UEMMN-YUXM2A":{"avg_price":"0.00000","cost":"0.00000","descr":{"close":null,"leverage":null,"order":"buy 0.00010000 XBT/USDT @ market 0.00000","ordertype":"market","pair":"XBT/USDT","price":"0.00000","price2":"0.00000","type":"buy"},"expiretm":null,"fee":"0.00000","limitprice":"0.00000","misc":"","oflags":"fciq","opentm":"1692851641.361371","refid":null,"starttm":null,"status":"pending","stopprice":"0.00000","timeinforce":"GTC","userref":0,"vol":"0.00010000","vol_exec":"0.00000000"}}],"openOrders",{"sequence":1}]
+[[{"OGTT3Y-C6I3P-XRI6HR":{"cost":"0.00000","descr":{"close":"","leverage":"0.1","order":"sell 10.00345345 XBT/USD @ limit 34.50000 with 0:1 leverage","ordertype":"limit","pair":"XBT/USD","price":"34.50000","price2":"0.00000","type":"sell"},"expiretm":"0.000000","fee":"0.00000","limitprice":"34.50000","misc":"","oflags":"fcib","opentm":"0.000000","avg_price":"0.00000","refid":"LIMIT-OPEN","starttm":"0.000000","status":"open","stopprice":"0.000000","userref":0,"vol":"10.00345345","vol_exec":"0.00000000"}},{"OKB55A-UEMMN-YUXM2A":{"avg_price":"0.00000","cost":"0.00000","descr":{"close":null,"leverage":null,"order":"buy 0.00010000 XBT/USD @ market 0.00000","ordertype":"market","pair":"XBT/USD","price":"0.00000","price2":"0.00000","type":"buy"},"expiretm":null,"fee":"0.00000","limitprice":"0.00000","misc":"","oflags":"fciq","opentm":"1692851641.361371","refid":null,"starttm":null,"status":"pending","stopprice":"0.00000","timeinforce":"GTC","userref":0,"vol":"0.00010000","vol_exec":"0.00000000"}}],"openOrders",{"sequence":1}]
 [[{"OKB55A-UEMMN-YUXM2A":{"status":"open","userref":0}}],"openOrders",{"sequence":2}]
 [[{"OKB55A-UEMMN-YUXM2A":{"vol_exec":"0.00010000","cost":"2.64252","fee":"0.00687","avg_price":"26425.20000","userref":0}}],"openOrders",{"sequence":3}]
 [[{"OKB55A-UEMMN-YUXM2A":{"lastupdated":"1692851641.361447","status":"closed","vol_exec":"0.00010000","cost":"2.64252","fee":"0.00687","avg_price":"26425.20000","userref":0}}],"openOrders",{"sequence":4}]

exchanges/kucoin/kucoin_websocket.go

@@ -211,7 +211,7 @@ func (ku *Kucoin) wsHandleData(respData []byte) error {
 	}
 	if resp.ID != "" {
 		if !ku.Websocket.Match.IncomingWithData("msgID:"+resp.ID, respData) {
-			return fmt.Errorf("message listener not found: %s", resp.ID)
+			return fmt.Errorf("%w: %s", stream.ErrNoMessageListener, resp.ID)
 		}
 		return nil
 	}

exchanges/stream/websocket.go

@@ -966,6 +966,9 @@ func (w *Websocket) checkSubscriptions(subs subscription.List) error {
 	}
 
 	for _, s := range subs {
+		if s.State() == subscription.ResubscribingState {
+			continue
+		}
 		if found := w.subscriptions.Get(s); found != nil {
 			return fmt.Errorf("%w: %s", subscription.ErrDuplicate, s)
 		}

exchanges/subscription/keys.go

@@ -87,3 +87,38 @@ func (k IgnoringPairsKey) Match(eachKey MatchableKey) bool {
 		eachSub.Levels == k.Levels &&
 		eachSub.Interval == k.Interval
 }
+
+// IgnoringAssetKey is a key type for finding subscriptions to group together for requests
+type IgnoringAssetKey struct {
+	*Subscription
+}
+
+var _ MatchableKey = IgnoringAssetKey{} // Enforce IgnoringAssetKey must implement MatchableKey
+
+// GetSubscription returns the underlying subscription
+func (k IgnoringAssetKey) GetSubscription() *Subscription {
+	return k.Subscription
+}
+
+// String implements Stringer; returns the asset and Channel name but no pairs
+func (k IgnoringAssetKey) String() string {
+	s := k.Subscription
+	if s == nil {
+		return "Uninitialised IgnoringAssetKey"
+	}
+	return fmt.Sprintf("%s %s", s.Channel, s.Pairs)
+}
+
+// Match implements MatchableKey
+func (k IgnoringAssetKey) Match(eachKey MatchableKey) bool {
+	if eachKey == nil {
+		return false
+	}
+	eachSub := eachKey.GetSubscription()
+
+	return eachSub != nil &&
+		eachSub.Channel == k.Channel &&
+		eachSub.Pairs.Equal(k.Pairs) &&
+		eachSub.Levels == k.Levels &&
+		eachSub.Interval == k.Interval
+}

exchanges/subscription/keys_test.go

@@ -110,6 +110,39 @@ func TestIgnoringPairsKeyString(t *testing.T) {
 	assert.Equal(t, "ticker spot", key.String())
 }
 
+// TestIgnoringAssetKeyMatch exercises IgnoringAssetKey.Match
+func TestIgnoringAssetKeyMatch(t *testing.T) {
+	t.Parallel()
+
+	key := &IgnoringAssetKey{&Subscription{Channel: TickerChannel, Asset: asset.Spot}}
+	try := &DummyKey{&Subscription{Channel: OrderbookChannel}, t}
+
+	require.False(t, key.Match(nil), "Match on a nil must return false")
+	require.False(t, key.Match(try), "Gate 1: Match must reject a bad Channel")
+	try.Channel = TickerChannel
+	require.True(t, key.Match(try), "Gate 1: Match must accept a good Channel")
+	key.Pairs = currency.Pairs{btcusdtPair}
+	require.False(t, key.Match(try), "Gate 2: Match must reject bad Pairs")
+	try.Pairs = currency.Pairs{btcusdtPair}
+	require.True(t, key.Match(try), "Gate 2: Match must accept a good Pairs")
+	key.Levels = 4
+	require.False(t, key.Match(try), "Gate 3: Match must reject a bad Level")
+	try.Levels = 4
+	require.True(t, key.Match(try), "Gate 3: Match must accept a good Level")
+	key.Interval = kline.FiveMin
+	require.False(t, key.Match(try), "Gate 4: Match must reject a bad Interval")
+	try.Interval = kline.FiveMin
+	require.True(t, key.Match(try), "Gate 4: Match must accept a good Interval")
+}
+
+// TestIgnoringAssetKeyString exercises IgnoringAssetKey.String
+func TestIgnoringAssetKeyString(t *testing.T) {
+	t.Parallel()
+	assert.Equal(t, "Uninitialised IgnoringAssetKey", IgnoringAssetKey{}.String())
+	key := &IgnoringAssetKey{&Subscription{Asset: asset.Spot, Channel: TickerChannel, Pairs: currency.Pairs{ethusdcPair, btcusdtPair}}}
+	assert.Equal(t, "ticker [ETHUSDC BTCUSDT]", key.String())
+}
+
 // TestGetSubscription exercises GetSubscription
 func TestGetSubscription(t *testing.T) {
 	t.Parallel()

exchanges/subscription/subscription.go

@@ -35,11 +35,13 @@ const (
 
 // Public errors
 var (
-	ErrNotFound       = errors.New("subscription not found")
-	ErrNotSinglePair  = errors.New("only single pair subscriptions expected")
-	ErrInStateAlready = errors.New("subscription already in state")
-	ErrInvalidState   = errors.New("invalid subscription state")
-	ErrDuplicate      = errors.New("duplicate subscription")
+	ErrNotFound             = errors.New("subscription not found")
+	ErrNotSinglePair        = errors.New("only single pair subscriptions expected")
+	ErrBatchingNotSupported = errors.New("subscription batching not supported")
+	ErrInStateAlready       = errors.New("subscription already in state")
+	ErrInvalidState         = errors.New("invalid subscription state")
+	ErrDuplicate            = errors.New("duplicate subscription")
+	ErrPrivateChannelName   = errors.New("must use standard channel name constants")
 )
 
 // State tracks the status of a subscription channel