Files
gocryptotrader/engine/routines.go
Ryan O'Hara-Reid db317a2447 Engine: Dispatch service (#346)
* Added dispatch service

* Added orderbook streaming capabilities

* Assigned correct orderbook.base exchange name

* Fixed Requested niterinos
Add in cli orderbook QA tool to gctcli
Add exchange orderbook streaming

* Add ticker streaming support through dispatch package

* Added in some more info on error returns for orderbook.go

* fix linter issues

* Fix some issues

* Update

* Fix requested

* move dispatch out of exchanges folder to its own independant folder

* Fix requested

* change orderbook string to tickers

* Limit orderbooks to 50 and made dispatch system more stateless in operation

* lower cases for update/retrieve/sub exchange name

* Adds in asset validation and lower case conversion on cli

* Remove comment

* Moved timer to a higher scope so its not constantly initialised just reset per instance and removed returning unused channel on error

* Rm unused release function in dispatch.go
Reset timer and bleed buffered timer chan if needed in dispatch.go
Added in ticker.Stop() and timer.Stop() functions for worker routine return in dispatch.go
Index aggregated bid and ask functions for orderbook.go
Added in dummy slice for wsorderbook_test.go

* Moved drain to above Reset so potential race would not occur in dispatch.go
Fix various linter issues dispatch.go

* Fix some issues

* change to start/stop service, added in service state change via cli, updated logger

* fix requested

* Add worker amount init spawning

* fix linter issues

* Fix more linter issues

* More fixes

* Fix race issue on releasing pipe channel on a close after shutting down dispatcher system

* Moved all types to dispatch_types.go && remove panic

* Moved types into serperate file && improve test coverage

* RM unnecessary select case for draining channel && fixed error string

* Added orderbook_types file and improved code coverage

* gofmt file

* reinstated select cases on drain because I am silly

* Remove error for drop worker

* Added more test cases

* not checking error issue fix

* remove func causing race in test, this has required protection via an exported function

* set Gemini websocket orderbook exchange name
2019-10-03 09:47:37 +10:00

348 lines
10 KiB
Go

