Merge branch 'master' into gateio_rate_limits

exchanges/bithumb/bithumb_websocket.go

@@ -5,11 +5,16 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"strings"
+	"text/template"
 	"time"
 
+	"github.com/Masterminds/sprig/v3"
 	"github.com/gorilla/websocket"
 	"github.com/thrasher-corp/gocryptotrader/common"
+	"github.com/thrasher-corp/gocryptotrader/currency"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/kline"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/request"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
@@ -24,10 +29,15 @@ const (
 )
 
 var (
-	wsDefaultTickTypes = []string{"30M"} // alternatives "1H", "12H", "24H", "MID"
-	location           *time.Location
+	location *time.Location
 )
 
+var defaultSubscriptions = subscription.List{
+	{Enabled: true, Asset: asset.Spot, Channel: subscription.TickerChannel, Interval: kline.ThirtyMin}, // alternatives "1H", "12H", "24H"
+	{Enabled: true, Asset: asset.Spot, Channel: subscription.OrderbookChannel},
+	{Enabled: true, Asset: asset.Spot, Channel: subscription.AllTradesChannel},
+}
+
 // WsConnect initiates a websocket connection
 func (b *Bithumb) WsConnect() error {
 	if !b.Websocket.IsEnabled() || !b.IsEnabled() {
@@ -171,41 +181,19 @@ func (b *Bithumb) wsHandleData(respRaw []byte) error {
 
 // generateSubscriptions generates the default subscription set
 func (b *Bithumb) generateSubscriptions() (subscription.List, error) {
-	var channels = []string{"ticker", "transaction", "orderbookdepth"}
-	var subscriptions subscription.List
-	pairs, err := b.GetEnabledPairs(asset.Spot)
-	if err != nil {
-		return nil, err
-	}
+	return b.Features.Subscriptions.ExpandTemplates(b)
+}
 
-	pFmt, err := b.GetPairFormat(asset.Spot, true)
-	if err != nil {
-		return nil, err
-	}
-	pairs = pairs.Format(pFmt)
-
-	for y := range channels {
-		subscriptions = append(subscriptions, &subscription.Subscription{
-			Channel: channels[y],
-			Pairs:   pairs,
-			Asset:   asset.Spot,
-		})
-	}
-	return subscriptions, nil
+// GetSubscriptionTemplate returns a subscription channel template
+func (b *Bithumb) GetSubscriptionTemplate(_ *subscription.Subscription) (*template.Template, error) {
+	return template.New("master.tmpl").Funcs(sprig.FuncMap()).Funcs(template.FuncMap{"subToReq": subToReq}).Parse(subTplText)
 }
 
 // Subscribe subscribes to a set of channels
-func (b *Bithumb) Subscribe(channelsToSubscribe subscription.List) error {
+func (b *Bithumb) Subscribe(subs subscription.List) error {
 	var errs error
-	for _, s := range channelsToSubscribe {
-		req := &WsSubscribe{
-			Type:    s.Channel,
-			Symbols: s.Pairs,
-		}
-		if s.Channel == "ticker" {
-			req.TickTypes = wsDefaultTickTypes
-		}
-		err := b.Websocket.Conn.SendJSONMessage(context.TODO(), request.Unset, req)
+	for _, s := range subs {
+		err := b.Websocket.Conn.SendJSONMessage(context.TODO(), request.Unset, json.RawMessage(s.QualifiedChannel))
 		if err == nil {
 			err = b.Websocket.AddSuccessfulSubscriptions(b.Websocket.Conn, s)
 		}
@@ -215,3 +203,32 @@ func (b *Bithumb) Subscribe(channelsToSubscribe subscription.List) error {
 	}
 	return errs
 }
+
+// subToReq returns the subscription as a map to populate WsSubscribe
+func subToReq(s *subscription.Subscription, p currency.Pairs) *WsSubscribe {
+	req := &WsSubscribe{
+		Type:    s.Channel,
+		Symbols: common.SortStrings(p),
+	}
+	switch s.Channel {
+	case subscription.TickerChannel:
+		// As-is
+	case subscription.OrderbookChannel:
+		req.Type = "orderbookdepth"
+	case subscription.AllTradesChannel:
+		req.Type = "transaction"
+	default:
+		panic(fmt.Errorf("%w: %s", subscription.ErrNotSupported, s.Channel))
+	}
+	if s.Interval > 0 {
+		req.TickTypes = []string{strings.ToUpper(s.Interval.Short())}
+	}
+	return req
+}
+
+const subTplText = `
+{{ range $asset, $pairs := $.AssetPairs }}
+	{{- subToReq $.S $pairs | mustToJson }}
+	{{- $.AssetSeparator }}
+{{- end }}
+`

exchanges/bithumb/bithumb_websocket_test.go

@@ -4,11 +4,17 @@ import (
 	"errors"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"github.com/thrasher-corp/gocryptotrader/currency"
 	exchange "github.com/thrasher-corp/gocryptotrader/exchanges"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/kline"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/ticker"
+	testexch "github.com/thrasher-corp/gocryptotrader/internal/testing/exchange"
+	testsubs "github.com/thrasher-corp/gocryptotrader/internal/testing/subscriptions"
 )
 
 var (
@@ -87,13 +93,40 @@ func TestWsHandleData(t *testing.T) {
 	}
 }
 
+func TestSubToReq(t *testing.T) {
+	t.Parallel()
+	p := currency.Pairs{currency.NewPairWithDelimiter("BTC", "KRW", "_"), currency.NewPairWithDelimiter("ETH", "KRW", "_")}
+	r := subToReq(&subscription.Subscription{Channel: subscription.AllTradesChannel}, p)
+	assert.Equal(t, "transaction", r.Type)
+	assert.True(t, p.Equal(r.Symbols))
+	r = subToReq(&subscription.Subscription{Channel: subscription.OrderbookChannel}, p)
+	assert.Equal(t, "orderbookdepth", r.Type)
+	assert.True(t, p.Equal(r.Symbols))
+	r = subToReq(&subscription.Subscription{Channel: subscription.TickerChannel, Interval: kline.OneHour}, p)
+	assert.Equal(t, "ticker", r.Type)
+	assert.True(t, p.Equal(r.Symbols))
+	assert.Equal(t, []string{"1H"}, r.TickTypes)
+	assert.PanicsWithError(t,
+		"subscription channel not supported: myTrades",
+		func() { subToReq(&subscription.Subscription{Channel: subscription.MyTradesChannel}, p) },
+		"should panic on invalid channel",
+	)
+}
+
 func TestGenerateSubscriptions(t *testing.T) {
 	t.Parallel()
-	sub, err := b.generateSubscriptions()
-	if err != nil {
-		t.Fatal(err)
-	}
-	if sub == nil {
-		t.Fatal("unexpected value")
+	b := new(Bithumb)
+	require.NoError(t, testexch.Setup(b), "Test instance Setup must not error")
+	p := currency.Pairs{currency.NewPairWithDelimiter("BTC", "KRW", "_"), currency.NewPairWithDelimiter("ETH", "KRW", "_")}
+	require.NoError(t, b.CurrencyPairs.StorePairs(asset.Spot, p, false))
+	require.NoError(t, b.CurrencyPairs.StorePairs(asset.Spot, p, true))
+	subs, err := b.generateSubscriptions()
+	require.NoError(t, err)
+	exp := subscription.List{
+		{Asset: asset.Spot, Channel: subscription.AllTradesChannel, Pairs: p, QualifiedChannel: `{"type":"transaction","symbols":["BTC_KRW","ETH_KRW"]}`},
+		{Asset: asset.Spot, Channel: subscription.OrderbookChannel, Pairs: p, QualifiedChannel: `{"type":"orderbookdepth","symbols":["BTC_KRW","ETH_KRW"]}`},
+		{Asset: asset.Spot, Channel: subscription.TickerChannel, Pairs: p, Interval: kline.ThirtyMin,
+			QualifiedChannel: `{"type":"ticker","symbols":["BTC_KRW","ETH_KRW"],"tickTypes":["30M"]}`},
 	}
+	testsubs.EqualLists(t, exp, subs)
 }

exchanges/bithumb/bithumb_wrapper.go

@@ -109,6 +109,7 @@ func (b *Bithumb) SetDefaults() {
 				GlobalResultLimit: 1500,
 			},
 		},
+		Subscriptions: defaultSubscriptions.Clone(),
 	}
 	b.Requester, err = request.New(b.Name,
 		common.NewHTTPClientWithTimeout(exchange.DefaultHTTPTimeout),

exchanges/btcmarkets/btcmarkets_test.go

@@ -19,7 +19,9 @@ import (
 	"github.com/thrasher-corp/gocryptotrader/exchanges/order"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/orderbook"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/sharedtestvalues"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
 	testexch "github.com/thrasher-corp/gocryptotrader/internal/testing/exchange"
+	testsubs "github.com/thrasher-corp/gocryptotrader/internal/testing/subscriptions"
 )
 
 var b = &BTCMarkets{}
@@ -1134,3 +1136,33 @@ func TestGetCurrencyTradeURL(t *testing.T) {
 		assert.NotEmpty(t, resp)
 	}
 }
+
+func TestGenerateSubscriptions(t *testing.T) {
+	t.Parallel()
+	b := new(BTCMarkets)
+	require.NoError(t, testexch.Setup(b), "Test instance Setup must not error")
+	p := currency.Pairs{currency.NewPairWithDelimiter("BTC", "USD", "_"), currency.NewPairWithDelimiter("ETH", "BTC", "_")}
+	require.NoError(t, b.CurrencyPairs.StorePairs(asset.Spot, p, false))
+	require.NoError(t, b.CurrencyPairs.StorePairs(asset.Spot, p, true))
+	b.Websocket.SetCanUseAuthenticatedEndpoints(true)
+	require.True(t, b.Websocket.CanUseAuthenticatedEndpoints(), "CanUseAuthenticatedEndpoints must return true")
+	subs, err := b.generateSubscriptions()
+	require.NoError(t, err, "generateSubscriptions must not error")
+	pairs, err := b.GetEnabledPairs(asset.Spot)
+	require.NoError(t, err, "GetEnabledPairs must not error")
+	exp := subscription.List{}
+	for _, baseSub := range b.Features.Subscriptions {
+		s := baseSub.Clone()
+		if !s.Authenticated && s.Channel != subscription.HeartbeatChannel {
+			s.Pairs = pairs
+		}
+		s.QualifiedChannel = channelName(s)
+		exp = append(exp, s)
+	}
+	testsubs.EqualLists(t, exp, subs)
+	assert.PanicsWithError(t,
+		"subscription channel not supported: wibble",
+		func() { channelName(&subscription.Subscription{Channel: "wibble"}) },
+		"should panic on invalid channel",
+	)
+}

exchanges/btcmarkets/btcmarkets_websocket.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"text/template"
 	"time"
 
 	"github.com/gorilla/websocket"
@@ -33,10 +34,26 @@ const (
 var (
 	errTypeAssertionFailure = errors.New("type assertion failure")
 	errChecksumFailure      = errors.New("crc32 checksum failure")
-
-	authChannels = []string{fundChange, heartbeat, orderChange}
 )
 
+var defaultSubscriptions = subscription.List{
+	{Enabled: true, Asset: asset.Spot, Channel: subscription.TickerChannel},
+	{Enabled: true, Asset: asset.Spot, Channel: subscription.OrderbookChannel},
+	{Enabled: true, Asset: asset.Spot, Channel: subscription.AllTradesChannel},
+	{Enabled: true, Channel: subscription.MyOrdersChannel, Authenticated: true},
+	{Enabled: true, Channel: subscription.MyAccountChannel, Authenticated: true},
+	{Enabled: true, Channel: subscription.HeartbeatChannel},
+}
+
+var subscriptionNames = map[string]string{
+	subscription.OrderbookChannel: wsOrderbookUpdate,
+	subscription.TickerChannel:    tick,
+	subscription.AllTradesChannel: tradeEndPoint,
+	subscription.MyOrdersChannel:  orderChange,
+	subscription.MyAccountChannel: fundChange,
+	subscription.HeartbeatChannel: heartbeat,
+}
+
 // WsConnect connects to a websocket feed
 func (b *BTCMarkets) WsConnect() error {
 	if !b.Websocket.IsEnabled() || !b.IsEnabled() {
@@ -326,29 +343,13 @@ func (b *BTCMarkets) wsHandleData(respRaw []byte) error {
 	return nil
 }
 
-func (b *BTCMarkets) generateDefaultSubscriptions() (subscription.List, error) {
-	var channels = []string{wsOrderbookUpdate, tick, tradeEndPoint}
-	enabledCurrencies, err := b.GetEnabledPairs(asset.Spot)
-	if err != nil {
-		return nil, err
-	}
-	var subscriptions subscription.List
-	for i := range channels {
-		subscriptions = append(subscriptions, &subscription.Subscription{
-			Channel: channels[i],
-			Pairs:   enabledCurrencies,
-			Asset:   asset.Spot,
-		})
-	}
+func (b *BTCMarkets) generateSubscriptions() (subscription.List, error) {
+	return b.Features.Subscriptions.ExpandTemplates(b)
+}
 
-	if b.Websocket.CanUseAuthenticatedEndpoints() {
-		for i := range authChannels {
-			subscriptions = append(subscriptions, &subscription.Subscription{
-				Channel: authChannels[i],
-			})
-		}
-	}
-	return subscriptions, nil
+// GetSubscriptionTemplate returns a subscription channel template
+func (b *BTCMarkets) GetSubscriptionTemplate(_ *subscription.Subscription) (*template.Template, error) {
+	return template.New("master.tmpl").Funcs(template.FuncMap{"channelName": channelName}).Parse(subTplText)
 }
 
 // Subscribe sends a websocket message to receive data from the channel
@@ -358,26 +359,33 @@ func (b *BTCMarkets) Subscribe(subs subscription.List) error {
 	}
 
 	var errs error
-	for _, s := range subs {
-		if baseReq.Key == "" && common.StringSliceContains(authChannels, s.Channel) {
-			if err := b.authWsSubscibeReq(baseReq); err != nil {
-				return err
+	if authed := subs.Private(); len(authed) > 0 {
+		if err := b.signWsReq(baseReq); err != nil {
+			errs = err
+			for _, s := range authed {
+				errs = common.AppendError(errs, fmt.Errorf("%w: %s", request.ErrAuthRequestFailed, s))
 			}
+			subs = subs.Public()
 		}
+	}
 
+	for _, batch := range subs.GroupByPairs() {
 		if baseReq.MessageType == subscribe && len(b.Websocket.GetSubscriptions()) != 0 {
 			baseReq.MessageType = addSubscription // After first *successful* subscription API requires addSubscription
 			baseReq.ClientType = clientType       // Note: Only addSubscription requires/accepts clientType
 		}
 
 		r := baseReq
 
-		r.Channels = []string{s.Channel}
-		r.MarketIDs = s.Pairs.Strings()
+		r.MarketIDs = batch[0].Pairs.Strings()
+		r.Channels = make([]string, len(batch))
+		for i, s := range batch {
+			r.Channels[i] = s.QualifiedChannel
+		}
 
 		err := b.Websocket.Conn.SendJSONMessage(context.TODO(), request.Unset, r)
 		if err == nil {
-			err = b.Websocket.AddSuccessfulSubscriptions(b.Websocket.Conn, s)
+			err = b.Websocket.AddSuccessfulSubscriptions(b.Websocket.Conn, batch...)
 		}
 		if err != nil {
 			errs = common.AppendError(errs, err)
@@ -387,7 +395,7 @@ func (b *BTCMarkets) Subscribe(subs subscription.List) error {
 	return errs
 }
 
-func (b *BTCMarkets) authWsSubscibeReq(r *WsSubscribe) error {
+func (b *BTCMarkets) signWsReq(r *WsSubscribe) error {
 	creds, err := b.GetCredentials(context.TODO())
 	if err != nil {
 		return err
@@ -471,11 +479,24 @@ func concat(liquidity orderbook.Tranches) string {
 	return c
 }
 
-// trim turns value into string, removes the decimal point and all the leading
-// zeros.
+// trim turns value into string, removes the decimal point and all the leading zeros
 func trim(value float64) string {
 	valstr := strconv.FormatFloat(value, 'f', -1, 64)
 	valstr = strings.ReplaceAll(valstr, ".", "")
 	valstr = strings.TrimLeft(valstr, "0")
 	return valstr
 }
+
+func channelName(s *subscription.Subscription) string {
+	if n, ok := subscriptionNames[s.Channel]; ok {
+		return n
+	}
+	panic(fmt.Errorf("%w: %s", subscription.ErrNotSupported, s.Channel))
+}
+
+const subTplText = `
+{{ range $asset, $pairs := $.AssetPairs }}
+	{{- channelName $.S -}}
+	{{ $.AssetSeparator }}
+{{- end }}
+`