diff --git a/exchange/aevo.go b/exchange/aevo.go index aaa7bb0..a45bcbd 100644 --- a/exchange/aevo.go +++ b/exchange/aevo.go @@ -47,7 +47,7 @@ func NewAevoWS(tracked []TrackedSymbol) *AevoWS { } // Run connects to Aevo WS and streams ticker data. -func (a *AevoWS) Run(updateFn func(coin string, price float64)) error { +func (a *AevoWS) Run(updateFn func(coin string, price, bid, ask float64)) error { a.Conn.OnConnect = func() { log.Printf("[Aevo WS] Connected, subscribing to %d tickers", len(a.Tracked)) @@ -120,7 +120,7 @@ func (a *AevoWS) Run(updateFn func(coin string, price float64)) error { } if price > 0 { - updateFn(coin, price) + updateFn(coin, price, 0, 0) // B#7: pass bid=ask=0 for 4-arg signature } } } diff --git a/exchange/bitget_trade.go b/exchange/bitget_trade.go index 859730c..e1a9efe 100644 --- a/exchange/bitget_trade.go +++ b/exchange/bitget_trade.go @@ -7,6 +7,7 @@ import ( "encoding/json" "fmt" "io" + "math" "net/http" "strings" "time" @@ -95,6 +96,7 @@ func (b *BitgetTrade) sign(method, requestPath, timestamp, body string) string { // GetBitgetSize calculates the contract size for a given USD amount. // Returns size as a decimal string complying with Bitget's USDT-FUTURES precision. // Enforces the exchange's minimum: minTradeNum contracts AND $5 min notional. +// Uses math.Floor to round DOWN to the nearest valid step (B#5: prevent rounding up). func GetBitgetSize(symbol string, amountUSD, price float64) string { if amountUSD < 5 { amountUSD = 5 // Bitget minimum notional @@ -106,28 +108,34 @@ func GetBitgetSize(symbol string, amountUSD, price float64) string { if sz < 1 { sz = 1 } + sz = math.Floor(sz) // step=1 return fmt.Sprintf("%.0f", sz) // minTradeNum=1, sizeMultiplier=1 case "LINKUSDT": if sz < 1 { sz = 1 } + sz = math.Floor(sz) // step=1 return fmt.Sprintf("%.0f", sz) // minTradeNum=1, sizeMultiplier=1 case "ONDOUSDT": + sz = math.Floor(sz*10) / 10 // step=0.1 if sz < 0.1 { sz = 0.1 } return fmt.Sprintf("%.1f", sz) // minTradeNum=0.1, sizeMultiplier=0.1 case "OPUSDT": + sz = math.Floor(sz*10) / 10 // step=0.1 if sz < 0.1 { sz = 0.1 } return fmt.Sprintf("%.1f", sz) // minTradeNum=0.1, sizeMultiplier=0.1 case "WIFUSDT": + sz = math.Floor(sz*10) / 10 // step=0.1 if sz < 0.1 { sz = 0.1 } return fmt.Sprintf("%.1f", sz) // minTradeNum=0.1, sizeMultiplier=0.1 case "ARBUSDT": + sz = math.Floor(sz*100) / 100 // step=0.01 if sz < 0.01 { sz = 0.01 } diff --git a/exchange/connector.go b/exchange/connector.go index b79b00c..e2ae6c0 100644 --- a/exchange/connector.go +++ b/exchange/connector.go @@ -1,8 +1,8 @@ package exchange import ( - "fmt" "log" + "strconv" "time" "github.com/gorilla/websocket" @@ -148,9 +148,10 @@ func (pc *PriceConnector) SendJSON(v interface{}) error { } // Helper: parse float from string +// B#8: use strconv.ParseFloat instead of fmt.Sscanf func parseFloat(s string) float64 { - var f float64 - if _, err := fmt.Sscanf(s, "%f", &f); err != nil { + f, err := strconv.ParseFloat(s, 64) + if err != nil { return 0 } return f diff --git a/exchange/dydx.go b/exchange/dydx.go index ffe7500..cda04cd 100644 --- a/exchange/dydx.go +++ b/exchange/dydx.go @@ -3,13 +3,16 @@ package exchange import ( "encoding/json" "log" + "sync" "time" ) // DydxWS connects to dYdX v4 WebSocket for market data (oracle prices). type DydxWS struct { - Conn *PriceConnector - Tracked []string // coin names like ["BTC", "ETH", ...] + Conn *PriceConnector + Tracked []string // coin names like ["BTC", "ETH", ...] + stopHeartbeat chan struct{} + heartbeatMu sync.Mutex } type dydxSubscribeMsg struct { @@ -53,6 +56,15 @@ func (d *DydxWS) Run(updateFn func(coin string, price, bid, ask float64)) error log.Printf("[dYdX WS] Subscribe error: %v", err) } + // B#4: Stop any previous heartbeat goroutine before starting a new one + d.heartbeatMu.Lock() + if d.stopHeartbeat != nil { + close(d.stopHeartbeat) + } + d.stopHeartbeat = make(chan struct{}) + hbStop := d.stopHeartbeat + d.heartbeatMu.Unlock() + // dYdX requires JSON {"type":"ping"} every ~30s go func() { heartbeat := time.NewTicker(15 * time.Second) @@ -65,6 +77,8 @@ func (d *DydxWS) Run(updateFn func(coin string, price, bid, ask float64)) error if err := d.Conn.SendJSON(map[string]string{"type": "ping"}); err != nil { log.Printf("[dYdX WS] Heartbeat send error: %v", err) } + case <-hbStop: + return case <-d.Conn.Done(): return } diff --git a/main.go b/main.go index 3c73fb6..e26b0f1 100644 --- a/main.go +++ b/main.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "context" "fmt" "io" "log" @@ -61,16 +62,21 @@ func main() { log.Printf("[Trader] Automated trading DISABLED (set TRADE_ENABLED=1 or TEST_MODE=true in .env)") } + // Context for graceful shutdown — replaces shared sigCh (B#1) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, os.Interrupt, syscall.SIGUSR1) // Collect symbols - var bnSymbols, bgSymbols, hlSymbols []string + var bnSymbols, bgSymbols, hlSymbols, dydxSymbols []string var aevoSymbols []exchange.TrackedSymbol for _, c := range TrackedCoins { bnSymbols = append(bnSymbols, c.BN) bgSymbols = append(bgSymbols, c.BG) hlSymbols = append(hlSymbols, c.HL) + dydxSymbols = append(dydxSymbols, c.HL) aevoSymbols = append(aevoSymbols, exchange.TrackedSymbol{ Coin: c.Name, InstrumentID: c.Name + "-PERP", @@ -87,7 +93,7 @@ func main() { }) log.Printf("[%s] WS error: %v (reconnecting...)", name, err) select { - case <-sigCh: + case <-ctx.Done(): return case <-time.After(3 * time.Second): } @@ -98,7 +104,7 @@ func main() { startExchange("Binance", exchange.NewBinanceWS(bnSymbols).Run) startExchange("HyperLiquid", exchange.NewHyperLiquidWS(hlSymbols).Run) startExchange("Bitget", exchange.NewBitgetWS(bgSymbols).Run) - startExchange("dYdX", exchange.NewDydxWS(hlSymbols).Run) + startExchange("dYdX", exchange.NewDydxWS(dydxSymbols).Run) // B#9: use dedicated symbol list log.Println("[Monitor] Waiting for initial data...") time.Sleep(10 * time.Second) @@ -131,6 +137,7 @@ func main() { continue } log.Println("[Monitor] Shutting down...") + cancel() // B#1: cancel context to stop all WS goroutines runLoop = false case <-statusTick.C: diff --git a/scanner.go b/scanner.go index b651c9e..6102db9 100644 --- a/scanner.go +++ b/scanner.go @@ -4,6 +4,8 @@ import ( "log" "sort" "time" + + "exchange-monitor/exchange" ) // Exchange names @@ -44,16 +46,9 @@ var TrackedCoins = []TrackedCoin{ } // netProfit calculates net profit % after fees for a complete round trip (entry + exit). +// B#6: Delegates to exchange.CalcNetProfit to eliminate formula duplication. func netProfit(buyPrice, sellPrice, buyFee, sellFee float64) float64 { - if buyPrice <= 0 || sellPrice <= 0 { - return 0 - } - // Entry: buy at buyPrice (pay buyFee), sell short at sellPrice (pay sellFee) - 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) + return exchange.CalcNetProfit(buyPrice, sellPrice, buyFee, sellFee, buyFee, sellFee) } // ScanArbWithFees checks all coins for arbitrage opportunities using a custom fee map. diff --git a/trader.go b/trader.go index 93eca21..8fb4436 100644 --- a/trader.go +++ b/trader.go @@ -625,6 +625,7 @@ func (t *Trader) restoreOpenPositions() { AmountUSD: tr.AmountUSD, EntrySpread: *tr.EntrySpread, ScaleLevels: tr.ScaleCount, + LastScaleAt: tr.OpenedAt, // B#3: prevent immediate scale-in bypass StartedAt: tr.OpenedAt, Status: "open", }