Okx: Fix sending trades to the websocket DataHandler (#1833)

* Send trades down the DataHandler, add test

* Send the trades down to the DataHandler

Update the test to process trades for each asset type
Add raw trade data from OKX
Update the timestamp in the method

* Replace pair repetition with p var

* Update the test, remove repetition

* Rename trade var in the test

* Fix nits and slice sorting

* Compare trades

* Fix nits

exchanges/okx/okx_test.go

@@ -6,6 +6,8 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"slices"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -30,6 +32,7 @@ import (
 	"github.com/thrasher-corp/gocryptotrader/exchanges/request"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/sharedtestvalues"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/trade"
 	testexch "github.com/thrasher-corp/gocryptotrader/internal/testing/exchange"
 	testsubs "github.com/thrasher-corp/gocryptotrader/internal/testing/subscriptions"
 	"github.com/thrasher-corp/gocryptotrader/portfolio/withdraw"
@@ -4045,6 +4048,75 @@ func setupWS() {
 	}
 }
 
+func TestWSProcessTrades(t *testing.T) {
+	t.Parallel()
+
+	ok := new(Okx) //nolint:govet // Intentional shadow to avoid future copy/paste mistakes
+	require.NoError(t, testexch.Setup(ok), "Test instance Setup must not error")
+	assets, err := ok.getAssetsFromInstrumentID("BTC-USDT")
+	require.NoError(t, err, "getAssetsFromInstrumentID must not error")
+
+	p := currency.NewPairWithDelimiter("BTC", "USDT", currency.DashDelimiter)
+
+	for _, a := range assets {
+		err := ok.Websocket.AddSubscriptions(ok.Websocket.Conn, &subscription.Subscription{
+			Asset:   a,
+			Pairs:   currency.Pairs{p},
+			Channel: subscription.AllTradesChannel,
+			Key:     fmt.Sprintf("%s-%s", p, a)})
+		require.NoError(t, err, "AddSubscriptions must not error")
+	}
+	testexch.FixtureToDataHandler(t, "testdata/wsAllTrades.json", ok.WsHandleData)
+
+	exp := []trade.Data{
+		{
+			Timestamp: time.UnixMilli(1740394561685).UTC(),
+			Price:     95634.9,
+			Amount:    0.00011186,
+			Side:      order.Buy,
+			TID:       "674510826",
+		},
+		{
+			Timestamp: time.UnixMilli(1740394561686).UTC(),
+			Price:     95635.3,
+			Amount:    0.00011194,
+			Side:      order.Sell,
+			TID:       "674510827",
+		},
+	}
+
+	total := len(assets) * len(exp)
+	require.Len(t, ok.Websocket.DataHandler, total, "Must see correct number of trades")
+
+	trades := make(map[asset.Item][]trade.Data)
+
+	for len(ok.Websocket.DataHandler) > 0 {
+		resp := <-ok.Websocket.DataHandler
+		switch v := resp.(type) {
+		case trade.Data:
+			trades[v.AssetType] = append(trades[v.AssetType], v)
+		case error:
+			t.Error(v)
+		default:
+			t.Errorf("Unexpected type in DataHandler: %T (%s)", v, v)
+		}
+	}
+
+	for _, assetType := range assets {
+		require.Len(t, trades[assetType], len(exp), "Must have received %d trades for asset %v", len(exp), assetType)
+		slices.SortFunc(trades[assetType], func(a, b trade.Data) int {
+			return strings.Compare(a.TID, b.TID)
+		})
+		for i, tradeData := range trades[assetType] {
+			expected := exp[i]
+			expected.AssetType = assetType
+			expected.Exchange = ok.Name
+			expected.CurrencyPair = p
+			require.Equal(t, expected, tradeData, "Trade %d (TID: %s) for asset %v must match expected data", i, tradeData.TID, assetType)
+		}
+	}
+}
+
 // ************************** Public Channel Subscriptions *****************************
 
 func TestInstrumentsSubscription(t *testing.T) {

exchanges/okx/okx_websocket.go

@@ -1219,6 +1219,13 @@ func (ok *Okx) wsProcessTrades(data []byte) error {
 	if err != nil {
 		return err
 	}
+
+	saveTradeData := ok.IsSaveTradeDataEnabled()
+	tradeFeed := ok.IsTradeFeedEnabled()
+	if !saveTradeData && !tradeFeed {
+		return nil
+	}
+
 	var assets []asset.Item
 	if response.Argument.InstrumentType != "" {
 		assetType, err := assetTypeFromInstrumentType(response.Argument.InstrumentType)
@@ -1245,13 +1252,21 @@ func (ok *Okx) wsProcessTrades(data []byte) error {
 				CurrencyPair: pair,
 				Exchange:     ok.Name,
 				Side:         response.Data[i].Side,
-				Timestamp:    response.Data[i].Timestamp.Time(),
+				Timestamp:    response.Data[i].Timestamp.Time().UTC(),
 				TID:          response.Data[i].TradeID,
 				Price:        response.Data[i].Price.Float64(),
 			})
 		}
 	}
-	return trade.AddTradesToBuffer(trades...)
+	if tradeFeed {
+		for i := range trades {
+			ok.Websocket.DataHandler <- trades[i]
+		}
+	}
+	if saveTradeData {
+		return trade.AddTradesToBuffer(trades...)
+	}
+	return nil
 }
 
 // wsProcessOrders handles websocket order push data responses.

exchanges/okx/testdata/wsAllTrades.json

@@ -0,0 +1,2 @@
+{"arg":{"channel":"trades","instId":"BTC-USDT"},"data":[{"instId":"BTC-USDT","tradeId":"674510826","px":"95634.9","sz":"0.00011186","side":"buy","ts":"1740394561685","count":"1"}]}
+{"arg":{"channel":"trades","instId":"BTC-USDT"},"data":[{"instId":"BTC-USDT","tradeId":"674510827","px":"95635.3","sz":"0.00011194","side":"sell","ts":"1740394561686","count":"1"}]}