bbgo_origin/pkg/exchange/binance/stream.go

473 lines
13 KiB
Go
Raw Normal View History

2020-07-11 05:02:53 +00:00
package binance
import (
"context"
2021-05-25 11:13:10 +00:00
"net"
2020-07-13 08:03:47 +00:00
"time"
"github.com/c9s/bbgo/pkg/depth"
"github.com/c9s/bbgo/pkg/util"
"github.com/adshao/go-binance/v2"
"github.com/adshao/go-binance/v2/futures"
2020-10-11 08:46:15 +00:00
"github.com/c9s/bbgo/pkg/types"
2020-07-11 05:02:53 +00:00
)
// from Binance document:
// The websocket server will send a ping frame every 3 minutes.
// If the websocket server does not receive a pong frame back from the connection within a 10 minute period, the connection will be disconnected.
// Unsolicited pong frames are allowed.
// WebSocket connections have a limit of 5 incoming messages per second. A message is considered:
// A PING frame
// A PONG frame
// A JSON controlled message (e.g. subscribe, unsubscribe)
const listenKeyKeepAliveInterval = 15 * time.Minute
2021-05-29 12:40:47 +00:00
type WebSocketCommand struct {
2020-07-11 05:02:53 +00:00
// request ID is required
ID int `json:"id"`
Method string `json:"method"`
Params []string `json:"params"`
}
2020-10-03 10:35:28 +00:00
//go:generate callbackgen -type Stream -interface
type Stream struct {
2021-01-19 18:09:12 +00:00
types.MarginSettings
2021-11-15 17:24:36 +00:00
types.FuturesSettings
2020-10-03 10:35:28 +00:00
types.StandardStream
client *binance.Client
futuresClient *futures.Client
2020-07-11 08:07:09 +00:00
// custom callbacks
2020-10-03 12:09:22 +00:00
depthEventCallbacks []func(e *DepthEvent)
kLineEventCallbacks []func(e *KLineEvent)
kLineClosedEventCallbacks []func(e *KLineEvent)
2020-07-11 08:07:09 +00:00
markPriceUpdateEventCallbacks []func(e *MarkPriceUpdateEvent)
marketTradeEventCallbacks []func(e *MarketTradeEvent)
continuousKLineEventCallbacks []func(e *ContinuousKLineEvent)
continuousKLineClosedEventCallbacks []func(e *ContinuousKLineEvent)
2021-11-15 17:24:36 +00:00
2020-12-21 09:48:30 +00:00
balanceUpdateEventCallbacks []func(event *BalanceUpdateEvent)
outboundAccountInfoEventCallbacks []func(event *OutboundAccountInfoEvent)
outboundAccountPositionEventCallbacks []func(event *OutboundAccountPositionEvent)
executionReportEventCallbacks []func(event *ExecutionReportEvent)
bookTickerEventCallbacks []func(event *BookTickerEvent)
2021-12-30 14:32:03 +00:00
orderTradeUpdateEventCallbacks []func(e *OrderTradeUpdateEvent)
accountUpdateEventCallbacks []func(e *AccountUpdateEvent)
accountConfigUpdateEventCallbacks []func(e *AccountConfigUpdateEvent)
depthBuffers map[string]*depth.Buffer
2020-07-11 05:02:53 +00:00
}
func NewStream(ex *Exchange, client *binance.Client, futuresClient *futures.Client) *Stream {
stream := &Stream{
StandardStream: types.NewStandardStream(),
2022-01-02 03:25:34 +00:00
client: client,
futuresClient: futuresClient,
depthBuffers: make(map[string]*depth.Buffer),
}
stream.SetParser(parseWebSocketEvent)
stream.SetDispatcher(stream.dispatchEvent)
stream.SetEndpointCreator(stream.createEndpoint)
2020-10-05 06:16:55 +00:00
stream.OnDepthEvent(func(e *DepthEvent) {
f, ok := stream.depthBuffers[e.Symbol]
if ok {
2022-01-12 13:55:26 +00:00
err := f.AddUpdate(types.SliceOrderBook{
Symbol: e.Symbol,
Bids: e.Bids,
Asks: e.Asks,
}, e.FirstUpdateID, e.FinalUpdateID)
2022-01-12 13:55:26 +00:00
if err != nil {
log.WithError(err).Errorf("found missing %s update event", e.Symbol)
}
} else {
f = depth.NewBuffer(func() (types.SliceOrderBook, int64, error) {
2022-01-12 13:55:26 +00:00
log.Infof("fetching %s depth...", e.Symbol)
return ex.QueryDepth(context.Background(), e.Symbol)
2020-10-05 06:16:55 +00:00
})
f.SetBufferingPeriod(time.Second)
f.OnReady(func(snapshot types.SliceOrderBook, updates []depth.Update) {
if valid, err := snapshot.IsValid(); !valid {
2022-01-12 13:55:26 +00:00
log.Errorf("%s depth snapshot is invalid, error: %v", e.Symbol, err)
2020-10-05 06:16:55 +00:00
return
}
stream.EmitBookSnapshot(snapshot)
for _, u := range updates {
stream.EmitBookUpdate(u.Object)
}
})
f.OnPush(func(update depth.Update) {
stream.EmitBookUpdate(update.Object)
2020-10-05 06:16:55 +00:00
})
2022-01-02 03:25:34 +00:00
stream.depthBuffers[e.Symbol] = f
2020-10-05 06:16:55 +00:00
}
})
stream.OnOutboundAccountPositionEvent(stream.handleOutboundAccountPositionEvent)
stream.OnKLineEvent(stream.handleKLineEvent)
stream.OnBookTickerEvent(stream.handleBookTickerEvent)
stream.OnExecutionReportEvent(stream.handleExecutionReportEvent)
stream.OnContinuousKLineEvent(stream.handleContinuousKLineEvent)
stream.OnMarketTradeEvent(stream.handleMarketTradeEvent)
2021-12-30 14:32:03 +00:00
// Event type ACCOUNT_UPDATE from user data stream updates Balance and FuturesPosition.
stream.OnAccountUpdateEvent(stream.handleAccountUpdateEvent)
stream.OnAccountConfigUpdateEvent(stream.handleAccountConfigUpdateEvent)
stream.OnOrderTradeUpdateEvent(stream.handleOrderTradeUpdateEvent)
2022-01-02 03:25:34 +00:00
stream.OnDisconnect(stream.handleDisconnect)
stream.OnConnect(stream.handleConnect)
return stream
}
2022-01-02 03:25:34 +00:00
func (s *Stream) handleDisconnect() {
log.Debugf("resetting depth snapshots...")
for _, f := range s.depthBuffers {
f.Reset()
}
}
2022-01-02 03:25:34 +00:00
func (s *Stream) handleConnect() {
if !s.PublicOnly {
return
}
2022-01-02 03:25:34 +00:00
var params []string
for _, subscription := range s.Subscriptions {
params = append(params, convertSubscription(subscription))
}
if len(params) == 0 {
return
}
log.Infof("subscribing channels: %+v", params)
err := s.Conn.WriteJSON(WebSocketCommand{
Method: "SUBSCRIBE",
Params: params,
ID: 1,
2020-10-03 12:09:22 +00:00
})
2022-01-02 03:25:34 +00:00
if err != nil {
log.WithError(err).Error("subscribe error")
}
2020-10-03 12:09:22 +00:00
}
func (s *Stream) handleContinuousKLineEvent(e *ContinuousKLineEvent) {
kline := e.KLine.KLine()
if e.KLine.Closed {
s.EmitContinuousKLineClosedEvent(e)
s.EmitKLineClosed(kline)
} else {
s.EmitKLine(kline)
}
}
func (s *Stream) handleExecutionReportEvent(e *ExecutionReportEvent) {
switch e.CurrentExecutionType {
case "NEW", "CANCELED", "REJECTED", "EXPIRED", "REPLACED":
order, err := e.Order()
if err != nil {
log.WithError(err).Error("order convert error")
return
}
s.EmitOrderUpdate(*order)
case "TRADE":
trade, err := e.Trade()
if err != nil {
log.WithError(err).Error("trade convert error")
return
}
s.EmitTradeUpdate(*trade)
order, err := e.Order()
if err != nil {
log.WithError(err).Error("order convert error")
return
}
// Update Order with FILLED event
if order.Status == types.OrderStatusFilled {
s.EmitOrderUpdate(*order)
}
}
}
func (s *Stream) handleBookTickerEvent(e *BookTickerEvent) {
s.EmitBookTickerUpdate(e.BookTicker())
}
func (s *Stream) handleMarketTradeEvent(e *MarketTradeEvent) {
s.EmitMarketTrade(e.Trade())
}
func (s *Stream) handleKLineEvent(e *KLineEvent) {
kline := e.KLine.KLine()
if e.KLine.Closed {
s.EmitKLineClosedEvent(e)
s.EmitKLineClosed(kline)
} else {
s.EmitKLine(kline)
}
}
func (s *Stream) handleOutboundAccountPositionEvent(e *OutboundAccountPositionEvent) {
snapshot := types.BalanceMap{}
for _, balance := range e.Balances {
snapshot[balance.Asset] = types.Balance{
Currency: balance.Asset,
Available: balance.Free,
Locked: balance.Locked,
}
}
s.EmitBalanceSnapshot(snapshot)
}
func (s *Stream) handleAccountUpdateEvent(e *AccountUpdateEvent) {
futuresPositionSnapshot := toGlobalFuturesPositions(e.AccountUpdate.Positions)
s.EmitFuturesPositionSnapshot(futuresPositionSnapshot)
balanceSnapshot := toGlobalFuturesBalance(e.AccountUpdate.Balances)
s.EmitBalanceSnapshot(balanceSnapshot)
}
// TODO: emit account config leverage updates
func (s *Stream) handleAccountConfigUpdateEvent(e *AccountConfigUpdateEvent) {
}
func (s *Stream) handleOrderTradeUpdateEvent(e *OrderTradeUpdateEvent) {
switch e.OrderTrade.CurrentExecutionType {
case "NEW", "CANCELED", "EXPIRED":
order, err := e.OrderFutures()
if err != nil {
2022-01-13 11:28:12 +00:00
log.WithError(err).Error("futures order convert error")
return
}
s.EmitOrderUpdate(*order)
case "TRADE":
2022-01-13 11:28:12 +00:00
trade, err := e.TradeFutures()
if err != nil {
log.WithError(err).Error("futures trade convert error")
return
}
2022-01-13 11:28:12 +00:00
s.EmitTradeUpdate(*trade)
case "CALCULATED - Liquidation Execution":
log.Infof("CALCULATED - Liquidation Execution not support yet.")
}
}
2021-12-30 08:22:29 +00:00
func (s *Stream) getEndpointUrl(listenKey string) string {
2020-12-21 07:43:54 +00:00
var url string
2021-12-30 15:46:43 +00:00
if s.IsFutures {
url = FuturesWebSocketURL + "/ws"
2021-12-30 15:46:43 +00:00
} else if isBinanceUs() {
url = BinanceUSWebSocketURL + "/ws"
2020-12-21 07:43:54 +00:00
} else {
url = WebSocketURL + "/ws"
2021-12-30 15:46:43 +00:00
}
if !s.PublicOnly {
url += "/" + listenKey
2020-12-21 07:43:54 +00:00
}
2021-12-30 08:22:29 +00:00
return url
}
func (s *Stream) createEndpoint(ctx context.Context) (string, error) {
var err error
var listenKey string
if s.PublicOnly {
log.Debugf("stream is set to public only mode")
} else {
listenKey, err = s.fetchListenKey(ctx)
if err != nil {
return "", err
2021-12-29 09:27:37 +00:00
}
2021-02-11 00:13:50 +00:00
log.Debugf("listen key is created: %s", util.MaskKey(listenKey))
go s.listenKeyKeepAlive(ctx, listenKey)
}
url := s.getEndpointUrl(listenKey)
return url, nil
2020-07-12 14:52:24 +00:00
}
2021-12-30 08:30:02 +00:00
func (s *Stream) dispatchEvent(e interface{}) {
switch e := e.(type) {
2020-12-21 09:48:30 +00:00
2021-12-30 08:30:02 +00:00
case *OutboundAccountPositionEvent:
s.EmitOutboundAccountPositionEvent(e)
2020-07-11 08:07:09 +00:00
2021-12-30 08:30:02 +00:00
case *OutboundAccountInfoEvent:
s.EmitOutboundAccountInfoEvent(e)
2020-07-11 08:07:09 +00:00
2021-12-30 08:30:02 +00:00
case *BalanceUpdateEvent:
s.EmitBalanceUpdateEvent(e)
2020-07-11 08:07:09 +00:00
case *MarketTradeEvent:
s.EmitMarketTradeEvent(e)
2021-12-30 08:30:02 +00:00
case *KLineEvent:
s.EmitKLineEvent(e)
2021-12-30 08:30:02 +00:00
case *BookTickerEvent:
s.EmitBookTickerEvent(e)
2020-10-03 12:09:22 +00:00
2021-12-30 08:30:02 +00:00
case *DepthEvent:
s.EmitDepthEvent(e)
2021-12-30 08:30:02 +00:00
case *ExecutionReportEvent:
s.EmitExecutionReportEvent(e)
2021-12-30 08:30:02 +00:00
case *MarkPriceUpdateEvent:
s.EmitMarkPriceUpdateEvent(e)
2021-12-30 08:30:02 +00:00
case *ContinuousKLineEvent:
s.EmitContinuousKLineEvent(e)
case *OrderTradeUpdateEvent:
s.EmitOrderTradeUpdateEvent(e)
2021-12-30 14:32:03 +00:00
case *AccountUpdateEvent:
s.EmitAccountUpdateEvent(e)
case *AccountConfigUpdateEvent:
s.EmitAccountConfigUpdateEvent(e)
2020-07-11 05:02:53 +00:00
}
}
func (s *Stream) fetchListenKey(ctx context.Context) (string, error) {
if s.IsMargin {
if s.IsIsolatedMargin {
log.Debugf("isolated margin %s is enabled, requesting margin user stream listen key...", s.IsolatedMarginSymbol)
req := s.client.NewStartIsolatedMarginUserStreamService()
req.Symbol(s.IsolatedMarginSymbol)
return req.Do(ctx)
}
log.Debugf("margin mode is enabled, requesting margin user stream listen key...")
req := s.client.NewStartMarginUserStreamService()
return req.Do(ctx)
} else if s.IsFutures {
log.Debugf("futures mode is enabled, requesting futures user stream listen key...")
req := s.futuresClient.NewStartUserStreamService()
return req.Do(ctx)
}
log.Debugf("spot mode is enabled, requesting user stream listen key...")
return s.client.NewStartUserStreamService().Do(ctx)
}
func (s *Stream) keepaliveListenKey(ctx context.Context, listenKey string) error {
log.Debugf("keepalive listen key: %s", util.MaskKey(listenKey))
if s.IsMargin {
if s.IsIsolatedMargin {
req := s.client.NewKeepaliveIsolatedMarginUserStreamService().ListenKey(listenKey)
req.Symbol(s.IsolatedMarginSymbol)
return req.Do(ctx)
}
req := s.client.NewKeepaliveMarginUserStreamService().ListenKey(listenKey)
return req.Do(ctx)
} else if s.IsFutures {
req := s.futuresClient.NewKeepaliveUserStreamService().ListenKey(listenKey)
return req.Do(ctx)
}
return s.client.NewKeepaliveUserStreamService().ListenKey(listenKey).Do(ctx)
}
func (s *Stream) closeListenKey(ctx context.Context, listenKey string) (err error) {
2020-12-02 14:19:31 +00:00
// should use background context to invalidate the user stream
log.Debugf("closing listen key: %s", util.MaskKey(listenKey))
2020-12-02 14:19:31 +00:00
2021-01-19 18:09:12 +00:00
if s.IsMargin {
if s.IsIsolatedMargin {
req := s.client.NewCloseIsolatedMarginUserStreamService().ListenKey(listenKey)
2021-01-19 18:09:12 +00:00
req.Symbol(s.IsolatedMarginSymbol)
err = req.Do(ctx)
} else {
req := s.client.NewCloseMarginUserStreamService().ListenKey(listenKey)
err = req.Do(ctx)
2020-12-21 09:48:30 +00:00
}
} else if s.IsFutures {
req := s.futuresClient.NewCloseUserStreamService().ListenKey(listenKey)
err = req.Do(ctx)
2020-12-02 14:19:31 +00:00
} else {
err = s.client.NewCloseUserStreamService().ListenKey(listenKey).Do(ctx)
2020-07-11 05:02:53 +00:00
}
return err
2020-07-11 05:02:53 +00:00
}
2021-12-30 08:30:02 +00:00
// listenKeyKeepAlive
// From Binance
// Keepalive a user data stream to prevent a time out. User data streams will close after 60 minutes.
// It's recommended to send a ping about every 30 minutes.
func (s *Stream) listenKeyKeepAlive(ctx context.Context, listenKey string) {
keepAliveTicker := time.NewTicker(listenKeyKeepAliveInterval)
defer keepAliveTicker.Stop()
// if we exit, we should invalidate the existing listen key
defer func() {
log.Debugf("keepalive worker stopped")
if err := s.closeListenKey(context.Background(), listenKey); err != nil {
log.WithError(err).Errorf("close listen key error: %v key: %s", err, util.MaskKey(listenKey))
2021-12-30 08:30:02 +00:00
}
}()
log.Debugf("starting listen key keep alive worker with interval %s, listen key = %s", listenKeyKeepAliveInterval, util.MaskKey(listenKey))
2021-12-30 08:30:02 +00:00
for {
select {
case <-s.CloseC:
return
case <-ctx.Done():
return
case <-keepAliveTicker.C:
for i := 0; i < 5; i++ {
err := s.keepaliveListenKey(ctx, listenKey)
2022-01-11 05:38:03 +00:00
if err != nil {
time.Sleep(5 * time.Second)
2021-12-30 08:30:02 +00:00
switch err.(type) {
case net.Error:
log.WithError(err).Errorf("listen key keep-alive network error: %v key: %s", err, util.MaskKey(listenKey))
continue
default:
log.WithError(err).Errorf("listen key keep-alive unexpected error: %v key: %s", err, util.MaskKey(listenKey))
s.Reconnect()
return
}
2022-01-11 05:38:03 +00:00
} else {
break
2021-12-30 08:30:02 +00:00
}
}
}
}
}