mirror of
https://github.com/NoFxAiOS/nofx.git
synced 2026-07-07 04:50:57 +08:00
fix(trader): stop order-sync goroutine leak and rate-limit hammering
Every StartOrderSync spawned a ticker goroutine that ran forever — it survived trader stop AND deletion, so each quick-created trader left a permanent 30s Hyperliquid poll behind. Stacked leaks turned into an ~8s effective hammer that tripped Hyperliquid's 429 rate limit, which then broke the symbol board, trader creation, and order sync itself. - new trader/syncloop package: shared stoppable sync loop with exponential failure backoff (30s base, 5min cap) - all 9 exchanges' StartOrderSync now take the trader's stop channel and stop when the trader stops (close broadcast from AutoTrader.Stop) - provider/hyperliquid: GetPerpDexCoins now serves a 5min TTL cache and falls back to the stale board when the upstream returns 429, so the symbol panel keeps working through rate limiting
This commit is contained in:
@@ -164,9 +164,59 @@ func fetchPerpDexCoins(ctx context.Context, client *http.Client, dex string) ([]
|
||||
return coins, nil
|
||||
}
|
||||
|
||||
// GetPerpDexCoins fetches current tradable USDC perp assets for a given Hyperliquid dex.
|
||||
// perpDexCacheTTL bounds how often the perp-dex symbol board is re-fetched.
|
||||
// The tradable symbol list changes rarely; prices/volume on the board are
|
||||
// display hints, so short staleness is far better than hammering the
|
||||
// Hyperliquid API (which rate-limits with 429) on every panel render.
|
||||
const perpDexCacheTTL = 5 * time.Minute
|
||||
|
||||
type perpDexCacheEntry struct {
|
||||
coins []CoinInfo
|
||||
fetchedAt time.Time
|
||||
}
|
||||
|
||||
type perpDexCacheStore struct {
|
||||
mu sync.Mutex
|
||||
entries map[string]perpDexCacheEntry
|
||||
}
|
||||
|
||||
var perpDexCoinCache = &perpDexCacheStore{entries: map[string]perpDexCacheEntry{}}
|
||||
|
||||
// fetchPerpDexCoinsFn is swappable in tests.
|
||||
var fetchPerpDexCoinsFn = fetchPerpDexCoins
|
||||
|
||||
// GetPerpDexCoins returns current tradable USDC perp assets for a given
|
||||
// Hyperliquid dex, served from a TTL cache. When the upstream fetch fails
|
||||
// (e.g. HTTP 429 rate limiting) and stale data exists, the stale board is
|
||||
// served instead of an error so the UI keeps working.
|
||||
func GetPerpDexCoins(ctx context.Context, dex string) ([]CoinInfo, error) {
|
||||
return fetchPerpDexCoins(ctx, &http.Client{Timeout: 30 * time.Second}, dex)
|
||||
perpDexCoinCache.mu.Lock()
|
||||
defer perpDexCoinCache.mu.Unlock()
|
||||
|
||||
entry, hasCache := perpDexCoinCache.entries[dex]
|
||||
if hasCache && time.Since(entry.fetchedAt) < perpDexCacheTTL {
|
||||
return copyCoins(entry.coins), nil
|
||||
}
|
||||
|
||||
coins, err := fetchPerpDexCoinsFn(ctx, &http.Client{Timeout: 30 * time.Second}, dex)
|
||||
if err != nil {
|
||||
if hasCache {
|
||||
logger.Infof("⚠️ Hyperliquid perp-dex fetch failed (%v); serving cached board for dex %q from %s",
|
||||
err, dex, entry.fetchedAt.Format(time.RFC3339))
|
||||
return copyCoins(entry.coins), nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
perpDexCoinCache.entries[dex] = perpDexCacheEntry{coins: coins, fetchedAt: time.Now()}
|
||||
return copyCoins(coins), nil
|
||||
}
|
||||
|
||||
// copyCoins returns a defensive copy so callers cannot mutate the cache.
|
||||
func copyCoins(coins []CoinInfo) []CoinInfo {
|
||||
out := make([]CoinInfo, len(coins))
|
||||
copy(out, coins)
|
||||
return out
|
||||
}
|
||||
|
||||
// fetchCoins fetches all default Hyperliquid crypto coins and sorts by volume
|
||||
|
||||
113
provider/hyperliquid/perp_dex_cache_test.go
Normal file
113
provider/hyperliquid/perp_dex_cache_test.go
Normal file
@@ -0,0 +1,113 @@
|
||||
package hyperliquid
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// withStubbedPerpDexFetch swaps the live fetch function and resets the cache,
|
||||
// restoring both when the test finishes.
|
||||
func withStubbedPerpDexFetch(t *testing.T, fn func(ctx context.Context, client *http.Client, dex string) ([]CoinInfo, error)) {
|
||||
t.Helper()
|
||||
original := fetchPerpDexCoinsFn
|
||||
fetchPerpDexCoinsFn = fn
|
||||
perpDexCoinCache.mu.Lock()
|
||||
perpDexCoinCache.entries = map[string]perpDexCacheEntry{}
|
||||
perpDexCoinCache.mu.Unlock()
|
||||
t.Cleanup(func() {
|
||||
fetchPerpDexCoinsFn = original
|
||||
perpDexCoinCache.mu.Lock()
|
||||
perpDexCoinCache.entries = map[string]perpDexCacheEntry{}
|
||||
perpDexCoinCache.mu.Unlock()
|
||||
})
|
||||
}
|
||||
|
||||
func TestGetPerpDexCoinsCachesWithinTTL(t *testing.T) {
|
||||
calls := 0
|
||||
withStubbedPerpDexFetch(t, func(ctx context.Context, client *http.Client, dex string) ([]CoinInfo, error) {
|
||||
calls++
|
||||
return []CoinInfo{{Symbol: "xyz:TSLA", MarkPrice: 400}}, nil
|
||||
})
|
||||
|
||||
first, err := GetPerpDexCoins(context.Background(), "xyz")
|
||||
if err != nil {
|
||||
t.Fatalf("first call: %v", err)
|
||||
}
|
||||
second, err := GetPerpDexCoins(context.Background(), "xyz")
|
||||
if err != nil {
|
||||
t.Fatalf("second call: %v", err)
|
||||
}
|
||||
if calls != 1 {
|
||||
t.Fatalf("fetch calls = %d, want 1 (second call must hit cache)", calls)
|
||||
}
|
||||
if len(first) != 1 || len(second) != 1 || second[0].Symbol != "xyz:TSLA" {
|
||||
t.Fatalf("unexpected results: first=%v second=%v", first, second)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetPerpDexCoinsServesStaleOnUpstreamError(t *testing.T) {
|
||||
calls := 0
|
||||
withStubbedPerpDexFetch(t, func(ctx context.Context, client *http.Client, dex string) ([]CoinInfo, error) {
|
||||
calls++
|
||||
if calls == 1 {
|
||||
return []CoinInfo{{Symbol: "xyz:NVDA", MarkPrice: 1000}}, nil
|
||||
}
|
||||
return nil, errors.New("API returned status 429")
|
||||
})
|
||||
|
||||
if _, err := GetPerpDexCoins(context.Background(), "xyz"); err != nil {
|
||||
t.Fatalf("first call: %v", err)
|
||||
}
|
||||
|
||||
// Expire the cache so the next call must attempt a refresh.
|
||||
perpDexCoinCache.mu.Lock()
|
||||
entry := perpDexCoinCache.entries["xyz"]
|
||||
entry.fetchedAt = time.Now().Add(-2 * perpDexCacheTTL)
|
||||
perpDexCoinCache.entries["xyz"] = entry
|
||||
perpDexCoinCache.mu.Unlock()
|
||||
|
||||
coins, err := GetPerpDexCoins(context.Background(), "xyz")
|
||||
if err != nil {
|
||||
t.Fatalf("expected stale data instead of error, got: %v", err)
|
||||
}
|
||||
if len(coins) != 1 || coins[0].Symbol != "xyz:NVDA" {
|
||||
t.Fatalf("expected stale NVDA entry, got %v", coins)
|
||||
}
|
||||
if calls != 2 {
|
||||
t.Fatalf("fetch calls = %d, want 2 (refresh attempted)", calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetPerpDexCoinsErrorsWithoutAnyCache(t *testing.T) {
|
||||
withStubbedPerpDexFetch(t, func(ctx context.Context, client *http.Client, dex string) ([]CoinInfo, error) {
|
||||
return nil, errors.New("API returned status 429")
|
||||
})
|
||||
|
||||
if _, err := GetPerpDexCoins(context.Background(), "xyz"); err == nil {
|
||||
t.Fatal("expected error when upstream fails and no cache exists")
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetPerpDexCoinsCachesPerDex(t *testing.T) {
|
||||
withStubbedPerpDexFetch(t, func(ctx context.Context, client *http.Client, dex string) ([]CoinInfo, error) {
|
||||
if dex == "xyz" {
|
||||
return []CoinInfo{{Symbol: "xyz:AAPL"}}, nil
|
||||
}
|
||||
return []CoinInfo{{Symbol: "BTC"}}, nil
|
||||
})
|
||||
|
||||
xyz, err := GetPerpDexCoins(context.Background(), "xyz")
|
||||
if err != nil {
|
||||
t.Fatalf("xyz: %v", err)
|
||||
}
|
||||
def, err := GetPerpDexCoins(context.Background(), "")
|
||||
if err != nil {
|
||||
t.Fatalf("default dex: %v", err)
|
||||
}
|
||||
if xyz[0].Symbol != "xyz:AAPL" || def[0].Symbol != "BTC" {
|
||||
t.Fatalf("cache keys collided: xyz=%v default=%v", xyz, def)
|
||||
}
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"nofx/logger"
|
||||
"nofx/market"
|
||||
"nofx/store"
|
||||
"nofx/trader/syncloop"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -180,14 +181,8 @@ func deriveAsterOrderAction(side, positionSide string, realizedPnL float64) stri
|
||||
}
|
||||
|
||||
// StartOrderSync starts background order sync task for Aster
|
||||
func (t *AsterTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
|
||||
ticker := time.NewTicker(interval)
|
||||
go func() {
|
||||
for range ticker.C {
|
||||
if err := t.SyncOrdersFromAster(traderID, exchangeID, exchangeType, st); err != nil {
|
||||
logger.Infof("⚠️ Aster order sync failed: %v", err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
logger.Infof("🔄 Aster order sync started (interval: %v)", interval)
|
||||
func (t *AsterTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) {
|
||||
syncloop.Run(stop, interval, "Aster", func() error {
|
||||
return t.SyncOrdersFromAster(traderID, exchangeID, exchangeType, st)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -421,7 +421,7 @@ func (at *AutoTrader) Run() error {
|
||||
// Start Lighter order sync if using Lighter exchange
|
||||
if at.exchange == "lighter" {
|
||||
if lighterTrader, ok := at.trader.(*lighter.LighterTraderV2); ok && at.store != nil {
|
||||
lighterTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
|
||||
lighterTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh)
|
||||
at.logInfof("🔄 Lighter order+position sync enabled (every 30s)")
|
||||
}
|
||||
}
|
||||
@@ -429,7 +429,7 @@ func (at *AutoTrader) Run() error {
|
||||
// Start Hyperliquid order sync if using Hyperliquid exchange
|
||||
if at.exchange == "hyperliquid" {
|
||||
if hyperliquidTrader, ok := at.trader.(*hyperliquid.HyperliquidTrader); ok && at.store != nil {
|
||||
hyperliquidTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
|
||||
hyperliquidTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh)
|
||||
at.logInfof("🔄 Hyperliquid order+position sync enabled (every 30s)")
|
||||
}
|
||||
}
|
||||
@@ -437,7 +437,7 @@ func (at *AutoTrader) Run() error {
|
||||
// Start Bybit order sync if using Bybit exchange
|
||||
if at.exchange == "bybit" {
|
||||
if bybitTrader, ok := at.trader.(*bybit.BybitTrader); ok && at.store != nil {
|
||||
bybitTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
|
||||
bybitTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh)
|
||||
at.logInfof("🔄 Bybit order+position sync enabled (every 30s)")
|
||||
}
|
||||
}
|
||||
@@ -445,7 +445,7 @@ func (at *AutoTrader) Run() error {
|
||||
// Start OKX order sync if using OKX exchange
|
||||
if at.exchange == "okx" {
|
||||
if okxTrader, ok := at.trader.(*okx.OKXTrader); ok && at.store != nil {
|
||||
okxTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
|
||||
okxTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh)
|
||||
at.logInfof("🔄 OKX order+position sync enabled (every 30s)")
|
||||
}
|
||||
}
|
||||
@@ -453,7 +453,7 @@ func (at *AutoTrader) Run() error {
|
||||
// Start Bitget order sync if using Bitget exchange
|
||||
if at.exchange == "bitget" {
|
||||
if bitgetTrader, ok := at.trader.(*bitget.BitgetTrader); ok && at.store != nil {
|
||||
bitgetTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
|
||||
bitgetTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh)
|
||||
at.logInfof("🔄 Bitget order+position sync enabled (every 30s)")
|
||||
}
|
||||
}
|
||||
@@ -461,7 +461,7 @@ func (at *AutoTrader) Run() error {
|
||||
// Start Aster order sync if using Aster exchange
|
||||
if at.exchange == "aster" {
|
||||
if asterTrader, ok := at.trader.(*aster.AsterTrader); ok && at.store != nil {
|
||||
asterTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
|
||||
asterTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh)
|
||||
at.logInfof("🔄 Aster order+position sync enabled (every 30s)")
|
||||
}
|
||||
}
|
||||
@@ -469,7 +469,7 @@ func (at *AutoTrader) Run() error {
|
||||
// Start Binance order sync if using Binance exchange
|
||||
if at.exchange == "binance" {
|
||||
if binanceTrader, ok := at.trader.(*binance.FuturesTrader); ok && at.store != nil {
|
||||
binanceTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
|
||||
binanceTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh)
|
||||
at.logInfof("🔄 Binance order+position sync enabled (every 30s)")
|
||||
}
|
||||
}
|
||||
@@ -477,7 +477,7 @@ func (at *AutoTrader) Run() error {
|
||||
// Start Gate order sync if using Gate exchange
|
||||
if at.exchange == "gate" {
|
||||
if gateTrader, ok := at.trader.(*gate.GateTrader); ok && at.store != nil {
|
||||
gateTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
|
||||
gateTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh)
|
||||
at.logInfof("🔄 Gate order+position sync enabled (every 30s)")
|
||||
}
|
||||
}
|
||||
@@ -485,7 +485,7 @@ func (at *AutoTrader) Run() error {
|
||||
// Start KuCoin order sync if using KuCoin exchange
|
||||
if at.exchange == "kucoin" {
|
||||
if kucoinTrader, ok := at.trader.(*kucoin.KuCoinTrader); ok && at.store != nil {
|
||||
kucoinTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
|
||||
kucoinTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second, at.stopMonitorCh)
|
||||
at.logInfof("🔄 KuCoin order+position sync enabled (every 30s)")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"nofx/logger"
|
||||
"nofx/market"
|
||||
"nofx/store"
|
||||
"nofx/trader/syncloop"
|
||||
"nofx/trader/types"
|
||||
"sort"
|
||||
"strings"
|
||||
@@ -349,7 +350,7 @@ func (t *FuturesTrader) determineOrderAction(side, positionSide string, realized
|
||||
}
|
||||
|
||||
// StartOrderSync starts background order sync task for Binance
|
||||
func (t *FuturesTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
|
||||
func (t *FuturesTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) {
|
||||
// Run first sync immediately
|
||||
go func() {
|
||||
logger.Infof("🔄 Running initial Binance order sync...")
|
||||
@@ -359,13 +360,7 @@ func (t *FuturesTrader) StartOrderSync(traderID string, exchangeID string, excha
|
||||
}()
|
||||
|
||||
// Then run periodically
|
||||
ticker := time.NewTicker(interval)
|
||||
go func() {
|
||||
for range ticker.C {
|
||||
if err := t.SyncOrdersFromBinance(traderID, exchangeID, exchangeType, st); err != nil {
|
||||
logger.Infof("⚠️ Binance order sync failed: %v", err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
logger.Infof("🔄 Binance order sync started (interval: %v)", interval)
|
||||
syncloop.Run(stop, interval, "Binance", func() error {
|
||||
return t.SyncOrdersFromBinance(traderID, exchangeID, exchangeType, st)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"nofx/logger"
|
||||
"nofx/market"
|
||||
"nofx/store"
|
||||
"nofx/trader/syncloop"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -48,7 +49,6 @@ func (t *BitgetTrader) GetTrades(startTime time.Time, limit int) ([]BitgetTrade,
|
||||
return nil, fmt.Errorf("failed to get fill history: %w", err)
|
||||
}
|
||||
|
||||
|
||||
// Bitget fill structure - supports both one-way and hedge mode
|
||||
type BitgetFill struct {
|
||||
TradeID string `json:"tradeId"`
|
||||
@@ -279,14 +279,8 @@ func (t *BitgetTrader) SyncOrdersFromBitget(traderID string, exchangeID string,
|
||||
}
|
||||
|
||||
// StartOrderSync starts background order sync task for Bitget
|
||||
func (t *BitgetTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
|
||||
ticker := time.NewTicker(interval)
|
||||
go func() {
|
||||
for range ticker.C {
|
||||
if err := t.SyncOrdersFromBitget(traderID, exchangeID, exchangeType, st); err != nil {
|
||||
logger.Infof("⚠️ Bitget order sync failed: %v", err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
logger.Infof("🔄 Bitget order sync started (interval: %v)", interval)
|
||||
func (t *BitgetTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) {
|
||||
syncloop.Run(stop, interval, "Bitget", func() error {
|
||||
return t.SyncOrdersFromBitget(traderID, exchangeID, exchangeType, st)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"nofx/logger"
|
||||
"nofx/market"
|
||||
"nofx/store"
|
||||
"nofx/trader/syncloop"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -298,14 +299,8 @@ func (t *BybitTrader) SyncOrdersFromBybit(traderID string, exchangeID string, ex
|
||||
}
|
||||
|
||||
// StartOrderSync starts background order sync task for Bybit
|
||||
func (t *BybitTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
|
||||
ticker := time.NewTicker(interval)
|
||||
go func() {
|
||||
for range ticker.C {
|
||||
if err := t.SyncOrdersFromBybit(traderID, exchangeID, exchangeType, st); err != nil {
|
||||
logger.Infof("⚠️ Bybit order sync failed: %v", err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
logger.Infof("🔄 Bybit order sync started (interval: %v)", interval)
|
||||
func (t *BybitTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) {
|
||||
syncloop.Run(stop, interval, "Bybit", func() error {
|
||||
return t.SyncOrdersFromBybit(traderID, exchangeID, exchangeType, st)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"nofx/logger"
|
||||
"nofx/market"
|
||||
"nofx/store"
|
||||
"nofx/trader/syncloop"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -291,14 +292,8 @@ func (t *GateTrader) SyncOrdersFromGate(traderID string, exchangeID string, exch
|
||||
}
|
||||
|
||||
// StartOrderSync starts background order sync task for Gate
|
||||
func (t *GateTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
|
||||
ticker := time.NewTicker(interval)
|
||||
go func() {
|
||||
for range ticker.C {
|
||||
if err := t.SyncOrdersFromGate(traderID, exchangeID, exchangeType, st); err != nil {
|
||||
logger.Infof("⚠️ Gate order sync failed: %v", err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
logger.Infof("🔄 Gate order sync started (interval: %v)", interval)
|
||||
func (t *GateTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) {
|
||||
syncloop.Run(stop, interval, "Gate", func() error {
|
||||
return t.SyncOrdersFromGate(traderID, exchangeID, exchangeType, st)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"nofx/logger"
|
||||
"nofx/market"
|
||||
"nofx/store"
|
||||
"nofx/trader/syncloop"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -44,87 +45,87 @@ func (t *HyperliquidTrader) SyncOrdersFromHyperliquid(traderID string, exchangeI
|
||||
syncedCount := 0
|
||||
|
||||
for _, trade := range trades {
|
||||
// Check if trade already exists (use exchangeID which is UUID, not exchange type)
|
||||
existing, err := orderStore.GetOrderByExchangeID(exchangeID, trade.TradeID)
|
||||
if err == nil && existing != nil {
|
||||
continue // Order already exists, skip
|
||||
}
|
||||
// Check if trade already exists (use exchangeID which is UUID, not exchange type)
|
||||
existing, err := orderStore.GetOrderByExchangeID(exchangeID, trade.TradeID)
|
||||
if err == nil && existing != nil {
|
||||
continue // Order already exists, skip
|
||||
}
|
||||
|
||||
// Normalize symbol
|
||||
symbol := market.Normalize(trade.Symbol)
|
||||
// Normalize symbol
|
||||
symbol := market.Normalize(trade.Symbol)
|
||||
|
||||
// Use order action from trade (parsed from Hyperliquid Dir field)
|
||||
// Dir field values: "Open Long", "Open Short", "Close Long", "Close Short"
|
||||
orderAction := trade.OrderAction
|
||||
positionSide := "LONG"
|
||||
if strings.Contains(orderAction, "short") {
|
||||
positionSide = "SHORT"
|
||||
}
|
||||
// Use order action from trade (parsed from Hyperliquid Dir field)
|
||||
// Dir field values: "Open Long", "Open Short", "Close Long", "Close Short"
|
||||
orderAction := trade.OrderAction
|
||||
positionSide := "LONG"
|
||||
if strings.Contains(orderAction, "short") {
|
||||
positionSide = "SHORT"
|
||||
}
|
||||
|
||||
// Create order record - use Unix milliseconds UTC
|
||||
tradeTimeMs := trade.Time.UTC().UnixMilli()
|
||||
orderRecord := &store.TraderOrder{
|
||||
TraderID: traderID,
|
||||
ExchangeID: exchangeID, // UUID
|
||||
ExchangeType: exchangeType, // Exchange type
|
||||
ExchangeOrderID: trade.TradeID,
|
||||
Symbol: symbol,
|
||||
Side: trade.Side,
|
||||
PositionSide: "BOTH", // Hyperliquid uses one-way position mode
|
||||
Type: "MARKET",
|
||||
OrderAction: orderAction,
|
||||
Quantity: trade.Quantity,
|
||||
Price: trade.Price,
|
||||
Status: "FILLED",
|
||||
FilledQuantity: trade.Quantity,
|
||||
AvgFillPrice: trade.Price,
|
||||
Commission: trade.Fee,
|
||||
FilledAt: tradeTimeMs,
|
||||
CreatedAt: tradeTimeMs,
|
||||
UpdatedAt: tradeTimeMs,
|
||||
}
|
||||
// Create order record - use Unix milliseconds UTC
|
||||
tradeTimeMs := trade.Time.UTC().UnixMilli()
|
||||
orderRecord := &store.TraderOrder{
|
||||
TraderID: traderID,
|
||||
ExchangeID: exchangeID, // UUID
|
||||
ExchangeType: exchangeType, // Exchange type
|
||||
ExchangeOrderID: trade.TradeID,
|
||||
Symbol: symbol,
|
||||
Side: trade.Side,
|
||||
PositionSide: "BOTH", // Hyperliquid uses one-way position mode
|
||||
Type: "MARKET",
|
||||
OrderAction: orderAction,
|
||||
Quantity: trade.Quantity,
|
||||
Price: trade.Price,
|
||||
Status: "FILLED",
|
||||
FilledQuantity: trade.Quantity,
|
||||
AvgFillPrice: trade.Price,
|
||||
Commission: trade.Fee,
|
||||
FilledAt: tradeTimeMs,
|
||||
CreatedAt: tradeTimeMs,
|
||||
UpdatedAt: tradeTimeMs,
|
||||
}
|
||||
|
||||
// Insert order record
|
||||
if err := orderStore.CreateOrder(orderRecord); err != nil {
|
||||
logger.Infof(" ⚠️ Failed to sync trade %s: %v", trade.TradeID, err)
|
||||
continue
|
||||
}
|
||||
// Insert order record
|
||||
if err := orderStore.CreateOrder(orderRecord); err != nil {
|
||||
logger.Infof(" ⚠️ Failed to sync trade %s: %v", trade.TradeID, err)
|
||||
continue
|
||||
}
|
||||
|
||||
// Create fill record - use Unix milliseconds UTC
|
||||
fillRecord := &store.TraderFill{
|
||||
TraderID: traderID,
|
||||
ExchangeID: exchangeID, // UUID
|
||||
ExchangeType: exchangeType, // Exchange type
|
||||
OrderID: orderRecord.ID,
|
||||
ExchangeOrderID: trade.TradeID,
|
||||
ExchangeTradeID: trade.TradeID,
|
||||
Symbol: symbol,
|
||||
Side: trade.Side,
|
||||
Price: trade.Price,
|
||||
Quantity: trade.Quantity,
|
||||
QuoteQuantity: trade.Price * trade.Quantity,
|
||||
Commission: trade.Fee,
|
||||
CommissionAsset: "USDT",
|
||||
RealizedPnL: trade.RealizedPnL,
|
||||
IsMaker: false, // Hyperliquid GetTrades doesn't provide maker/taker info
|
||||
CreatedAt: tradeTimeMs,
|
||||
}
|
||||
// Create fill record - use Unix milliseconds UTC
|
||||
fillRecord := &store.TraderFill{
|
||||
TraderID: traderID,
|
||||
ExchangeID: exchangeID, // UUID
|
||||
ExchangeType: exchangeType, // Exchange type
|
||||
OrderID: orderRecord.ID,
|
||||
ExchangeOrderID: trade.TradeID,
|
||||
ExchangeTradeID: trade.TradeID,
|
||||
Symbol: symbol,
|
||||
Side: trade.Side,
|
||||
Price: trade.Price,
|
||||
Quantity: trade.Quantity,
|
||||
QuoteQuantity: trade.Price * trade.Quantity,
|
||||
Commission: trade.Fee,
|
||||
CommissionAsset: "USDT",
|
||||
RealizedPnL: trade.RealizedPnL,
|
||||
IsMaker: false, // Hyperliquid GetTrades doesn't provide maker/taker info
|
||||
CreatedAt: tradeTimeMs,
|
||||
}
|
||||
|
||||
if err := orderStore.CreateFill(fillRecord); err != nil {
|
||||
logger.Infof(" ⚠️ Failed to sync fill for trade %s: %v", trade.TradeID, err)
|
||||
}
|
||||
if err := orderStore.CreateFill(fillRecord); err != nil {
|
||||
logger.Infof(" ⚠️ Failed to sync fill for trade %s: %v", trade.TradeID, err)
|
||||
}
|
||||
|
||||
// Create/update position record using PositionBuilder
|
||||
if err := posBuilder.ProcessTrade(
|
||||
traderID, exchangeID, exchangeType,
|
||||
symbol, positionSide, orderAction,
|
||||
trade.Quantity, trade.Price, trade.Fee, trade.RealizedPnL,
|
||||
tradeTimeMs, trade.TradeID,
|
||||
); err != nil {
|
||||
logger.Infof(" ⚠️ Failed to sync position for trade %s: %v", trade.TradeID, err)
|
||||
} else {
|
||||
logger.Infof(" 📍 Position updated for trade: %s (action: %s, qty: %.6f)", trade.TradeID, orderAction, trade.Quantity)
|
||||
}
|
||||
// Create/update position record using PositionBuilder
|
||||
if err := posBuilder.ProcessTrade(
|
||||
traderID, exchangeID, exchangeType,
|
||||
symbol, positionSide, orderAction,
|
||||
trade.Quantity, trade.Price, trade.Fee, trade.RealizedPnL,
|
||||
tradeTimeMs, trade.TradeID,
|
||||
); err != nil {
|
||||
logger.Infof(" ⚠️ Failed to sync position for trade %s: %v", trade.TradeID, err)
|
||||
} else {
|
||||
logger.Infof(" 📍 Position updated for trade: %s (action: %s, qty: %.6f)", trade.TradeID, orderAction, trade.Quantity)
|
||||
}
|
||||
|
||||
syncedCount++
|
||||
logger.Infof(" ✅ Synced trade: %s %s %s qty=%.6f price=%.6f pnl=%.2f fee=%.6f action=%s",
|
||||
@@ -136,14 +137,8 @@ func (t *HyperliquidTrader) SyncOrdersFromHyperliquid(traderID string, exchangeI
|
||||
}
|
||||
|
||||
// StartOrderSync starts background order sync task
|
||||
func (t *HyperliquidTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
|
||||
ticker := time.NewTicker(interval)
|
||||
go func() {
|
||||
for range ticker.C {
|
||||
if err := t.SyncOrdersFromHyperliquid(traderID, exchangeID, exchangeType, st); err != nil {
|
||||
logger.Infof("⚠️ Hyperliquid order sync failed: %v", err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
logger.Infof("🔄 Hyperliquid order sync started (interval: %v)", interval)
|
||||
func (t *HyperliquidTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) {
|
||||
syncloop.Run(stop, interval, "Hyperliquid", func() error {
|
||||
return t.SyncOrdersFromHyperliquid(traderID, exchangeID, exchangeType, st)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"fmt"
|
||||
"nofx/logger"
|
||||
"nofx/store"
|
||||
"nofx/trader/syncloop"
|
||||
"nofx/trader/types"
|
||||
"sort"
|
||||
"strings"
|
||||
@@ -399,14 +400,8 @@ func (t *KuCoinTrader) SyncOrdersFromKuCoin(traderID string, exchangeID string,
|
||||
}
|
||||
|
||||
// StartOrderSync starts background order sync task for KuCoin
|
||||
func (t *KuCoinTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
|
||||
ticker := time.NewTicker(interval)
|
||||
go func() {
|
||||
for range ticker.C {
|
||||
if err := t.SyncOrdersFromKuCoin(traderID, exchangeID, exchangeType, st); err != nil {
|
||||
logger.Infof("⚠️ KuCoin order sync failed: %v", err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
logger.Infof("🔄 KuCoin order sync started (interval: %v)", interval)
|
||||
func (t *KuCoinTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) {
|
||||
syncloop.Run(stop, interval, "KuCoin", func() error {
|
||||
return t.SyncOrdersFromKuCoin(traderID, exchangeID, exchangeType, st)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"nofx/logger"
|
||||
"nofx/market"
|
||||
"nofx/store"
|
||||
"nofx/trader/syncloop"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -145,17 +146,14 @@ func (t *LighterTraderV2) SyncOrdersFromLighter(traderID string, exchangeID stri
|
||||
}
|
||||
|
||||
// StartOrderSync starts background order sync task
|
||||
func (t *LighterTraderV2) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
|
||||
ticker := time.NewTicker(interval)
|
||||
go func() {
|
||||
for range ticker.C {
|
||||
if err := t.SyncOrdersFromLighter(traderID, exchangeID, exchangeType, st); err != nil {
|
||||
// Only log non-404 errors to reduce log spam
|
||||
if !strings.Contains(err.Error(), "status 404") {
|
||||
logger.Infof("⚠️ Order sync failed: %v", err)
|
||||
}
|
||||
}
|
||||
func (t *LighterTraderV2) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) {
|
||||
syncloop.Run(stop, interval, "Lighter", func() error {
|
||||
err := t.SyncOrdersFromLighter(traderID, exchangeID, exchangeType, st)
|
||||
// A 404 just means the account has no fills yet — treat as a
|
||||
// successful empty sync to avoid log spam and pointless backoff.
|
||||
if err != nil && strings.Contains(err.Error(), "status 404") {
|
||||
return nil
|
||||
}
|
||||
}()
|
||||
logger.Infof("🔄 Lighter order+position sync started (interval: %v)", interval)
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"nofx/logger"
|
||||
"nofx/market"
|
||||
"nofx/store"
|
||||
"nofx/trader/syncloop"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -272,14 +273,8 @@ func (t *OKXTrader) SyncOrdersFromOKX(traderID string, exchangeID string, exchan
|
||||
}
|
||||
|
||||
// StartOrderSync starts background order sync task for OKX
|
||||
func (t *OKXTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
|
||||
ticker := time.NewTicker(interval)
|
||||
go func() {
|
||||
for range ticker.C {
|
||||
if err := t.SyncOrdersFromOKX(traderID, exchangeID, exchangeType, st); err != nil {
|
||||
logger.Infof("⚠️ OKX order sync failed: %v", err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
logger.Infof("🔄 OKX order sync started (interval: %v)", interval)
|
||||
func (t *OKXTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration, stop <-chan struct{}) {
|
||||
syncloop.Run(stop, interval, "OKX", func() error {
|
||||
return t.SyncOrdersFromOKX(traderID, exchangeID, exchangeType, st)
|
||||
})
|
||||
}
|
||||
|
||||
48
trader/syncloop/syncloop.go
Normal file
48
trader/syncloop/syncloop.go
Normal file
@@ -0,0 +1,48 @@
|
||||
// Package syncloop runs background exchange order-sync loops with a shared
|
||||
// lifecycle: loops stop when the owning trader stops, and consecutive
|
||||
// failures back off exponentially so a rate-limiting exchange (HTTP 429)
|
||||
// is not hammered at full frequency.
|
||||
package syncloop
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"nofx/logger"
|
||||
)
|
||||
|
||||
// maxBackoff caps the failure backoff so a recovered exchange is picked up
|
||||
// within a few minutes at worst.
|
||||
const maxBackoff = 5 * time.Minute
|
||||
|
||||
// Run starts a background loop calling syncFn at the given interval until
|
||||
// stop is closed. After each consecutive failure the wait doubles (capped at
|
||||
// maxBackoff); the first success resets it to the base interval.
|
||||
func Run(stop <-chan struct{}, interval time.Duration, name string, syncFn func() error) {
|
||||
if interval <= 0 {
|
||||
interval = 30 * time.Second
|
||||
}
|
||||
go func() {
|
||||
wait := interval
|
||||
timer := time.NewTimer(wait)
|
||||
defer timer.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
logger.Infof("⏹ %s order sync stopped", name)
|
||||
return
|
||||
case <-timer.C:
|
||||
if err := syncFn(); err != nil {
|
||||
wait *= 2
|
||||
if wait > maxBackoff {
|
||||
wait = maxBackoff
|
||||
}
|
||||
logger.Infof("⚠️ %s order sync failed: %v (backing off, next attempt in %v)", name, err, wait)
|
||||
} else {
|
||||
wait = interval
|
||||
}
|
||||
timer.Reset(wait)
|
||||
}
|
||||
}
|
||||
}()
|
||||
logger.Infof("🔄 %s order sync started (interval: %v)", name, interval)
|
||||
}
|
||||
78
trader/syncloop/syncloop_test.go
Normal file
78
trader/syncloop/syncloop_test.go
Normal file
@@ -0,0 +1,78 @@
|
||||
package syncloop
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestRunStopsWhenStopChannelCloses(t *testing.T) {
|
||||
stop := make(chan struct{})
|
||||
var calls atomic.Int64
|
||||
|
||||
Run(stop, 5*time.Millisecond, "test", func() error {
|
||||
calls.Add(1)
|
||||
return nil
|
||||
})
|
||||
|
||||
// Let a few ticks happen, then stop.
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
close(stop)
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
after := calls.Load()
|
||||
if after == 0 {
|
||||
t.Fatal("sync function never ran")
|
||||
}
|
||||
|
||||
// No further calls after stop.
|
||||
time.Sleep(40 * time.Millisecond)
|
||||
if calls.Load() != after {
|
||||
t.Fatalf("sync kept running after stop: %d -> %d", after, calls.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunBacksOffOnConsecutiveFailures(t *testing.T) {
|
||||
stop := make(chan struct{})
|
||||
defer close(stop)
|
||||
var calls atomic.Int64
|
||||
|
||||
Run(stop, 10*time.Millisecond, "test", func() error {
|
||||
calls.Add(1)
|
||||
return errors.New("API returned status 429")
|
||||
})
|
||||
|
||||
// With a 10ms base interval and exponential backoff (10, 20, 40, 80...),
|
||||
// 100ms allows at most ~4 failing attempts. Without backoff there would
|
||||
// be ~10.
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
got := calls.Load()
|
||||
if got == 0 {
|
||||
t.Fatal("sync function never ran")
|
||||
}
|
||||
if got > 5 {
|
||||
t.Fatalf("expected backoff to throttle failing sync, got %d calls in 100ms", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunRecoversIntervalAfterSuccess(t *testing.T) {
|
||||
stop := make(chan struct{})
|
||||
defer close(stop)
|
||||
var calls atomic.Int64
|
||||
|
||||
// Fail twice, then succeed forever.
|
||||
Run(stop, 5*time.Millisecond, "test", func() error {
|
||||
n := calls.Add(1)
|
||||
if n <= 2 {
|
||||
return errors.New("transient")
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
// Failures at 5ms and 15ms (backoff 10ms), success at ~35ms (backoff 20ms),
|
||||
// then the interval resets to 5ms — plenty of successful runs by 150ms.
|
||||
time.Sleep(150 * time.Millisecond)
|
||||
if got := calls.Load(); got < 8 {
|
||||
t.Fatalf("expected interval to reset after success, got only %d calls", got)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user