aopoltorzhicky / go_kraken Goto Github PK

package kraken

import (
	"errors"
	"fmt"
	"hash/crc32"
	"os"
	"os/signal"
	e "router/exchanges"
	log "router/logging"
	"sort"
	"strings"
	"syscall"

	ws "github.com/aopoltorzhicky/go_kraken/websocket"
	"github.com/shopspring/decimal"
)

var orderbook_btc_usd = e.IndexedBook{
	Bids: make(map[float64]e.SubBook),
	Asks: make(map[float64]e.SubBook),
}

func ConnectKrakenWS() {
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt, syscall.SIGTERM, syscall.SIGINT)

	kraken := ws.NewKraken(ws.ProdBaseURL)
	if err := kraken.Connect(); err != nil {
		log.LogFatal(errors.New("Error connecting to web socket: " + err.Error()))
	}

	// subscribe to BTCUSD`s ticker
	if err := kraken.SubscribeBook([]string{ws.BTCUSD}, ws.Depth10); err != nil {
		log.LogFatal(errors.New("SubscribeTicker error: " + err.Error()))
	}

	orderbook_btc_usd = e.IndexedBook{
		Bids: make(map[float64]e.SubBook),
		Asks: make(map[float64]e.SubBook),
	}

	var processedCounter = 0
	var recievedCounter = 0
	// busy := false

	for {
		select {
		case <-signals:
			log.LogInfo("Stopping kraken WS...")
			if err := kraken.Close(); err != nil {
				log.LogFatal(err)
			}
			return
		case update := <-kraken.Listen():

			switch data := update.Data.(type) {
			case ws.OrderBookUpdate:
				recievedCounter += 1
				// if !busy {
				// 	busy = true
				switch update.Pair {
				case "XBT/USD":
					handleBTCUSDUpdateEvent(&data)
				}
				processedCounter += 1
				// 	busy = false
				// }
			default:
			}
		}
	}
}

func handleBTCUSDUpdateEvent(v *ws.OrderBookUpdate) {

	capture(orderbook_btc_usd.Bids, v.Bids)
	capture(orderbook_btc_usd.Asks, v.Asks)

	newBook := sortorderbook()
	newBook.Pair.Currency1 = e.BTC
	newBook.Pair.Currency2 = e.USD

	if !v.IsSnapshot {
		if !isBookValid(newBook, v.CheckSum) {
			fmt.Println("Checksum failure. Expected", v.CheckSum)
			// log.LogFatal(errors.New("Kraken checksum failure"))
			return
		}
	}

	e.Kraken.Book = newBook
	e.Kraken.Active = true
}

func capture(liveSideCollection map[float64]e.SubBook, deltaCollection []ws.OrderBookItem) {
	// for k, v := range liveSideCollection {
	for i := range deltaCollection {
		price, _ := deltaCollection[i].Price.Float64()
		qty, _ := deltaCollection[i].Volume.Float64()
		if qty == 0 {
			delete(liveSideCollection, price)
		} else {
			liveSideCollection[price] = e.SubBook{
				Price: price,
				Qty:   qty,
			}
		}
	}
	// }
}

func sortorderbook() (Book e.Book) {
	c := orderbook_btc_usd

	Book.Bids = []e.SubBook{}
	for i := range c.Bids {
		Book.Bids = append(Book.Bids, e.SubBook{
			Price: c.Bids[i].Price,
			Qty:   c.Bids[i].Qty,
		})
	}
	sort.Sort(e.BidSubBook(Book.Bids))
	Book.Bids[0].Total = Book.Bids[0].Qty
	for i := 1; i < 10; i++ {
		Book.Bids[i].Total = Book.Bids[i-1].Total + Book.Bids[i].Qty
	}

	Book.Asks = []e.SubBook{}
	for i := range c.Asks {
		Book.Asks = append(Book.Asks, e.SubBook{
			Price: c.Asks[i].Price,
			Qty:   c.Asks[i].Qty,
		})
	}
	sort.Sort(e.AskSubBook(Book.Asks))
	Book.Asks[0].Total = Book.Asks[0].Qty
	for i := 1; i < 10; i++ {
		Book.Asks[i].Total = Book.Asks[i-1].Total + Book.Asks[i].Qty
	}

	// Book.Bids = Book.Bids[:10]
	// Book.Asks = Book.Asks[:10]

	return Book
}

