Fix 8 bugs from code review

B#1 — sigCh shared across goroutines, SIGINT unreliable
  → context.WithCancel: main loop cancels ctx on SIGINT,
    4 WS goroutines select on ctx.Done() instead of shared sigCh

B#3 — restoreOpenPositions missing LastScaleAt
  → Set LastScaleAt = tr.OpenedAt on restore so scale-in cooldown works

B#4 — dYdX heartbeat goroutine leaks on reconnect
  → Added stopHeartbeat chan + heartbeatMu mutex; close old channel
    before spawning new heartbeat goroutine

B#5 — GetBitgetSize fmt.Sprintf rounds up, may exceed amountUSD
  → Added math.Floor(sz*multiplier)/multiplier before format to round
    DOWN to nearest valid step size for every coin

B#6 — netProfit and CalcNetProfit duplicate formula
  → scanner.go netProfit now delegates to exchange.CalcNetProfit

B#7 — Aevo Run callback only 2 params, incompatible with startExchange
  → Changed to 4-arg callback func(coin, price, bid, ask) with bid=ask=0

B#8 — parseFloat uses fmt.Sscanf (slow, locale-sensitive)
  → Replaced with strconv.ParseFloat

B#9 — dYdX receives hlSymbols instead of its own symbol list
  → Added dydxSymbols var, built from c.HL like other exchanges
