lychiyu/bbgo_origin
1
0
Fork 0
mirror of https://github.com/c9s/bbgo.git synced 2024-11-26 08:45:16 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

okex: refactor okex stream

Browse Source
This commit is contained in:
c9s 2022-01-02 02:37:33 +08:00
parent 9d382a6b8c
commit 96fedfd311
3 changed files with 166 additions and 322 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
pkg/exchange/okex/parse.go
View File

@ -8,14 +8,15 @@ import (
"strings"
"time"
"github.com/valyala/fastjson"
"github.com/c9s/bbgo/pkg/exchange/okex/okexapi"
"github.com/c9s/bbgo/pkg/fixedpoint"
"github.com/c9s/bbgo/pkg/types"
"github.com/valyala/fastjson"
)
func Parse(str string) (interface{}, error) {
v, err := fastjson.Parse(str)
func parseWebSocketEvent(str []byte) (interface{}, error) {
v, err := fastjson.ParseBytes(str)
if err != nil {
return nil, err
}
@ -52,7 +53,7 @@ func parseEvent(v *fastjson.Value) (*WebSocketEvent, error) {
}, nil
}
type BookData struct {
type BookEvent struct {
InstrumentID string
Symbol string
Action string
@ -63,7 +64,7 @@ type BookData struct {
channel string
}
func (data *BookData) BookTicker() types.BookTicker {
func (data *BookEvent) BookTicker() types.BookTicker {
var askBookData BookEntry = data.Asks[0]
var bidBookData BookEntry = data.Bids[0]
@ -77,7 +78,7 @@ func (data *BookData) BookTicker() types.BookTicker {
}
}
func (data *BookData) Book() types.SliceOrderBook {
func (data *BookEvent) Book() types.SliceOrderBook {
book := types.SliceOrderBook{
Symbol: data.Symbol,
}
@ -130,7 +131,7 @@ func parseBookEntry(v *fastjson.Value) (*BookEntry, error) {
}, nil
}
func parseBookData(v *fastjson.Value) (*BookData, error) {
func parseBookData(v *fastjson.Value) (*BookEvent, error) {
instrumentId := string(v.GetStringBytes("arg", "instId"))
data := v.GetArray("data")
if len(data) == 0 {
@ -166,7 +167,7 @@ func parseBookData(v *fastjson.Value) (*BookData, error) {
bids = append(bids, *entry)
}
return &BookData{
return &BookEvent{
InstrumentID: instrumentId,
Symbol: toGlobalSymbol(instrumentId),
Action: action,

361
pkg/exchange/okex/stream.go
View File

@ -2,18 +2,13 @@ package okex
import (
"context"
"net"
"strconv"
"sync"
"time"
"github.com/c9s/bbgo/pkg/exchange/okex/okexapi"
"github.com/c9s/bbgo/pkg/types"
"github.com/gorilla/websocket"
)
const readTimeout = 15 * time.Second
type WebsocketOp struct {
Op string `json:"op"`
Args interface{} `json:"args"`
@ -30,18 +25,14 @@ type WebsocketLogin struct {
type Stream struct {
types.StandardStream
Client *okexapi.RestClient
Conn *websocket.Conn
connLock sync.Mutex
connCtx context.Context
connCancel context.CancelFunc
client *okexapi.RestClient
// public callbacks
candleDataCallbacks []func(candle Candle)
bookDataCallbacks []func(book BookData)
candleEventCallbacks []func(candle Candle)
bookEventCallbacks []func(book BookEvent)
eventCallbacks []func(event WebSocketEvent)
accountCallbacks []func(account okexapi.Account)
orderDetailsCallbacks []func(orderDetails []okexapi.OrderDetails)
accountEventCallbacks []func(account okexapi.Account)
orderDetailsEventCallbacks []func(orderDetails []okexapi.OrderDetails)
lastCandle map[CandleKey]Candle
}
@ -53,92 +44,28 @@ type CandleKey struct {
func NewStream(client *okexapi.RestClient) *Stream {
stream := &Stream{
Client: client,
StandardStream: types.StandardStream{
ReconnectC: make(chan struct{}, 1),
},
client: client,
StandardStream: types.NewStandardStream(),
lastCandle: make(map[CandleKey]Candle),
}
stream.OnCandleData(func(candle Candle) {
key := CandleKey{Channel: candle.Channel, InstrumentID: candle.InstrumentID}
kline := candle.KLine()
stream.SetParser(parseWebSocketEvent)
stream.SetDispatcher(stream.dispatchEvent)
stream.SetEndpointCreator(stream.createEndpoint)
// check if we need to close previous kline
lastCandle, ok := stream.lastCandle[key]
if ok && candle.StartTime.After(lastCandle.StartTime) {
lastKline := lastCandle.KLine()
lastKline.Closed = true
stream.EmitKLineClosed(lastKline)
stream.OnCandleEvent(stream.handleCandleEvent)
stream.OnBookEvent(stream.handleBookEvent)
stream.OnAccountEvent(stream.handleAccountEvent)
stream.OnOrderDetailsEvent(stream.handleOrderDetailsEvent)
stream.OnEvent(stream.handleEvent)
stream.OnConnect(stream.handleConnect)
return stream
}
stream.EmitKLine(kline)
stream.lastCandle[key] = candle
})
stream.OnBookData(func(data BookData) {
book := data.Book()
switch data.Action {
case "snapshot":
stream.EmitBookSnapshot(book)
case "update":
stream.EmitBookUpdate(book)
}
})
stream.OnAccount(func(account okexapi.Account) {
balances := toGlobalBalance(&account)
stream.EmitBalanceSnapshot(balances)
})
stream.OnOrderDetails(func(orderDetails []okexapi.OrderDetails) {
detailTrades, detailOrders := segmentOrderDetails(orderDetails)
trades, err := toGlobalTrades(detailTrades)
if err != nil {
log.WithError(err).Errorf("error converting order details into trades")
} else {
for _, trade := range trades {
stream.EmitTradeUpdate(trade)
}
}
orders, err := toGlobalOrders(detailOrders)
if err != nil {
log.WithError(err).Errorf("error converting order details into orders")
} else {
for _, order := range orders {
stream.EmitOrderUpdate(order)
}
}
})
stream.OnEvent(func(event WebSocketEvent) {
switch event.Event {
case "login":
if event.Code == "0" {
var subs = []WebsocketSubscription{
{Channel: "account"},
{Channel: "orders", InstrumentType: string(okexapi.InstrumentTypeSpot)},
}
log.Infof("subscribing private channels: %+v", subs)
err := stream.Conn.WriteJSON(WebsocketOp{
Op: "subscribe",
Args: subs,
})
if err != nil {
log.WithError(err).Error("private channel subscribe error")
}
}
}
})
stream.OnConnect(func() {
if stream.PublicOnly {
func (s *Stream) handleConnect() {
if s.PublicOnly {
var subs []WebsocketSubscription
for _, subscription := range stream.Subscriptions {
for _, subscription := range s.Subscriptions {
sub, err := convertSubscription(subscription)
if err != nil {
log.WithError(err).Errorf("subscription convert error")
@ -152,7 +79,7 @@ func NewStream(client *okexapi.RestClient) *Stream {
}
log.Infof("subscribing channels: %+v", subs)
err := stream.Conn.WriteJSON(WebsocketOp{
err := s.Conn.WriteJSON(WebsocketOp{
Op: "subscribe",
Args: subs,
})
@ -166,13 +93,13 @@ func NewStream(client *okexapi.RestClient) *Stream {
// sign=CryptoJS.enc.Base64.Stringify(CryptoJS.HmacSHA256(timestamp +'GET'+'/users/self/verify', secretKey))
msTimestamp := strconv.FormatFloat(float64(time.Now().UnixNano())/float64(time.Second), 'f', -1, 64)
payload := msTimestamp + "GET" + "/users/self/verify"
sign := okexapi.Sign(payload, stream.Client.Secret)
sign := okexapi.Sign(payload, s.client.Secret)
op := WebsocketOp{
Op: "login",
Args: []WebsocketLogin{
{
Key: stream.Client.Key,
Passphrase: stream.Client.Passphrase,
Key: s.client.Key,
Passphrase: s.client.Passphrase,
Timestamp: msTimestamp,
Sign: sign,
},
@ -180,201 +107,117 @@ func NewStream(client *okexapi.RestClient) *Stream {
}
log.Infof("sending okex login request")
err := stream.Conn.WriteJSON(op)
err := s.Conn.WriteJSON(op)
if err != nil {
log.WithError(err).Errorf("can not send login message")
}
}
}
func (s *Stream) handleEvent(event WebSocketEvent) {
switch event.Event {
case "login":
if event.Code == "0" {
var subs = []WebsocketSubscription{
{Channel: "account"},
{Channel: "orders", InstrumentType: string(okexapi.InstrumentTypeSpot)},
}
log.Infof("subscribing private channels: %+v", subs)
err := s.Conn.WriteJSON(WebsocketOp{
Op: "subscribe",
Args: subs,
})
return stream
}
func (s *Stream) Close() error {
return nil
}
func (s *Stream) Connect(ctx context.Context) error {
err := s.connect(ctx)
if err != nil {
return err
}
// start one re-connector goroutine with the base context
go s.Reconnector(ctx)
s.EmitStart()
return nil
}
func (s *Stream) Reconnector(ctx context.Context) {
for {
select {
case <-ctx.Done():
return
case <-s.ReconnectC:
log.Warnf("received reconnect signal, reconnecting...")
time.Sleep(3 * time.Second)
if err := s.connect(ctx); err != nil {
log.WithError(err).Errorf("connect error, try to reconnect again...")
s.Reconnect()
log.WithError(err).Error("private channel subscribe error")
}
}
}
}
func (s *Stream) connect(ctx context.Context) error {
// when in public mode, the listen key is an empty string
func (s *Stream) handleOrderDetailsEvent(orderDetails []okexapi.OrderDetails) {
detailTrades, detailOrders := segmentOrderDetails(orderDetails)
trades, err := toGlobalTrades(detailTrades)
if err != nil {
log.WithError(err).Errorf("error converting order details into trades")
} else {
for _, trade := range trades {
s.EmitTradeUpdate(trade)
}
}
orders, err := toGlobalOrders(detailOrders)
if err != nil {
log.WithError(err).Errorf("error converting order details into orders")
} else {
for _, order := range orders {
s.EmitOrderUpdate(order)
}
}
}
func (s *Stream) handleAccountEvent(account okexapi.Account) {
balances := toGlobalBalance(&account)
s.EmitBalanceSnapshot(balances)
}
func (s *Stream) handleBookEvent(data BookEvent) {
book := data.Book()
switch data.Action {
case "snapshot":
s.EmitBookSnapshot(book)
case "update":
s.EmitBookUpdate(book)
}
}
func (s *Stream) handleCandleEvent(candle Candle) {
key := CandleKey{Channel: candle.Channel, InstrumentID: candle.InstrumentID}
kline := candle.KLine()
// check if we need to close previous kline
lastCandle, ok := s.lastCandle[key]
if ok && candle.StartTime.After(lastCandle.StartTime) {
lastKline := lastCandle.KLine()
lastKline.Closed = true
s.EmitKLineClosed(lastKline)
}
s.EmitKLine(kline)
s.lastCandle[key] = candle
}
func (s *Stream) createEndpoint(ctx context.Context) (string, error) {
var url string
if s.PublicOnly {
url = okexapi.PublicWebSocketURL
} else {
url = okexapi.PrivateWebSocketURL
}
conn, err := s.StandardStream.Dial(ctx, url)
if err != nil {
return err
return url, nil
}
log.Infof("websocket connected: %s", url)
// should only start one connection one time, so we lock the mutex
s.connLock.Lock()
// ensure the previous context is cancelled
if s.connCancel != nil {
s.connCancel()
}
// create a new context
s.connCtx, s.connCancel = context.WithCancel(ctx)
conn.SetReadDeadline(time.Now().Add(readTimeout))
conn.SetPongHandler(func(string) error {
conn.SetReadDeadline(time.Now().Add(readTimeout))
return nil
})
s.Conn = conn
s.connLock.Unlock()
s.EmitConnect()
go s.read(s.connCtx)
go s.ping(s.connCtx)
return nil
}
func (s *Stream) read(ctx context.Context) {
defer func() {
if s.connCancel != nil {
s.connCancel()
}
s.EmitDisconnect()
}()
for {
select {
case <-ctx.Done():
return
default:
s.connLock.Lock()
conn := s.Conn
s.connLock.Unlock()
if err := conn.SetReadDeadline(time.Now().Add(readTimeout)); err != nil {
log.WithError(err).Errorf("set read deadline error: %s", err.Error())
}
mt, message, err := conn.ReadMessage()
if err != nil {
// if it's a network timeout error, we should re-connect
switch err := err.(type) {
// if it's a websocket related error
case *websocket.CloseError:
if err.Code == websocket.CloseNormalClosure {
return
}
// for unexpected close error, we should re-connect
// emit reconnect to start a new connection
s.Reconnect()
return
case net.Error:
log.WithError(err).Error("network error")
s.Reconnect()
return
default:
log.WithError(err).Error("unexpected connection error")
s.Reconnect()
return
}
}
// skip non-text messages
if mt != websocket.TextMessage {
continue
}
e, err := Parse(string(message))
if err != nil {
log.WithError(err).Error("message parse error")
}
if e != nil {
func (s *Stream) dispatchEvent(e interface{}) {
switch et := e.(type) {
case *WebSocketEvent:
s.EmitEvent(*et)
case *BookData:
case *BookEvent:
// there's "books" for 400 depth and books5 for 5 depth
if et.channel != "books5" {
s.EmitBookData(*et)
s.EmitBookEvent(*et)
}
s.EmitBookTickerUpdate(et.BookTicker())
case *Candle:
s.EmitCandleData(*et)
s.EmitCandleEvent(*et)
case *okexapi.Account:
s.EmitAccount(*et)
s.EmitAccountEvent(*et)
case []okexapi.OrderDetails:
s.EmitOrderDetails(et)
s.EmitOrderDetailsEvent(et)
}
}
}
}
}
func (s *Stream) ping(ctx context.Context) {
pingTicker := time.NewTicker(readTimeout / 2)
defer pingTicker.Stop()
for {
select {
case <-ctx.Done():
log.Debug("ping worker stopped")
return
case <-pingTicker.C:
s.connLock.Lock()
conn := s.Conn
s.connLock.Unlock()
if err := conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(3*time.Second)); err != nil {
log.WithError(err).Error("ping error", err)
s.Reconnect()
}
}
}
}

40
pkg/exchange/okex/stream_callbacks.go
View File

@ -6,22 +6,22 @@ import (
"github.com/c9s/bbgo/pkg/exchange/okex/okexapi"
)
func (s *Stream) OnCandleData(cb func(candle Candle)) {
s.candleDataCallbacks = append(s.candleDataCallbacks, cb)
func (s *Stream) OnCandleEvent(cb func(candle Candle)) {
s.candleEventCallbacks = append(s.candleEventCallbacks, cb)
}
func (s *Stream) EmitCandleData(candle Candle) {
for _, cb := range s.candleDataCallbacks {
func (s *Stream) EmitCandleEvent(candle Candle) {
for _, cb := range s.candleEventCallbacks {
cb(candle)
}
}
func (s *Stream) OnBookData(cb func(book BookData)) {
s.bookDataCallbacks = append(s.bookDataCallbacks, cb)
func (s *Stream) OnBookEvent(cb func(book BookEvent)) {
s.bookEventCallbacks = append(s.bookEventCallbacks, cb)
}
func (s *Stream) EmitBookData(book BookData) {
for _, cb := range s.bookDataCallbacks {
func (s *Stream) EmitBookEvent(book BookEvent) {
for _, cb := range s.bookEventCallbacks {
cb(book)
}
}
@ -36,34 +36,34 @@ func (s *Stream) EmitEvent(event WebSocketEvent) {
}
}
func (s *Stream) OnAccount(cb func(account okexapi.Account)) {
s.accountCallbacks = append(s.accountCallbacks, cb)
func (s *Stream) OnAccountEvent(cb func(account okexapi.Account)) {
s.accountEventCallbacks = append(s.accountEventCallbacks, cb)
}
func (s *Stream) EmitAccount(account okexapi.Account) {
for _, cb := range s.accountCallbacks {
func (s *Stream) EmitAccountEvent(account okexapi.Account) {
for _, cb := range s.accountEventCallbacks {
cb(account)
}
}
func (s *Stream) OnOrderDetails(cb func(orderDetails []okexapi.OrderDetails)) {
s.orderDetailsCallbacks = append(s.orderDetailsCallbacks, cb)
func (s *Stream) OnOrderDetailsEvent(cb func(orderDetails []okexapi.OrderDetails)) {
s.orderDetailsEventCallbacks = append(s.orderDetailsEventCallbacks, cb)
}
func (s *Stream) EmitOrderDetails(orderDetails []okexapi.OrderDetails) {
for _, cb := range s.orderDetailsCallbacks {
func (s *Stream) EmitOrderDetailsEvent(orderDetails []okexapi.OrderDetails) {
for _, cb := range s.orderDetailsEventCallbacks {
cb(orderDetails)
}
}
type StreamEventHub interface {
OnCandleData(cb func(candle Candle))
OnCandleEvent(cb func(candle Candle))
OnBookData(cb func(book BookData))
OnBookEvent(cb func(book BookEvent))
OnEvent(cb func(event WebSocketEvent))
OnAccount(cb func(account okexapi.Account))
OnAccountEvent(cb func(account okexapi.Account))
OnOrderDetails(cb func(orderDetails []okexapi.OrderDetails))
OnOrderDetailsEvent(cb func(orderDetails []okexapi.OrderDetails))
}