exchange: binance orderbook fix (thrasher-corp#599)

* port orderbook binance management from draft singular asset (spot) processing add additional updates to buffer management

* integrate port

* shifted burden of proof to exchange and remove repairing techniques that obfuscate issues and could caause artifacts

* WIP

* Update exchanges, update tests, update configuration so we can default off on buffer util.

* Add buffer enabled switching to all exchanges and some that are missing, default to off.

* lbtc set not aggregate books

* Addr linter issues

* EOD wip

* optimization and bug fix pass

* clean before test and benchmarking

* add testing/benchmarks to sorting/reversing functions, dropped pointer to slice as we aren't changing slice len or cap

* Add tests and removed ptr for main book as we just ammend amount

* addr exchange test issues

* ci issues

* addr glorious issues

* Addr MCB nits, fixed funding rate book for bitfinex and fixed potential panic on nil book return

* addr linter issues

* updated mistakes

* Fix more tests

* revert bypass

* Addr mcb nits

* fix zero price bug caused by exchange. Filted out bid result rather then unsubscribing. Updated orderbook to L2 so there is no aggregation.

* Allow for zero bid and ask books to be loaded and warn if found.

* remove authentication subscription conflicts as they do not have a channel ID return

* WIP - Batching outbound requests for kraken as they do not give you the partial if you subscribe to do many things.

* finalised outbound request for kraken

* filter zero value due to invalid returned data from exchange, add in max subscription amount and increased outbound batch limit

* expand to max allowed book length & fix issue where they were sending a zero length ask side when we sent a depth of zero

* Updated function comments and added in more realistic book sizing for sort cases

* change map ordering

* amalgamate maps in buffer

* Rm ln

* fix kraken linter issues

* add in buffer initialisation

* increase timout by 30seconds

* Coinbene: Add websocket orderbook length check.

* Engine: Improve switch statement for orderbook summary dissplay.

* Binance: Added tests, remove deadlock

* Exchanges: Change orderbook field -> IsFundingRate

* Orderbook Buffer: Added method to orderbookHolder

* Kraken: removed superfluous integer for sleep

* Bitmex: fixed error return

* cmd/gctcli: force 8 decimal place usage for orderbook streaming

* Kraken: Add checksum and fix bug where we were dropping returned data which was causing artifacts

* Kraken: As per orderbook documentation added in maxdepth field to update to filter depth that goes beyond current scope

* Bitfinex: Tracking down bug on margin-funding, added sequence and checksum validation websocket config on connect (WIP)

* Bitfinex: Complete implementation of checksum

* Bitfinex: Fix funding book insertion and checksum - Dropped updates and deleting items not on book are continuously occuring from stream

* Bitfinex: Fix linter issues

* Bitfinex: Fix even more linter issues.

* Bitmex: Populate orderbook base identification fields to be passed back when error occurrs

* OkGroup: Populate orderbook base identification fields to be passed back when error occurrs

* BTSE: Change string check to 'connect success' to capture multiple user successful strings

* Bitfinex: Updated handling of funding tickers

* Bitfinex: Fix undocumented alignment bug for funding rates

* Bitfinex: Updated error return with more information

* Bitfinex: Change REST fetching to Raw book to keep it in line with websocket implementation. Fix woopsy.

* Localbitcoins: Had to impose a rate limiter to stop errors, fixed return for easier error identification.

* Exchanges: Update failing tests

* LocalBitcoins: Addr nit and bumped time by 1 second for fetching books

* Kraken: Dynamically scale precision based on str return for checksum calculations

* Kraken: Add pair and asset type to validateCRC32 error reponse

* BTSE: Filter out zero amount orderbook price levels in websocket return

* Exchanges: Update orderbook functions to return orderbook base to differentiate errors.

* BTSE: Fix spelling

* Bitmex: Fix error return string

* BTSE: Add orderbook filtering function

* Coinbene: Change wording

* BTSE: Add test for filtering

* Binance: Addr nits, added in variables for buffers and worker amounts and fixed error log messages

* GolangCI: Remove excess 0

* Binance: Reduces double ups on asset and pair in errors

* Binance: Fix error checking

.golangci.yml

@@ -1,5 +1,5 @@
 run:
-  timeout: 1m30s
+  timeout: 2m0s
   issues-exit-code: 1
   tests: true
   skip-dirs:

cmd/exchange_template/wrapper_file.tmpl

@@ -272,37 +272,32 @@ func ({{.Variable}} *{{.CapitalName}}) FetchOrderbook(currency currency.Pair, as
 
 // UpdateOrderbook updates and returns the orderbook for a currency pair
 func ({{.Variable}} *{{.CapitalName}}) UpdateOrderbook(p currency.Pair, assetType asset.Item) (*orderbook.Base, error) {
-	orderBook := new(orderbook.Base)
+	book := &orderbook.Base{ExchangeName: {{.Variable}}.Name, Pair: p, AssetType: assetType}
 	// NOTE: UPDATE ORDERBOOK EXAMPLE
 	/* 
 	orderbookNew, err := {{.Variable}}.GetOrderBook(exchange.FormatExchangeCurrency({{.Variable}}.Name, p).String(), 1000)
 	if err != nil {
-		return orderBook, err
+		return book, err
 	}
 
 	for x := range orderbookNew.Bids {
-		orderBook.Bids = append(orderBook.Bids, orderbook.Item{
+		book.Bids = append(book.Bids, orderbook.Item{
 			Amount: orderbookNew.Bids[x].Quantity, 
 			Price: orderbookNew.Bids[x].Price,
 		})
 	}
 
 	for x := range orderbookNew.Asks {
-		orderBook.Asks = append(orderBook.Asks, orderbook.Item{
-			Amount: orderBook.Asks[x].Quantity,
-			Price: orderBook.Asks[x].Price,
+		book.Asks = append(book.Asks, orderbook.Item{
+			Amount: orderBookNew.Asks[x].Quantity,
+			Price: orderBookNew.Asks[x].Price,
 		})
 	}
 	*/
 
-
-	orderBook.Pair = p
-	orderBook.ExchangeName = {{.Variable}}.Name
-	orderBook.AssetType = assetType
-
-	err := orderBook.Process()
+	err := book.Process()
 	if err != nil {
-		return orderBook, err
+		return book, err
 	}
 
 	return orderbook.Get({{.Variable}}.Name, p, assetType)

cmd/gctcli/commands.go

@@ -3152,7 +3152,7 @@ func getOrderbookStream(c *cli.Context) error {
 				askPrice = resp.Asks[i].Price
 			}
 
-			fmt.Printf("%f %s @ %f %s\t\t%f %s @ %f %s\n",
+			fmt.Printf("%.8f %s @ %.8f %s\t\t%.8f %s @ %.8f %s\n",
 				bidAmount,
 				resp.Pair.Base,
 				bidPrice,

config/config_types.go

@@ -113,23 +113,24 @@ type ConnectionMonitorConfig struct {
 
 // ExchangeConfig holds all the information needed for each enabled Exchange.
 type ExchangeConfig struct {
-	Name                          string                 `json:"name"`
-	Enabled                       bool                   `json:"enabled"`
-	Verbose                       bool                   `json:"verbose"`
-	UseSandbox                    bool                   `json:"useSandbox,omitempty"`
-	HTTPTimeout                   time.Duration          `json:"httpTimeout"`
-	HTTPUserAgent                 string                 `json:"httpUserAgent,omitempty"`
-	HTTPDebugging                 bool                   `json:"httpDebugging,omitempty"`
-	WebsocketResponseCheckTimeout time.Duration          `json:"websocketResponseCheckTimeout"`
-	WebsocketResponseMaxLimit     time.Duration          `json:"websocketResponseMaxLimit"`
-	WebsocketTrafficTimeout       time.Duration          `json:"websocketTrafficTimeout"`
-	WebsocketOrderbookBufferLimit int                    `json:"websocketOrderbookBufferLimit"`
-	ProxyAddress                  string                 `json:"proxyAddress,omitempty"`
-	BaseCurrencies                currency.Currencies    `json:"baseCurrencies"`
-	CurrencyPairs                 *currency.PairsManager `json:"currencyPairs"`
-	API                           APIConfig              `json:"api"`
-	Features                      *FeaturesConfig        `json:"features"`
-	BankAccounts                  []banking.Account      `json:"bankAccounts,omitempty"`
+	Name                            string                 `json:"name"`
+	Enabled                         bool                   `json:"enabled"`
+	Verbose                         bool                   `json:"verbose"`
+	UseSandbox                      bool                   `json:"useSandbox,omitempty"`
+	HTTPTimeout                     time.Duration          `json:"httpTimeout"`
+	HTTPUserAgent                   string                 `json:"httpUserAgent,omitempty"`
+	HTTPDebugging                   bool                   `json:"httpDebugging,omitempty"`
+	WebsocketResponseCheckTimeout   time.Duration          `json:"websocketResponseCheckTimeout"`
+	WebsocketResponseMaxLimit       time.Duration          `json:"websocketResponseMaxLimit"`
+	WebsocketTrafficTimeout         time.Duration          `json:"websocketTrafficTimeout"`
+	WebsocketOrderbookBufferLimit   int                    `json:"websocketOrderbookBufferLimit"`
+	WebsocketOrderbookBufferEnabled bool                   `json:"websocketOrderbookBufferEnabled"`
+	ProxyAddress                    string                 `json:"proxyAddress,omitempty"`
+	BaseCurrencies                  currency.Currencies    `json:"baseCurrencies"`
+	CurrencyPairs                   *currency.PairsManager `json:"currencyPairs"`
+	API                             APIConfig              `json:"api"`
+	Features                        *FeaturesConfig        `json:"features"`
+	BankAccounts                    []banking.Account      `json:"bankAccounts,omitempty"`
 
 	// Deprecated settings which will be removed in a future update
 	AvailablePairs                   *currency.Pairs      `json:"availablePairs,omitempty"`

engine/routines.go

@@ -3,6 +3,7 @@ package engine
 import (
 	"errors"
 	"fmt"
+	"strconv"
 	"strings"
 	"sync"
 
@@ -115,74 +116,66 @@ func printTickerSummary(result *ticker.Price, protocol string, err error) {
 	}
 }
 
+const (
+	book = "%s %s %s %s: ORDERBOOK: Bids len: %d Amount: %f %s. Total value: %s Asks len: %d Amount: %f %s. Total value: %s\n"
+)
+
 func printOrderbookSummary(result *orderbook.Base, protocol string, err error) {
 	if err != nil {
+		if result == nil {
+			log.Errorf(log.OrderBook, "Failed to get %s orderbook. Error: %s\n",
+				protocol,
+				err)
+			return
+		}
 		if err == common.ErrNotYetImplemented {
-			log.Warnf(log.Ticker, "Failed to get %s ticker. Error: %s\n",
+			log.Warnf(log.OrderBook, "Failed to get %s orderbook for %s %s %s. Error: %s\n",
 				protocol,
+				result.ExchangeName,
+				result.Pair,
+				result.AssetType,
 				err)
 			return
 		}
-		log.Errorf(log.OrderBook, "Failed to get %s orderbook. Error: %s\n",
+		log.Errorf(log.OrderBook, "Failed to get %s orderbook for %s %s %s. Error: %s\n",
 			protocol,
+			result.ExchangeName,
+			result.Pair,
+			result.AssetType,
 			err)
 		return
 	}
 
 	bidsAmount, bidsValue := result.TotalBidsAmount()
 	asksAmount, asksValue := result.TotalAsksAmount()
 
-	if result.Pair.Quote.IsFiatCurrency() &&
-		result.Pair.Quote != Bot.Config.Currency.FiatDisplayCurrency {
+	var bidValueResult, askValueResult string
+	switch {
+	case result.Pair.Quote.IsFiatCurrency() && result.Pair.Quote != Bot.Config.Currency.FiatDisplayCurrency:
 		origCurrency := result.Pair.Quote.Upper()
-		log.Infof(log.OrderBook, "%s %s %s %s: ORDERBOOK: Bids len: %d Amount: %f %s. Total value: %s Asks len: %d Amount: %f %s. Total value: %s\n",
-			result.ExchangeName,
-			protocol,
-			FormatCurrency(result.Pair),
-			strings.ToUpper(result.AssetType.String()),
-			len(result.Bids),
-			bidsAmount,
-			result.Pair.Base,
-			printConvertCurrencyFormat(origCurrency, bidsValue),
-			len(result.Asks),
-			asksAmount,
-			result.Pair.Base,
-			printConvertCurrencyFormat(origCurrency, asksValue),
-		)
-	} else {
-		if result.Pair.Quote.IsFiatCurrency() &&
-			result.Pair.Quote == Bot.Config.Currency.FiatDisplayCurrency {
-			log.Infof(log.OrderBook, "%s %s %s %s: ORDERBOOK: Bids len: %d Amount: %f %s. Total value: %s Asks len: %d Amount: %f %s. Total value: %s\n",
-				result.ExchangeName,
-				protocol,
-				FormatCurrency(result.Pair),
-				strings.ToUpper(result.AssetType.String()),
-				len(result.Bids),
-				bidsAmount,
-				result.Pair.Base,
-				printCurrencyFormat(bidsValue),
-				len(result.Asks),
-				asksAmount,
-				result.Pair.Base,
-				printCurrencyFormat(asksValue),
-			)
-		} else {
-			log.Infof(log.OrderBook, "%s %s %s %s: ORDERBOOK: Bids len: %d Amount: %f %s. Total value: %f Asks len: %d Amount: %f %s. Total value: %f\n",
-				result.ExchangeName,
-				protocol,
-				FormatCurrency(result.Pair),
-				strings.ToUpper(result.AssetType.String()),
-				len(result.Bids),
-				bidsAmount,
-				result.Pair.Base,
-				bidsValue,
-				len(result.Asks),
-				asksAmount,
-				result.Pair.Base,
-				asksValue,
-			)
-		}
+		bidValueResult = printConvertCurrencyFormat(origCurrency, bidsValue)
+		askValueResult = printConvertCurrencyFormat(origCurrency, asksValue)
+	case result.Pair.Quote.IsFiatCurrency() && result.Pair.Quote == Bot.Config.Currency.FiatDisplayCurrency:
+		bidValueResult = printCurrencyFormat(bidsValue)
+		askValueResult = printCurrencyFormat(asksValue)
+	default:
+		bidValueResult = strconv.FormatFloat(bidsValue, 'f', -1, 64)
+		askValueResult = strconv.FormatFloat(asksValue, 'f', -1, 64)
 	}
+	log.Infof(log.OrderBook, book,
+		result.ExchangeName,
+		protocol,
+		FormatCurrency(result.Pair),
+		strings.ToUpper(result.AssetType.String()),
+		len(result.Bids),
+		bidsAmount,
+		result.Pair.Base,
+		bidValueResult,
+		len(result.Asks),
+		asksAmount,
+		result.Pair.Base,
+		askValueResult,
+	)
 }
 
 func relayWebsocketEvent(result interface{}, event, assetType, exchangeName string) {

exchanges/binance/binance.go

@@ -65,6 +65,8 @@ type Binance struct {
 
 	// Valid string list that is required by the exchange
 	validLimits []int
+
+	obm *orderbookManager
 }
 
 // GetExchangeInfo returns exchange information. Check binance_types for more

exchanges/binance/binance_live_test.go

@@ -34,6 +34,7 @@ func TestMain(m *testing.M) {
 	if err != nil {
 		log.Fatal("Binance setup error", err)
 	}
+	b.setupOrderbookManager()
 	b.Websocket.DataHandler = sharedtestvalues.GetWebsocketInterfaceChannelOverride()
 	log.Printf(sharedtestvalues.LiveTesting, b.Name, b.API.Endpoints.URL)
 	os.Exit(m.Run())

exchanges/binance/binance_mock_test.go

@@ -39,6 +39,8 @@ func TestMain(m *testing.M) {
 		log.Fatal("Binance setup error", err)
 	}
 
+	b.setupOrderbookManager()
+
 	serverDetails, newClient, err := mock.NewVCRServer(mockfile)
 	if err != nil {
 		log.Fatalf("Mock server error %s", err)

exchanges/binance/binance_test.go

@@ -1,6 +1,7 @@
 package binance
 
 import (
+	"encoding/json"
 	"testing"
 	"time"
 
@@ -63,7 +64,7 @@ func TestGetOrderBook(t *testing.T) {
 	t.Parallel()
 	_, err := b.GetOrderBook(OrderBookDataRequestParams{
 		Symbol: currency.NewPair(currency.BTC, currency.USDT),
-		Limit:  10,
+		Limit:  1000,
 	})
 
 	if err != nil {
@@ -911,6 +912,7 @@ func TestWsTradeUpdate(t *testing.T) {
 }
 
 func TestWsDepthUpdate(t *testing.T) {
+	b.setupOrderbookManager()
 	seedLastUpdateID := int64(161)
 	book := OrderBook{
 		Asks: []OrderbookItem{
@@ -1184,3 +1186,50 @@ func TestGetRecentTrades(t *testing.T) {
 		t.Error(err)
 	}
 }
+
+func TestSeedLocalCache(t *testing.T) {
+	t.Parallel()
+	err := b.SeedLocalCache(currency.NewPair(currency.BTC, currency.USDT))
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestGenerateSubscriptions(t *testing.T) {
+	t.Parallel()
+	subs, err := b.GenerateSubscriptions()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(subs) != 4 {
+		t.Fatal("unexpected subscription length")
+	}
+}
+
+var websocketDepthUpdate = []byte(`{"E":1608001030784,"U":7145637266,"a":[["19455.19000000","0.59490200"],["19455.37000000","0.00000000"],["19456.11000000","0.00000000"],["19456.16000000","0.00000000"],["19458.67000000","0.06400000"],["19460.73000000","0.05139800"],["19461.43000000","0.00000000"],["19464.59000000","0.00000000"],["19466.03000000","0.45000000"],["19466.36000000","0.00000000"],["19508.67000000","0.00000000"],["19572.96000000","0.00217200"],["24386.00000000","0.00256600"]],"b":[["19455.18000000","2.94649200"],["19453.15000000","0.01233600"],["19451.18000000","0.00000000"],["19446.85000000","0.11427900"],["19446.74000000","0.00000000"],["19446.73000000","0.00000000"],["19444.45000000","0.14937800"],["19426.75000000","0.00000000"],["19416.36000000","0.36052100"]],"e":"depthUpdate","s":"BTCUSDT","u":7145637297}`)
+
+func TestProcessUpdate(t *testing.T) {
+	t.Parallel()
+	p := currency.NewPair(currency.BTC, currency.USDT)
+	var depth WebsocketDepthStream
+	err := json.Unmarshal(websocketDepthUpdate, &depth)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = b.obm.stageWsUpdate(&depth, p, asset.Spot)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = b.obm.fetchBookViaREST(p)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = b.obm.cleanup(p)
+	if err != nil {
+		t.Fatal(err)
+	}
+}

exchanges/binance/binance_types.go

@@ -1,9 +1,11 @@
 package binance
 
 import (
+	"sync"
 	"time"
 
 	"github.com/thrasher-corp/gocryptotrader/currency"
+	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
 )
 
 // withdrawals status codes description
@@ -107,13 +109,13 @@ type DepthUpdateParams []struct {
 
 // WebsocketDepthStream is the difference for the update depth stream
 type WebsocketDepthStream struct {
-	Event         string          `json:"e"`
-	Timestamp     time.Time       `json:"E"`
-	Pair          string          `json:"s"`
-	FirstUpdateID int64           `json:"U"`
-	LastUpdateID  int64           `json:"u"`
-	UpdateBids    [][]interface{} `json:"b"`
-	UpdateAsks    [][]interface{} `json:"a"`
+	Event         string           `json:"e"`
+	Timestamp     time.Time        `json:"E"`
+	Pair          string           `json:"s"`
+	FirstUpdateID int64            `json:"U"`
+	LastUpdateID  int64            `json:"u"`
+	UpdateBids    [][2]interface{} `json:"b"`
+	UpdateAsks    [][2]interface{} `json:"a"`
 }
 
 // RecentTradeRequestParams represents Klines request data.
@@ -746,3 +748,24 @@ type WsPayload struct {
 	Params []string `json:"params"`
 	ID     int64    `json:"id"`
 }
+
+// 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 *WebsocketDepthStream
+	fetchingBook bool
+	initialSync  bool
+}
+
+// job defines a synchonisation job that tells a go routine to fetch an
+// orderbook via the REST protocol
+type job struct {
+	Pair currency.Pair
+}