Merge pull request #994 from zenixls2/feature/binance_aggTrade

feature: add aggTrade for binance

Author: zenixls2

pkg/exchange/binance/convert.go

@@ -295,6 +295,8 @@ func convertSubscription(s types.Subscription) string {
 	// binance uses lower case symbol name,
 	// for kline, it's "<symbol>@kline_<interval>"
 	// for depth, it's "<symbol>@depth OR <symbol>@depth@100ms"
+	// for trade, it's "<symbol>@trade"
+	// for aggregated trade, it's "<symbol>@aggTrade"
 	switch s.Channel {
 	case types.KLineChannel:
 		return fmt.Sprintf("%s@%s_%s", strings.ToLower(s.Symbol), s.Channel, s.Options.String())
@@ -326,6 +328,10 @@ func convertSubscription(s types.Subscription) string {
 		return n
 	case types.BookTickerChannel:
 		return fmt.Sprintf("%s@bookTicker", strings.ToLower(s.Symbol))
+	case types.MarketTradeChannel:
+		return fmt.Sprintf("%s@trade", strings.ToLower(s.Symbol))
+	case types.AggTradeChannel:
+		return fmt.Sprintf("%s@aggTrade", strings.ToLower(s.Symbol))
 	}
 
 	return fmt.Sprintf("%s@%s", strings.ToLower(s.Symbol), s.Channel)

pkg/exchange/binance/parse.go

@@ -350,6 +350,11 @@ func parseWebSocketEvent(message []byte) (interface{}, error) {
 		err = json.Unmarshal([]byte(message), &event)
 		return &event, err
 
+	case "aggTrade":
+		var event AggTradeEvent
+		err = json.Unmarshal([]byte(message), &event)
+		return &event, err
+
 	default:
 		id := val.GetInt("id")
 		if id > 0 {
@@ -542,6 +547,63 @@ func (e *MarketTradeEvent) Trade() types.Trade {
 	}
 }
 
+type AggTradeEvent struct {
+	EventBase
+	Symbol         string           `json:"s"`
+	Quantity       fixedpoint.Value `json:"q"`
+	Price          fixedpoint.Value `json:"p"`
+	FirstTradeId   int64            `json:"f"`
+	LastTradeId    int64            `json:"l"`
+	OrderTradeTime int64            `json:"T"`
+	IsMaker        bool             `json:"m"`
+	Dummy          bool             `json:"M"`
+}
+
+/*
+aggregate trade
+{
+  "e": "aggTrade",  // Event type
+  "E": 123456789,   // Event time
+  "s": "BNBBTC",    // Symbol
+  "a": 12345,       // Aggregate trade ID
+  "p": "0.001",     // Price
+  "q": "100",       // Quantity
+  "f": 100,         // First trade ID
+  "l": 105,         // Last trade ID
+  "T": 123456785,   // Trade time
+  "m": true,        // Is the buyer the market maker?
+  "M": true         // Ignore
+}
+*/
+
+func (e *AggTradeEvent) Trade() types.Trade {
+	tt := time.Unix(0, e.OrderTradeTime*int64(time.Millisecond))
+	var side types.SideType
+	var isBuyer bool
+	if e.IsMaker {
+		side = types.SideTypeSell
+		isBuyer = false
+	} else {
+		side = types.SideTypeBuy
+		isBuyer = true
+	}
+	return types.Trade{
+		ID:            uint64(e.LastTradeId),
+		Exchange:      types.ExchangeBinance,
+		Symbol:        e.Symbol,
+		OrderID:       0,
+		Side:          side,
+		Price:         e.Price,
+		Quantity:      e.Quantity,
+		QuoteQuantity: e.Quantity,
+		IsBuyer:       isBuyer,
+		IsMaker:       e.IsMaker,
+		Time:          types.Time(tt),
+		Fee:           fixedpoint.Zero,
+		FeeCurrency:   "",
+	}
+}
+
 type KLine struct {
 	StartTime int64 `json:"t"`
 	EndTime   int64 `json:"T"`

pkg/exchange/binance/stream.go

@@ -48,6 +48,7 @@ type Stream struct {
 
 	markPriceUpdateEventCallbacks []func(e *MarkPriceUpdateEvent)
 	marketTradeEventCallbacks     []func(e *MarketTradeEvent)
+	aggTradeEventCallbacks        []func(e *AggTradeEvent)
 
 	continuousKLineEventCallbacks       []func(e *ContinuousKLineEvent)
 	continuousKLineClosedEventCallbacks []func(e *ContinuousKLineEvent)
@@ -120,6 +121,7 @@ func NewStream(ex *Exchange, client *binance.Client, futuresClient *futures.Clie
 	stream.OnExecutionReportEvent(stream.handleExecutionReportEvent)
 	stream.OnContinuousKLineEvent(stream.handleContinuousKLineEvent)
 	stream.OnMarketTradeEvent(stream.handleMarketTradeEvent)
+	stream.OnAggTradeEvent(stream.handleAggTradeEvent)
 
 	// Event type ACCOUNT_UPDATE from user data stream updates Balance and FuturesPosition.
 	stream.OnAccountUpdateEvent(stream.handleAccountUpdateEvent)
@@ -218,6 +220,10 @@ func (s *Stream) handleMarketTradeEvent(e *MarketTradeEvent) {
 	s.EmitMarketTrade(e.Trade())
 }
 
+func (s *Stream) handleAggTradeEvent(e *AggTradeEvent) {
+	s.EmitAggTrade(e.Trade())
+}
+
 func (s *Stream) handleKLineEvent(e *KLineEvent) {
 	kline := e.KLine.KLine()
 	if e.KLine.Closed {
@@ -342,6 +348,9 @@ func (s *Stream) dispatchEvent(e interface{}) {
 	case *MarketTradeEvent:
 		s.EmitMarketTradeEvent(e)
 
+	case *AggTradeEvent:
+		s.EmitAggTradeEvent(e)
+
 	case *KLineEvent:
 		s.EmitKLineEvent(e)
 

pkg/exchange/binance/stream_callbacks.go

@@ -6,3 +6,4 @@ var BookChannel = Channel("book")
 var KLineChannel = Channel("kline")
 var BookTickerChannel = Channel("bookticker")
 var MarketTradeChannel = Channel("trade")
+var AggTradeChannel = Channel("aggTrade")

pkg/types/channel.go

@@ -100,6 +100,8 @@ type StandardStream struct {
 
 	marketTradeCallbacks []func(trade Trade)
 
+	aggTradeCallbacks []func(trade Trade)
+
 	// Futures
 	FuturesPositionUpdateCallbacks []func(futuresPositions FuturesPositionMap)
 
@@ -121,6 +123,7 @@ type StandardStreamEmitter interface {
 	EmitBookTickerUpdate(BookTicker)
 	EmitBookSnapshot(SliceOrderBook)
 	EmitMarketTrade(Trade)
+	EmitAggTrade(Trade)
 	EmitFuturesPositionUpdate(FuturesPositionMap)
 	EmitFuturesPositionSnapshot(FuturesPositionMap)
 }