This commit is contained in:
jackyu66git
2026-05-03 17:48:06 +08:00
parent c2489614a7
commit eb74495470
7 changed files with 45 additions and 19 deletions
+2 -2
View File
@@ -47,7 +47,7 @@ func NewAevoWS(tracked []TrackedSymbol) *AevoWS {
} }
// Run connects to Aevo WS and streams ticker data. // 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() { a.Conn.OnConnect = func() {
log.Printf("[Aevo WS] Connected, subscribing to %d tickers", len(a.Tracked)) 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 { if price > 0 {
updateFn(coin, price) updateFn(coin, price, 0, 0) // B#7: pass bid=ask=0 for 4-arg signature
} }
} }
} }
+8
View File
@@ -7,6 +7,7 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"io" "io"
"math"
"net/http" "net/http"
"strings" "strings"
"time" "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. // GetBitgetSize calculates the contract size for a given USD amount.
// Returns size as a decimal string complying with Bitget's USDT-FUTURES precision. // Returns size as a decimal string complying with Bitget's USDT-FUTURES precision.
// Enforces the exchange's minimum: minTradeNum contracts AND $5 min notional. // 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 { func GetBitgetSize(symbol string, amountUSD, price float64) string {
if amountUSD < 5 { if amountUSD < 5 {
amountUSD = 5 // Bitget minimum notional amountUSD = 5 // Bitget minimum notional
@@ -106,28 +108,34 @@ func GetBitgetSize(symbol string, amountUSD, price float64) string {
if sz < 1 { if sz < 1 {
sz = 1 sz = 1
} }
sz = math.Floor(sz) // step=1
return fmt.Sprintf("%.0f", sz) // minTradeNum=1, sizeMultiplier=1 return fmt.Sprintf("%.0f", sz) // minTradeNum=1, sizeMultiplier=1
case "LINKUSDT": case "LINKUSDT":
if sz < 1 { if sz < 1 {
sz = 1 sz = 1
} }
sz = math.Floor(sz) // step=1
return fmt.Sprintf("%.0f", sz) // minTradeNum=1, sizeMultiplier=1 return fmt.Sprintf("%.0f", sz) // minTradeNum=1, sizeMultiplier=1
case "ONDOUSDT": case "ONDOUSDT":
sz = math.Floor(sz*10) / 10 // step=0.1
if sz < 0.1 { if sz < 0.1 {
sz = 0.1 sz = 0.1
} }
return fmt.Sprintf("%.1f", sz) // minTradeNum=0.1, sizeMultiplier=0.1 return fmt.Sprintf("%.1f", sz) // minTradeNum=0.1, sizeMultiplier=0.1
case "OPUSDT": case "OPUSDT":
sz = math.Floor(sz*10) / 10 // step=0.1
if sz < 0.1 { if sz < 0.1 {
sz = 0.1 sz = 0.1
} }
return fmt.Sprintf("%.1f", sz) // minTradeNum=0.1, sizeMultiplier=0.1 return fmt.Sprintf("%.1f", sz) // minTradeNum=0.1, sizeMultiplier=0.1
case "WIFUSDT": case "WIFUSDT":
sz = math.Floor(sz*10) / 10 // step=0.1
if sz < 0.1 { if sz < 0.1 {
sz = 0.1 sz = 0.1
} }
return fmt.Sprintf("%.1f", sz) // minTradeNum=0.1, sizeMultiplier=0.1 return fmt.Sprintf("%.1f", sz) // minTradeNum=0.1, sizeMultiplier=0.1
case "ARBUSDT": case "ARBUSDT":
sz = math.Floor(sz*100) / 100 // step=0.01
if sz < 0.01 { if sz < 0.01 {
sz = 0.01 sz = 0.01
} }
+4 -3
View File
@@ -1,8 +1,8 @@
package exchange package exchange
import ( import (
"fmt"
"log" "log"
"strconv"
"time" "time"
"github.com/gorilla/websocket" "github.com/gorilla/websocket"
@@ -148,9 +148,10 @@ func (pc *PriceConnector) SendJSON(v interface{}) error {
} }
// Helper: parse float from string // Helper: parse float from string
// B#8: use strconv.ParseFloat instead of fmt.Sscanf
func parseFloat(s string) float64 { func parseFloat(s string) float64 {
var f float64 f, err := strconv.ParseFloat(s, 64)
if _, err := fmt.Sscanf(s, "%f", &f); err != nil { if err != nil {
return 0 return 0
} }
return f return f
+16 -2
View File
@@ -3,13 +3,16 @@ package exchange
import ( import (
"encoding/json" "encoding/json"
"log" "log"
"sync"
"time" "time"
) )
// DydxWS connects to dYdX v4 WebSocket for market data (oracle prices). // DydxWS connects to dYdX v4 WebSocket for market data (oracle prices).
type DydxWS struct { type DydxWS struct {
Conn *PriceConnector Conn *PriceConnector
Tracked []string // coin names like ["BTC", "ETH", ...] Tracked []string // coin names like ["BTC", "ETH", ...]
stopHeartbeat chan struct{}
heartbeatMu sync.Mutex
} }
type dydxSubscribeMsg struct { 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) 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 // dYdX requires JSON {"type":"ping"} every ~30s
go func() { go func() {
heartbeat := time.NewTicker(15 * time.Second) 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 { if err := d.Conn.SendJSON(map[string]string{"type": "ping"}); err != nil {
log.Printf("[dYdX WS] Heartbeat send error: %v", err) log.Printf("[dYdX WS] Heartbeat send error: %v", err)
} }
case <-hbStop:
return
case <-d.Conn.Done(): case <-d.Conn.Done():
return return
} }
+10 -3
View File
@@ -2,6 +2,7 @@ package main
import ( import (
"bytes" "bytes"
"context"
"fmt" "fmt"
"io" "io"
"log" "log"
@@ -61,16 +62,21 @@ func main() {
log.Printf("[Trader] Automated trading DISABLED (set TRADE_ENABLED=1 or TEST_MODE=true in .env)") 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) sigCh := make(chan os.Signal, 1)
signal.Notify(sigCh, os.Interrupt, syscall.SIGUSR1) signal.Notify(sigCh, os.Interrupt, syscall.SIGUSR1)
// Collect symbols // Collect symbols
var bnSymbols, bgSymbols, hlSymbols []string var bnSymbols, bgSymbols, hlSymbols, dydxSymbols []string
var aevoSymbols []exchange.TrackedSymbol var aevoSymbols []exchange.TrackedSymbol
for _, c := range TrackedCoins { for _, c := range TrackedCoins {
bnSymbols = append(bnSymbols, c.BN) bnSymbols = append(bnSymbols, c.BN)
bgSymbols = append(bgSymbols, c.BG) bgSymbols = append(bgSymbols, c.BG)
hlSymbols = append(hlSymbols, c.HL) hlSymbols = append(hlSymbols, c.HL)
dydxSymbols = append(dydxSymbols, c.HL)
aevoSymbols = append(aevoSymbols, exchange.TrackedSymbol{ aevoSymbols = append(aevoSymbols, exchange.TrackedSymbol{
Coin: c.Name, Coin: c.Name,
InstrumentID: c.Name + "-PERP", InstrumentID: c.Name + "-PERP",
@@ -87,7 +93,7 @@ func main() {
}) })
log.Printf("[%s] WS error: %v (reconnecting...)", name, err) log.Printf("[%s] WS error: %v (reconnecting...)", name, err)
select { select {
case <-sigCh: case <-ctx.Done():
return return
case <-time.After(3 * time.Second): case <-time.After(3 * time.Second):
} }
@@ -98,7 +104,7 @@ func main() {
startExchange("Binance", exchange.NewBinanceWS(bnSymbols).Run) startExchange("Binance", exchange.NewBinanceWS(bnSymbols).Run)
startExchange("HyperLiquid", exchange.NewHyperLiquidWS(hlSymbols).Run) startExchange("HyperLiquid", exchange.NewHyperLiquidWS(hlSymbols).Run)
startExchange("Bitget", exchange.NewBitgetWS(bgSymbols).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...") log.Println("[Monitor] Waiting for initial data...")
time.Sleep(10 * time.Second) time.Sleep(10 * time.Second)
@@ -131,6 +137,7 @@ func main() {
continue continue
} }
log.Println("[Monitor] Shutting down...") log.Println("[Monitor] Shutting down...")
cancel() // B#1: cancel context to stop all WS goroutines
runLoop = false runLoop = false
case <-statusTick.C: case <-statusTick.C:
+4 -9
View File
@@ -4,6 +4,8 @@ import (
"log" "log"
"sort" "sort"
"time" "time"
"exchange-monitor/exchange"
) )
// Exchange names // Exchange names
@@ -44,16 +46,9 @@ var TrackedCoins = []TrackedCoin{
} }
// netProfit calculates net profit % after fees for a complete round trip (entry + exit). // 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 { func netProfit(buyPrice, sellPrice, buyFee, sellFee float64) float64 {
if buyPrice <= 0 || sellPrice <= 0 { return exchange.CalcNetProfit(buyPrice, sellPrice, buyFee, sellFee, buyFee, sellFee)
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)
} }
// ScanArbWithFees checks all coins for arbitrage opportunities using a custom fee map. // ScanArbWithFees checks all coins for arbitrage opportunities using a custom fee map.
+1
View File
@@ -625,6 +625,7 @@ func (t *Trader) restoreOpenPositions() {
AmountUSD: tr.AmountUSD, AmountUSD: tr.AmountUSD,
EntrySpread: *tr.EntrySpread, EntrySpread: *tr.EntrySpread,
ScaleLevels: tr.ScaleCount, ScaleLevels: tr.ScaleCount,
LastScaleAt: tr.OpenedAt, // B#3: prevent immediate scale-in bypass
StartedAt: tr.OpenedAt, StartedAt: tr.OpenedAt,
Status: "open", Status: "open",
} }