bithumb: Add websocket support (#745)

* bithumb: Add websocket support (WIP)

* bithumb: cont impl.

* bithumb: finish, issues with orderbook needs review

* linter issues

* bithumb: move to separate file

* bithumb: change to pointer for book

* bithumb: Add tests

* bithumb: Address nits

* websocket: Export subscription error and wrap returns

* bithumb: cleanup/ fix tests

* gctrpc/bithumb: fix misspelling

* bithumb: gofmt

* readme: update support table, regen doc

* tradesReadme: update

* readme: update template

CONTRIBUTORS

@@ -13,8 +13,8 @@ MadCozBadd | https://github.com/MadCozBadd
 vadimzhukck | https://github.com/vadimzhukck
 140am | https://github.com/140am
 marcofranssen | https://github.com/marcofranssen
-dackroyd | https://github.com/dackroyd
 ydm | https://github.com/ydm
+dackroyd | https://github.com/dackroyd
 cranktakular | https://github.com/cranktakular
 woshidama323 | https://github.com/woshidama323
 crackcomm | https://github.com/crackcomm

README.md

@@ -22,7 +22,7 @@ Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader
 | Binance| Yes  | Yes        | NA  |
 | Bitfinex | Yes  | Yes        | NA  |
 | Bitflyer | Yes  | No      | NA  |
-| Bithumb | Yes  | NA       | NA  |
+| Bithumb | Yes  | Yes       | NA  |
 | BitMEX | Yes | Yes | NA |
 | Bitstamp | Yes  | Yes       | No  |
 | Bittrex | Yes | Yes | NA |
@@ -143,11 +143,11 @@ Binaries will be published once the codebase reaches a stable condition.
 |User|Contribution Amount|
 |--|--|
 | [thrasher-](https://github.com/thrasher-) | 656 |
-| [shazbert](https://github.com/shazbert) | 212 |
+| [shazbert](https://github.com/shazbert) | 214 |
-| [gloriousCode](https://github.com/gloriousCode) | 186 |
+| [gloriousCode](https://github.com/gloriousCode) | 189 |
 | [dependabot-preview[bot]](https://github.com/apps/dependabot-preview) | 88 |
 | [xtda](https://github.com/xtda) | 47 |
-| [dependabot[bot]](https://github.com/apps/dependabot) | 18 |
+| [dependabot[bot]](https://github.com/apps/dependabot) | 20 |
 | [Rots](https://github.com/Rots) | 15 |
 | [vazha](https://github.com/vazha) | 15 |
 | [ermalguni](https://github.com/ermalguni) | 14 |
@@ -155,8 +155,8 @@ Binaries will be published once the codebase reaches a stable condition.
 | [vadimzhukck](https://github.com/vadimzhukck) | 10 |
 | [140am](https://github.com/140am) | 8 |
 | [marcofranssen](https://github.com/marcofranssen) | 8 |
+| [ydm](https://github.com/ydm) | 8 |
 | [dackroyd](https://github.com/dackroyd) | 5 |
-| [ydm](https://github.com/ydm) | 5 |
 | [cranktakular](https://github.com/cranktakular) | 5 |
 | [woshidama323](https://github.com/woshidama323) | 3 |
 | [crackcomm](https://github.com/crackcomm) | 3 |

cmd/documentation/exchanges_templates/exchanges_trade_readme.tmpl

@@ -45,7 +45,7 @@ _b in this context is an `IBotExchange` implemented struct_
 | Binance| Yes  | Yes        | Yes  |
 | Bitfinex | Yes  | Yes        | Yes  |
 | Bitflyer | Yes  | No      | No  |
-| Bithumb | Yes  | NA       | No  |
+| Bithumb | Yes  | Yes       | No  |
 | BitMEX | Yes | Yes | Yes |
 | Bitstamp | Yes  | Yes       | No  |
 | Bittrex | Yes | Yes | No |

cmd/documentation/root_templates/root_readme.tmpl

@@ -23,7 +23,7 @@ Join our slack to discuss all things related to GoCryptoTrader! [GoCryptoTrader
 | Binance| Yes  | Yes        | NA  |
 | Bitfinex | Yes  | Yes        | NA  |
 | Bitflyer | Yes  | No      | NA  |
-| Bithumb | Yes  | NA       | NA  |
+| Bithumb | Yes  | Yes       | NA  |
 | BitMEX | Yes | Yes | NA |
 | Bitstamp | Yes  | Yes       | No  |
 | Bittrex | Yes | Yes | NA |

engine/rpcserver.go

@@ -285,9 +285,9 @@ func (s *RPCServer) EnableExchange(_ context.Context, r *gctrpc.GenericExchangeN
 }
 
 // GetExchangeOTPCode retrieves an exchanges OTP code
-func (s *RPCServer) GetExchangeOTPCode(_ context.Context, r *gctrpc.GenericExchangeNameRequest) (*gctrpc.GetExchangeOTPReponse, error) {
+func (s *RPCServer) GetExchangeOTPCode(_ context.Context, r *gctrpc.GenericExchangeNameRequest) (*gctrpc.GetExchangeOTPResponse, error) {
 	result, err := s.GetExchangeOTPByName(r.Exchange)
-	return &gctrpc.GetExchangeOTPReponse{OtpCode: result}, err
+	return &gctrpc.GetExchangeOTPResponse{OtpCode: result}, err
 }
 
 // GetExchangeOTPCodes retrieves OTP codes for all exchanges which have an

exchanges/bithumb/bithumb.go

@@ -51,6 +51,7 @@ const (
 // Bithumb is the overarching type across the Bithumb package
 type Bithumb struct {
 	exchange.Base
+	obm orderbookManager
 }
 
 // GetTradablePairs returns a list of tradable currencies
@@ -114,18 +115,18 @@ func (b *Bithumb) GetAllTickers() (map[string]Ticker, error) {
 // GetOrderBook returns current orderbook
 //
 // symbol e.g. "btc"
-func (b *Bithumb) GetOrderBook(symbol string) (Orderbook, error) {
+func (b *Bithumb) GetOrderBook(symbol string) (*Orderbook, error) {
 	response := Orderbook{}
 	err := b.SendHTTPRequest(exchange.RestSpot, publicOrderBook+strings.ToUpper(symbol), &response)
 	if err != nil {
-		return response, err
+		return nil, err
 	}
 
 	if response.Status != noError {
-		return response, errors.New(response.Message)
+		return nil, errors.New(response.Message)
 	}
 
-	return response, nil
+	return &response, nil
 }
 
 // GetTransactionHistory returns recent transactions

exchanges/bithumb/bithumb_websocket.go

@@ -0,0 +1,210 @@
+package bithumb
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/ticker"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/trade"
+)
+
+const (
+	wsEndpoint       = "wss://pubwss.bithumb.com/pub/ws"
+	tickerTimeLayout = "20060102150405"
+	tradeTimeLayout  = "2006-01-02 15:04:05.000000"
+)
+
+var (
+	wsDefaultTickTypes = []string{"30M"} // alternatives "1H", "12H", "24H", "MID"
+	location           *time.Location
+)
+
+// WsConnect initiates a websocket connection
+func (b *Bithumb) WsConnect() error {
+	if !b.Websocket.IsEnabled() || !b.IsEnabled() {
+		return errors.New(stream.WebsocketNotEnabled)
+	}
+
+	var dialer websocket.Dialer
+	dialer.HandshakeTimeout = b.Config.HTTPTimeout
+	dialer.Proxy = http.ProxyFromEnvironment
+
+	err := b.Websocket.Conn.Dial(&dialer, http.Header{})
+	if err != nil {
+		return fmt.Errorf("%v - Unable to connect to Websocket. Error: %w",
+			b.Name,
+			err)
+	}
+	go b.wsReadData()
+	b.setupOrderbookManager()
+	return nil
+}
+
+// wsReadData receives and passes on websocket messages for processing
+func (b *Bithumb) wsReadData() {
+	b.Websocket.Wg.Add(1)
+	defer b.Websocket.Wg.Done()
+
+	for {
+		resp := b.Websocket.Conn.ReadMessage()
+		if resp.Raw == nil {
+			return
+		}
+		err := b.wsHandleData(resp.Raw)
+		if err != nil {
+			b.Websocket.DataHandler <- err
+		}
+	}
+}
+
+func (b *Bithumb) wsHandleData(respRaw []byte) error {
+	var resp WsResponse
+	err := json.Unmarshal(respRaw, &resp)
+	if err != nil {
+		return err
+	}
+
+	if len(resp.Status) > 0 {
+		if resp.Status == "0000" {
+			return nil
+		}
+		return fmt.Errorf("%s: %w",
+			resp.ResponseMessage,
+			stream.ErrSubscriptionFailure)
+	}
+
+	switch resp.Type {
+	case "ticker":
+		var tick WsTicker
+		err = json.Unmarshal(resp.Content, &tick)
+		if err != nil {
+			return err
+		}
+		var lu time.Time
+		lu, err = time.ParseInLocation(tickerTimeLayout,
+			tick.Date+tick.Time,
+			location)
+		if err != nil {
+			return err
+		}
+		b.Websocket.DataHandler <- &ticker.Price{
+			ExchangeName: b.Name,
+			AssetType:    asset.Spot,
+			Last:         tick.PreviousClosePrice,
+			Pair:         tick.Symbol,
+			Open:         tick.OpenPrice,
+			Close:        tick.ClosePrice,
+			Low:          tick.LowPrice,
+			High:         tick.HighPrice,
+			QuoteVolume:  tick.Value,
+			Volume:       tick.Volume,
+			LastUpdated:  lu,
+		}
+	case "transaction":
+		if !b.IsSaveTradeDataEnabled() {
+			return nil
+		}
+
+		var trades WsTransactions
+		err = json.Unmarshal(resp.Content, &trades)
+		if err != nil {
+			return err
+		}
+
+		toBuffer := make([]trade.Data, len(trades.List))
+		var lu time.Time
+		for x := range trades.List {
+			lu, err = time.ParseInLocation(tradeTimeLayout,
+				trades.List[x].ContractTime,
+				location)
+			if err != nil {
+				return err
+			}
+
+			toBuffer[x] = trade.Data{
+				Exchange:     b.Name,
+				AssetType:    asset.Spot,
+				CurrencyPair: trades.List[x].Symbol,
+				Timestamp:    lu,
+				Price:        trades.List[x].ContractPrice,
+				Amount:       trades.List[x].ContractAmount,
+			}
+		}
+
+		err = b.AddTradesToBuffer(toBuffer...)
+		if err != nil {
+			return err
+		}
+	case "orderbookdepth":
+		var orderbooks WsOrderbooks
+		err = json.Unmarshal(resp.Content, &orderbooks)
+		if err != nil {
+			return err
+		}
+		init, err := b.UpdateLocalBuffer(&orderbooks)
+		if err != nil && !init {
+			return fmt.Errorf("%v - UpdateLocalCache error: %s", b.Name, err)
+		}
+		return nil
+	default:
+		return fmt.Errorf("unhandled response type %s", resp.Type)
+	}
+
+	return nil
+}
+
+// GenerateSubscriptions generates the default subscription set
+func (b *Bithumb) GenerateSubscriptions() ([]stream.ChannelSubscription, error) {
+	var channels = []string{"ticker", "transaction", "orderbookdepth"}
+	var subscriptions []stream.ChannelSubscription
+	pairs, err := b.GetEnabledPairs(asset.Spot)
+	if err != nil {
+		return nil, err
+	}
+
+	for x := range pairs {
+		for y := range channels {
+			subscriptions = append(subscriptions, stream.ChannelSubscription{
+				Channel:  channels[y],
+				Currency: pairs[x].Format("_", true),
+				Asset:    asset.Spot,
+			})
+		}
+	}
+	return subscriptions, nil
+}
+
+// Subscribe subscribes to a set of channels
+func (b *Bithumb) Subscribe(channelsToSubscribe []stream.ChannelSubscription) error {
+	subs := make(map[string]*WsSubscribe)
+	for i := range channelsToSubscribe {
+		s, ok := subs[channelsToSubscribe[i].Channel]
+		if !ok {
+			s = &WsSubscribe{
+				Type: channelsToSubscribe[i].Channel,
+			}
+			subs[channelsToSubscribe[i].Channel] = s
+		}
+		s.Symbols = append(s.Symbols, channelsToSubscribe[i].Currency)
+	}
+
+	tSub, ok := subs["ticker"]
+	if ok {
+		tSub.TickTypes = wsDefaultTickTypes
+	}
+
+	for _, s := range subs {
+		err := b.Websocket.Conn.SendJSONMessage(s)
+		if err != nil {
+			return err
+		}
+	}
+	b.Websocket.AddSuccessfulSubscriptions(channelsToSubscribe...)
+	return nil
+}

exchanges/bithumb/bithumb_websocket_test.go

@@ -0,0 +1,101 @@
+package bithumb
+
+import (
+	"errors"
+	"sync"
+	"testing"
+
+	"github.com/thrasher-corp/gocryptotrader/currency"
+	exchange "github.com/thrasher-corp/gocryptotrader/exchanges"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/ticker"
+)
+
+var (
+	wsTickerResp    = []byte(`{"type":"ticker","content":{"tickType":"24H","date":"20210811","time":"132017","openPrice":"33400","closePrice":"34010","lowPrice":"32660","highPrice":"34510","value":"45741663716.89916828275244531","volume":"1359398.496892086826189907","sellVolume":"198021.237915860451480504","buyVolume":"1161377.258976226374709403","prevClosePrice":"33530","chgRate":"1.83","chgAmt":"610","volumePower":"500","symbol":"UNI_KRW"}}`)
+	wsTransResp     = []byte(`{"type":"transaction","content":{"list":[{"buySellGb":"1","contPrice":"1166","contQty":"125.2400","contAmt":"146029.8400","contDtm":"2021-08-13 15:23:42.911273","updn":"dn","symbol":"DAI_KRW"}]}}`)
+	wsOrderbookResp = []byte(`{"type":"orderbookdepth","content":{"list":[{"symbol":"XLM_KRW","orderType":"ask","price":"401.2","quantity":"0","total":"0"},{"symbol":"XLM_KRW","orderType":"ask","price":"401.6","quantity":"21277.735","total":"1"},{"symbol":"XLM_KRW","orderType":"ask","price":"403.3","quantity":"4000","total":"1"},{"symbol":"XLM_KRW","orderType":"bid","price":"399.5","quantity":"0","total":"0"},{"symbol":"XLM_KRW","orderType":"bid","price":"398.2","quantity":"0","total":"0"},{"symbol":"XLM_KRW","orderType":"bid","price":"399.8","quantity":"31416.8779","total":"1"},{"symbol":"XLM_KRW","orderType":"bid","price":"398.5","quantity":"34328.387","total":"1"}],"datetime":"1628835823604483"}}`)
+)
+
+func TestWsHandleData(t *testing.T) {
+	t.Parallel()
+	pairs := currency.Pairs{
+		currency.Pair{
+			Base:  currency.BTC,
+			Quote: currency.USDT,
+		},
+	}
+
+	dummy := Bithumb{
+		Base: exchange.Base{
+			Name: "dummy",
+			Features: exchange.Features{
+				Enabled: exchange.FeaturesEnabled{SaveTradeData: true},
+			},
+			CurrencyPairs: currency.PairsManager{
+				Pairs: map[asset.Item]*currency.PairStore{
+					asset.Spot: {
+						Available: pairs,
+						Enabled:   pairs,
+						ConfigFormat: &currency.PairFormat{
+							Uppercase: true,
+						},
+					},
+				},
+			},
+			Websocket: &stream.Websocket{
+				Wg:          new(sync.WaitGroup),
+				DataHandler: make(chan interface{}, 1),
+			},
+		},
+	}
+
+	dummy.setupOrderbookManager()
+	dummy.API.Endpoints = b.NewEndpoints()
+
+	welcomeMsg := []byte(`{"status":"0000","resmsg":"Connected Successfully"}`)
+	err := dummy.wsHandleData(welcomeMsg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = dummy.wsHandleData([]byte(`{"status":"1336","resmsg":"Failed"}`))
+	if !errors.Is(err, stream.ErrSubscriptionFailure) {
+		t.Fatalf("received: %v but expected: %v",
+			err,
+			stream.ErrSubscriptionFailure)
+	}
+
+	err = dummy.wsHandleData(wsTickerResp)
+	if !errors.Is(err, nil) {
+		t.Fatalf("received: %v but expected: %v", err, nil)
+	}
+
+	handled := <-dummy.Websocket.DataHandler
+	_, ok := handled.(*ticker.Price)
+	if !ok {
+		t.Fatal("unexpected value")
+	}
+
+	err = dummy.wsHandleData(wsTransResp) // This doesn't pipe to datahandler
+	if !errors.Is(err, nil) {
+		t.Fatalf("received: %v but expected: %v", err, nil)
+	}
+
+	err = dummy.wsHandleData(wsOrderbookResp) // This doesn't pipe to datahandler
+	if !errors.Is(err, nil) {
+		t.Fatalf("received: %v but expected: %v", err, nil)
+	}
+}
+
+func TestGenerateSubscriptions(t *testing.T) {
+	t.Parallel()
+	sub, err := b.GenerateSubscriptions()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if sub == nil {
+		t.Fatal("unexpected value")
+	}
+}

exchanges/bithumb/bithumb_websocket_types.go

@@ -0,0 +1,100 @@
+package bithumb
+
+import (
+	"encoding/json"
+	"sync"
+	"time"
+
+	"github.com/thrasher-corp/gocryptotrader/currency"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/order"
+)
+
+// WsResponse is a generalised response data structure which will defer
+// unmarshalling of different contents.
+type WsResponse struct {
+	Status          string          `json:"status"`
+	ResponseMessage string          `json:"resmsg"`
+	Type            string          `json:"type"`
+	Content         json.RawMessage `json:"content"`
+}
+
+// WsTicker defines a websocket ticker
+type WsTicker struct {
+	Symbol             currency.Pair `json:"symbol"`
+	TickType           string        `json:"tickType"`
+	Date               string        `json:"date"`
+	Time               string        `json:"time"`
+	OpenPrice          float64       `json:"openPrice,string"`
+	ClosePrice         float64       `json:"closePrice,string"`
+	LowPrice           float64       `json:"lowPrice,string"`
+	HighPrice          float64       `json:"highPrice,string"`
+	Value              float64       `json:"value,string"`
+	Volume             float64       `json:"volume,string"`
+	SellVolume         float64       `json:"sellVolume,string"`
+	BuyVolume          float64       `json:"buyVolume,string"`
+	PreviousClosePrice float64       `json:"prevClosePrice,string"`
+	ChangeRate         float64       `json:"chgRate,string"`
+	ChangeAmount       float64       `json:"chgAmt,string"`
+	VolumePower        float64       `json:"volumePower,string"`
+}
+
+// WsOrderbooks defines an amalgamated bid ask orderbook tranche list
+type WsOrderbooks struct {
+	List     []WsOrderbook `json:"list"`
+	DateTime bithumbTime   `json:"datetime"`
+}
+
+// WsOrderbook defines a singular orderbook tranche
+type WsOrderbook struct {
+	Symbol    currency.Pair `json:"symbol"`
+	OrderSide order.Side    `json:"orderType"`
+	Price     float64       `json:"price,string"`
+	Quantity  float64       `json:"quantity,string"`
+	Total     int32         `json:"total,string"`
+}
+
+// WsTransactions defines a transaction list
+type WsTransactions struct {
+	List []WsTransaction `json:"list"`
+}
+
+// WsTransaction defines a trade that has executed via their matching engine
+type WsTransaction struct {
+	Symbol           currency.Pair `json:"symbol"`
+	BuySell          int32         `json:"buySellGb,string"` // 1: Sell 2: Buy
+	ContractPrice    float64       `json:"contPrice,string"`
+	ContractQuantity float64       `json:"contQty,string"`
+	ContractAmount   float64       `json:"contAmt,string"`
+	ContractTime     string        `json:"contDtm"` // 2020-01-29 12:24:18.830039
+	UpOrDown         string        `json:"updn"`
+}
+
+// WsSubscribe is used to subscribe to the ws channel.
+type WsSubscribe struct {
+	Type      string          `json:"type"`
+	Symbols   []currency.Pair `json:"symbols"`
+	TickTypes []string        `json:"tickTypes,omitempty"`
+}
+
+// orderbookManager defines a way of managing and maintaining synchronisation
+// across connections and assets.
+type orderbookManager struct {
+	state map[currency.Code]map[currency.Code]map[asset.Item]*update
+	sync.Mutex
+
+	jobs chan job
+}
+
+type update struct {
+	buffer       chan *WsOrderbooks
+	fetchingBook bool
+	initialSync  bool
+	lastUpdated  time.Time
+}
+
+// job defines a synchonisation job that tells a go routine to fetch an
+// orderbook via the REST protocol
+type job struct {
+	Pair currency.Pair
+}

exchanges/bithumb/bithumb_wrapper.go

@@ -20,12 +20,15 @@ import (
 	"github.com/thrasher-corp/gocryptotrader/exchanges/orderbook"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/protocol"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/request"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/ticker"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/trade"
 	"github.com/thrasher-corp/gocryptotrader/log"
 	"github.com/thrasher-corp/gocryptotrader/portfolio/withdraw"
 )
 
+const wsRateLimitMillisecond = 1000
+
 var errNotEnoughPairs = errors.New("at least one currency is required to fetch order history")
 
 // GetDefaultConfig returns a default exchange config
@@ -92,6 +95,13 @@ func (b *Bithumb) SetDefaults() {
 				CryptoWithdrawalFee: true,
 				KlineFetching:       true,
 			},
+			Websocket: true,
+			WebsocketCapabilities: protocol.Features{
+				TradeFetching:     true,
+				TickerFetching:    true,
+				OrderbookFetching: true,
+				Subscribe:         true,
+			},
 			WithdrawPermissions: exchange.AutoWithdrawCrypto |
 				exchange.AutoWithdrawFiat,
 			Kline: kline.ExchangeCapabilitiesSupported{
@@ -120,11 +130,16 @@ func (b *Bithumb) SetDefaults() {
 		request.WithLimiter(SetRateLimit()))
 	b.API.Endpoints = b.NewEndpoints()
 	err = b.API.Endpoints.SetDefaultEndpoints(map[exchange.URL]string{
-		exchange.RestSpot: apiURL,
+		exchange.RestSpot:      apiURL,
+		exchange.WebsocketSpot: wsEndpoint,
 	})
 	if err != nil {
 		log.Errorln(log.ExchangeSys, err)
 	}
+
+	b.Websocket = stream.New()
+	b.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
+	b.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
 }
 
 // Setup takes in the supplied exchange configuration details and sets params
@@ -133,7 +148,44 @@ func (b *Bithumb) Setup(exch *config.ExchangeConfig) error {
 		b.SetEnabled(false)
 		return nil
 	}
-	return b.SetupDefaults(exch)
+	err := b.SetupDefaults(exch)
+	if err != nil {
+		return err
+	}
+
+	location, err = time.LoadLocation("Asia/Seoul")
+	if err != nil {
+		return err
+	}
+
+	ePoint, err := b.API.Endpoints.GetURL(exchange.WebsocketSpot)
+	if err != nil {
+		return err
+	}
+	err = b.Websocket.Setup(&stream.WebsocketSetup{
+		Enabled:                          exch.Features.Enabled.Websocket,
+		Verbose:                          exch.Verbose,
+		AuthenticatedWebsocketAPISupport: exch.API.AuthenticatedWebsocketSupport,
+		WebsocketTimeout:                 exch.WebsocketTrafficTimeout,
+		DefaultURL:                       wsEndpoint,
+		ExchangeName:                     exch.Name,
+		RunningURL:                       ePoint,
+		Connector:                        b.WsConnect,
+		Subscriber:                       b.Subscribe,
+		GenerateSubscriptions:            b.GenerateSubscriptions,
+		Features:                         &b.Features.Supports.WebsocketCapabilities,
+		OrderbookBufferLimit:             exch.OrderbookConfig.WebsocketBufferLimit,
+		BufferEnabled:                    exch.OrderbookConfig.WebsocketBufferEnabled,
+	})
+	if err != nil {
+		return err
+	}
+
+	return b.Websocket.SetupNewConnection(stream.ConnectionSetup{
+		ResponseCheckTimeout: exch.WebsocketResponseCheckTimeout,
+		ResponseMaxLimit:     exch.WebsocketResponseMaxLimit,
+		RateLimit:            wsRateLimitMillisecond,
+	})
 }
 
 // Start starts the Bithumb go routine

exchanges/bithumb/bithumb_ws_orderbook.go

@@ -0,0 +1,391 @@
+package bithumb
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/thrasher-corp/gocryptotrader/currency"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/orderbook"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/stream/buffer"
+	"github.com/thrasher-corp/gocryptotrader/log"
+)
+
+const (
+	// maxWSUpdateBuffer defines max websocket updates to apply when an
+	// orderbook is initially fetched
+	maxWSUpdateBuffer = 150
+	// maxWSOrderbookJobs defines max websocket orderbook jobs in queue to fetch
+	// an orderbook snapshot via REST
+	maxWSOrderbookJobs = 2000
+	// maxWSOrderbookWorkers defines a max amount of workers allowed to execute
+	// jobs from the job channel
+	maxWSOrderbookWorkers = 10
+)
+
+func (b *Bithumb) processBooks(updates *WsOrderbooks) error {
+	var bids, asks []orderbook.Item
+	for x := range updates.List {
+		i := orderbook.Item{Price: updates.List[x].Price, Amount: updates.List[x].Quantity}
+		if updates.List[x].OrderSide == "bid" {
+			bids = append(bids, i)
+			continue
+		}
+		asks = append(asks, i)
+	}
+	return b.Websocket.Orderbook.Update(&buffer.Update{
+		Pair:       updates.List[0].Symbol,
+		Asset:      asset.Spot,
+		Bids:       bids,
+		Asks:       asks,
+		UpdateTime: updates.DateTime.Time(),
+	})
+}
+
+// UpdateLocalBuffer updates and returns the most recent iteration of the orderbook
+func (b *Bithumb) UpdateLocalBuffer(wsdp *WsOrderbooks) (bool, error) {
+	if len(wsdp.List) < 1 {
+		return false, errors.New("insufficient data to process")
+	}
+	err := b.obm.stageWsUpdate(wsdp, wsdp.List[0].Symbol, asset.Spot)
+	if err != nil {
+		init, err2 := b.obm.checkIsInitialSync(wsdp.List[0].Symbol)
+		if err2 != nil {
+			return false, err2
+		}
+		return init, err
+	}
+
+	err = b.applyBufferUpdate(wsdp.List[0].Symbol)
+	if err != nil {
+		b.flushAndCleanup(wsdp.List[0].Symbol)
+	}
+	return false, err
+}
+
+// applyBufferUpdate applies the buffer to the orderbook or initiates a new
+// orderbook sync by the REST protocol which is off handed to go routine.
+func (b *Bithumb) applyBufferUpdate(pair currency.Pair) error {
+	fetching, err := b.obm.checkIsFetchingBook(pair)
+	if err != nil {
+		return err
+	}
+	if fetching {
+		return nil
+	}
+
+	recent, err := b.Websocket.Orderbook.GetOrderbook(pair, asset.Spot)
+	if err != nil || (recent.Asks == nil && recent.Bids == nil) {
+		return b.obm.fetchBookViaREST(pair)
+	}
+
+	return b.obm.checkAndProcessUpdate(b.processBooks, pair, recent)
+}
+
+// SynchroniseWebsocketOrderbook synchronises full orderbook for currency pair
+// asset
+func (b *Bithumb) SynchroniseWebsocketOrderbook() {
+	b.Websocket.Wg.Add(1)
+	go func() {
+		defer b.Websocket.Wg.Done()
+		for {
+			select {
+			case j := <-b.obm.jobs:
+				err := b.processJob(j.Pair)
+				if err != nil {
+					log.Errorf(log.WebsocketMgr,
+						"%s processing websocket orderbook error %v",
+						b.Name, err)
+				}
+			case <-b.Websocket.ShutdownC:
+				return
+			}
+		}
+	}()
+}
+
+// processJob fetches and processes orderbook updates
+func (b *Bithumb) processJob(p currency.Pair) error {
+	err := b.SeedLocalCache(p)
+	if err != nil {
+		return fmt.Errorf("%s %s seeding local cache for orderbook error: %v",
+			p, asset.Spot, err)
+	}
+
+	err = b.obm.stopFetchingBook(p)
+	if err != nil {
+		return err
+	}
+
+	// Immediately apply the buffer updates so we don't wait for a
+	// new update to initiate this.
+	err = b.applyBufferUpdate(p)
+	if err != nil {
+		b.flushAndCleanup(p)
+		return err
+	}
+	return nil
+}
+
+// flushAndCleanup flushes orderbook and clean local cache
+func (b *Bithumb) flushAndCleanup(p currency.Pair) {
+	errClean := b.Websocket.Orderbook.FlushOrderbook(p, asset.Spot)
+	if errClean != nil {
+		log.Errorf(log.WebsocketMgr,
+			"%s flushing websocket error: %v",
+			b.Name,
+			errClean)
+	}
+	errClean = b.obm.cleanup(p)
+	if errClean != nil {
+		log.Errorf(log.WebsocketMgr, "%s cleanup websocket error: %v",
+			b.Name,
+			errClean)
+	}
+}
+
+func (b *Bithumb) setupOrderbookManager() {
+	if b.obm.state == nil {
+		b.obm.state = make(map[currency.Code]map[currency.Code]map[asset.Item]*update)
+		b.obm.jobs = make(chan job, maxWSOrderbookJobs)
+
+		for i := 0; i < maxWSOrderbookWorkers; i++ {
+			// 10 workers for synchronising book
+			b.SynchroniseWebsocketOrderbook()
+		}
+	}
+}
+
+// stageWsUpdate stages websocket update to roll through updates that need to
+// be applied to a fetched orderbook via REST.
+func (o *orderbookManager) stageWsUpdate(u *WsOrderbooks, pair currency.Pair, a asset.Item) error {
+	o.Lock()
+	defer o.Unlock()
+	m1, ok := o.state[pair.Base]
+	if !ok {
+		m1 = make(map[currency.Code]map[asset.Item]*update)
+		o.state[pair.Base] = m1
+	}
+
+	m2, ok := m1[pair.Quote]
+	if !ok {
+		m2 = make(map[asset.Item]*update)
+		m1[pair.Quote] = m2
+	}
+
+	state, ok := m2[a]
+	if !ok {
+		state = &update{
+			buffer:       make(chan *WsOrderbooks, maxWSUpdateBuffer),
+			fetchingBook: false,
+			initialSync:  true,
+		}
+		m2[a] = state
+	}
+
+	if !state.lastUpdated.IsZero() && u.DateTime.Time().Before(state.lastUpdated) {
+		return fmt.Errorf("websocket orderbook synchronisation failure for pair %s and asset %s", pair, a)
+	}
+	state.lastUpdated = u.DateTime.Time()
+
+	select {
+	// Put update in the channel buffer to be processed
+	case state.buffer <- u:
+		return nil
+	default:
+		<-state.buffer    // pop one element
+		state.buffer <- u // to shift buffer on fail
+		return fmt.Errorf("channel blockage for %s, asset %s and connection",
+			pair, a)
+	}
+}
+
+// checkIsFetchingBook checks status if the book is currently being via the REST
+// protocol.
+func (o *orderbookManager) checkIsFetchingBook(pair currency.Pair) (bool, error) {
+	o.Lock()
+	defer o.Unlock()
+	state, ok := o.state[pair.Base][pair.Quote][asset.Spot]
+	if !ok {
+		return false,
+			fmt.Errorf("check is fetching book cannot match currency pair %s asset type %s",
+				pair,
+				asset.Spot)
+	}
+	return state.fetchingBook, nil
+}
+
+// stopFetchingBook completes the book fetching.
+func (o *orderbookManager) stopFetchingBook(pair currency.Pair) error {
+	o.Lock()
+	defer o.Unlock()
+	state, ok := o.state[pair.Base][pair.Quote][asset.Spot]
+	if !ok {
+		return fmt.Errorf("could not match pair %s and asset type %s in hash table",
+			pair,
+			asset.Spot)
+	}
+	if !state.fetchingBook {
+		return fmt.Errorf("fetching book already set to false for %s %s",
+			pair,
+			asset.Spot)
+	}
+	state.fetchingBook = false
+	return nil
+}
+
+// completeInitialSync sets if an asset type has completed its initial sync
+func (o *orderbookManager) completeInitialSync(pair currency.Pair) error {
+	o.Lock()
+	defer o.Unlock()
+	state, ok := o.state[pair.Base][pair.Quote][asset.Spot]
+	if !ok {
+		return fmt.Errorf("complete initial sync cannot match currency pair %s asset type %s",
+			pair,
+			asset.Spot)
+	}
+	if !state.initialSync {
+		return fmt.Errorf("initital sync already set to false for %s %s",
+			pair,
+			asset.Spot)
+	}
+	state.initialSync = false
+	return nil
+}
+
+// checkIsInitialSync checks status if the book is Initial Sync being via the REST
+// protocol.
+func (o *orderbookManager) checkIsInitialSync(pair currency.Pair) (bool, error) {
+	o.Lock()
+	defer o.Unlock()
+	state, ok := o.state[pair.Base][pair.Quote][asset.Spot]
+	if !ok {
+		return false,
+			fmt.Errorf("checkIsInitialSync of orderbook cannot match currency pair %s asset type %s",
+				pair,
+				asset.Spot)
+	}
+	return state.initialSync, nil
+}
+
+// fetchBookViaREST pushes a job of fetching the orderbook via the REST protocol
+// to get an initial full book that we can apply our buffered updates too.
+func (o *orderbookManager) fetchBookViaREST(pair currency.Pair) error {
+	o.Lock()
+	defer o.Unlock()
+
+	state, ok := o.state[pair.Base][pair.Quote][asset.Spot]
+	if !ok {
+		return fmt.Errorf("fetch book via rest cannot match currency pair %s asset type %s",
+			pair,
+			asset.Spot)
+	}
+
+	state.initialSync = true
+	state.fetchingBook = true
+
+	select {
+	case o.jobs <- job{pair}:
+		return nil
+	default:
+		return fmt.Errorf("%s %s book synchronisation channel blocked up",
+			pair,
+			asset.Spot)
+	}
+}
+
+func (o *orderbookManager) checkAndProcessUpdate(processor func(*WsOrderbooks) error, pair currency.Pair, recent *orderbook.Base) error {
+	o.Lock()
+	defer o.Unlock()
+	state, ok := o.state[pair.Base][pair.Quote][asset.Spot]
+	if !ok {
+		return fmt.Errorf("could not match pair [%s] asset type [%s] in hash table to process websocket orderbook update",
+			pair, asset.Spot)
+	}
+
+	// This will continuously remove updates from the buffered channel and
+	// apply them to the current orderbook.
+buffer:
+	for {
+		select {
+		case d := <-state.buffer:
+			if !state.validate(d, recent) {
+				continue
+			}
+			err := processor(d)
+			if err != nil {
+				return fmt.Errorf("%s %s processing update error: %w",
+					pair, asset.Spot, err)
+			}
+		default:
+			break buffer
+		}
+	}
+	return nil
+}
+
+// validate checks for correct update alignment
+func (u *update) validate(updt *WsOrderbooks, recent *orderbook.Base) bool {
+	return updt.DateTime.Time().After(recent.LastUpdated)
+}
+
+// cleanup cleans up buffer and reset fetch and init
+func (o *orderbookManager) cleanup(pair currency.Pair) error {
+	o.Lock()
+	state, ok := o.state[pair.Base][pair.Quote][asset.Spot]
+	if !ok {
+		o.Unlock()
+		return fmt.Errorf("cleanup cannot match %s %s to hash table",
+			pair,
+			asset.Spot)
+	}
+
+bufferEmpty:
+	for {
+		select {
+		case <-state.buffer:
+			// bleed and discard buffer
+		default:
+			break bufferEmpty
+		}
+	}
+	o.Unlock()
+	// disable rest orderbook synchronisation
+	_ = o.stopFetchingBook(pair)
+	_ = o.completeInitialSync(pair)
+	return nil
+}
+
+// SeedLocalCache seeds depth data
+func (b *Bithumb) SeedLocalCache(p currency.Pair) error {
+	ob, err := b.GetOrderBook(p.String())
+	if err != nil {
+		return err
+	}
+	return b.SeedLocalCacheWithBook(p, ob)
+}
+
+// SeedLocalCacheWithBook seeds the local orderbook cache
+func (b *Bithumb) SeedLocalCacheWithBook(p currency.Pair, o *Orderbook) error {
+	var newOrderBook orderbook.Base
+	for i := range o.Data.Bids {
+		newOrderBook.Bids = append(newOrderBook.Bids, orderbook.Item{
+			Amount: o.Data.Bids[i].Quantity,
+			Price:  o.Data.Bids[i].Price,
+		})
+	}
+	for i := range o.Data.Asks {
+		newOrderBook.Asks = append(newOrderBook.Asks, orderbook.Item{
+			Amount: o.Data.Asks[i].Quantity,
+			Price:  o.Data.Asks[i].Price,
+		})
+	}
+
+	newOrderBook.Pair = p
+	newOrderBook.Asset = asset.Spot
+	newOrderBook.Exchange = b.Name
+	newOrderBook.LastUpdated = time.Unix(0, o.Data.Timestamp*int64(time.Millisecond))
+	newOrderBook.VerifyOrderbook = b.CanVerifyOrderbook
+	return b.Websocket.Orderbook.LoadSnapshot(&newOrderBook)
+}

exchanges/bithumb/convert.go

@@ -0,0 +1,31 @@
+package bithumb
+
+import (
+	"encoding/json"
+	"strconv"
+	"time"
+)
+
+// bithumbMSTime provides an internal conversion helper for microsecond parsing
+type bithumbTime time.Time
+
+// UnmarshalJSON implements the unmarshal interface
+func (t *bithumbTime) UnmarshalJSON(data []byte) error {
+	var timestamp string
+	if err := json.Unmarshal(data, &timestamp); err != nil {
+		return err
+	}
+
+	i, err := strconv.ParseInt(timestamp, 10, 64)
+	if err != nil {
+		return err
+	}
+
+	*t = bithumbTime(time.Unix(0, i*int64(time.Microsecond)))
+	return nil
+}
+
+// Time returns a time.Time object
+func (t bithumbTime) Time() time.Time {
+	return time.Time(t)
+}

exchanges/bithumb/convert_test.go

@@ -0,0 +1,27 @@
+package bithumb
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+func TestBithumbTime(t *testing.T) {
+	var newTime bithumbTime
+	err := json.Unmarshal([]byte("bad news"), &newTime)
+	if err == nil {
+		t.Fatal(err)
+	}
+
+	strData := []byte(`"1628739590000000"`) // Thursday, August 12, 2021 3:39:50 AM UTC
+	err = json.Unmarshal(strData, &newTime)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tt := newTime.Time()
+	if tt.UTC().String() != "2021-08-12 03:39:50 +0000 UTC" {
+		t.Fatalf("expected: %s but receieved: %s",
+			"2021-08-12 03:39:50 +0000 UTC",
+			tt.UTC().String())
+	}
+}

exchanges/stream/websocket.go

@@ -23,7 +23,11 @@ const (
 	defaultTrafficPeriod = time.Second
 )
 
-var errClosedConnection = errors.New("use of closed network connection")
+var (
+	errClosedConnection = errors.New("use of closed network connection")
+	// ErrSubscriptionFailure defines an error when a subscription fails
+	ErrSubscriptionFailure = errors.New("subscription failure")
+)
 
 // New initialises the websocket struct
 func New() *Websocket {
@@ -210,7 +214,7 @@ func (w *Websocket) Connect() error {
 	if len(w.subscriptions) != 0 {
 		err = w.Subscriber(w.subscriptions)
 		if err != nil {
-			return fmt.Errorf("%v Error subscribing %s", w.exchangeName, err)
+			return fmt.Errorf("%v %w: %v", w.exchangeName, ErrSubscriptionFailure, err)
 		}
 	}
 
@@ -836,7 +840,11 @@ func (w *Websocket) SubscribeToChannels(channels []ChannelSubscription) error {
 			}
 		}
 	}
-	return w.Subscriber(channels)
+	err := w.Subscriber(channels)
+	if err != nil {
+		return fmt.Errorf("%v %w: %v", w.exchangeName, ErrSubscriptionFailure, err)
+	}
+	return nil
 }
 
 // AddSuccessfulSubscriptions adds subscriptions to the subscription lists that

exchanges/trade/README.md

@@ -63,7 +63,7 @@ _b in this context is an `IBotExchange` implemented struct_
 | Binance| Yes  | Yes        | Yes  |
 | Bitfinex | Yes  | Yes        | Yes  |
 | Bitflyer | Yes  | No      | No  |
-| Bithumb | Yes  | NA       | No  |
+| Bithumb | Yes  | Yes       | No  |
 | BitMEX | Yes | Yes | Yes |
 | Bitstamp | Yes  | Yes       | No  |
 | Bittrex | Yes | Yes | No |

gctrpc/rpc.proto

@@ -61,7 +61,7 @@ message GetExchangesResponse {
     string exchanges = 1;
 }
 
-message GetExchangeOTPReponse {
+message GetExchangeOTPResponse {
     string otp_code = 1;
 }
 
@@ -1016,7 +1016,7 @@ service GoCryptoTrader {
         };
     }
 
-    rpc GetExchangeOTPCode (GenericExchangeNameRequest) returns (GetExchangeOTPReponse) {
+    rpc GetExchangeOTPCode (GenericExchangeNameRequest) returns (GetExchangeOTPResponse) {
         option (google.api.http) = {
             get: "/v1/getexchangeotp"
         };

gctrpc/rpc.swagger.json

@@ -1285,7 +1285,7 @@
           "200": {
             "description": "A successful response.",
             "schema": {
-              "$ref": "#/definitions/gctrpcGetExchangeOTPReponse"
+              "$ref": "#/definitions/gctrpcGetExchangeOTPResponse"
             }
           },
           "default": {
@@ -3793,7 +3793,7 @@
         }
       }
     },
-    "gctrpcGetExchangeOTPReponse": {
+    "gctrpcGetExchangeOTPResponse": {
       "type": "object",
       "properties": {
         "otpCode": {

gctrpc/rpc_grpc.pb.go

@@ -27,7 +27,7 @@ type GoCryptoTraderClient interface {
 	GetExchanges(ctx context.Context, in *GetExchangesRequest, opts ...grpc.CallOption) (*GetExchangesResponse, error)
 	DisableExchange(ctx context.Context, in *GenericExchangeNameRequest, opts ...grpc.CallOption) (*GenericResponse, error)
 	GetExchangeInfo(ctx context.Context, in *GenericExchangeNameRequest, opts ...grpc.CallOption) (*GetExchangeInfoResponse, error)
-	GetExchangeOTPCode(ctx context.Context, in *GenericExchangeNameRequest, opts ...grpc.CallOption) (*GetExchangeOTPReponse, error)
+	GetExchangeOTPCode(ctx context.Context, in *GenericExchangeNameRequest, opts ...grpc.CallOption) (*GetExchangeOTPResponse, error)
 	GetExchangeOTPCodes(ctx context.Context, in *GetExchangeOTPsRequest, opts ...grpc.CallOption) (*GetExchangeOTPsResponse, error)
 	EnableExchange(ctx context.Context, in *GenericExchangeNameRequest, opts ...grpc.CallOption) (*GenericResponse, error)
 	GetTicker(ctx context.Context, in *GetTickerRequest, opts ...grpc.CallOption) (*TickerResponse, error)
@@ -197,8 +197,8 @@ func (c *goCryptoTraderClient) GetExchangeInfo(ctx context.Context, in *GenericE
 	return out, nil
 }
 
-func (c *goCryptoTraderClient) GetExchangeOTPCode(ctx context.Context, in *GenericExchangeNameRequest, opts ...grpc.CallOption) (*GetExchangeOTPReponse, error) {
+func (c *goCryptoTraderClient) GetExchangeOTPCode(ctx context.Context, in *GenericExchangeNameRequest, opts ...grpc.CallOption) (*GetExchangeOTPResponse, error) {
-	out := new(GetExchangeOTPReponse)
+	out := new(GetExchangeOTPResponse)
 	err := c.cc.Invoke(ctx, "/gctrpc.GoCryptoTrader/GetExchangeOTPCode", in, out, opts...)
 	if err != nil {
 		return nil, err
@@ -1059,7 +1059,7 @@ type GoCryptoTraderServer interface {
 	GetExchanges(context.Context, *GetExchangesRequest) (*GetExchangesResponse, error)
 	DisableExchange(context.Context, *GenericExchangeNameRequest) (*GenericResponse, error)
 	GetExchangeInfo(context.Context, *GenericExchangeNameRequest) (*GetExchangeInfoResponse, error)
-	GetExchangeOTPCode(context.Context, *GenericExchangeNameRequest) (*GetExchangeOTPReponse, error)
+	GetExchangeOTPCode(context.Context, *GenericExchangeNameRequest) (*GetExchangeOTPResponse, error)
 	GetExchangeOTPCodes(context.Context, *GetExchangeOTPsRequest) (*GetExchangeOTPsResponse, error)
 	EnableExchange(context.Context, *GenericExchangeNameRequest) (*GenericResponse, error)
 	GetTicker(context.Context, *GetTickerRequest) (*TickerResponse, error)
@@ -1172,7 +1172,7 @@ func (UnimplementedGoCryptoTraderServer) DisableExchange(context.Context, *Gener
 func (UnimplementedGoCryptoTraderServer) GetExchangeInfo(context.Context, *GenericExchangeNameRequest) (*GetExchangeInfoResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method GetExchangeInfo not implemented")
 }
-func (UnimplementedGoCryptoTraderServer) GetExchangeOTPCode(context.Context, *GenericExchangeNameRequest) (*GetExchangeOTPReponse, error) {
+func (UnimplementedGoCryptoTraderServer) GetExchangeOTPCode(context.Context, *GenericExchangeNameRequest) (*GetExchangeOTPResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method GetExchangeOTPCode not implemented")
 }
 func (UnimplementedGoCryptoTraderServer) GetExchangeOTPCodes(context.Context, *GetExchangeOTPsRequest) (*GetExchangeOTPsResponse, error) {