pkg/exchange: add order event

pkg/exchange/bybit/stream.go

@@ -34,6 +34,7 @@ type Stream struct {
 
 	bookEventCallbacks   []func(e BookEvent)
 	walletEventCallbacks []func(e []*WalletEvent)
+	orderEventCallbacks  []func(e []*OrderEvent)
 }
 
 func NewStream(key, secret string) *Stream {
@@ -52,6 +53,7 @@ func NewStream(key, secret string) *Stream {
 	stream.OnConnect(stream.handlerConnect)
 	stream.OnBookEvent(stream.handleBookEvent)
 	stream.OnWalletEvent(stream.handleWalletEvent)
+	stream.OnOrderEvent(stream.handleOrderEvent)
 	return stream
 }
 
@@ -77,6 +79,9 @@ func (s *Stream) dispatchEvent(event interface{}) {
 
 	case []*WalletEvent:
 		s.EmitWalletEvent(e)
+
+	case []*OrderEvent:
+		s.EmitOrderEvent(e)
 	}
 }
 
@@ -107,6 +112,11 @@ func (s *Stream) parseWebSocketEvent(in []byte) (interface{}, error) {
 		case TopicTypeWallet:
 			var wallets []*WalletEvent
 			return wallets, json.Unmarshal(e.WebSocketTopicEvent.Data, &wallets)
+
+		case TopicTypeOrder:
+			var orders []*OrderEvent
+			return orders, json.Unmarshal(e.WebSocketTopicEvent.Data, &orders)
+
 		}
 	}
 
@@ -194,6 +204,7 @@ func (s *Stream) handlerConnect() {
 			Op: WsOpTypeSubscribe,
 			Args: []string{
 				string(TopicTypeWallet),
+				string(TopicTypeOrder),
 			},
 		}); err != nil {
 			log.WithError(err).Error("failed to send subscription request")
@@ -246,3 +257,18 @@ func (s *Stream) handleWalletEvent(events []*WalletEvent) {
 
 	s.StandardStream.EmitBalanceSnapshot(bm)
 }
+
+func (s *Stream) handleOrderEvent(events []*OrderEvent) {
+	for _, event := range events {
+		if event.Category != bybitapi.CategorySpot {
+			return
+		}
+
+		gOrder, err := toGlobalOrder(event.Order)
+		if err != nil {
+			log.WithError(err).Error("failed to convert to global order")
+			continue
+		}
+		s.StandardStream.EmitOrderUpdate(*gOrder)
+	}
+}

pkg/exchange/bybit/stream_callbacks.go

@@ -23,3 +23,13 @@ func (s *Stream) EmitWalletEvent(e []*WalletEvent) {
 		cb(e)
 	}
 }
+
+func (s *Stream) OnOrderEvent(cb func(e []*OrderEvent)) {
+	s.orderEventCallbacks = append(s.orderEventCallbacks, cb)
+}
+
+func (s *Stream) EmitOrderEvent(e []*OrderEvent) {
+	for _, cb := range s.orderEventCallbacks {
+		cb(e)
+	}
+}

pkg/exchange/bybit/stream_test.go

@@ -65,6 +65,17 @@ func TestStream(t *testing.T) {
 		c := make(chan struct{})
 		<-c
 	})
+
+	t.Run("order test", func(t *testing.T) {
+		err := s.Connect(context.Background())
+		assert.NoError(t, err)
+
+		s.OnOrderUpdate(func(order types.Order) {
+			t.Log("got update", order)
+		})
+		c := make(chan struct{})
+		<-c
+	})
 }
 
 func TestStream_parseWebSocketEvent(t *testing.T) {

pkg/exchange/bybit/types.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/c9s/bbgo/pkg/exchange/bybit/bybitapi"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/c9s/bbgo/pkg/types"
 )
@@ -80,6 +81,7 @@ type TopicType string
 const (
 	TopicTypeOrderBook TopicType = "orderbook"
 	TopicTypeWallet    TopicType = "wallet"
+	TopicTypeOrder     TopicType = "order"
 )
 
 type DataType string
@@ -201,3 +203,9 @@ type WalletEvent struct {
 		MarginCollateral bool `json:"marginCollateral"`
 	} `json:"coin"`
 }
+
+type OrderEvent struct {
+	bybitapi.Order
+
+	Category bybitapi.Category `json:"category"`
+}