d0zingcat/gocryptotrader
1
0
Fork 0
mirror of https://github.com/d0zingcat/gocryptotrader.git synced 2026-05-31 15:10:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

futures: Implement GetLatestFundingRates across exchanges (#1339)

Browse Source 
* adds funding rate implementations and improvements

* merge fixes x1

* lint

* kucoin funding rates func make

* migrate sync-manager to keys

* some kucoin work

* adds some kucoin wrapper funcs

* ehhh, todo

* kucoin position

* start of orders

* adds the kucoin tests yay

* multiplier

* nits, EWS includes order limits

* NotYetImplemented, IsPerp improvements, cleaning

* lint, test fix, huobi time

* fixes issues, improves testing

* fixes linters I WRECKED

* local lint but remote lint, lint, lint, lint

* fixes err

* skip CI

* lint

* Supported rates, binance endpoints

* fixes weird mocktest problems

* no, CZ is invalid

* fixes some new EWS test errors
This commit is contained in:
Scott
2023-11-03 10:01:32 +10:00
committed by GitHub
parent f9437dbd08
commit 70690d9a04
78 changed files with 4088 additions and 527 deletions
Download Patch File Download Diff File Expand all files Collapse all files

154
engine/sync_manager.go
View File

@@ -11,6 +11,7 @@ import (
"time"
"github.com/thrasher-corp/gocryptotrader/common"
"github.com/thrasher-corp/gocryptotrader/common/key"
"github.com/thrasher-corp/gocryptotrader/config"
"github.com/thrasher-corp/gocryptotrader/currency"
exchange "github.com/thrasher-corp/gocryptotrader/exchanges"
@@ -39,8 +40,8 @@ var (
errCouldNotSyncNewData = errors.New("could not sync new data")
)
// setupSyncManager starts a new CurrencyPairSyncer
func setupSyncManager(c *config.SyncManagerConfig, exchangeManager iExchangeManager, remoteConfig *config.RemoteControlConfig, websocketRoutineManagerEnabled bool) (*syncManager, error) {
// SetupSyncManager creates a new CurrencyPairSyncer
func SetupSyncManager(c *config.SyncManagerConfig, exchangeManager iExchangeManager, remoteConfig *config.RemoteControlConfig, websocketRoutineManagerEnabled bool) (*SyncManager, error) {
if c == nil {
return nil, fmt.Errorf("%T %w", c, common.ErrNilPointer)
}
@@ -79,15 +80,15 @@ func setupSyncManager(c *config.SyncManagerConfig, exchangeManager iExchangeMana
return nil, fmt.Errorf("%T %w", c.PairFormatDisplay, common.ErrNilPointer)
}
s := &syncManager{
s := &SyncManager{
config: *c,
remoteConfig: remoteConfig,
exchangeManager: exchangeManager,
websocketRoutineManagerEnabled: websocketRoutineManagerEnabled,
fiatDisplayCurrency: c.FiatDisplayCurrency,
format: *c.PairFormatDisplay,
tickerBatchLastRequested: make(map[string]time.Time),
currencyPairs: make(map[currencyPairKey]*currencyPairSyncAgent),
tickerBatchLastRequested: make(map[key.ExchangeAsset]time.Time),
currencyPairs: make(map[key.ExchangePairAsset]*currencyPairSyncAgent),
}
log.Debugf(log.SyncMgr,
@@ -102,12 +103,12 @@ func setupSyncManager(c *config.SyncManagerConfig, exchangeManager iExchangeMana
}
// IsRunning safely checks whether the subsystem is running
func (m *syncManager) IsRunning() bool {
func (m *SyncManager) IsRunning() bool {
return m != nil && atomic.LoadInt32(&m.started) == 1
}
// Start runs the subsystem
func (m *syncManager) Start() error {
func (m *SyncManager) Start() error {
if m == nil {
return fmt.Errorf("exchange CurrencyPairSyncer %w", ErrNilSubsystem)
}
@@ -175,10 +176,11 @@ func (m *syncManager) Start() error {
continue
}
for i := range enabledPairs {
k := currencyPairKey{
AssetType: assetTypes[y],
Exchange: exchangeName,
Pair: enabledPairs[i].Format(currency.PairFormat{Uppercase: true}),
k := key.ExchangePairAsset{
Asset: assetTypes[y],
Exchange: exchangeName,
Base: enabledPairs[i].Base.Item,
Quote: enabledPairs[i].Quote.Item,
}
if e := m.get(k); e != nil {
continue
@@ -235,7 +237,7 @@ func (m *syncManager) Start() error {
}
// Stop shuts down the exchange currency pair syncer
func (m *syncManager) Stop() error {
func (m *SyncManager) Stop() error {
if m == nil {
return fmt.Errorf("exchange CurrencyPairSyncer %w", ErrNilSubsystem)
}
@@ -248,22 +250,23 @@ func (m *syncManager) Stop() error {
return nil
}
func (m *syncManager) get(k currencyPairKey) *currencyPairSyncAgent {
func (m *SyncManager) get(k key.ExchangePairAsset) *currencyPairSyncAgent {
m.mux.Lock()
defer m.mux.Unlock()
return m.currencyPairs[k]
}
func newCurrencyPairSyncAgent(k currencyPairKey) *currencyPairSyncAgent {
func newCurrencyPairSyncAgent(k key.ExchangePairAsset) *currencyPairSyncAgent {
return &currencyPairSyncAgent{
currencyPairKey: k,
Created: time.Now(),
locks: make([]sync.Mutex, SyncItemTrade+1),
trackers: make([]*syncBase, SyncItemTrade+1),
Key: k,
Pair: currency.NewPair(k.Base.Currency(), k.Quote.Currency()),
Created: time.Now(),
locks: make([]sync.Mutex, SyncItemTrade+1),
trackers: make([]*syncBase, SyncItemTrade+1),
}
}
func (m *syncManager) add(k currencyPairKey, s syncBase) *currencyPairSyncAgent {
func (m *SyncManager) add(k key.ExchangePairAsset, s syncBase) *currencyPairSyncAgent {
m.mux.Lock()
defer m.mux.Unlock()
@@ -288,7 +291,9 @@ func (m *syncManager) add(k currencyPairKey, s syncBase) *currencyPairSyncAgent
if m.config.Verbose {
log.Debugf(log.SyncMgr,
"%s: Added ticker sync item %v: using websocket: %v using REST: %v",
c.Exchange, m.FormatCurrency(c.Pair).String(), c.trackers[SyncItemTicker].IsUsingWebsocket,
c.Key.Exchange,
m.FormatCurrency(c.Pair),
c.trackers[SyncItemTicker].IsUsingWebsocket,
c.trackers[SyncItemTicker].IsUsingREST)
}
if atomic.LoadInt32(&m.initSyncCompleted) != 1 {
@@ -301,7 +306,9 @@ func (m *syncManager) add(k currencyPairKey, s syncBase) *currencyPairSyncAgent
if m.config.Verbose {
log.Debugf(log.SyncMgr,
"%s: Added orderbook sync item %v: using websocket: %v using REST: %v",
c.Exchange, m.FormatCurrency(c.Pair).String(), c.trackers[SyncItemOrderbook].IsUsingWebsocket,
k.Exchange,
m.FormatCurrency(c.Pair),
c.trackers[SyncItemOrderbook].IsUsingWebsocket,
c.trackers[SyncItemOrderbook].IsUsingREST)
}
if atomic.LoadInt32(&m.initSyncCompleted) != 1 {
@@ -314,7 +321,9 @@ func (m *syncManager) add(k currencyPairKey, s syncBase) *currencyPairSyncAgent
if m.config.Verbose {
log.Debugf(log.SyncMgr,
"%s: Added trade sync item %v: using websocket: %v using REST: %v",
c.Exchange, m.FormatCurrency(c.Pair).String(), c.trackers[SyncItemTrade].IsUsingWebsocket,
k.Exchange,
m.FormatCurrency(c.Pair),
c.trackers[SyncItemTrade].IsUsingWebsocket,
c.trackers[SyncItemTrade].IsUsingREST)
}
if atomic.LoadInt32(&m.initSyncCompleted) != 1 {
@@ -324,7 +333,7 @@ func (m *syncManager) add(k currencyPairKey, s syncBase) *currencyPairSyncAgent
}
if m.currencyPairs == nil {
m.currencyPairs = make(map[currencyPairKey]*currencyPairSyncAgent)
m.currencyPairs = make(map[key.ExchangePairAsset]*currencyPairSyncAgent)
}
m.currencyPairs[k] = c
@@ -332,9 +341,9 @@ func (m *syncManager) add(k currencyPairKey, s syncBase) *currencyPairSyncAgent
return c
}
// WebsocketUpdate notifies the syncManager to change the last updated time for a exchange asset pair
// WebsocketUpdate notifies the SyncManager to change the last updated time for a exchange asset pair
// And set IsUsingWebsocket to true. It should be used externally only from websocket updaters
func (m *syncManager) WebsocketUpdate(exchangeName string, p currency.Pair, a asset.Item, syncType syncItemType, err error) error {
func (m *SyncManager) WebsocketUpdate(exchangeName string, p currency.Pair, a asset.Item, syncType syncItemType, err error) error {
if m == nil {
return fmt.Errorf("exchange CurrencyPairSyncer %w", ErrNilSubsystem)
}
@@ -362,15 +371,22 @@ func (m *syncManager) WebsocketUpdate(exchangeName string, p currency.Pair, a as
return fmt.Errorf("%v %w", syncType, errUnknownSyncItem)
}
k := currencyPairKey{
AssetType: a,
Exchange: exchangeName,
Pair: p.Format(currency.PairFormat{Uppercase: true}),
k := key.ExchangePairAsset{
Asset: a,
Exchange: exchangeName,
Base: p.Base.Item,
Quote: p.Quote.Item,
}
c, exists := m.currencyPairs[k]
if !exists {
return fmt.Errorf("%w for %s %s %s %s", errCouldNotSyncNewData, k.Exchange, k.Pair, k.AssetType, syncType)
return fmt.Errorf("%w for %s %s %s %s %s",
errCouldNotSyncNewData,
k.Exchange,
k.Base,
k.Quote,
k.Asset,
syncType)
}
c.locks[syncType].Lock()
@@ -387,9 +403,9 @@ func (m *syncManager) WebsocketUpdate(exchangeName string, p currency.Pair, a as
if m.config.LogSwitchProtocolEvents {
log.Warnf(log.SyncMgr,
"%s %s %s: %s Websocket re-enabled, switching from rest to websocket",
c.Exchange,
k.Exchange,
m.FormatCurrency(c.Pair),
strings.ToUpper(c.AssetType.String()),
strings.ToUpper(k.Asset.String()),
syncType,
)
}
@@ -398,8 +414,8 @@ func (m *syncManager) WebsocketUpdate(exchangeName string, p currency.Pair, a as
return m.update(c, syncType, err)
}
// update notifies the syncManager to change the last updated time for a exchange asset pair
func (m *syncManager) update(c *currencyPairSyncAgent, syncType syncItemType, err error) error {
// update notifies the SyncManager to change the last updated time for a exchange asset pair
func (m *SyncManager) update(c *currencyPairSyncAgent, syncType syncItemType, err error) error {
if syncType < SyncItemTicker || syncType > SyncItemTrade {
return fmt.Errorf("%v %w", syncType, errUnknownSyncItem)
}
@@ -416,7 +432,7 @@ func (m *syncManager) update(c *currencyPairSyncAgent, syncType syncItemType, er
removedCounter++
if m.config.LogInitialSyncEvents {
log.Debugf(log.SyncMgr, "%s %s sync complete %v [%d/%d].",
c.Exchange,
c.Key.Exchange,
syncType,
m.FormatCurrency(c.Pair),
removedCounter,
@@ -428,7 +444,7 @@ func (m *syncManager) update(c *currencyPairSyncAgent, syncType syncItemType, er
return nil
}
func (m *syncManager) worker() {
func (m *SyncManager) worker() {
cleanup := func() {
log.Debugln(log.SyncMgr,
"Exchange CurrencyPairSyncer worker shutting down.")
@@ -492,10 +508,11 @@ func (m *syncManager) worker() {
return
}
k := currencyPairKey{
AssetType: assetTypes[y],
Exchange: exchangeName,
Pair: enabledPairs[i].Format(currency.PairFormat{Uppercase: true}),
k := key.ExchangePairAsset{
Asset: assetTypes[y],
Exchange: exchangeName,
Base: enabledPairs[i].Base.Item,
Quote: enabledPairs[i].Quote.Item,
}
c := m.get(k)
if c == nil {
@@ -521,7 +538,7 @@ func (m *syncManager) worker() {
}
}
func (m *syncManager) syncTicker(c *currencyPairSyncAgent, e exchange.IBotExchange) {
func (m *SyncManager) syncTicker(c *currencyPairSyncAgent, e exchange.IBotExchange) {
if !c.locks[SyncItemTicker].TryLock() {
return
}
@@ -541,9 +558,9 @@ func (m *syncManager) syncTicker(c *currencyPairSyncAgent, e exchange.IBotExchan
if m.config.LogSwitchProtocolEvents {
log.Warnf(log.SyncMgr,
"%s %s %s: No ticker update after %s, switching from websocket to rest",
c.Exchange,
c.Key.Exchange,
m.FormatCurrency(c.Pair),
strings.ToUpper(c.AssetType.String()),
strings.ToUpper(c.Key.Asset.String()),
m.config.TimeoutWebsocket,
)
}
@@ -555,9 +572,15 @@ func (m *syncManager) syncTicker(c *currencyPairSyncAgent, e exchange.IBotExchan
if e.SupportsRESTTickerBatchUpdates() {
m.mux.Lock()
batchLastDone, ok := m.tickerBatchLastRequested[e.GetName()]
batchLastDone, ok := m.tickerBatchLastRequested[key.ExchangeAsset{
Exchange: c.Key.Exchange,
Asset: c.Key.Asset,
}]
if !ok {
m.tickerBatchLastRequested[exchangeName] = time.Time{}
m.tickerBatchLastRequested[key.ExchangeAsset{
Exchange: c.Key.Exchange,
Asset: c.Key.Asset,
}] = time.Time{}
}
m.mux.Unlock()
@@ -566,11 +589,14 @@ func (m *syncManager) syncTicker(c *currencyPairSyncAgent, e exchange.IBotExchan
if m.config.Verbose {
log.Debugf(log.SyncMgr, "Initialising %s REST ticker batching", exchangeName)
}
err = e.UpdateTickers(context.TODO(), c.AssetType)
err = e.UpdateTickers(context.TODO(), c.Key.Asset)
if err == nil {
result, err = e.FetchTicker(context.TODO(), c.Pair, c.AssetType)
result, err = e.FetchTicker(context.TODO(), c.Pair, c.Key.Asset)
}
m.tickerBatchLastRequested[exchangeName] = time.Now()
m.tickerBatchLastRequested[key.ExchangeAsset{
Exchange: c.Key.Exchange,
Asset: c.Key.Asset,
}] = time.Now()
m.mux.Unlock()
} else {
if m.config.Verbose {
@@ -578,17 +604,17 @@ func (m *syncManager) syncTicker(c *currencyPairSyncAgent, e exchange.IBotExchan
}
result, err = e.FetchTicker(context.TODO(),
c.Pair,
c.AssetType)
c.Key.Asset)
}
} else {
result, err = e.UpdateTicker(context.TODO(),
c.Pair,
c.AssetType)
c.Key.Asset)
}
m.PrintTickerSummary(result, "REST", err)
if err == nil {
if m.remoteConfig.WebsocketRPC.Enabled {
relayWebsocketEvent(result, "ticker_update", c.AssetType.String(), exchangeName)
relayWebsocketEvent(result, "ticker_update", c.Key.Asset.String(), exchangeName)
}
}
updateErr := m.update(c, SyncItemTicker, err)
@@ -598,7 +624,7 @@ func (m *syncManager) syncTicker(c *currencyPairSyncAgent, e exchange.IBotExchan
}
}
func (m *syncManager) syncOrderbook(c *currencyPairSyncAgent, e exchange.IBotExchange) {
func (m *SyncManager) syncOrderbook(c *currencyPairSyncAgent, e exchange.IBotExchange) {
if !c.locks[SyncItemOrderbook].TryLock() {
return
}
@@ -622,9 +648,9 @@ func (m *syncManager) syncOrderbook(c *currencyPairSyncAgent, e exchange.IBotExc
if m.config.LogSwitchProtocolEvents {
log.Warnf(log.SyncMgr,
"%s %s %s: No orderbook update after %s, switching from websocket to rest",
c.Exchange,
m.FormatCurrency(c.Pair).String(),
strings.ToUpper(c.AssetType.String()),
c.Key.Exchange,
m.FormatCurrency(c.Pair),
strings.ToUpper(c.Key.Asset.String()),
m.config.TimeoutWebsocket,
)
}
@@ -633,11 +659,11 @@ func (m *syncManager) syncOrderbook(c *currencyPairSyncAgent, e exchange.IBotExc
if s.IsUsingREST && time.Since(s.LastUpdated) > m.config.TimeoutREST {
result, err := e.UpdateOrderbook(context.TODO(),
c.Pair,
c.AssetType)
c.Key.Asset)
m.PrintOrderbookSummary(result, "REST", err)
if err == nil {
if m.remoteConfig.WebsocketRPC.Enabled {
relayWebsocketEvent(result, "orderbook_update", c.AssetType.String(), e.GetName())
relayWebsocketEvent(result, "orderbook_update", c.Key.Asset.String(), e.GetName())
}
}
updateErr := m.update(c, SyncItemOrderbook, err)
@@ -647,7 +673,7 @@ func (m *syncManager) syncOrderbook(c *currencyPairSyncAgent, e exchange.IBotExc
}
}
func (m *syncManager) syncTrades(c *currencyPairSyncAgent) {
func (m *SyncManager) syncTrades(c *currencyPairSyncAgent) {
if !c.locks[SyncItemTrade].TryLock() {
return
}
@@ -703,7 +729,7 @@ func printConvertCurrencyFormat(origPrice float64, origCurrency, displayCurrency
}
// PrintTickerSummary outputs the ticker results
func (m *syncManager) PrintTickerSummary(result *ticker.Price, protocol string, err error) {
func (m *SyncManager) PrintTickerSummary(result *ticker.Price, protocol string, err error) {
if m == nil || atomic.LoadInt32(&m.started) == 0 {
return
}
@@ -777,11 +803,11 @@ func (m *syncManager) PrintTickerSummary(result *ticker.Price, protocol string,
// FormatCurrency is a method that formats and returns a currency pair
// based on the user currency display preferences
func (m *syncManager) FormatCurrency(p currency.Pair) currency.Pair {
func (m *SyncManager) FormatCurrency(cp currency.Pair) string {
if m == nil || atomic.LoadInt32(&m.started) == 0 {
return p
return ""
}
return p.Format(m.format)
return m.format.Format(cp)
}
const (
@@ -789,7 +815,7 @@ const (
)
// PrintOrderbookSummary outputs orderbook results
func (m *syncManager) PrintOrderbookSummary(result *orderbook.Base, protocol string, err error) {
func (m *SyncManager) PrintOrderbookSummary(result *orderbook.Base, protocol string, err error) {
if m == nil || atomic.LoadInt32(&m.started) == 0 {
return
}
@@ -863,7 +889,7 @@ func (m *syncManager) PrintOrderbookSummary(result *orderbook.Base, protocol str
// WaitForInitialSync allows for a routine to wait for an initial sync to be
// completed without exposing the underlying type. This needs to be called in a
// separate routine.
func (m *syncManager) WaitForInitialSync() error {
func (m *SyncManager) WaitForInitialSync() error {
if m == nil {
return fmt.Errorf("sync manager %w", ErrNilSubsystem)
}