Kucoin: Add subscription templating and various fixes

README.md

@@ -142,12 +142,12 @@ Binaries will be published once the codebase reaches a stable condition.
 
 |User|Contribution Amount|
 |--|--|
-| [thrasher-](https://github.com/thrasher-) | 690 |
-| [shazbert](https://github.com/shazbert) | 330 |
-| [dependabot[bot]](https://github.com/apps/dependabot) | 287 |
+| [thrasher-](https://github.com/thrasher-) | 692 |
+| [shazbert](https://github.com/shazbert) | 333 |
+| [dependabot[bot]](https://github.com/apps/dependabot) | 293 |
 | [gloriousCode](https://github.com/gloriousCode) | 234 |
 | [dependabot-preview[bot]](https://github.com/apps/dependabot-preview) | 88 |
-| [gbjk](https://github.com/gbjk) | 76 |
+| [gbjk](https://github.com/gbjk) | 80 |
 | [xtda](https://github.com/xtda) | 47 |
 | [lrascao](https://github.com/lrascao) | 27 |
 | [Beadko](https://github.com/Beadko) | 17 |
@@ -162,8 +162,8 @@ Binaries will be published once the codebase reaches a stable condition.
 | [marcofranssen](https://github.com/marcofranssen) | 8 |
 | [140am](https://github.com/140am) | 8 |
 | [TaltaM](https://github.com/TaltaM) | 6 |
+| [cranktakular](https://github.com/cranktakular) | 6 |
 | [dackroyd](https://github.com/dackroyd) | 5 |
-| [cranktakular](https://github.com/cranktakular) | 5 |
 | [khcchiu](https://github.com/khcchiu) | 5 |
 | [yangrq1018](https://github.com/yangrq1018) | 4 |
 | [woshidama323](https://github.com/woshidama323) | 3 |

cmd/documentation/exchanges_templates/kucoin.tmpl

@@ -0,0 +1,40 @@
+{{define "exchanges kucoin" -}}
+{{template "header" .}}
+## Kucoin Exchange
+
+### Current Features
+
++ REST Support
++ Websocket Support
+
+### Subscriptions
+
+Default Public Subscriptions:
+- Ticker for spot, margin and futures
+- Orderbook for spot, margin and futures
+- All trades for spot and margin
+
+Default Authenticated Subscriptions:
+- All trades for futures
+- Stop Order Lifecycle events for futures
+- Account Balance events for spot, margin and futures
+- Margin Position updates
+- Margin Loan updates
+
+Subscriptions are subject to enabled assets and pairs.
+
+Limitations:
+- 100 symbols per subscription
+- 300 symbols per connection
+
+Due to these limitations, if more than 10 symbols are enabled, ticker will subscribe to ticker:all.
+
+Unimplemented subscriptions:
+- Candles for Futures
+- Market snapshot for currency
+
+### Please click GoDocs chevron above to view current GoDoc information for this package
+
+{{template "contributions"}}
+{{template "donations" .}}
+{{end}}

cmd/documentation/exchanges_templates/subscription.tmpl

@@ -20,12 +20,30 @@ The template is provided with a single context structure:
   AssetPairs     map[asset.Item]currency.Pairs
   AssetSeparator string
   PairSeparator  string
+  BatchSize      string
 ```
 
 Subscriptions may fan out many channels for assets and pairs, to support exchanges which require individual subscriptions.  
-To allow the template to communicate how to handle its output it should use the provided separators:
+To allow the template to communicate how to handle its output it should use the provided directives:
 - AssetSeparator should be added at the end of each section related to assets
 - PairSeparator should be added at the end of each pair
+- BatchSize should be added with a number directly before AssetSeparator to indicate pairs have been batched
+
+Example:
+```{{`
+{{- range $asset, $pairs := $.AssetPairs }}
+    {{- range $b := batch $pairs 30 -}}
+        {{- $.S.Channel -}} : {{- $b.Join -}}
+        {{ $.PairSeparator }}
+    {{- end -}}
+    {{- $.BatchSize -}} 30
+    {{- $.AssetSeparator }}
+{{- end }}
+`}}```
+
+Assets and pairs should be output in the sequence in AssetPairs since text/template range function uses an sorted order for map keys.
+
+Template functions may modify AssetPairs to update the subscription's pairs, e.g. Filtering out margin pairs already in spot subscription
 
 We use separators like this because it allows mono-templates to decide at runtime whether to fan out.
 

common/common.go

@@ -15,6 +15,7 @@ import (
 	"path/filepath"
 	"reflect"
 	"regexp"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -388,20 +389,6 @@ func ChangePermission(directory string) error {
 	})
 }
 
-// SplitStringSliceByLimit splits a slice of strings into slices by input limit and returns a slice of slice of strings
-func SplitStringSliceByLimit(in []string, limit uint) [][]string {
-	var stringSlice []string
-	sliceSlice := make([][]string, 0, len(in)/int(limit)+1)
-	for len(in) >= int(limit) {
-		stringSlice, in = in[:limit], in[limit:]
-		sliceSlice = append(sliceSlice, stringSlice)
-	}
-	if len(in) > 0 {
-		sliceSlice = append(sliceSlice, in)
-	}
-	return sliceSlice
-}
-
 // AddPaddingOnUpperCase adds padding to a string when detecting an upper case letter. If
 // there are multiple upper case items like `ThisIsHTTPExample`, it will only
 // pad between like this `This Is HTTP Example`.
@@ -653,3 +640,34 @@ func GetTypeAssertError(required string, received interface{}, fieldDescription
 	}
 	return fmt.Errorf("%w from %T to %s%s", ErrTypeAssertFailure, received, required, description)
 }
+
+// Batch takes a slice type and converts it into a slice of containing slices of length batchSize, and any remainder in the final batch
+// batchSize <= 0 will return the entire input slice in one batch
+func Batch[S ~[]E, E any](blobs S, batchSize int) []S {
+	if len(blobs) == 0 {
+		return []S{}
+	}
+	blobs = slices.Clone(blobs)
+	if batchSize <= 0 {
+		return []S{blobs}
+	}
+	i := 0
+	batches := make([]S, (len(blobs)+batchSize-1)/batchSize)
+	for batchSize < len(blobs) {
+		blobs, batches[i] = blobs[batchSize:], blobs[:batchSize:batchSize]
+		i++
+	}
+	if len(blobs) > 0 {
+		batches[i] = blobs
+	}
+	return batches
+}
+
+// SortStrings takes a slice of fmt.Stringer implementers and returns a new ascending sorted slice
+func SortStrings[S ~[]E, E fmt.Stringer](x S) S {
+	n := slices.Clone(x)
+	slices.SortFunc(n, func(a, b E) int {
+		return strings.Compare(a.String(), b.String())
+	})
+	return n
+}

common/common_test.go

@@ -565,26 +565,6 @@ func initStringSlice(size int) (out []string) {
 	return
 }
 
-func TestSplitStringSliceByLimit(t *testing.T) {
-	t.Parallel()
-	slice50 := initStringSlice(50)
-	out := SplitStringSliceByLimit(slice50, 20)
-	if len(out) != 3 {
-		t.Errorf("expected len() to be 3 instead received: %v", len(out))
-	}
-	if len(out[0]) != 20 {
-		t.Errorf("expected len() to be 20 instead received: %v", len(out[0]))
-	}
-
-	out = SplitStringSliceByLimit(slice50, 50)
-	if len(out) != 1 {
-		t.Errorf("expected len() to be 3 instead received: %v", len(out))
-	}
-	if len(out[0]) != 50 {
-		t.Errorf("expected len() to be 20 instead received: %v", len(out[0]))
-	}
-}
-
 func TestAddPaddingOnUpperCase(t *testing.T) {
 	t.Parallel()
 
@@ -856,3 +836,36 @@ func TestErrorCollector(t *testing.T) {
 	require.True(t, ok, "Must return a multiError")
 	assert.Len(t, errs.Unwrap(), 2, "Should have 2 errors")
 }
+
+// TestBatch ensures the Batch function does not regress into common behavioural faults if implementation changes
+func TestBatch(t *testing.T) {
+	s := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
+	b := Batch(s, 3)
+	require.Len(t, b, 4)
+	assert.Len(t, b[0], 3)
+	assert.Len(t, b[3], 1)
+
+	b[0][0] = 42
+	assert.Equal(t, 1, s[0], "Changing the batches must not change the source")
+
+	require.NotPanics(t, func() { Batch(s, -1) }, "Must not panic on negative batch size")
+	done := make(chan any, 1)
+	go func() { done <- Batch(s, 0) }()
+	require.Eventually(t, func() bool { return len(done) > 0 }, time.Second, time.Millisecond, "Batch 0 must not hang")
+
+	for _, i := range []int{-1, 0, 50} {
+		b = Batch(s, i)
+		require.Lenf(t, b, 1, "A batch size of %v should produce a single batch", i)
+		assert.Lenf(t, b[0], len(s), "A batch size of %v should produce a single batch", i)
+	}
+}
+
+type A int
+
+func (a A) String() string {
+	return strconv.Itoa(int(a))
+}
+
+func TestSortStrings(t *testing.T) {
+	assert.Equal(t, []A{1, 2, 5, 6}, SortStrings([]A{6, 2, 5, 1}))
+}

currency/manager.go

@@ -289,11 +289,14 @@ func (p *PairsManager) DisablePair(a asset.Item, pair Pair) error {
 		return err
 	}
 
-	enabled, err := pairStore.Enabled.Remove(pair)
-	if err != nil {
-		return err
+	enabledLen := len(pairStore.Enabled)
+
+	pairStore.Enabled = pairStore.Enabled.Remove(pair)
+
+	if enabledLen == len(pairStore.Enabled) {
+		return fmt.Errorf("%w %s", ErrPairNotFound, pair)
 	}
-	pairStore.Enabled = enabled
+
 	return nil
 }
 

currency/manager_test.go

@@ -394,42 +394,30 @@ func TestDisablePair(t *testing.T) {
 	t.Parallel()
 	p := initTest(t)
 
-	if err := p.DisablePair(asset.Empty, EMPTYPAIR); !errors.Is(err, asset.ErrNotSupported) {
-		t.Fatalf("received: '%v' but expected: '%v'", err, asset.ErrNotSupported)
-	}
+	err := p.DisablePair(asset.Empty, EMPTYPAIR)
+	assert.ErrorIs(t, err, asset.ErrNotSupported, "Empty asset should error")
 
-	if err := p.DisablePair(asset.Spot, EMPTYPAIR); !errors.Is(err, ErrCurrencyPairEmpty) {
-		t.Fatalf("received: '%v' but expected: '%v'", err, ErrCurrencyPairEmpty)
-	}
+	err = p.DisablePair(asset.Spot, EMPTYPAIR)
+	assert.ErrorIs(t, err, ErrCurrencyPairEmpty, "Empty pair should error")
 
 	p.Pairs = nil
-	// Test disabling a pair when the pair manager is not initialised
-	if err := p.DisablePair(asset.Spot, NewPair(BTC, USD)); err == nil {
-		t.Error("unexpected result")
-	}
+	err = p.DisablePair(asset.Spot, NewPair(BTC, USD))
+	assert.ErrorIs(t, err, ErrPairManagerNotInitialised, "Uninitialised PairManager should error")
 
-	// Test asset type which doesn't exist
 	p = initTest(t)
-	if err := p.DisablePair(asset.Futures, EMPTYPAIR); err == nil {
-		t.Error("unexpected result")
-	}
+	err = p.DisablePair(asset.CoinMarginedFutures, EMPTYPAIR)
+	assert.ErrorIs(t, err, ErrCurrencyPairEmpty, "Non-existent asset type should error")
 
-	// Test asset type which has an empty pair store
 	p.Pairs[asset.Spot] = nil
-	if err := p.DisablePair(asset.Spot, EMPTYPAIR); err == nil {
-		t.Error("unexpected result")
-	}
+	err = p.DisablePair(asset.Spot, EMPTYPAIR)
+	assert.ErrorIs(t, err, ErrCurrencyPairEmpty, "Empty pair store should error")
 
-	// Test disabling a pair which isn't enabled
 	p = initTest(t)
-	if err := p.DisablePair(asset.Spot, NewPair(LTC, USD)); err == nil {
-		t.Error("unexpected result")
-	}
+	err = p.DisablePair(asset.Spot, NewPair(LTC, USD))
+	assert.ErrorIs(t, err, ErrPairNotFound, "Not Enabled pair should error")
 
-	// Test disabling a valid pair and ensure nil is empty
-	if err := p.DisablePair(asset.Spot, NewPair(BTC, USD)); err != nil {
-		t.Error("unexpected result")
-	}
+	err = p.DisablePair(asset.Spot, NewPair(BTC, USD))
+	assert.NoError(t, err, "DisablePair should not error")
 }
 
 func TestEnablePair(t *testing.T) {

currency/pairs.go

@@ -199,28 +199,26 @@ func (p Pairs) GetPairsByCurrencies(currencies Currencies) Pairs {
 	return pairs
 }
 
-// Remove removes the specified pair from the list of pairs if it exists
-func (p Pairs) Remove(pair Pair) (Pairs, error) {
-	pairs := slices.Clone(p)
-	for x := range p {
-		if p[x].Equal(pair) {
-			return append(pairs[:x], pairs[x+1:]...), nil
+// Remove removes the specified pairs from the list of pairs if they exist
+func (p Pairs) Remove(rem ...Pair) Pairs {
+	n := make(Pairs, 0, len(p))
+	for _, pN := range p {
+		if !slices.ContainsFunc(rem, func(pX Pair) bool { return pX.Equal(pN) }) {
+			n = append(n, pN)
 		}
 	}
-	return nil, fmt.Errorf("%s %w", pair, ErrPairNotFound)
+	return slices.Clip(n)
 }
 
-// Add adds a specified pair to the list of pairs if it doesn't exist
+// Add adds pairs to the list of pairs ignoring duplicates
 func (p Pairs) Add(pairs ...Pair) Pairs {
-	merge := append(slices.Clone(p), pairs...)
-	var filterInt int
-	for x := len(p); x < len(merge); x++ {
-		if !merge[:len(p)+filterInt].Contains(merge[x], true) {
-			merge[len(p)+filterInt] = merge[x]
-			filterInt++
+	n := slices.Clone(p)
+	for _, a := range pairs {
+		if !n.Contains(a, true) {
+			n = append(n, a)
 		}
 	}
-	return merge[:len(p)+filterInt]
+	return n
 }
 
 // GetMatch returns either the pair that is equal including the reciprocal for