Kraken Subscription Improvements

.golangci.yml

@@ -146,6 +146,8 @@ linters-settings:
     disable:
       - require-error
       - float-compare
+      # We deliberately use Equal over Len to avoid spamming the contents of large Slices
+      - len
 
 issues:
   max-issues-per-linter: 0

common/convert/convert.go

@@ -59,11 +59,10 @@ func TimeFromUnixTimestampFloat(raw interface{}) (time.Time, error) {
 	return time.UnixMilli(int64(ts)), nil
 }
 
-// TimeFromUnixTimestampDecimal converts a unix timestamp in decimal form to
-// a time.Time
+// TimeFromUnixTimestampDecimal converts a unix timestamp in decimal form to a time.Time
 func TimeFromUnixTimestampDecimal(input float64) time.Time {
 	i, f := math.Modf(input)
-	return time.Unix(int64(i), int64(f*(1e9)))
+	return time.Unix(int64(i), int64(f*(1e9))).UTC()
 }
 
 // UnixTimestampToTime returns time.time

common/convert/convert_test.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/shopspring/decimal"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestFloatFromString(t *testing.T) {
@@ -98,18 +99,17 @@ func TestTimeFromUnixTimestampFloat(t *testing.T) {
 }
 
 func TestTimeFromUnixTimestampDecimal(t *testing.T) {
-	r := TimeFromUnixTimestampDecimal(1590633982.5714)
-	if r.Year() != 2020 ||
-		r.Month().String() != "May" ||
-		r.Day() != 28 {
-		t.Error("unexpected result")
-	}
-
-	r = TimeFromUnixTimestampDecimal(1560516023.070651)
-	if r.Year() != 2019 ||
-		r.Month().String() != "June" ||
-		r.Day() != 14 {
-		t.Error("unexpected result")
+	for in, exp := range map[float64]time.Time{
+		1590633982.5714:   time.Date(2020, 5, 28, 2, 46, 22, 571400000, time.UTC),
+		1560516023.070651: time.Date(2019, 6, 14, 12, 40, 23, 70651000, time.UTC),
+		// Examples from Kraken
+		1373750306.9819:   time.Date(2013, 7, 13, 21, 18, 26, 981900000, time.UTC),
+		1534614098.345543: time.Date(2018, 8, 18, 17, 41, 38, 345543000, time.UTC),
+	} {
+		got := TimeFromUnixTimestampDecimal(in)
+		z, _ := got.Zone()
+		assert.Equal(t, "UTC", z, "TimeFromUnixTimestampDecimal should return a UTC time")
+		assert.WithinRangef(t, got, exp.Add(-time.Microsecond), exp.Add(time.Microsecond), "TimeFromUnixTimestampDecimal(%f) should parse a unix timestamp correctly", in)
 	}
 }
 

config/config_types.go

@@ -317,9 +317,9 @@ type FeaturesEnabledConfig struct {
 
 // FeaturesConfig stores the exchanges supported and enabled features
 type FeaturesConfig struct {
-	Supports      FeaturesSupportedConfig      `json:"supports"`
-	Enabled       FeaturesEnabledConfig        `json:"enabled"`
-	Subscriptions []*subscription.Subscription `json:"subscriptions,omitempty"`
+	Supports      FeaturesSupportedConfig `json:"supports"`
+	Enabled       FeaturesEnabledConfig   `json:"enabled"`
+	Subscriptions subscription.List       `json:"subscriptions,omitempty"`
 }
 
 // APIEndpointsConfig stores the API endpoint addresses

currency/pair.go

@@ -19,8 +19,7 @@ func NewBTCUSD() Pair {
 	return NewPair(BTC, USD)
 }
 
-// NewPairDelimiter splits the desired currency string at delimiter, the returns
-// a Pair struct
+// NewPairDelimiter splits the desired currency string at delimiter, the returns a Pair struct
 func NewPairDelimiter(currencyPair, delimiter string) (Pair, error) {
 	if !strings.Contains(currencyPair, delimiter) {
 		return EMPTYPAIR,
@@ -77,8 +76,7 @@ func NewPairWithDelimiter(base, quote, delimiter string) Pair {
 	}
 }
 
-// NewPairFromIndex returns a CurrencyPair via a currency string and specific
-// index
+// NewPairFromIndex returns a CurrencyPair via a currency string and specific index
 func NewPairFromIndex(currencyPair, index string) (Pair, error) {
 	i := strings.Index(currencyPair, index)
 	if i == -1 {

currency/pairs.go

@@ -1,10 +1,12 @@
 package currency
 
 import (
+	"cmp"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"math/rand"
+	"slices"
 	"strings"
 
 	"github.com/thrasher-corp/gocryptotrader/log"
@@ -59,8 +61,7 @@ func (p Pairs) Join() string {
 
 // Format formats the pair list to the exchange format configuration
 func (p Pairs) Format(pairFmt PairFormat) Pairs {
-	pairs := make(Pairs, len(p))
-	copy(pairs, p)
+	pairs := slices.Clone(p)
 
 	var err error
 	for x := range pairs {
@@ -120,8 +121,7 @@ func (p Pairs) Lower() Pairs {
 	return newSlice
 }
 
-// Contains checks to see if a specified pair exists inside a currency pair
-// array
+// Contains checks to see if a specified pair exists inside a currency pair array
 func (p Pairs) Contains(check Pair, exact bool) bool {
 	for i := range p {
 		if (exact && p[i].Equal(check)) ||
@@ -132,15 +132,13 @@ func (p Pairs) Contains(check Pair, exact bool) bool {
 	return false
 }
 
-// ContainsAll checks to see if all pairs supplied are contained within the
-// original pairs list.
+// ContainsAll checks to see if all pairs supplied are contained within the original pairs list
 func (p Pairs) ContainsAll(check Pairs, exact bool) error {
 	if len(check) == 0 {
 		return ErrCurrencyPairsEmpty
 	}
 
-	comparative := make(Pairs, len(p))
-	copy(comparative, p)
+	comparative := slices.Clone(p)
 list:
 	for x := range check {
 		for y := range comparative {
@@ -215,8 +213,7 @@ func (p Pairs) GetPairsByCurrencies(currencies Currencies) Pairs {
 
 // Remove removes the specified pair from the list of pairs if it exists
 func (p Pairs) Remove(pair Pair) (Pairs, error) {
-	pairs := make(Pairs, len(p))
-	copy(pairs, p)
+	pairs := slices.Clone(p)
 	for x := range p {
 		if p[x].Equal(pair) {
 			return append(pairs[:x], pairs[x+1:]...), nil
@@ -492,3 +489,21 @@ func (p Pairs) GetPairsByBase(baseTerm Code) (Pairs, error) {
 	}
 	return pairs, nil
 }
+
+// Sort sorts the Pairs in place by String comparison
+func (p Pairs) Sort() {
+	slices.SortFunc(p, func(a, b Pair) int {
+		return cmp.Compare(a.String(), b.String())
+	})
+}
+
+// Equal checks to see if two lists of pairs contain the only the same pairs, ignoring delimiter and case
+// Does not check for inverted/reciprocal pairs
+func (p Pairs) Equal(b Pairs) bool {
+	pFmt := PairFormat{Uppercase: true, Delimiter: ""}
+	p = p.Format(pFmt)
+	b = b.Format(pFmt)
+	p.Sort()
+	b.Sort()
+	return slices.Equal(p, b)
+}

currency/pairs_test.go

@@ -4,6 +4,8 @@
 	"encoding/json"
 	"errors"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestPairsUpper(t *testing.T) {
@@ -898,3 +900,19 @@
 		t.Fatalf("received: '%v' but expected '%v'", len(got), 3)
 	}
 }
+
+// TestPairsSort exercises Pairs.Sort
+func TestPairsSort(t *testing.T) {
+	p := Pairs{NewPair(USDT, BTC), NewPair(DAI, XRP), NewPair(DAI, BTC)}
+	p.Sort()
+	assert.Equal(t, []string{"DAIBTC", "DAIXRP", "USDTBTC"}, p.Strings(), "Pairs should be sorted")
+}
+
+// TestPairsEqual exercises Pairs.Equal
+func TestPairsEqual(t *testing.T) {
+
+	orig := Pairs{NewPairWithDelimiter("USDT", "BTC", "-"), NewPair(DAI, XRP), NewPair(DAI, BTC)}
+	assert.True(t, orig.Equal(Pairs{NewPair(DAI, XRP), NewPair(DAI, BTC), NewPair(USDT, BTC)}), "Equal Pairs should return true")
+	assert.Equal(t, "USDT-BTC", orig[0].String(), "Equal Pairs should not effect original order or format")
+	assert.False(t, orig.Equal(Pairs{NewPair(DAI, XRP), NewPair(DAI, BTC), NewPair(USD, LTC)}), "UnEqual Pairs should return false")
+}

docs/ADD_NEW_EXCHANGE.md

@@ -719,7 +719,7 @@ func (f *FTX) WsConnect() error {
 		}
 	}
 	// Generates the default subscription set, based off enabled pairs.
-	subs, err := f.GenerateDefaultSubscriptions()
+	subs, err := f.generateSubscriptions()
 	if err != nil {
 		return err
 	}
@@ -731,10 +731,10 @@ func (f *FTX) WsConnect() error {
 - Create function to generate default subscriptions:
 
 ```go
-// GenerateDefaultSubscriptions generates default subscription
-func (f *FTX) GenerateDefaultSubscriptions() ([]subscription.Subscription, error) {
-	var subscriptions []subscription.Subscription
-	subscriptions = append(subscriptions, subscription.Subscription{
+// generateSubscriptions generates default subscription
+func (f *FTX) generateSubscriptions() (subscription.List, error) {
+	var subscriptions subscription.List
+	subscriptions = append(subscriptions, &subscription.Subscription{
 		Channel: wsMarkets,
 	})
 	// Ranges over available channels, pairs and asset types to produce a full
@@ -752,9 +752,9 @@ func (f *FTX) GenerateDefaultSubscriptions() ([]subscription.Subscription, error
 				"-")
 			for x := range channels {
 				subscriptions = append(subscriptions,
-					subscription.Subscription{
+					&subscription.Subscription{
 						Channel:  channels[x],
-						Pair: newPair,
+						Pair:     currency.Pairs{newPair},
 						Asset:    assets[a],
 					})
 			}
@@ -764,9 +764,7 @@ func (f *FTX) GenerateDefaultSubscriptions() ([]subscription.Subscription, error
 	if f.IsWebsocketAuthenticationSupported() {
 		var authchan = []string{wsOrders, wsFills}
 		for x := range authchan {
-			subscriptions = append(subscriptions, subscription.Subscription{
-				Channel: authchan[x],
-			})
+			subscriptions = append(subscriptions, &subscription.Subscription{Channel: authchan[x]})
 		}
 	}
 	return subscriptions, nil
@@ -807,7 +805,7 @@ type WsSub struct {
 
 ```go
 // Subscribe sends a websocket message to receive data from the channel
-func (f *FTX) Subscribe(channelsToSubscribe []subscription.Subscription) error {
+func (f *FTX) Subscribe(channelsToSubscribe subscription.List) error {
 	// For subscriptions we try to batch as much as possible to limit the amount
 	// of connection usage but sometimes this is not supported on the exchange 
 	// API.
@@ -823,13 +821,8 @@ channels:
 		case wsFills, wsOrders, wsMarkets:
 		// Authenticated wsFills && wsOrders or wsMarkets which is a channel subscription for the full set of tradable markets do not need a currency pair association. 
 		default:
-			a, err := f.GetPairAssetType(channelsToSubscribe[i].Pair)
-			if err != nil {
-				errs = append(errs, err)
-				continue channels
-			}
 			// Ensures our outbound currency pair is formatted correctly, sometimes our configuration format is different from what our request format needs to be.
-			formattedPair, err := f.FormatExchangeCurrency(channelsToSubscribe[i].Pair, a)
+			formattedPair, err := f.FormatExchangeCurrency(channelsToSubscribe[i].Pair, channelsToSubscribe[i].Asset)
 			if err != nil {
 				errs = append(errs, err)
 				continue channels
@@ -844,10 +837,7 @@ channels:
 		// When we have a successful subscription, we can alert our internal management system of the success.
 		f.Websocket.AddSuccessfulSubscriptions(channelsToSubscribe[i])
 	}
-	if errs != nil {
-		return errs
-	}
-	return nil
+    return errs
 }
 ```
 
@@ -1061,7 +1051,7 @@ func (f *FTX) WsAuth(ctx context.Context) error {
 
 ```go
 // Unsubscribe sends a websocket message to stop receiving data from the channel
-func (f *FTX) Unsubscribe(channelsToUnsubscribe []subscription.Subscription) error {
+func (f *FTX) Unsubscribe(channelsToUnsubscribe subscription.List) error {
 	// As with subscribing we want to batch as much as possible, but sometimes this cannot be achieved due to API shortfalls. 
 	var errs common.Errors
 channels:
@@ -1072,13 +1062,7 @@ channels:
 		switch channelsToUnsubscribe[i].Channel {
 		case wsFills, wsOrders, wsMarkets:
 		default:
-			a, err := f.GetPairAssetType(channelsToUnsubscribe[i].Pair)
-			if err != nil {
-				errs = append(errs, err)
-				continue channels
-			}
-
-			formattedPair, err := f.FormatExchangeCurrency(channelsToUnsubscribe[i].Pair, a)
+			formattedPair, err := f.FormatExchangeCurrency(channelsToUnsubscribe[i].Pair, channelsToUnsubscribe[i].Asset)
 			if err != nil {
 				errs = append(errs, err)
 				continue channels
@@ -1133,8 +1117,8 @@ func (f *FTX) Setup(exch *config.Exchange) error {
 		Subscriber:             f.Subscribe, 
 		// Unsubscriber function outlined above.
 		UnSubscriber:           f.Unsubscribe,
-		// GenerateDefaultSubscriptions function outlined above. 
-		GenerateSubscriptions:  f.GenerateDefaultSubscriptions, 
+		// GenerateSubscriptions function outlined above. 
+		GenerateSubscriptions:  f.generateSubscriptions, 
 		// Defines the capabilities of the websocket outlined in supported 
 		// features struct. This allows the websocket connection to be flushed 
 		// appropriately if we have a pair/asset enable/disable change. This is 

engine/rpcserver.go

@@ -3121,7 +3121,7 @@ func (s *RPCServer) WebsocketGetSubscriptions(_ context.Context, r *gctrpc.Webso
 		payload.Subscriptions = append(payload.Subscriptions,
 			&gctrpc.WebsocketSubscription{
 				Channel: subs[i].Channel,
-				Pair:    subs[i].Pair.String(),
+				Pairs:   subs[i].Pairs.Join(),
 				Asset:   subs[i].Asset.String(),
 				Params:  string(params),
 			})

exchanges/asset/asset.go

@@ -41,9 +41,8 @@ const (
 	USDTMarginedFutures
 	USDCMarginedFutures
 	Options
-
-	// Added to represent a USDT and USDC based linear derivatives(futures/perpetual) assets in Bybit V5.
-	LinearContract
+	LinearContract // Added to represent a USDT and USDC based linear derivatives(futures/perpetual) assets in Bybit V5.
+	All
 
 	futuresFlag   = PerpetualContract | PerpetualSwap | Futures | DeliveryFutures | UpsideProfitContract | DownsideProfitContract | CoinMarginedFutures | USDTMarginedFutures | USDCMarginedFutures | LinearContract
 	supportedFlag = Spot | Margin | CrossMargin | MarginFunding | Index | Binary | PerpetualContract | PerpetualSwap | Futures | DeliveryFutures | UpsideProfitContract | DownsideProfitContract | CoinMarginedFutures | USDTMarginedFutures | USDCMarginedFutures | Options | LinearContract
@@ -65,6 +64,7 @@ const (
 	usdtMarginedFutures    = "usdtmarginedfutures"
 	usdcMarginedFutures    = "usdcmarginedfutures"
 	options                = "options"
+	all                    = "all"
 )
 
 var (
@@ -111,6 +111,8 @@ func (a Item) String() string {
 		return usdcMarginedFutures
 	case Options:
 		return options
+	case All:
+		return all
 	default:
 		return ""
 	}
@@ -212,11 +214,10 @@ func New(input string) (Item, error) {
 		return USDCMarginedFutures, nil
 	case options, "option":
 		return Options, nil
+	case all:
+		return All, nil
 	default:
-		return 0, fmt.Errorf("%w '%v', only supports %s",
-			ErrNotSupported,
-			input,
-			supportedList)
+		return 0, fmt.Errorf("%w '%v', only supports %s", ErrNotSupported, input, supportedList)
 	}
 }