func isBookValid(newBook e.Book, challengeChecksum string) bool {
	var str strings.Builder

	// Literally here so I can print the current local orderbook for checksum comparisons
	dict := map[string][][]string{}

	for _, level := range newBook.Asks[:10] {
		price := decimal.NewFromFloat(level.Price).StringFixed(5)
		price = strings.Replace(price, ".", "", 1)
		price = strings.TrimLeft(price, "0")
		str.WriteString(price)

		volume := decimal.NewFromFloat(level.Qty).StringFixed(8)
		volume = strings.Replace(volume, ".", "", 1)
		volume = strings.TrimLeft(volume, "0")
		str.WriteString(volume)

		dict["a"] = append(dict["a"], []string{
			decimal.NewFromFloat(level.Price).StringFixed(5), decimal.NewFromFloat(level.Qty).StringFixed(8), "",
		})
	}
	for _, level := range newBook.Bids[:10] {
		price := decimal.NewFromFloat(level.Price).StringFixed(5)
		price = strings.Replace(price, ".", "", 1)
		price = strings.TrimLeft(price, "0")
		str.WriteString(price)

		volume := decimal.NewFromFloat(level.Qty).StringFixed(8)
		volume = strings.Replace(volume, ".", "", 1)
		volume = strings.TrimLeft(volume, "0")
		str.WriteString(volume)

		dict["b"] = append(dict["b"], []string{
			decimal.NewFromFloat(level.Price).StringFixed(5), decimal.NewFromFloat(level.Qty).StringFixed(8), "",
		})
	}

	checksum := str.String()
	checksumInt := crc32.ChecksumIEEE([]byte(checksum))
	checksumPass := fmt.Sprint(checksumInt) == challengeChecksum

	if !checksumPass {
		fmt.Println("Kraken checksum fail")
	}

	return checksumPass
}

func (k *Kraken) resubscribe() error {
	for _, sub := range k.subscriptions {
		switch sub.Subscription.Name {
		// Private Channels
		case ChanOwnTrades, ChanOpenOrders:
			return k.subscribeToPrivate(sub.Subscription.Name)
		default:
			//All other channels can do normal auth
			if err := k.send(SubscriptionRequest{
				Event:        EventSubscribe,
				Pairs:        []string{sub.Pair},
				Subscription: sub.Subscription,
			}); err != nil {
				return err
			}
		}
	}
	return nil
}

2020/12/12 21:49:10 [WARN]: invalid data length: []interface {}{[]interface {}{
*** snip many go-objects with info of requests from the past ***
}}, "openOrders", map[string]interface {}{"sequence":1}}

goroutine 72 [running]:
github.com/aopoltorzhicky/go_kraken/websocket.(*Client).close(0xc001112500, 0x0, 0x0)
        /home/dvp/go/pkg/mod/github.com/aopoltorzhicky/go_kraken/[email protected]/client.go:285 +0xc8
github.com/aopoltorzhicky/go_kraken/websocket.(*Client).exit(0xc001112500, 0x0, 0x0, 0x0, 0x0)
        /home/dvp/go/pkg/mod/github.com/aopoltorzhicky/go_kraken/[email protected]/client.go:258 +0x52
github.com/aopoltorzhicky/go_kraken/websocket.(*Client).reconnect(0xc001112500, 0x0, 0x0, 0x0, 0x0)
        /home/dvp/go/pkg/mod/github.com/aopoltorzhicky/go_kraken/[email protected]/client.go:222 +0x7a
github.com/aopoltorzhicky/go_kraken/websocket.(*Client).listenDisconnect(0xc001112500)
        /home/dvp/go/pkg/mod/github.com/aopoltorzhicky/go_kraken/[email protected]/client.go:149 +0x357
created by github.com/aopoltorzhicky/go_kraken/websocket.(*Client).reset
        /home/dvp/go/pkg/mod/github.com/aopoltorzhicky/go_kraken/[email protected]/client.go:189 +0x67
panic: send on closed channel

goroutine 30 [running]:
github.com/aopoltorzhicky/go_kraken/websocket.(*Client).close(0xc001112500, 0xdc0ca0, 0xc00018e230)
        /home/dvp/go/pkg/mod/github.com/aopoltorzhicky/go_kraken/[email protected]/client.go:283 +0xad
github.com/aopoltorzhicky/go_kraken/websocket.(*Client).exit(0xc001112500, 0xdc0ca0, 0xc00018e230, 0x0, 0x0)
        /home/dvp/go/pkg/mod/github.com/aopoltorzhicky/go_kraken/[email protected]/client.go:258 +0x52
github.com/aopoltorzhicky/go_kraken/websocket.(*Client).reconnect(0xc001112500, 0xdc0ca0, 0xc00018e230, 0x0, 0x0)
        /home/dvp/go/pkg/mod/github.com/aopoltorzhicky/go_kraken/[email protected]/client.go:245 +0x5d0
github.com/aopoltorzhicky/go_kraken/websocket.(*Client).listenDisconnect(0xc001112500)
        /home/dvp/go/pkg/mod/github.com/aopoltorzhicky/go_kraken/[email protected]/client.go:149 +0x357
created by github.com/aopoltorzhicky/go_kraken/websocket.(*Client).reset
        /home/dvp/go/pkg/mod/github.com/aopoltorzhicky/go_kraken/[email protected]/client.go:189 +0x67

          Just heard back from Kraken support and it appears the website was under maintenance