package engine
import (
"errors"
"fmt"
"strings"
"sync"
"time"
"github.com/thrasher-corp/gocryptotrader/common"
"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/stats"
"github.com/thrasher-corp/gocryptotrader/exchanges/ticker"
"github.com/thrasher-corp/gocryptotrader/exchanges/websocket/wshandler"
log "github.com/thrasher-corp/gocryptotrader/logger"
)
func printCurrencyFormat(price float64) string {
displaySymbol, err := currency.GetSymbolByCurrencyName(Bot.Config.Currency.FiatDisplayCurrency)
if err != nil {
log.Errorf(log.Global, "Failed to get display symbol: %s\n", err)
}
return fmt.Sprintf("%s%.8f", displaySymbol, price)
}
func printConvertCurrencyFormat(origCurrency currency.Code, origPrice float64) string {
displayCurrency := Bot.Config.Currency.FiatDisplayCurrency
conv, err := currency.ConvertCurrency(origPrice,
origCurrency,
displayCurrency)
if err != nil {
log.Errorf(log.Global, "Failed to convert currency: %s\n", err)
}
displaySymbol, err := currency.GetSymbolByCurrencyName(displayCurrency)
if err != nil {
log.Errorf(log.Global, "Failed to get display symbol: %s\n", err)
}
origSymbol, err := currency.GetSymbolByCurrencyName(origCurrency)
if err != nil {
log.Errorf(log.Global, "Failed to get original currency symbol for %s: %s\n",
origCurrency,
err)
}
return fmt.Sprintf("%s%.2f %s (%s%.2f %s)",
displaySymbol,
conv,
displayCurrency,
origSymbol,
origPrice,
origCurrency,
)
}
func printTickerSummary(result *ticker.Price, p currency.Pair, assetType asset.Item, exchangeName string, err error) {
if err != nil {
log.Errorf(log.Ticker, "Failed to get %s %s ticker. Error: %s\n",
p.String(),
exchangeName,
err)
return
}
stats.Add(exchangeName, p, assetType, result.Last, result.Volume)
if p.Quote.IsFiatCurrency() &&
p.Quote != Bot.Config.Currency.FiatDisplayCurrency {
origCurrency := p.Quote.Upper()
log.Infof(log.Ticker, "%s %s %s: TICKER: Last %s Ask %s Bid %s High %s Low %s Volume %.8f\n",
exchangeName,
FormatCurrency(p).String(),
strings.ToUpper(assetType.String()),
printConvertCurrencyFormat(origCurrency, result.Last),
printConvertCurrencyFormat(origCurrency, result.Ask),
printConvertCurrencyFormat(origCurrency, result.Bid),
printConvertCurrencyFormat(origCurrency, result.High),
printConvertCurrencyFormat(origCurrency, result.Low),
result.Volume)
} else {
if p.Quote.IsFiatCurrency() &&
p.Quote == Bot.Config.Currency.FiatDisplayCurrency {
log.Infof(log.Ticker, "%s %s %s: TICKER: Last %s Ask %s Bid %s High %s Low %s Volume %.8f\n",
exchangeName,
FormatCurrency(p).String(),
strings.ToUpper(assetType.String()),
printCurrencyFormat(result.Last),
printCurrencyFormat(result.Ask),
printCurrencyFormat(result.Bid),
printCurrencyFormat(result.High),
printCurrencyFormat(result.Low),
result.Volume)
} else {
log.Infof(log.Ticker, "%s %s %s: TICKER: Last %.8f Ask %.8f Bid %.8f High %.8f Low %.8f Volume %.8f\n",
exchangeName,
FormatCurrency(p).String(),
strings.ToUpper(assetType.String()),
result.Last,
result.Ask,
result.Bid,
result.High,
result.Low,
result.Volume)
}
}
}
func printOrderbookSummary(result *orderbook.Base, p currency.Pair, assetType asset.Item, exchangeName string, err error) {
if err != nil {
log.Errorf(log.OrderBook, "Failed to get %s %s orderbook of type %s. Error: %s\n",
p,
exchangeName,
assetType,
err)
return
}
bidsAmount, bidsValue := result.TotalBidsAmount()
asksAmount, asksValue := result.TotalAsksAmount()
if p.Quote.IsFiatCurrency() &&
p.Quote != Bot.Config.Currency.FiatDisplayCurrency {
origCurrency := p.Quote.Upper()
log.Infof(log.OrderBook, "%s %s %s: ORDERBOOK: Bids len: %d Amount: %f %s. Total value: %s Asks len: %d Amount: %f %s. Total value: %s\n",
exchangeName,
FormatCurrency(p).String(),
strings.ToUpper(assetType.String()),
len(result.Bids),
bidsAmount,
p.Base.String(),
printConvertCurrencyFormat(origCurrency, bidsValue),
len(result.Asks),
asksAmount,
p.Base.String(),
printConvertCurrencyFormat(origCurrency, asksValue),
)
} else {
if p.Quote.IsFiatCurrency() &&
p.Quote == Bot.Config.Currency.FiatDisplayCurrency {
log.Infof(log.OrderBook, "%s %s %s: ORDERBOOK: Bids len: %d Amount: %f %s. Total value: %s Asks len: %d Amount: %f %s. Total value: %s\n",
exchangeName,
FormatCurrency(p).String(),
strings.ToUpper(assetType.String()),
len(result.Bids),
bidsAmount,
p.Base.String(),
printCurrencyFormat(bidsValue),
len(result.Asks),
asksAmount,
p.Base.String(),
printCurrencyFormat(asksValue),
)
} else {
log.Infof(log.OrderBook, "%s %s %s: ORDERBOOK: Bids len: %d Amount: %f %s. Total value: %f Asks len: %d Amount: %f %s. Total value: %f\n",
exchangeName,
FormatCurrency(p).String(),
strings.ToUpper(assetType.String()),
len(result.Bids),
bidsAmount,
p.Base.String(),
bidsValue,
len(result.Asks),
asksAmount,
p.Base.String(),
asksValue,
)
}
}
}
func relayWebsocketEvent(result interface{}, event, assetType, exchangeName string) {
evt := WebsocketEvent{
Data: result,
Event: event,
AssetType: assetType,
Exchange: exchangeName,
}
err := BroadcastWebsocketMessage(evt)
if err != nil {
log.Errorf(log.WebsocketMgr, "Failed to broadcast websocket event %v. Error: %s\n",
event, err)
}
}
// WebsocketRoutine Initial routine management system for websocket
func WebsocketRoutine() {
if Bot.Settings.Verbose {
log.Debugln(log.WebsocketMgr, "Connecting exchange websocket services...")
}
for i := range Bot.Exchanges {
go func(i int) {
if Bot.Exchanges[i].SupportsWebsocket() {
if Bot.Settings.Verbose {
log.Debugf(log.WebsocketMgr, "Exchange %s websocket support: Yes Enabled: %v\n", Bot.Exchanges[i].GetName(),
common.IsEnabled(Bot.Exchanges[i].IsWebsocketEnabled()))
}
// TO-DO: expose IsConnected() and IsConnecting so this can be simplified
if Bot.Exchanges[i].IsWebsocketEnabled() {
ws, err := Bot.Exchanges[i].GetWebsocket()
if err != nil {
log.Errorf(log.WebsocketMgr, "Exchange %s GetWebsocket error: %s\n",
Bot.Exchanges[i].GetName(), err)
return
}
// Exchange sync manager might have already started ws
// service or is in the process of connecting, so check
if ws.IsConnected() || ws.IsConnecting() {
return
}
// Data handler routine
go WebsocketDataHandler(ws)
err = ws.Connect()
if err != nil {
log.Errorf(log.WebsocketMgr, "%v\n", err)
}
}
} else if Bot.Settings.Verbose {
log.Debugf(log.WebsocketMgr, "Exchange %s websocket support: No\n", Bot.Exchanges[i].GetName())
}
}(i)
}
}
var shutdowner = make(chan struct{}, 1)
var wg sync.WaitGroup
// Websocketshutdown shuts down the exchange routines and then shuts down
// governing routines
func Websocketshutdown(ws *wshandler.Websocket) error {
err := ws.Shutdown() // shutdown routines on the exchange
if err != nil {
log.Errorf(log.WebsocketMgr, "routines.go error - failed to shutdown %s\n", err)
}
timer := time.NewTimer(5 * time.Second)
c := make(chan struct{}, 1)
go func(c chan struct{}) {
close(shutdowner)
wg.Wait()
c <- struct{}{}
}(c)
select {
case <-timer.C:
return errors.New("routines.go error - failed to shutdown routines")
case <-c:
return nil
}
}
// WebsocketDataHandler handles websocket data coming from a websocket feed
// associated with an exchange
func WebsocketDataHandler(ws *wshandler.Websocket) {
wg.Add(1)
defer wg.Done()
for {
select {
case <-shutdowner:
return
case data := <-ws.DataHandler:
switch d := data.(type) {
case string:
switch d {
case wshandler.WebsocketNotEnabled:
if Bot.Settings.Verbose {
log.Warnf(log.WebsocketMgr, "routines.go warning - exchange %s websocket not enabled\n",
ws.GetName())
}
default:
log.Info(log.WebsocketMgr, d)
}
case error:
log.Errorf(log.WebsocketMgr, "routines.go exchange %s websocket error - %s", ws.GetName(), data)
case wshandler.TradeData:
// Trade Data
// if Bot.Settings.Verbose {
// log.Println("Websocket trades Updated: ", data.(exchange.TradeData))
// }
case wshandler.TickerData:
// Ticker data
// if Bot.Settings.Verbose {
// log.Println("Websocket Ticker Updated: ", data.(exchange.TickerData))
// }
tickerNew := ticker.Price{
Last: d.Last,
High: d.High,
Low: d.Low,
Bid: d.Bid,
Ask: d.Ask,
Volume: d.Volume,
QuoteVolume: d.QuoteVolume,
PriceATH: d.PriceATH,
Open: d.Open,
Close: d.Close,
Pair: d.Pair,
LastUpdated: d.Timestamp,
}
if Bot.Settings.EnableExchangeSyncManager && Bot.ExchangeCurrencyPairManager != nil {
Bot.ExchangeCurrencyPairManager.update(ws.GetName(),
d.Pair, d.AssetType, SyncItemTicker, nil)
}
err := ticker.ProcessTicker(ws.GetName(), &tickerNew, d.AssetType)
if err != nil {
log.Errorf(log.WebsocketMgr, "routines.go exchange %s websocket error - %s", ws.GetName(), err)
}
printTickerSummary(&tickerNew, tickerNew.Pair, d.AssetType, ws.GetName(), nil)
case wshandler.KlineData:
// Kline data
if Bot.Settings.Verbose {
log.Infof(log.WebsocketMgr, "Websocket Kline Updated: %v\n", d)
}
case wshandler.WebsocketOrderbookUpdate:
// Orderbook data
result := data.(wshandler.WebsocketOrderbookUpdate)
if Bot.Settings.EnableExchangeSyncManager && Bot.ExchangeCurrencyPairManager != nil {
Bot.ExchangeCurrencyPairManager.update(ws.GetName(),
result.Pair, result.Asset, SyncItemOrderbook, nil)
}
// TO-DO: printOrderbookSummary
//nolint:gocritic
if Bot.Settings.Verbose {
log.Infof(log.WebsocketMgr, "Websocket %s %s orderbook updated\n", ws.GetName(), result.Pair.String())
}
default:
if Bot.Settings.Verbose {
log.Warnf(log.WebsocketMgr, "Websocket Unknown type: %s\n", d)
}
}
}
}
}