Files
nofx/trader/hyperliquid/order_sync.go
shinchan-zhai 5e06037fa2 reliability: wrap all 27 bare goroutines with safe.Go/GoNamed panic recovery
Applied safe.GoNamed to:
- 9 exchange order_sync goroutines (OKX, Hyperliquid, Aster, Bybit, KuCoin, Gate, Bitget, Lighter, Binance×2)
- Drawdown monitor (auto_trader_risk.go)
- Brain news scanner + market briefs (agent/brain.go)
- Sentinel scanner (agent/sentinel.go)
- Agent scheduler (agent/scheduler.go)
- x402 idle watchdog (mcp/payment/x402.go)
- MCP stream idle watchdog (mcp/client.go)
- Rate limiter cleanup (api/rate_limiter.go)
- 3 telemetry fire-and-forget sends (telemetry/experience.go)
- CoinAnk WS handler (provider/coinank/coinank_api/kline_ws.go)
- API server goroutine (main.go)

Added manual defer/recover with error reporting to:
- Telegram AI agent handler (sends error msg to user on panic)
- Trader data fetch (returns error result on panic to prevent deadlock)

Before: a panic in ANY of these 27 goroutines would crash the entire
trading process with zero diagnostics. Now all panics are caught, logged
with stack traces, and the process continues running.
2026-03-23 10:30:13 +08:00

151 lines
5.2 KiB
Go

package hyperliquid
import (
"fmt"
"nofx/logger"
"nofx/safe"
"nofx/market"
"nofx/store"
"sort"
"strings"
"time"
)
// SyncOrdersFromHyperliquid syncs Hyperliquid exchange order history to local database
// Also creates/updates position records to ensure orders/fills/positions data consistency
// exchangeID: Exchange account UUID (from exchanges.id)
// exchangeType: Exchange type ("hyperliquid")
func (t *HyperliquidTrader) SyncOrdersFromHyperliquid(traderID string, exchangeID string, exchangeType string, st *store.Store) error {
if st == nil {
return fmt.Errorf("store is nil")
}
// Get recent trades (last 24 hours)
startTime := time.Now().Add(-24 * time.Hour)
logger.Infof("🔄 Syncing Hyperliquid trades from: %s", startTime.Format(time.RFC3339))
// Use GetTrades method to fetch trade records
trades, err := t.GetTrades(startTime, 1000)
if err != nil {
return fmt.Errorf("failed to get trades: %w", err)
}
logger.Infof("📥 Received %d trades from Hyperliquid", len(trades))
// Sort trades by time ASC (oldest first) for proper position building
sort.Slice(trades, func(i, j int) bool {
return trades[i].Time.UnixMilli() < trades[j].Time.UnixMilli()
})
// Process trades one by one (no transaction to avoid deadlock)
orderStore := st.Order()
positionStore := st.Position()
posBuilder := store.NewPositionBuilder(positionStore)
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
}
// 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"
}
// 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
}
// 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)
}
// 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",
trade.TradeID, symbol, trade.Side, trade.Quantity, trade.Price, trade.RealizedPnL, trade.Fee, orderAction)
}
logger.Infof("✅ Order sync completed: %d new trades synced", syncedCount)
return nil
}
// 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)
safe.GoNamed("hyperliquid-order-sync", 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)
}