subscriptions: Encapsulate, replace Pair with Pairs and refactor; improve exchange support

.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

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/pairs.go

@@ -109,8 +109,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)) ||
@@ -121,15 +120,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 {
@@ -204,8 +201,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
@@ -481,3 +477,30 @@ func (p Pairs) GetPairsByBase(baseTerm Code) (Pairs, error) {
 	}
 	return pairs, nil
 }
+
+// equalKey is a small key for testing pair equality without delimiter
+type equalKey struct {
+	Base  *Item
+	Quote *Item
+}
+
+// Equal checks to see if two lists of pairs contain only the same pairs, ignoring delimiter and case
+// Does not check for inverted/reciprocal pairs
+func (p Pairs) Equal(b Pairs) bool {
+	if len(p) != len(b) {
+		return false
+	}
+	if len(p) == 0 {
+		return true
+	}
+	m := map[equalKey]struct{}{}
+	for i := range p {
+		m[equalKey{Base: p[i].Base.Item, Quote: p[i].Quote.Item}] = struct{}{}
+	}
+	for i := range b {
+		if _, ok := m[equalKey{Base: b[i].Base.Item, Quote: b[i].Quote.Item}]; !ok {
+			return false
+		}
+	}
+	return true
+}

currency/pairs_test.go

@@ -870,3 +870,12 @@ func TestGetPairsByBase(t *testing.T) {
 		t.Fatalf("received: '%v' but expected '%v'", len(got), 3)
 	}
 }
+
+// TestPairsEqual exercises Pairs.Equal
+func TestPairsEqual(t *testing.T) {
+	t.Parallel()
+	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

@@ -721,7 +721,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
 	}
@@ -733,10 +733,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
@@ -754,9 +754,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],
 					})
 			}
@@ -766,9 +766,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
@@ -809,7 +807,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.
@@ -825,13 +823,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
@@ -846,10 +839,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
 }
 ```
 
@@ -1063,7 +1053,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:
@@ -1074,13 +1064,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
@@ -1135,8 +1119,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

@@ -43,9 +43,8 @@ const (
 	Options
 	OptionCombo
 	FutureCombo
-
-	// 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
 
 	optionsFlag   = OptionCombo | Options
 	futuresFlag   = PerpetualContract | PerpetualSwap | Futures | DeliveryFutures | UpsideProfitContract | DownsideProfitContract | CoinMarginedFutures | USDTMarginedFutures | USDCMarginedFutures | LinearContract | FutureCombo
@@ -70,6 +69,7 @@ const (
 	options                = "options"
 	optionCombo            = "option_combo"
 	futureCombo            = "future_combo"
+	all                    = "all"
 )
 
 var (
@@ -120,6 +120,8 @@ func (a Item) String() string {
 		return optionCombo
 	case FutureCombo:
 		return futureCombo
+	case All:
+		return all
 	default:
 		return ""
 	}
@@ -225,11 +227,10 @@ func New(input string) (Item, error) {
 		return OptionCombo, nil
 	case futureCombo:
 		return FutureCombo, 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)
 	}
 }
 

exchanges/binance/binance_test.go

@@ -1462,7 +1462,7 @@ func TestGetHistoricTrades(t *testing.T) {
 	if mockTests {
 		expected = 1002
 	}
-	assert.Equal(t, expected, len(result), "GetHistoricTrades should return correct number of entries") //nolint:testifylint // assert.Len doesn't produce clear messages on result
+	assert.Equal(t, expected, len(result), "GetHistoricTrades should return correct number of entries")
 	for _, r := range result {
 		if !assert.WithinRange(t, r.Timestamp, start, end, "All trades should be within time range") {
 			break
@@ -1982,7 +1982,7 @@ func BenchmarkWsHandleData(bb *testing.B) {
 func TestSubscribe(t *testing.T) {
 	t.Parallel()
 	b := b
-	channels := []subscription.Subscription{
+	channels := subscription.List{
 		{Channel: "btcusdt@ticker"},
 		{Channel: "btcusdt@trade"},
 	}
@@ -2008,7 +2008,7 @@ func TestSubscribe(t *testing.T) {
 
 func TestSubscribeBadResp(t *testing.T) {
 	t.Parallel()
-	channels := []subscription.Subscription{
+	channels := subscription.List{
 		{Channel: "moons@ticker"},
 	}
 	mock := func(msg []byte, w *websocket.Conn) error {
@@ -2434,19 +2434,19 @@ func TestSeedLocalCache(t *testing.T) {
 
 func TestGenerateSubscriptions(t *testing.T) {
 	t.Parallel()
-	expected := []subscription.Subscription{}
+	expected := subscription.List{}
 	pairs, err := b.GetEnabledPairs(asset.Spot)
 	assert.NoError(t, err, "GetEnabledPairs should not error")
 	for _, p := range pairs {
 		for _, c := range []string{"kline_1m", "depth@100ms", "ticker", "trade"} {
-			expected = append(expected, subscription.Subscription{
+			expected = append(expected, &subscription.Subscription{
 				Channel: p.Format(currency.PairFormat{Delimiter: "", Uppercase: false}).String() + "@" + c,
-				Pair:    p,
+				Pairs:   currency.Pairs{p},
 				Asset:   asset.Spot,
 			})
 		}
 	}
-	subs, err := b.GenerateSubscriptions()
+	subs, err := b.generateSubscriptions()
 	assert.NoError(t, err, "GenerateSubscriptions should not error")
 	if assert.Len(t, subs, len(expected), "Should have the correct number of subs") {
 		assert.ElementsMatch(t, subs, expected, "Should get the correct subscriptions")