Files
gocryptotrader/exchanges/orderbook/orderbook.go
Scott 62a528a064 Bugfix/improvement: Orderbook/ticker update processing (#364)
* Greatly increases efficiency of web socket orderbook processing by minimising multiple map lookups. Changes buffer to use pointers. Ensures orderbook benchmarks are on equal footing and set correctly. Removes data race by not setting waitgroup adds inside go routine causing wait and add to happen simultaneously. Updates ticker and orderbook service to use a pointer var instead of map lookups when setting data.

* Removes misguided comment

* Removes waitgroups and goroutines for updating bids and asks for non-id orderbook updates via websocket. Updates benchmarks to be consistent
2019-10-04 16:24:29 +10:00

262 lines
6.4 KiB
Go

package orderbook
import (
"errors"
"fmt"
"sort"
"strings"
"time"
"github.com/gofrs/uuid"
"github.com/thrasher-corp/gocryptotrader/currency"
"github.com/thrasher-corp/gocryptotrader/dispatch"
"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
)
// Get checks and returns the orderbook given an exchange name and currency pair
// if it exists
func Get(exchange string, p currency.Pair, a asset.Item) (Base, error) {
o, err := service.Retrieve(exchange, p, a)
if err != nil {
return Base{}, err
}
return *o, nil
}
// SubscribeOrderbook subcribes to an orderbook and returns a communication
// channel to stream orderbook data updates
func SubscribeOrderbook(exchange string, p currency.Pair, a asset.Item) (dispatch.Pipe, error) {
exchange = strings.ToLower(exchange)
service.RLock()
defer service.RUnlock()
book, ok := service.Books[exchange][p.Base.Item][p.Quote.Item][a]
if !ok {
return dispatch.Pipe{}, fmt.Errorf("orderbook item not found for %s %s %s",
exchange,
p,
a)
}
return service.mux.Subscribe(book.Main)
}
// SubscribeToExchangeOrderbooks subcribes to all orderbooks on an exchange
func SubscribeToExchangeOrderbooks(exchange string) (dispatch.Pipe, error) {
exchange = strings.ToLower(exchange)
service.RLock()
defer service.RUnlock()
id, ok := service.Exchange[exchange]
if !ok {
return dispatch.Pipe{}, fmt.Errorf("%s exchange orderbooks not found",
exchange)
}
return service.mux.Subscribe(id)
}
// Update stores orderbook data
func (s *Service) Update(b *Base) error {
var ids []uuid.UUID
s.Lock()
switch {
case s.Books[b.ExchangeName] == nil:
s.Books[b.ExchangeName] = make(map[*currency.Item]map[*currency.Item]map[asset.Item]*Book)
s.Books[b.ExchangeName][b.Pair.Base.Item] = make(map[*currency.Item]map[asset.Item]*Book)
s.Books[b.ExchangeName][b.Pair.Base.Item][b.Pair.Quote.Item] = make(map[asset.Item]*Book)
err := s.SetNewData(b)
if err != nil {
s.Unlock()
return err
}
case s.Books[b.ExchangeName][b.Pair.Base.Item] == nil:
s.Books[b.ExchangeName][b.Pair.Base.Item] = make(map[*currency.Item]map[asset.Item]*Book)
s.Books[b.ExchangeName][b.Pair.Base.Item][b.Pair.Quote.Item] = make(map[asset.Item]*Book)
err := s.SetNewData(b)
if err != nil {
s.Unlock()
return err
}
case s.Books[b.ExchangeName][b.Pair.Base.Item][b.Pair.Quote.Item] == nil:
s.Books[b.ExchangeName][b.Pair.Base.Item][b.Pair.Quote.Item] = make(map[asset.Item]*Book)
err := s.SetNewData(b)
if err != nil {
s.Unlock()
return err
}
case s.Books[b.ExchangeName][b.Pair.Base.Item][b.Pair.Quote.Item][b.AssetType] == nil:
err := s.SetNewData(b)
if err != nil {
s.Unlock()
return err
}
default:
book := s.Books[b.ExchangeName][b.Pair.Base.Item][b.Pair.Quote.Item][b.AssetType]
book.b.Bids = b.Bids
book.b.Asks = b.Asks
book.b.LastUpdated = b.LastUpdated
ids = book.Assoc
ids = append(ids, book.Main)
}
s.Unlock()
return s.mux.Publish(ids, b)
}
// SetNewData sets new data
func (s *Service) SetNewData(b *Base) error {
ids, err := s.GetAssociations(b)
if err != nil {
return err
}
singleID, err := s.mux.GetID()
if err != nil {
return err
}
s.Books[b.ExchangeName][b.Pair.Base.Item][b.Pair.Quote.Item][b.AssetType] = &Book{b: b,
Main: singleID,
Assoc: ids}
return nil
}
// GetAssociations links a singular book with it's dispatch associations
func (s *Service) GetAssociations(b *Base) ([]uuid.UUID, error) {
if b == nil {
return nil, errors.New("orderbook is nil")
}
var ids []uuid.UUID
exchangeID, ok := s.Exchange[b.ExchangeName]
if !ok {
var err error
exchangeID, err = s.mux.GetID()
if err != nil {
return nil, err
}
s.Exchange[b.ExchangeName] = exchangeID
}
ids = append(ids, exchangeID)
return ids, nil
}
// Retrieve gets orderbook data from the slice
func (s *Service) Retrieve(exchange string, p currency.Pair, a asset.Item) (*Base, error) {
exchange = strings.ToLower(exchange)
s.RLock()
defer s.RUnlock()
if s.Books[exchange] == nil {
return nil, fmt.Errorf("no orderbooks for %s exchange", exchange)
}
if s.Books[exchange][p.Base.Item] == nil {
return nil, fmt.Errorf("no orderbooks associated with base currency %s",
p.Base)
}
if s.Books[exchange][p.Base.Item][p.Quote.Item] == nil {
return nil, fmt.Errorf("no orderbooks associated with quote currency %s",
p.Quote)
}
if s.Books[exchange][p.Base.Item][p.Quote.Item][a] == nil {
return nil, fmt.Errorf("no orderbooks associated with asset type %s",
a)
}
return s.Books[exchange][p.Base.Item][p.Quote.Item][a].b, nil
}
// TotalBidsAmount returns the total amount of bids and the total orderbook
// bids value
func (b *Base) TotalBidsAmount() (amountCollated, total float64) {
for x := range b.Bids {
amountCollated += b.Bids[x].Amount
total += b.Bids[x].Amount * b.Bids[x].Price
}
return amountCollated, total
}
// TotalAsksAmount returns the total amount of asks and the total orderbook
// asks value
func (b *Base) TotalAsksAmount() (amountCollated, total float64) {
for y := range b.Asks {
amountCollated += b.Asks[y].Amount
total += b.Asks[y].Amount * b.Asks[y].Price
}
return amountCollated, total
}
// Update updates the bids and asks
func (b *Base) Update(bids, asks []Item) {
b.Bids = bids
b.Asks = asks
b.LastUpdated = time.Now()
}
// Verify ensures that the orderbook items are correctly sorted
// Bids should always go from a high price to a low price and
// asks should always go from a low price to a higher price
func (b *Base) Verify() {
var lastPrice float64
var sortBids, sortAsks bool
for x := range b.Bids {
if lastPrice != 0 && b.Bids[x].Price >= lastPrice {
sortBids = true
break
}
lastPrice = b.Bids[x].Price
}
lastPrice = 0
for x := range b.Asks {
if lastPrice != 0 && b.Asks[x].Price <= lastPrice {
sortAsks = true
break
}
lastPrice = b.Asks[x].Price
}
if sortBids {
sort.Sort(sort.Reverse(byOBPrice(b.Bids)))
}
if sortAsks {
sort.Sort((byOBPrice(b.Asks)))
}
}
// Process processes incoming orderbooks, creating or updating the orderbook
// list
func (b *Base) Process() error {
if b.ExchangeName == "" {
return errors.New(errExchangeNameUnset)
}
b.ExchangeName = strings.ToLower(b.ExchangeName)
if b.Pair.IsEmpty() {
return errors.New(errPairNotSet)
}
if b.AssetType.String() == "" {
return errors.New(errAssetTypeNotSet)
}
if len(b.Asks) == 0 && len(b.Bids) == 0 {
return errors.New(errNoOrderbook)
}
if b.LastUpdated.IsZero() {
b.LastUpdated = time.Now()
}
b.Verify()
return service.Update(b)
}