feat: OKX Provider (#7)

* init

* hook up into provider factory

* reconnect timeout

Author: davidterpay

config/local/oracle.toml

@@ -44,6 +44,15 @@ name = "crypto_dot_com"
 path = "config/local/providers/crypto_dot_com.json"
 web_socket.enabled = true
 web_socket.max_buffer_size = 1000  # Replace "1000" with your desired maximum buffer size.
+web_socket.reconnection_timeout = "5s"  # Replace "5s" with your desired reconnection timeout duration.
+
+[[providers]]
+name = "okx"
+path = "config/local/providers/okx.json"
+web_socket.enabled = true
+web_socket.max_buffer_size = 1000
+web_socket.reconnection_timeout = "10s"
+
 
 [[providers]]
 name = "binanceus"

config/local/providers/okx.json

@@ -0,0 +1,12 @@
+{
+    "markets": {
+        "BITCOIN/USD": "BTC-USD",
+        "ETHEREUM/USD": "ETH-USD",
+        "SOLANA/USD": "SOL-USD",
+        "ATOM/USD": "ATOM-USD",
+        "POLKADOT/USD": "DOT-USD",
+        "DYDX/USD": "DYDX-USD",
+        "ETHEREUM/BITCOIN": "ETH-BTC"
+    },
+    "production": true
+}

oracle/config/provider_test.go

@@ -69,6 +69,7 @@ path = "testpath"
 [web_socket]
 enabled = true
 max_buffer_size = 100
+reconnection_timeout = "5s"
 `
 
 	invalidWebSocketConfigContent = `
@@ -78,6 +79,7 @@ path = "testpath"
 [web_socket]
 enabled = true
 max_buffer_size = -1
+reconnection_timeout = "5s"
 `
 
 	noHandlerSpecificationConfigContent = `
@@ -98,6 +100,16 @@ max_queries = 10
 [web_socket]
 enabled = true
 max_buffer_size = 100
+reconnection_timeout = "5s"
+`
+
+	badReconnectionTimeoutConfigContent = `
+name = "testname"
+path = "testpath"
+[web_socket]
+enabled = true
+max_buffer_size = 100
+reconnection_timeout = -1s
 `
 )
 
@@ -152,6 +164,11 @@ func TestReadProviderConfigFromFile(t *testing.T) {
 			config:      duplicateHandlerConfigContent,
 			expectedErr: true,
 		},
+		{
+			name:        "bad reconnection timeout config",
+			config:      badReconnectionTimeoutConfigContent,
+			expectedErr: true,
+		},
 	}
 
 	for _, tc := range testCases {

oracle/config/websocket.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"fmt"
+	"time"
 )
 
 // WebSocketConfig defines a config for a websocket based data provider.
@@ -13,6 +14,10 @@ type WebSocketConfig struct {
 	// at any given time. If the provider receives more messages than this, it will
 	// block receiving messages until the buffer is cleared.
 	MaxBufferSize int `mapstructure:"max_buffer_size" toml:"max_buffer_size"`
+
+	// ReconnectionTimeout is the timeout for the provider to attempt to reconnect
+	// to the websocket endpoint.
+	ReconnectionTimeout time.Duration `mapstructure:"reconnection_timeout" toml:"reconnection_timeout"`
 }
 
 func (c *WebSocketConfig) ValidateBasic() error {
@@ -24,5 +29,9 @@ func (c *WebSocketConfig) ValidateBasic() error {
 		return fmt.Errorf("websocket max buffer size must be greater than 0")
 	}
 
+	if c.ReconnectionTimeout <= 0 {
+		return fmt.Errorf("websocket reconnection timeout must be greater than 0")
+	}
+
 	return nil
 }

oracle/oracle.go

@@ -197,7 +197,8 @@ func (o *Oracle) fetchPrices(provider providertypes.Provider[oracletypes.Currenc
 	timeFilteredPrices := make(map[oracletypes.CurrencyPair]*big.Int)
 	for pair, result := range prices {
 		// If the price is older than the update interval, skip it.
-		if diff := time.Since(result.Timestamp); diff > o.cfg.UpdateInterval {
+		diff := time.Now().UTC().Sub(result.Timestamp)
+		if diff > o.cfg.UpdateInterval {
 			o.logger.Debug(
 				"skipping price",
 				zap.String("provider", provider.Name()),
@@ -212,6 +213,7 @@ func (o *Oracle) fetchPrices(provider providertypes.Provider[oracletypes.Currenc
 			zap.String("provider", provider.Name()),
 			zap.String("pair", pair.String()),
 			zap.String("price", result.Value.String()),
+			zap.Duration("diff", diff),
 		)
 		timeFilteredPrices[pair] = result.Value
 	}

oracle/oracle_test.go

@@ -33,8 +33,9 @@ var (
 		Path: "testpath2",
 		Name: "websocket1",
 		WebSocket: config.WebSocketConfig{
-			MaxBufferSize: 10,
-			Enabled:       true,
+			MaxBufferSize:       10,
+			Enabled:             true,
+			ReconnectionTimeout: 250 * time.Millisecond,
 		},
 	}
 )

providers/base/fetch.go

@@ -101,15 +101,24 @@ func (p *BaseProvider[K, V]) attemptDataUpdate(ctx context.Context, responseCh c
 
 // startWebSocket starts a connection to the web socket and handles the incoming messages.
 func (p *BaseProvider[K, V]) startWebSocket(ctx context.Context, responseCh chan<- providertypes.GetResponse[K, V]) error {
-	p.logger.Debug("starting web socket query handler")
+	// Start the web socket query handler. If the connection fails to start, then the query handler
+	// will be restarted after a timeout.
+	for {
+		select {
+		case <-ctx.Done():
+			p.logger.Info("provider stopped via context")
+			return ctx.Err()
+		default:
+			p.logger.Debug("starting web socket query handler")
+			if err := p.ws.Start(ctx, p.ids, responseCh); err != nil {
+				p.logger.Error("web socket query handler returned error", zap.Error(err))
+			}
 
-	if err := p.ws.Start(ctx, p.ids, responseCh); err != nil {
-		p.logger.Error("web socket query handler returned error", zap.Error(err))
-		return err
+			// If the web socket query handler returns, then the connection was closed. Wait for
+			// a bit before trying to reconnect.
+			time.Sleep(p.cfg.WebSocket.ReconnectionTimeout)
+		}
 	}
-
-	p.logger.Debug("web socket query handler finished")
-	return nil
 }
 
 // recv receives responses from the response channel and updates the data.

providers/base/provider_test.go

@@ -43,8 +43,9 @@ var (
 		Name: "websocket",
 		Path: "test",
 		WebSocket: config.WebSocketConfig{
-			Enabled:       true,
-			MaxBufferSize: 10,
+			Enabled:             true,
+			MaxBufferSize:       10,
+			ReconnectionTimeout: time.Millisecond * 200,
 		},
 	}
 	pairs = []oracletypes.CurrencyPair{
@@ -109,7 +110,7 @@ func TestStart(t *testing.T) {
 		handler.On("Start", mock.Anything, mock.Anything, mock.Anything).Return(func() error {
 			<-ctx.Done()
 			return ctx.Err()
-		}())
+		}()).Maybe()
 
 		provider, err := base.NewProvider(
 			wsConfig,
@@ -131,7 +132,7 @@ func TestStart(t *testing.T) {
 		handler.On("Start", mock.Anything, mock.Anything, mock.Anything).Return(func() error {
 			<-ctx.Done()
 			return ctx.Err()
-		}())
+		}()).Maybe()
 
 		provider, err := base.NewProvider(
 			wsConfig,
@@ -291,6 +292,20 @@ func TestWebSocketProvider(t *testing.T) {
 			},
 			expectedPrices: map[oracletypes.CurrencyPair]*big.Int{},
 		},
+		{
+			name: "continues restarting if the query handler returns",
+			handler: func() wshandlers.WebSocketQueryHandler[oracletypes.CurrencyPair, *big.Int] {
+				handler := wshandlermocks.NewWebSocketQueryHandler[oracletypes.CurrencyPair, *big.Int](t)
+
+				handler.On("Start", mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("no gib price updates")).Maybe()
+
+				return handler
+			},
+			pairs: []oracletypes.CurrencyPair{
+				pairs[0],
+			},
+			expectedPrices: map[oracletypes.CurrencyPair]*big.Int{},
+		},
 	}
 
 	for _, tc := range testCases {

providers/websockets/README.md

@@ -11,4 +11,5 @@ Web sockets are preferred over HTTP APIs for real-time data as they only require
 The current set of supported providers are:
 
 * [Crypto.com](./cryptodotcom/README.md) - Crypto.com is a cryptocurrency exchange that provides a free API for fetching cryptocurrency data. Crypto.com is a **primary data source** for the oracle.   
+* [OKX](./okx/README.md) - OKX is a cryptocurrency exchange that provides a free API for fetching cryptocurrency data. OKX is a **primary data source** for the oracle.
 

providers/websockets/cryptodotcom/parse.go

@@ -14,7 +14,7 @@ import (
 
 // parseInstrumentMessage is used to parse an instrument message received from the Crypto.com
 // web socket API. This message contains the latest price data for a set of instruments.
-func (h *CryptoWebSocketDataHandler) parseInstrumentMessage(
+func (h *WebSocketDataHandler) parseInstrumentMessage(
 	msg InstrumentResponseMessage,
 ) (providertypes.GetResponse[oracletypes.CurrencyPair, *big.Int], error) {
 	var (

providers/websockets/cryptodotcom/ws_data_handler.go

@@ -18,11 +18,11 @@ const (
 	Name = "crypto_dot_com"
 )
 
-var _ handlers.WebSocketDataHandler[oracletypes.CurrencyPair, *big.Int] = (*CryptoWebSocketDataHandler)(nil)
+var _ handlers.WebSocketDataHandler[oracletypes.CurrencyPair, *big.Int] = (*WebSocketDataHandler)(nil)
 
-// CryptoWebSocketDataHandler implements the WebSocketDataHandler interface. This is used to
+// WebSocketDataHandler implements the WebSocketDataHandler interface. This is used to
 // handle messages received from the Crypto.com websocket API.
-type CryptoWebSocketDataHandler struct {
+type WebSocketDataHandler struct {
 	logger *zap.Logger
 
 	// config is the config for the Crypto.com web socket API.
@@ -33,7 +33,7 @@ type CryptoWebSocketDataHandler struct {
 func NewWebSocketDataHandlerFromConfig(
 	logger *zap.Logger,
 	providerCfg config.ProviderConfig,
-) (*CryptoWebSocketDataHandler, error) {
+) (handlers.WebSocketDataHandler[oracletypes.CurrencyPair, *big.Int], error) {
 	if providerCfg.Name != Name {
 		return nil, fmt.Errorf("invalid provider name %s", providerCfg.Name)
 	}
@@ -43,7 +43,7 @@ func NewWebSocketDataHandlerFromConfig(
 		return nil, fmt.Errorf("failed to read config file %s: %s", providerCfg.Path, err)
 	}
 
-	return &CryptoWebSocketDataHandler{
+	return &WebSocketDataHandler{
 		config: cfg,
 		logger: logger.With(zap.String("web_socket_data_handler", Name)),
 	}, nil
@@ -53,12 +53,12 @@ func NewWebSocketDataHandlerFromConfig(
 func NewWebSocketDataHandler(
 	logger *zap.Logger,
 	cfg Config,
-) (*CryptoWebSocketDataHandler, error) {
+) (handlers.WebSocketDataHandler[oracletypes.CurrencyPair, *big.Int], error) {
 	if err := cfg.ValidateBasic(); err != nil {
 		return nil, fmt.Errorf("invalid config: %s", err)
 	}
 
-	return &CryptoWebSocketDataHandler{
+	return &WebSocketDataHandler{
 		config: cfg,
 		logger: logger.With(zap.String("web_socket_data_handler", Name)),
 	}, nil
@@ -69,7 +69,7 @@ func NewWebSocketDataHandler(
 // a heartbeat response message must be sent back to the Crypto.com web socket API, otherwise
 // the connection will be closed. If a subscribe message is received, the message must be parsed
 // and a response must be returned. No update message is required for subscribe messages.
-func (h *CryptoWebSocketDataHandler) HandleMessage(
+func (h *WebSocketDataHandler) HandleMessage(
 	message []byte,
 ) (providertypes.GetResponse[oracletypes.CurrencyPair, *big.Int], []byte, error) {
 	var (
@@ -117,7 +117,7 @@ func (h *CryptoWebSocketDataHandler) HandleMessage(
 // CreateMessage is used to create a message to send to the data provider. This is used to
 // subscribe to the given currency pairs. This is called when the connection to the data
 // provider is first established.
-func (h *CryptoWebSocketDataHandler) CreateMessage(
+func (h *WebSocketDataHandler) CreateMessage(
 	cps []oracletypes.CurrencyPair,
 ) ([]byte, error) {
 	instruments := make([]string, 0)
@@ -140,12 +140,12 @@ func (h *CryptoWebSocketDataHandler) CreateMessage(
 }
 
 // Name returns the name of the data provider.
-func (h *CryptoWebSocketDataHandler) Name() string {
+func (h *WebSocketDataHandler) Name() string {
 	return Name
 }
 
 // URL is used to get the URL of the data provider.
-func (h *CryptoWebSocketDataHandler) URL() string {
+func (h *WebSocketDataHandler) URL() string {
 	if h.config.Production {
 		return ProductionURL
 	}

providers/websockets/okx/README.md

@@ -0,0 +1,67 @@
+# OKX Provider
+
+## Overview
+
+The OKX provider is used to fetch the ticker price from the [OKX web socket API](https://www.okx.com/docs-v5/en/#overview-websocket-overview). The web socket request size for data transmission between the client and server is only 2 bytes. The total number of requests to subscribe to new markets is limited to 3 requests per second. The total number of 'subscribe'/'unsubscribe'/'login' requests per connection is limited to 480 times per hour. WebSocket login and subscription rate limits are based on connection.
+
+Connections will break automatically if the subscription is not established or data has not been pushed for more than 30 seconds. [Per OKX documentation](https://www.okx.com/docs-v5/en/#overview-websocket-overview),
+
+```text
+If there’s a network problem, the system will automatically disable the connection.
+
+The connection will break automatically if the subscription is not established or data has not been pushed for more than 30 seconds.
+
+To keep the connection stable:
+
+1. Set a timer of N seconds whenever a response message is received, where N is less than 30.
+2. If the timer is triggered, which means that no new message is received within N seconds, send the String 'ping'.
+3. Expect a 'pong' as a response. If the response message is not received within N seconds, please raise an error or reconnect.
+```
+
+OKX provides [public and private channels](https://www.okx.com/docs-v5/en/?shell#overview-websocket-subscribe). 
+
+* Public channels -- No authentication is required, include tickers channel, K-Line channel, limit price channel, order book channel, and mark price channel etc.
+* Private channels -- including account channel, order channel, and position channel, etc -- require log in.
+
+Users can choose to subscribe to one or more channels, and the total length of multiple channels cannot exceed 64 KB. This provider is implemented assuming that the user is only subscribing to public channels.
+
+The exact channel that is used to subscribe to the ticker price is the [`Index Tickers Channel`](https://www.okx.com/docs-v5/en/?shell#public-data-websocket-index-tickers-channel). This pushes data every 100ms if there are any price updates, otherwise it will push updates once a minute.
+
+To retrieve all of the supported [spot markets](https://www.okx.com/docs-v5/en/?shell#public-data-rest-api-get-instruments), please run the following command:
+
+```bash
+curl https://www.okx.com/api/v5/public/instruments?instType=SPOT
+```
+
+## Configuration
+
+The configuration structure for this provider looks like the following:
+
+```golang
+type Config struct {
+	// Markets is the list of markets to subscribe to. The key is the currency pair and the value
+	// is the instrument ID. The instrument ID must correspond to the spot market. For example,
+	// the instrument ID for the BITCOIN/USDT market is BTC-USDT.
+	Markets map[string]string `json:"markets"`
+
+	// Production is true if the config is for production.
+	Production bool `json:"production"`
+}
+```
+
+Note that if production is set to false, all prices returned by any subscribed markets will be static. A sample configuration is shown below:
+
+```json
+{
+    "markets": {
+        "BITCOIN/USD": "BTC-USD", // Spot market
+        "ETHEREUM/USD": "ETH-USD", // Spot market
+        "SOLANA/USD": "SOL-USD", // Spot market
+        "ATOM/USD": "ATOM-USD", // Spot market
+        "POLKADOT/USD": "DOT-USD", // Spot market
+        "DYDX/USD": "DYDX-USD", // Spot market
+        "ETHEREUM/BITCOIN": "ETH-BTC" // Spot market
+    },
+    "production": true
+}
+```