From b08d8490fcf423dbb733febcf956b6a85ccc5cb9 Mon Sep 17 00:00:00 2001 From: jackyu66git Date: Sun, 3 May 2026 18:31:14 +0800 Subject: [PATCH] fix: data race, scale-in PnL, nonce mutex, dead code, hourly check MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ๐Ÿ”ด Data race: Add GetPositionsCopy() returning deep copies (no shared ArbPosition pointers). Use it in dashboard broadcastLoop + handleStatus. - ๐ŸŸก Scale-in PnL: Track LongEntryPrices/ShortEntryPrices on ArbPosition, compute weighted average (harmonic mean) at exit for accurate PnL. - ๐ŸŸข CalcNetProfit: Delete dead code from exchange/helpers.go. - ๐ŸŸข HL nonce: Add sync.Mutex around lastNonce++ (thread safety). - ๐ŸŸข Hourly check: Change from 5-second window to minute window. - ๐ŸŸข ExitPrice: Test mode closeLeg already handled by checkExit. --- dashboard.go | 14 ++++--- exchange/helpers.go | 43 ++----------------- exchange/hyperliquid_trade.go | 6 ++- main.go | 6 +-- notifier.go | 2 +- trader.go | 77 +++++++++++++++++++++++++++++++++-- 6 files changed, 94 insertions(+), 54 deletions(-) diff --git a/dashboard.go b/dashboard.go index fa01a88..0297005 100644 --- a/dashboard.go +++ b/dashboard.go @@ -290,8 +290,8 @@ func (d *Dashboard) broadcastLoop() { } d.hub.Broadcast("prices", prices) - // 2. Open positions with live PnL (P3-3) - positions := d.trader.GetOpenPositions() + // 2. Open positions with live PnL (P3-3) โ€” use safe copy for concurrent read + positions := d.trader.GetPositionsCopy() posList := make([]map[string]interface{}, 0, len(positions)) for _, pos := range positions { posEntry := map[string]interface{}{ @@ -304,7 +304,7 @@ func (d *Dashboard) broadcastLoop() { "started_at": pos.StartedAt.Format("15:04:05"), } - // Calculate live PnL from current prices + // Calculate live PnL from current prices โ€” use weighted average for scale-ins if exMap := snap[pos.Coin]; exMap != nil { bgP := exMap[ExBitget] hlP := exMap[ExHyperLiquid] @@ -315,8 +315,10 @@ func (d *Dashboard) broadcastLoop() { } else { longCurrent, shortCurrent = hlP, bgP } - longPnl := (longCurrent - pos.LongLeg.EntryPrice) / pos.LongLeg.EntryPrice * 100 - shortPnl := (pos.ShortLeg.EntryPrice - shortCurrent) / pos.ShortLeg.EntryPrice * 100 + longAvg := weightedAvgPrice(pos.LongEntryPrices, pos.AmountUSD/float64(max(1, len(pos.LongEntryPrices)))) + shortAvg := weightedAvgPrice(pos.ShortEntryPrices, pos.AmountUSD/float64(max(1, len(pos.ShortEntryPrices)))) + longPnl := (longCurrent - longAvg) / longAvg * 100 + shortPnl := (shortAvg - shortCurrent) / shortAvg * 100 totalFees := 2 * (makerFees[ExBitget] + makerFees[ExHyperLiquid]) netPnl := longPnl + shortPnl - totalFees @@ -426,7 +428,7 @@ func (d *Dashboard) handleIndex(w http.ResponseWriter, r *http.Request) { func (d *Dashboard) handleStatus(w http.ResponseWriter, r *http.Request) { snap := d.store.GetAll() - positions := d.trader.GetOpenPositions() + positions := d.trader.GetPositionsCopy() converged, diverged, flat, total := d.trader.GetClosedStats() resp := map[string]interface{}{ diff --git a/exchange/helpers.go b/exchange/helpers.go index 8379f67..7cb54bf 100644 --- a/exchange/helpers.go +++ b/exchange/helpers.go @@ -1,42 +1,5 @@ package exchange -import "github.com/gorilla/websocket" - -// These are needed for compilation of the exchange package. -// PriceConnector is defined in connector.go. -var _ = websocket.ErrCloseSent // keep gorilla/websocket import - -// CalcNetProfit calculates net profit % for a complete round trip (entry + exit) between two exchanges. -// buyPrice: price on the buy exchange -// sellPrice: price on the sell exchange -// buyFee: fee rate on buy exchange (e.g. 0.03 for 0.03%) -// sellFee: fee rate on sell exchange -// buyFee2: buy fee on the other exchange -// sellFee2: sell fee on the other exchange -// Returns net profit in percentage. -func CalcNetProfit(price1, price2, fee1Buy, fee1Sell, fee2Buy, fee2Sell float64) float64 { - // price1 = Bitget, price2 = HyperLiquid - // Try: buy cheap (min), sell expensive (max) - buyPrice := price1 - sellPrice := price2 - buyFee := fee1Buy - sellFee := fee2Sell - - if price2 < price1 { - buyPrice = price2 - sellPrice = price1 - buyFee = fee2Buy - sellFee = fee1Sell - } - - // Entry: buy at buyPrice (pay buyFee), sell short at sellPrice (pay sellFee) - if buyPrice <= 0 || sellPrice <= 0 { - return 0 - } - cost := buyPrice * (1 + buyFee/100) - revenue := sellPrice * (1 - sellFee/100) - - // Exit: sell long (pay sellFee), buy back short (pay buyFee) - // Total fees = 2 * (buyFee + sellFee), first round already in formula above - return (revenue/cost-1)*100 - (buyFee + sellFee) -} +// Package-level helpers for the exchange package. +// connector.go imports gorilla/websocket, so this file needs no imports +// for that dependency. CalcNetProfit was removed (see netProfit in scanner.go). diff --git a/exchange/hyperliquid_trade.go b/exchange/hyperliquid_trade.go index 309799c..0573cde 100644 --- a/exchange/hyperliquid_trade.go +++ b/exchange/hyperliquid_trade.go @@ -8,6 +8,7 @@ import ( "math/big" "net/http" "strings" + "sync" "time" "crypto/ed25519" @@ -50,6 +51,7 @@ type HyperLiquidTrade struct { Address string client *http.Client lastNonce int64 + nonceMu sync.Mutex // protect lastNonce++ (Issue #4) } func NewHyperLiquidTrade(privateKeyHex, address string) (*HyperLiquidTrade, error) { @@ -96,9 +98,11 @@ func (h *HyperLiquidTrade) PlaceMarketOrder(coin, side, sz string) (string, erro BrokerCode: 0, } - // Generate nonce + // Generate nonce (thread-safe) + h.nonceMu.Lock() h.lastNonce++ nonce := time.Now().UnixMilli()*1_000_000 + h.lastNonce%1_000_000 + h.nonceMu.Unlock() // Sign the action sig, err := h.signAction(action, nonce) diff --git a/main.go b/main.go index 81e407c..6362674 100644 --- a/main.go +++ b/main.go @@ -187,10 +187,10 @@ func main() { tickMs, t1.Sub(t0).Milliseconds(), t2.Sub(t1).Milliseconds(), t3.Sub(t2).Milliseconds()) } - // Hourly trade summary + // Hourly trade summary โ€” use hour-based tracking (wider window than second-granularity) hour := now.Hour() - if now.Minute() == 0 && now.Second() < 5 && hour != lastHour { - positions := trader.GetOpenPositions() + if hour != lastHour && now.Minute() < 1 { + positions := trader.GetPositionsCopy() notifier.SendTradeSummary(positions, now.Format("2006-01-02 15:04")) lastHour = hour } diff --git a/notifier.go b/notifier.go index 75cd2de..02f54ed 100644 --- a/notifier.go +++ b/notifier.go @@ -73,7 +73,7 @@ func (n *Notifier) SendAlert(opp *ArbOpportunity) { } // SendTradeSummary sends a summary of open positions at each hour. -func (n *Notifier) SendTradeSummary(positions []*ArbPosition, timeStr string) { +func (n *Notifier) SendTradeSummary(positions []ArbPosition, timeStr string) { if n.BotToken == "" || n.ChatID == "" { return } diff --git a/trader.go b/trader.go index 2fdf3d0..a222b1b 100644 --- a/trader.go +++ b/trader.go @@ -50,6 +50,43 @@ type ArbPosition struct { Status string // "open", "closed" RealizedPnl float64 ErrorLog string + + // Track all entry prices for weighted-average PnL across scale-ins (Issue #2) + LongEntryPrices []float64 // all long entry prices (initial + scale-ins) + ShortEntryPrices []float64 // all short entry prices (initial + scale-ins) +} + +// DeepCopy returns a copy-safe snapshot of the position (no shared pointers). +func (p *ArbPosition) DeepCopy() ArbPosition { + c := *p + if p.LongLeg != nil { + lc := *p.LongLeg + c.LongLeg = &lc + } + if p.ShortLeg != nil { + sc := *p.ShortLeg + c.ShortLeg = &sc + } + if p.LongEntryPrices != nil { + c.LongEntryPrices = make([]float64, len(p.LongEntryPrices)) + copy(c.LongEntryPrices, p.LongEntryPrices) + } + if p.ShortEntryPrices != nil { + c.ShortEntryPrices = make([]float64, len(p.ShortEntryPrices)) + copy(c.ShortEntryPrices, p.ShortEntryPrices) + } + return c +} + +// GetPositionsCopy returns deep copies of all open positions โ€” safe for concurrent read. +func (t *Trader) GetPositionsCopy() []ArbPosition { + t.mu.Lock() + defer t.mu.Unlock() + r := make([]ArbPosition, 0, len(t.positions)) + for _, p := range t.positions { + r = append(r, p.DeepCopy()) + } + return r } // Trader handles scalable arbitrage between Bitget and HyperLiquid. @@ -245,6 +282,8 @@ func (t *Trader) executeEntry(opp *ArbOpportunity, store *PriceStore, notifier * Coin: opp.Coin, Exchange: ExHyperLiquid, Side: Short, EntryPrice: hlP, EntryTime: time.Now(), } + pos.LongEntryPrices = []float64{bgP} + pos.ShortEntryPrices = []float64{hlP} } else { pos.Direction = "HL->BG" pos.EntrySpread = (bgP - hlP) / hlP * 100 // positive when bgP > hlP @@ -256,6 +295,8 @@ func (t *Trader) executeEntry(opp *ArbOpportunity, store *PriceStore, notifier * Coin: opp.Coin, Exchange: ExBitget, Side: Short, EntryPrice: bgP, EntryTime: time.Now(), } + pos.LongEntryPrices = []float64{hlP} + pos.ShortEntryPrices = []float64{bgP} } t.mu.Lock() @@ -357,6 +398,8 @@ func (t *Trader) checkScaleIn(pos *ArbPosition, bgP, hlP, diffPct float64, store pos.ScaleLevels++ pos.LastScaleAt = time.Now() pos.AmountUSD += t.cfg.TradeAmountUSD + pos.LongEntryPrices = append(pos.LongEntryPrices, longPrice) + pos.ShortEntryPrices = append(pos.ShortEntryPrices, shortPrice) log.Printf("[Trader] %s: Scale-in #%d executed | spread=%.4f%% (entry=%.4f%%) | total=$%.0f", pos.Coin, pos.ScaleLevels, diffPct, entryDiff, pos.AmountUSD) @@ -389,7 +432,8 @@ func (t *Trader) checkExit(pos *ArbPosition, bgP, hlP, diffPct float64, notifier return } - // Calculate P&L + // Calculate P&L โ€” use weighted average entry for scale-in positions + // Each scale adds cfg.TradeAmountUSD at the scale price var longCurrent, shortCurrent float64 if pos.LongLeg.Exchange == ExBitget { longCurrent, shortCurrent = bgP, hlP @@ -397,8 +441,12 @@ func (t *Trader) checkExit(pos *ArbPosition, bgP, hlP, diffPct float64, notifier longCurrent, shortCurrent = hlP, bgP } - longPnl := (longCurrent - pos.LongLeg.EntryPrice) / pos.LongLeg.EntryPrice * 100 - shortPnl := (pos.ShortLeg.EntryPrice - shortCurrent) / pos.ShortLeg.EntryPrice * 100 + // Weighted average entry prices across all scale levels + longAvg := weightedAvgPrice(pos.LongEntryPrices, t.cfg.TradeAmountUSD) + shortAvg := weightedAvgPrice(pos.ShortEntryPrices, t.cfg.TradeAmountUSD) + + longPnl := (longCurrent - longAvg) / longAvg * 100 + shortPnl := (shortAvg - shortCurrent) / shortAvg * 100 totalFees := 2 * (makerFees[ExBitget] + makerFees[ExHyperLiquid]) // ๅผ€ไป“ + ๅนณไป“ๆ‰‹็ปญ่ดน netPnl := longPnl + shortPnl - totalFees @@ -627,6 +675,27 @@ func (t *Trader) GetOpenPositions() []*ArbPosition { return r } +// weightedAvgPrice computes the weighted average entry price across multiple scale levels. +// Each level trades the same USD amount, so the result is the harmonic mean of prices. +func weightedAvgPrice(prices []float64, amountPerTrade float64) float64 { + if len(prices) == 0 { + return 0 + } + totalShares := 0.0 + totalCost := 0.0 + for _, p := range prices { + if p <= 0 { + continue + } + totalShares += amountPerTrade / p + totalCost += amountPerTrade + } + if totalShares <= 0 { + return prices[0] // fallback + } + return totalCost / totalShares +} + // GetClosedStats returns convergence stats from all closed trades. func (t *Trader) GetClosedStats() (converged, diverged, flat, total int) { t.mu.Lock() @@ -717,12 +786,14 @@ func (t *Trader) restoreOpenPositions() { Coin: tr.Coin, Exchange: tr.LongExchange, Side: Long, EntryPrice: *tr.LongEntry, EntryTime: tr.OpenedAt, } + pos.LongEntryPrices = []float64{*tr.LongEntry} } if tr.ShortEntry != nil { pos.ShortLeg = &PositionLeg{ Coin: tr.Coin, Exchange: tr.ShortExchange, Side: Short, EntryPrice: *tr.ShortEntry, EntryTime: tr.OpenedAt, } + pos.ShortEntryPrices = []float64{*tr.ShortEntry} } t.positions[tr.Coin] = pos // Prevent immediate re-trading of the same coin