implement bitfinex stream

Author: c9s

pkg/exchange/bitfinex/stream.go

@@ -0,0 +1,64 @@
+package bitfinex
+
+import (
+	"context"
+	"os"
+
+	"github.com/sirupsen/logrus"
+
+	"github.com/c9s/bbgo/pkg/depth"
+	bfxapi "github.com/c9s/bbgo/pkg/exchange/bitfinex/bfxapi"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+// Stream represents the Bitfinex websocket stream.
+//
+//go:generate callbackgen -type Stream
+type Stream struct {
+	types.StandardStream
+
+	depthBuffers map[string]*depth.Buffer
+
+	tickerEventCallbacks      []func(e *bfxapi.TickerEvent)
+	bookEventCallbacks        []func(e *bfxapi.BookEvent)
+	candleEventCallbacks      []func(e *bfxapi.CandleEvent)
+	statusEventCallbacks      []func(e *bfxapi.StatusEvent)
+	marketTradeEventCallbacks []func(e *bfxapi.MarketTradeEvent)
+
+	parser *bfxapi.Parser
+}
+
+// NewStream creates a new Bitfinex Stream.
+func NewStream() *Stream {
+	stream := &Stream{
+		StandardStream: types.NewStandardStream(),
+		depthBuffers:   make(map[string]*depth.Buffer),
+
+		parser: bfxapi.NewParser(),
+	}
+	stream.SetParser(stream.parser.Parse)
+	stream.SetDispatcher(stream.dispatchEvent)
+	stream.SetEndpointCreator(stream.getEndpoint)
+	return stream
+}
+
+// getEndpoint returns the websocket endpoint URL.
+func (s *Stream) getEndpoint(ctx context.Context) (string, error) {
+	url := os.Getenv("BITFINEX_API_WS_URL")
+	if url == "" {
+		if s.PublicOnly {
+			url = "wss://api-pub.bitfinex.com/ws/2"
+		} else {
+			url = "wss://api.bitfinex.com/ws/2"
+		}
+	}
+	return url, nil
+}
+
+// dispatchEvent dispatches parsed events to corresponding callbacks.
+func (s *Stream) dispatchEvent(e interface{}) {
+	switch evt := e.(type) {
+	default:
+		logrus.Warnf("unhandled %T event: %+v", evt, evt)
+	}
+}

pkg/types/stream.go

@@ -196,14 +196,18 @@ func (s *StandardStream) GetPublicOnly() bool {
 	return s.PublicOnly
 }
 
+// SetEndpointCreator sets the endpoint creator function to create the websocket endpoint URL.
+// This is useful when the endpoint URL is dynamic or needs to be created based on some context.
 func (s *StandardStream) SetEndpointCreator(creator EndpointCreator) {
 	s.endpointCreator = creator
 }
 
+// SetDispatcher sets the dispatcher function to dispatch the parsed websocket message.
 func (s *StandardStream) SetDispatcher(dispatcher Dispatcher) {
 	s.dispatcher = dispatcher
 }
 
+// SetParser sets the parser function to parse the websocket message.
 func (s *StandardStream) SetParser(parser Parser) {
 	s.parser = parser
 }