feature: add forceOrder api for binance to show liquid info

pkg/exchange/binance/convert.go

@@ -340,6 +340,8 @@ func convertSubscription(s types.Subscription) string {
 		return fmt.Sprintf("%s@trade", strings.ToLower(s.Symbol))
 	case types.AggTradeChannel:
 		return fmt.Sprintf("%s@aggTrade", strings.ToLower(s.Symbol))
+	case types.ForceOrderChannel:
+		return fmt.Sprintf("%s@forceOrder", strings.ToLower(s.Symbol))
 	}
 
 	return fmt.Sprintf("%s@%s", strings.ToLower(s.Symbol), s.Channel)

pkg/exchange/binance/parse.go

@@ -362,7 +362,10 @@ func parseWebSocketEvent(message []byte) (interface{}, error) {
 		var event AggTradeEvent
 		err = json.Unmarshal([]byte(message), &event)
 		return &event, err
-
+	case "forceOrder":
+		var event ForceOrderEvent
+		err = json.Unmarshal([]byte(message), &event)
+		return &event, err
 	}
 
 	// futures stream
@@ -530,6 +533,63 @@ func parseDepthEvent(val *fastjson.Value) (*DepthEvent, error) {
 	return depth, err
 }
 
+type ForceOrderEventInner struct {
+	Symbol                string           `json:"s"`
+	TradeTime             int64            `json:"T"`
+	Side                  string           `json:"S"`
+	OrderType             string           `json:"o"`
+	TimeInForce           string           `json:"f"`
+	Quantity              fixedpoint.Value `json:"q"`
+	Price                 fixedpoint.Value `json:"p"`
+	AveragePrice          fixedpoint.Value `json:"ap"`
+	OrderStatus           string           `json:"X"`
+	LastFilledQuantity    fixedpoint.Value `json:"l"`
+	LastFilledAccQuantity fixedpoint.Value `json:"z"`
+}
+
+type ForceOrderEvent struct {
+	EventBase
+	Order ForceOrderEventInner `json:"o"`
+}
+
+func (e *ForceOrderEvent) LiquidationInfo() types.LiquidationInfo {
+	o := e.Order
+	tt := time.Unix(0, o.TradeTime*int64(time.Millisecond))
+	return types.LiquidationInfo{
+		Symbol:       o.Symbol,
+		Side:         types.SideType(o.Side),
+		OrderType:    types.OrderType(o.OrderType),
+		TimeInForce:  types.TimeInForce(o.TimeInForce),
+		Quantity:     o.Quantity,
+		Price:        o.Price,
+		AveragePrice: o.AveragePrice,
+		OrderStatus:  types.OrderStatus(o.OrderStatus),
+		TradeTime:    types.Time(tt),
+	}
+}
+
+/*
+ForceOrderEvent
+
+{
+   "E" : 1689303434028,
+   "e" : "forceOrder",
+   "o" : {
+      "S" : "BUY", // Side
+      "T" : 1689303434025, // Order Trade Time
+      "X" : "FILLED", // Order Status
+      "ap" : "2011.09", // Average Price
+      "f" : "IOC", // TimeInForce
+      "l" : "0.003", // Last filled Quantity
+      "o" : "LIMIT", // Order Type
+      "p" : "2021.37", // Price
+      "q" : "0.003", // Original Quantity
+      "s" : "ETHUSDT", // Symbol
+      "z" : "0.003" // Order Filed Accumulated Quantity
+   }
+}
+*/
+
 type MarketTradeEvent struct {
 	EventBase
 	Symbol   string           `json:"s"`

pkg/exchange/binance/stream.go

@@ -48,6 +48,7 @@ type Stream struct {
 
 	marketTradeEventCallbacks []func(e *MarketTradeEvent)
 	aggTradeEventCallbacks    []func(e *AggTradeEvent)
+	forceOrderEventCallbacks  []func(e *ForceOrderEvent)
 
 	balanceUpdateEventCallbacks           []func(event *BalanceUpdateEvent)
 	outboundAccountInfoEventCallbacks     []func(event *OutboundAccountInfoEvent)
@@ -126,6 +127,7 @@ func NewStream(ex *Exchange, client *binance.Client, futuresClient *futures.Clie
 	stream.OnContinuousKLineEvent(stream.handleContinuousKLineEvent)
 	stream.OnMarketTradeEvent(stream.handleMarketTradeEvent)
 	stream.OnAggTradeEvent(stream.handleAggTradeEvent)
+	stream.OnForceOrderEvent(stream.handleForceOrderEvent)
 
 	// Futures User Data Stream
 	// ===================================
@@ -233,6 +235,10 @@ func (s *Stream) handleAggTradeEvent(e *AggTradeEvent) {
 	s.EmitAggTrade(e.Trade())
 }
 
+func (s *Stream) handleForceOrderEvent(e *ForceOrderEvent) {
+	s.EmitForceOrder(e.LiquidationInfo())
+}
+
 func (s *Stream) handleKLineEvent(e *KLineEvent) {
 	kline := e.KLine.KLine()
 	if e.KLine.Closed {

pkg/exchange/binance/stream_callbacks.go

@@ -54,6 +54,16 @@ func (s *Stream) EmitAggTradeEvent(e *AggTradeEvent) {
 	}
 }
 
+func (s *Stream) OnForceOrderEvent(cb func(e *ForceOrderEvent)) {
+	s.forceOrderEventCallbacks = append(s.forceOrderEventCallbacks, cb)
+}
+
+func (s *Stream) EmitForceOrderEvent(e *ForceOrderEvent) {
+	for _, cb := range s.forceOrderEventCallbacks {
+		cb(e)
+	}
+}
+
 func (s *Stream) OnBalanceUpdateEvent(cb func(event *BalanceUpdateEvent)) {
 	s.balanceUpdateEventCallbacks = append(s.balanceUpdateEventCallbacks, cb)
 }
@@ -195,6 +205,8 @@ type StreamEventHub interface {
 
 	OnAggTradeEvent(cb func(e *AggTradeEvent))
 
+	OnForceOrderEvent(cb func(e *ForceOrderEvent))
+
 	OnBalanceUpdateEvent(cb func(event *BalanceUpdateEvent))
 
 	OnOutboundAccountInfoEvent(cb func(event *OutboundAccountInfoEvent))

pkg/types/channel.go

@@ -8,6 +8,7 @@ const (
 	BookTickerChannel  = Channel("bookTicker")
 	MarketTradeChannel = Channel("trade")
 	AggTradeChannel    = Channel("aggTrade")
+	ForceOrderChannel  = Channel("forceOrder")
 
 	// channels for futures
 	MarkPriceChannel = Channel("markPrice")

pkg/types/liquidation_info.go

@@ -0,0 +1,15 @@
+package types
+
+import "github.com/c9s/bbgo/pkg/fixedpoint"
+
+type LiquidationInfo struct {
+	Symbol       string
+	Side         SideType
+	OrderType    OrderType
+	TimeInForce  TimeInForce
+	Quantity     fixedpoint.Value
+	Price        fixedpoint.Value
+	AveragePrice fixedpoint.Value
+	OrderStatus  OrderStatus
+	TradeTime    Time
+}

pkg/types/standardstream_callbacks.go

@@ -164,6 +164,16 @@ func (s *StandardStream) EmitAggTrade(trade Trade) {
 	}
 }
 
+func (s *StandardStream) OnForceOrder(cb func(info LiquidationInfo)) {
+	s.forceOrderCallbacks = append(s.forceOrderCallbacks, cb)
+}
+
+func (s *StandardStream) EmitForceOrder(info LiquidationInfo) {
+	for _, cb := range s.forceOrderCallbacks {
+		cb(info)
+	}
+}
+
 func (s *StandardStream) OnFuturesPositionUpdate(cb func(futuresPositions FuturesPositionMap)) {
 	s.FuturesPositionUpdateCallbacks = append(s.FuturesPositionUpdateCallbacks, cb)
 }
@@ -217,6 +227,8 @@ type StandardStreamEventHub interface {
 
 	OnAggTrade(cb func(trade Trade))
 
+	OnForceOrder(cb func(info LiquidationInfo))
+
 	OnFuturesPositionUpdate(cb func(futuresPositions FuturesPositionMap))
 
 	OnFuturesPositionSnapshot(cb func(futuresPositions FuturesPositionMap))

pkg/types/stream.go

@@ -132,6 +132,8 @@ type StandardStream struct {
 
 	aggTradeCallbacks []func(trade Trade)
 
+	forceOrderCallbacks []func(info LiquidationInfo)
+
 	// Futures
 	FuturesPositionUpdateCallbacks []func(futuresPositions FuturesPositionMap)
 
@@ -159,6 +161,7 @@ type StandardStreamEmitter interface {
 	EmitBookSnapshot(SliceOrderBook)
 	EmitMarketTrade(Trade)
 	EmitAggTrade(Trade)
+	EmitForceOrder(LiquidationInfo)
 	EmitFuturesPositionUpdate(FuturesPositionMap)
 	EmitFuturesPositionSnapshot(FuturesPositionMap)
 }