kucoin: parse websocket messages

pkg/exchange/kucoin/convert.go

@@ -74,14 +74,14 @@ func convertSubscriptions(ss []types.Subscription) ([]kucoinapi.WebSocketCommand
 	for _, s := range ss {
 		id++
 
-		var subscribeType string
+		var subscribeTopic string
 		switch s.Channel {
 		case types.BookChannel:
 			// see https://docs.kucoin.com/#level-2-market-data
-			subscribeType = "/market/level2" + ":" + toLocalSymbol(s.Symbol)
+			subscribeTopic = "/market/level2" + ":" + toLocalSymbol(s.Symbol)
 
 		case types.KLineChannel:
-			subscribeType = "/market/candles" + ":" + toLocalSymbol(s.Symbol) + "_" + s.Options.Interval
+			subscribeTopic = "/market/candles" + ":" + toLocalSymbol(s.Symbol) + "_" + s.Options.Interval
 
 		default:
 			return nil, fmt.Errorf("websocket channel %s is not supported by kucoin", s.Channel)
@@ -89,8 +89,8 @@ func convertSubscriptions(ss []types.Subscription) ([]kucoinapi.WebSocketCommand
 
 		cmds = append(cmds, kucoinapi.WebSocketCommand{
 			Id:             id,
-			Type:           subscribeType,
-			Topic:          "subscribe",
+			Type:           kucoinapi.WebSocketMessageTypeSubscribe,
+			Topic:          subscribeTopic,
 			PrivateChannel: false,
 			Response:       true,
 		})

pkg/exchange/kucoin/kucoinapi/websocket.go

@@ -36,7 +36,7 @@ const (
 
 type WebSocketCommand struct {
 	Id             int64  `json:"id"`
-	Type           string `json:"type"`
+	Type           WebSocketMessageType `json:"type"`
 	Topic          string `json:"topic"`
 	PrivateChannel bool   `json:"privateChannel"`
 	Response       bool   `json:"response"`

pkg/exchange/kucoin/stream.go

@@ -74,6 +74,29 @@ func (s *Stream) handleConnect() {
 			log.WithError(err).Errorf("subscription error")
 			return
 		}
+	} else {
+		id := time.Now().UnixMilli()
+		cmds := []kucoinapi.WebSocketCommand{
+			{
+				Id:             id,
+				Type:           kucoinapi.WebSocketMessageTypeSubscribe,
+				Topic:          "/spotMarket/tradeOrders",
+				PrivateChannel: true,
+				Response:       true,
+			},
+			{
+				Id:             id + 1,
+				Type:           kucoinapi.WebSocketMessageTypeSubscribe,
+				Topic:          "/account/balance",
+				PrivateChannel: true,
+				Response:       true,
+			},
+		}
+		for _, cmd := range cmds {
+			if err := s.conn.WriteJSON(cmd); err != nil {
+				log.WithError(err).Errorf("private subscribe write error, cmd: %+v", cmd)
+			}
+		}
 	}
 }
 
@@ -164,7 +187,6 @@ func (s *Stream) connect(ctx context.Context) error {
 	// create a new context
 	s.connCtx, s.connCancel = context.WithCancel(ctx)
 
-
 	pingTimeout := s.bullet.PingTimeout()
 	conn.SetReadDeadline(time.Now().Add(pingTimeout))
 	conn.SetPongHandler(func(string) error {
@@ -236,12 +258,17 @@ func (s *Stream) read(ctx context.Context) {
 				continue
 			}
 
+			// used for debugging
+			// fmt.Println(string(message))
+
 			e, err := parseWebsocketPayload(message)
 			if err != nil {
 				log.WithError(err).Error("message parse error")
 				continue
 			}
 
+			// remove bytes, so we won't print them
+			e.Data = nil
 			log.Infof("event: %+v", e)
 
 			if e != nil && e.Object != nil {
@@ -321,35 +348,35 @@ func parseWebsocketPayload(in []byte) (*kucoinapi.WebSocketResponse, error) {
 		switch resp.Subject {
 		case kucoinapi.WebSocketSubjectOrderChange:
 			var o kucoinapi.WebSocketPrivateOrder
-			if err := json.Unmarshal(resp.Data, &o) ; err != nil {
+			if err := json.Unmarshal(resp.Data, &o); err != nil {
 				return &resp, err
 			}
 			resp.Object = &o
 
 		case kucoinapi.WebSocketSubjectAccountBalance:
 			var o kucoinapi.WebSocketAccountBalance
-			if err := json.Unmarshal(resp.Data, &o) ; err != nil {
+			if err := json.Unmarshal(resp.Data, &o); err != nil {
 				return &resp, err
 			}
 			resp.Object = &o
 
 		case kucoinapi.WebSocketSubjectTradeCandlesUpdate:
 			var o kucoinapi.WebSocketKLine
-			if err := json.Unmarshal(resp.Data, &o) ; err != nil {
+			if err := json.Unmarshal(resp.Data, &o); err != nil {
 				return &resp, err
 			}
 			resp.Object = &o
 
 		case kucoinapi.WebSocketSubjectTradeL2Update:
 			var o kucoinapi.WebSocketOrderBookL2
-			if err := json.Unmarshal(resp.Data, &o) ; err != nil {
+			if err := json.Unmarshal(resp.Data, &o); err != nil {
 				return &resp, err
 			}
 			resp.Object = &o
 
 		case kucoinapi.WebSocketSubjectTradeTicker:
 			var o kucoinapi.WebSocketTicker
-			if err := json.Unmarshal(resp.Data, &o) ; err != nil {
+			if err := json.Unmarshal(resp.Data, &o); err != nil {
 				return &resp, err
 			}
 			resp.Object = &o

pkg/exchange/kucoin/testdata/account-balance-01.json

@@ -0,0 +1 @@
+{"id":"61c3728cfd0c3c0001a16a64","type":"message","topic":"/account/balance","userId":"61af6413efeab1000113f08b","channelType":"private","subject":"account.balance","data":{"accountId":"61b48b6d94ab8d000103ea77","available":"42.240598061678","availableChange":"-14.099267307125","currency":"USDT","hold":"14.099267307125","holdChange":"14.099267307125","relationContext":{"symbol":"BTC-USDT","orderId":"61c3728cfd0c3c0001a16a62"},"relationEvent":"trade.hold","relationEventId":"61c3728cfd0c3c0001a16a64","time":"1640198796182","total":"56.339865368803"}}

pkg/exchange/kucoin/testdata/account-balance-02.json

@@ -0,0 +1 @@
+{"id":"61c3728c47d4ea0001c2238a","type":"message","topic":"/account/balance","userId":"61af6413efeab1000113f08b","channelType":"private","subject":"account.balance","data":{"accountId":"61c1fc287de2940001bd2aac","available":"0.00028975","availableChange":"0.00028975","currency":"BTC","hold":"0","holdChange":"0","relationContext":{"symbol":"BTC-USDT","orderId":"61c3728cfd0c3c0001a16a62","tradeId":"61c3728c2e113d2923db40a3"},"relationEvent":"trade.setted","relationEventId":"61c3728c47d4ea0001c2238a","time":"1640198796230","total":"0.00028975"}}

pkg/exchange/kucoin/testdata/ack.json

@@ -0,0 +1 @@
+{"id":"1640198781304","type":"ack"}

pkg/exchange/kucoin/testdata/trade-orders-01.json

@@ -0,0 +1 @@
+{"type":"message","topic":"/spotMarket/tradeOrders","userId":"61af6413efeab1000113f08b","channelType":"private","subject":"orderChange","data":{"symbol":"BTC-USDT","orderType":"limit","side":"buy","orderId":"61c3728cfd0c3c0001a16a62","liquidity":"taker","type":"match","orderTime":1640198796191168550,"size":"0.00028975","filledSize":"0.00028975","price":"48611.5","matchPrice":"48604.5","matchSize":"0.00028975","tradeId":"61c3728c2e113d2923db40a3","remainSize":"0","status":"match","ts":1640198796191168550}}

pkg/exchange/kucoin/testdata/trade-orders-02.json

@@ -0,0 +1 @@
+{"type":"message","topic":"/spotMarket/tradeOrders","userId":"61af6413efeab1000113f08b","channelType":"private","subject":"orderChange","data":{"symbol":"BTC-USDT","orderType":"limit","side":"buy","orderId":"61c3728cfd0c3c0001a16a62","type":"filled","orderTime":1640198796191168550,"size":"0.00028975","filledSize":"0.00028975","price":"48611.5","remainSize":"0","status":"done","ts":1640198796191168550}}

pkg/exchange/kucoin/testdata/welcome.json

@@ -0,0 +1 @@
+{"id":"TuhpZyeoee","type":"welcome"}