bbgo_origin/pkg/exchange/binance/stream.go

480 lines
11 KiB
Go
Raw Normal View History

2020-07-11 05:02:53 +00:00
package binance
import (
"context"
2020-10-03 11:14:15 +00:00
"fmt"
2020-12-21 06:43:40 +00:00
"math/rand"
2021-01-23 08:59:51 +00:00
"os"
"strconv"
2020-08-07 01:15:50 +00:00
"strings"
2021-01-25 05:53:11 +00:00
"sync"
2020-07-13 08:03:47 +00:00
"time"
"github.com/adshao/go-binance/v2"
2020-07-11 05:02:53 +00:00
"github.com/gorilla/websocket"
2020-07-13 08:03:47 +00:00
2020-11-10 06:19:33 +00:00
"github.com/c9s/bbgo/pkg/fixedpoint"
2020-10-11 08:46:15 +00:00
"github.com/c9s/bbgo/pkg/types"
2020-07-11 05:02:53 +00:00
)
2021-01-23 08:59:51 +00:00
var debugBinanceDepth bool
2020-12-21 06:43:40 +00:00
func init() {
2020-12-21 09:48:30 +00:00
// randomize pulling
2020-12-21 06:43:40 +00:00
rand.Seed(time.Now().UnixNano())
2021-01-23 08:59:51 +00:00
if s := os.Getenv("BINANCE_DEBUG_DEPTH"); len(s) > 0 {
v, err := strconv.ParseBool(s)
if err != nil {
log.Error(err)
} else {
debugBinanceDepth = v
if debugBinanceDepth {
log.Info("binance depth debugging is enabled")
}
}
}
2020-12-21 06:43:40 +00:00
}
2020-07-11 05:02:53 +00:00
type StreamRequest struct {
// 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
2020-12-21 09:48:30 +00:00
2020-10-03 10:35:28 +00:00
types.StandardStream
2020-08-07 01:15:50 +00:00
Client *binance.Client
ListenKey string
Conn *websocket.Conn
2021-01-25 05:53:11 +00:00
connLock sync.Mutex
2020-07-11 08:07:09 +00:00
2020-12-21 07:26:05 +00:00
publicOnly bool
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
2020-12-21 09:48:30 +00:00
balanceUpdateEventCallbacks []func(event *BalanceUpdateEvent)
outboundAccountInfoEventCallbacks []func(event *OutboundAccountInfoEvent)
outboundAccountPositionEventCallbacks []func(event *OutboundAccountPositionEvent)
executionReportEventCallbacks []func(event *ExecutionReportEvent)
depthFrames map[string]*DepthFrame
2020-07-11 05:02:53 +00:00
}
2020-10-03 12:09:22 +00:00
func NewStream(client *binance.Client) *Stream {
stream := &Stream{
Client: client,
depthFrames: make(map[string]*DepthFrame),
}
2020-10-05 06:16:55 +00:00
stream.OnDepthEvent(func(e *DepthEvent) {
f, ok := stream.depthFrames[e.Symbol]
2020-10-05 06:16:55 +00:00
if !ok {
f = &DepthFrame{
2021-01-24 06:12:44 +00:00
client: client,
context: context.Background(),
2021-01-24 06:12:44 +00:00
Symbol: e.Symbol,
2020-10-05 06:16:55 +00:00
}
stream.depthFrames[e.Symbol] = f
2020-10-05 06:16:55 +00:00
f.OnReady(func(e DepthEvent, bufEvents []DepthEvent) {
snapshot, err := e.OrderBook()
if err != nil {
2021-01-25 06:13:39 +00:00
log.WithError(err).Error("book snapshot convert error")
2020-10-05 06:16:55 +00:00
return
}
2021-01-25 05:53:11 +00:00
if valid, err := snapshot.IsValid(); !valid {
log.Warnf("depth snapshot is invalid, event: %+v, error: %v", e, err)
}
2020-10-05 06:16:55 +00:00
stream.EmitBookSnapshot(snapshot)
for _, e := range bufEvents {
book, err := e.OrderBook()
if err != nil {
log.WithError(err).Error("book convert error")
return
}
stream.EmitBookUpdate(book)
}
})
f.OnPush(func(e DepthEvent) {
book, err := e.OrderBook()
if err != nil {
log.WithError(err).Error("book convert error")
return
}
stream.EmitBookUpdate(book)
})
} else {
f.PushEvent(*e)
}
})
2020-12-21 09:48:30 +00:00
stream.OnOutboundAccountPositionEvent(func(e *OutboundAccountPositionEvent) {
snapshot := types.BalanceMap{}
for _, balance := range e.Balances {
2020-11-10 06:19:33 +00:00
available := fixedpoint.Must(fixedpoint.NewFromString(balance.Free))
locked := fixedpoint.Must(fixedpoint.NewFromString(balance.Locked))
snapshot[balance.Asset] = types.Balance{
Currency: balance.Asset,
Available: available,
Locked: locked,
}
}
stream.EmitBalanceSnapshot(snapshot)
})
stream.OnKLineEvent(func(e *KLineEvent) {
kline := e.KLine.KLine()
if e.KLine.Closed {
stream.EmitKLineClosedEvent(e)
stream.EmitKLineClosed(kline)
} else {
stream.EmitKLine(kline)
}
})
stream.OnExecutionReportEvent(func(e *ExecutionReportEvent) {
switch e.CurrentExecutionType {
2020-10-29 13:10:01 +00:00
case "NEW", "CANCELED", "REJECTED", "EXPIRED", "REPLACED":
order, err := e.Order()
if err != nil {
log.WithError(err).Error("order convert error")
return
}
stream.EmitOrderUpdate(*order)
case "TRADE":
trade, err := e.Trade()
if err != nil {
2020-10-05 06:16:55 +00:00
log.WithError(err).Error("trade convert error")
2020-10-29 13:10:01 +00:00
return
}
2020-10-05 06:16:55 +00:00
stream.EmitTradeUpdate(*trade)
}
})
2020-11-07 04:38:57 +00:00
stream.OnConnect(func() {
// reset the previous frames
for _, f := range stream.depthFrames {
2021-01-25 06:26:22 +00:00
f.reset()
2021-01-25 05:53:11 +00:00
f.loadDepthSnapshot()
}
2020-10-03 12:09:22 +00:00
var params []string
for _, subscription := range stream.Subscriptions {
params = append(params, convertSubscription(subscription))
}
if len(params) == 0 {
return
}
2020-12-21 06:53:34 +00:00
log.Infof("subscribing channels: %+v", params)
2020-10-03 12:09:22 +00:00
err := stream.Conn.WriteJSON(StreamRequest{
Method: "SUBSCRIBE",
Params: params,
ID: 1,
})
2020-10-03 12:09:22 +00:00
if err != nil {
log.WithError(err).Error("subscribe error")
}
})
return stream
}
2020-12-21 07:26:05 +00:00
func (s *Stream) SetPublicOnly() {
s.publicOnly = true
}
2020-10-03 10:35:28 +00:00
func (s *Stream) dial(listenKey string) (*websocket.Conn, error) {
2020-12-21 07:43:54 +00:00
var url string
if s.publicOnly {
url = "wss://stream.binance.com:9443/ws"
} else {
url = "wss://stream.binance.com:9443/ws/" + listenKey
}
2020-07-11 05:02:53 +00:00
conn, _, err := websocket.DefaultDialer.Dial(url, nil)
2020-07-12 14:52:24 +00:00
if err != nil {
return nil, err
}
2021-02-11 00:13:50 +00:00
// use the default ping handler
conn.SetPingHandler(nil)
2020-07-12 14:52:24 +00:00
return conn, nil
}
2020-12-02 14:19:31 +00:00
func (s *Stream) fetchListenKey(ctx context.Context) (string, error) {
2021-01-19 18:09:12 +00:00
if s.IsMargin {
if s.IsIsolatedMargin {
log.Infof("isolated margin %s is enabled, requesting margin user stream listen key...", s.IsolatedMarginSymbol)
req := s.Client.NewStartIsolatedMarginUserStreamService()
2021-01-19 18:09:12 +00:00
req.Symbol(s.IsolatedMarginSymbol)
return req.Do(ctx)
2020-12-21 09:48:30 +00:00
}
log.Infof("margin mode is enabled, requesting margin user stream listen key...")
req := s.Client.NewStartMarginUserStreamService()
2020-12-21 09:48:30 +00:00
return req.Do(ctx)
2020-12-02 14:19:31 +00:00
}
return s.Client.NewStartUserStreamService().Do(ctx)
}
func (s *Stream) keepaliveListenKey(ctx context.Context, listenKey string) error {
2021-01-19 18:09:12 +00:00
if s.IsMargin {
if s.IsIsolatedMargin {
req := s.Client.NewKeepaliveIsolatedMarginUserStreamService().ListenKey(listenKey)
2021-01-19 18:09:12 +00:00
req.Symbol(s.IsolatedMarginSymbol)
return req.Do(ctx)
2020-12-21 09:48:30 +00:00
}
req := s.Client.NewKeepaliveMarginUserStreamService().ListenKey(listenKey)
2020-12-21 09:48:30 +00:00
return req.Do(ctx)
2020-12-02 14:19:31 +00:00
}
return s.Client.NewKeepaliveUserStreamService().ListenKey(listenKey).Do(ctx)
}
2020-10-03 10:35:28 +00:00
func (s *Stream) connect(ctx context.Context) error {
2020-12-21 07:26:05 +00:00
if s.publicOnly {
log.Infof("stream is set to public only mode")
} else {
2020-12-02 14:19:31 +00:00
log.Infof("request listen key for creating user data stream...")
listenKey, err := s.fetchListenKey(ctx)
2020-12-21 07:26:05 +00:00
if err != nil {
return err
}
2020-07-12 14:52:24 +00:00
2020-12-21 07:26:05 +00:00
s.ListenKey = listenKey
2020-12-21 09:48:30 +00:00
log.Infof("user data stream created. listenKey: %s", maskListenKey(s.ListenKey))
2020-12-21 07:26:05 +00:00
}
2020-07-12 14:52:24 +00:00
conn, err := s.dial(s.ListenKey)
2020-07-11 05:02:53 +00:00
if err != nil {
return err
}
2020-12-21 06:53:34 +00:00
log.Infof("websocket connected")
2021-01-25 05:53:11 +00:00
s.connLock.Lock()
2020-07-11 05:02:53 +00:00
s.Conn = conn
2021-01-25 05:53:11 +00:00
s.connLock.Unlock()
2020-07-11 05:02:53 +00:00
2020-11-07 04:38:57 +00:00
s.EmitConnect()
2020-10-03 12:09:22 +00:00
return nil
2020-07-12 14:52:24 +00:00
}
2020-07-11 05:02:53 +00:00
2020-10-03 11:14:15 +00:00
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"
switch s.Channel {
case types.KLineChannel:
return fmt.Sprintf("%s@%s_%s", strings.ToLower(s.Symbol), s.Channel, s.Options.String())
case types.BookChannel:
return fmt.Sprintf("%s@depth", strings.ToLower(s.Symbol))
}
return fmt.Sprintf("%s@%s", strings.ToLower(s.Symbol), s.Channel)
}
2020-10-03 10:35:28 +00:00
func (s *Stream) Connect(ctx context.Context) error {
2020-07-12 14:52:24 +00:00
err := s.connect(ctx)
2020-07-11 05:02:53 +00:00
if err != nil {
return err
}
go s.read(ctx)
2020-07-11 05:02:53 +00:00
return nil
}
2020-10-03 10:35:28 +00:00
func (s *Stream) read(ctx context.Context) {
2020-07-11 05:02:53 +00:00
2021-01-24 11:08:33 +00:00
pingTicker := time.NewTicker(10 * time.Second)
2020-08-07 01:17:35 +00:00
defer pingTicker.Stop()
keepAliveTicker := time.NewTicker(5 * time.Minute)
defer keepAliveTicker.Stop()
2020-07-11 05:02:53 +00:00
go func() {
for {
select {
2020-07-11 05:02:53 +00:00
case <-ctx.Done():
return
2020-07-11 05:02:53 +00:00
case <-pingTicker.C:
2021-01-25 05:53:11 +00:00
s.connLock.Lock()
2021-01-24 11:08:33 +00:00
if err := s.Conn.WriteControl(websocket.PingMessage, []byte("hb"), time.Now().Add(3*time.Second)); err != nil {
log.WithError(err).Error("ping error", err)
}
2021-01-25 05:53:11 +00:00
s.connLock.Unlock()
case <-keepAliveTicker.C:
if !s.publicOnly {
if err := s.keepaliveListenKey(ctx, s.ListenKey); err != nil {
log.WithError(err).Errorf("listen key keep-alive error: %v key: %s", err, maskListenKey(s.ListenKey))
}
2020-12-21 07:26:05 +00:00
}
2020-08-07 01:17:35 +00:00
2020-07-18 02:43:27 +00:00
}
}
}()
for {
select {
case <-ctx.Done():
return
2020-07-18 02:43:27 +00:00
2020-07-11 05:02:53 +00:00
default:
2021-02-10 16:21:56 +00:00
if err := s.Conn.SetReadDeadline(time.Now().Add(3 * time.Minute)); err != nil {
2020-07-11 08:07:09 +00:00
log.WithError(err).Errorf("set read deadline error: %s", err.Error())
2020-07-11 05:02:53 +00:00
}
mt, message, err := s.Conn.ReadMessage()
if err != nil {
2020-07-12 17:06:04 +00:00
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
log.WithError(err).Errorf("read error: %s", err.Error())
2021-01-24 11:08:33 +00:00
} else {
log.Info("websocket connection closed, going away")
2020-07-12 17:06:04 +00:00
}
2020-07-12 14:52:24 +00:00
// reconnect
for err != nil {
2020-07-12 17:06:04 +00:00
select {
case <-ctx.Done():
return
default:
2020-12-21 07:26:05 +00:00
if !s.publicOnly {
if err := s.invalidateListenKey(ctx, s.ListenKey); err != nil {
log.WithError(err).Error("invalidate listen key error")
}
}
2020-07-13 05:25:48 +00:00
2020-07-12 17:06:04 +00:00
err = s.connect(ctx)
time.Sleep(5 * time.Second)
}
2020-07-12 14:52:24 +00:00
}
continue
2020-07-11 05:02:53 +00:00
}
// skip non-text messages
if mt != websocket.TextMessage {
continue
}
2020-12-21 09:48:30 +00:00
log.Debug(string(message))
2020-07-11 05:02:53 +00:00
e, err := ParseEvent(string(message))
if err != nil {
2020-07-11 08:07:09 +00:00
log.WithError(err).Errorf("[binance] event parse error")
2020-07-11 05:02:53 +00:00
continue
}
2020-10-22 02:47:54 +00:00
// log.NotifyTo("[binance] event: %+v", e)
2020-07-11 08:07:09 +00:00
switch e := e.(type) {
2020-12-21 09:48:30 +00:00
case *OutboundAccountPositionEvent:
log.Info(e.Event, " ", e.Balances)
s.EmitOutboundAccountPositionEvent(e)
2020-07-11 08:07:09 +00:00
case *OutboundAccountInfoEvent:
log.Info(e.Event, " ", e.Balances)
s.EmitOutboundAccountInfoEvent(e)
case *BalanceUpdateEvent:
log.Info(e.Event, " ", e.Asset, " ", e.Delta)
s.EmitBalanceUpdateEvent(e)
case *KLineEvent:
s.EmitKLineEvent(e)
2020-10-03 12:09:22 +00:00
case *DepthEvent:
s.EmitDepthEvent(e)
2020-07-11 08:07:09 +00:00
case *ExecutionReportEvent:
2020-11-17 05:25:59 +00:00
log.Info(e.Event, " ", e)
2020-07-11 08:07:09 +00:00
s.EmitExecutionReportEvent(e)
}
2020-07-11 05:02:53 +00:00
}
}
}
2020-12-02 14:19:31 +00:00
func (s *Stream) invalidateListenKey(ctx context.Context, listenKey string) (err error) {
// should use background context to invalidate the user stream
2020-12-21 09:48:30 +00:00
log.Info("closing listen key")
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
}
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
if err != nil {
2020-12-21 09:48:30 +00:00
log.WithError(err).Error("error deleting listen key")
2020-07-11 05:02:53 +00:00
return err
}
2020-07-11 09:02:37 +00:00
return nil
2020-07-11 05:02:53 +00:00
}
2020-07-12 14:52:24 +00:00
2020-10-03 10:35:28 +00:00
func (s *Stream) Close() error {
2020-12-21 06:53:34 +00:00
log.Infof("closing user data stream...")
2020-07-12 14:52:24 +00:00
2020-12-21 07:26:05 +00:00
if !s.publicOnly {
if err := s.invalidateListenKey(context.Background(), s.ListenKey); err != nil {
log.WithError(err).Error("invalidate listen key error")
}
log.Infof("user data stream closed")
}
2021-01-25 05:53:11 +00:00
s.connLock.Lock()
defer s.connLock.Unlock()
2020-12-21 07:26:05 +00:00
return s.Conn.Close()
2020-07-12 14:52:24 +00:00
}
2020-08-07 01:17:35 +00:00
func maskListenKey(listenKey string) string {
maskKey := listenKey[0:5]
return maskKey + strings.Repeat("*", len(listenKey)-1-5)
}
2020-10-05 06:16:55 +00:00
//go:generate callbackgen -type DepthFrame