decouple display from trading: snapMu + RefreshSnapshot/ReadSnapshot
- Add snapMu RWMutex + positionsSnapshot to Trader - RefreshSnapshot() called from main loop after Tick() — acquires t.mu briefly, stores deep copy under snapMu - ReadSnapshot() returns snapshot copy under snapMu.RLock — never touches t.mu, zero contention with trading path - Dashboard + handleStatus + hourly summary + status log all use ReadSnapshot() instead of GetPositionsCopy() - Trading path (Tick/TryEntry/executeEntry/checkExit/checkScaleIn) never blocked by display reads - Snapshot is at most 1 tick behind live state — acceptable delay
This commit is contained in:
+3
-3
@@ -290,8 +290,8 @@ func (d *Dashboard) broadcastLoop() {
|
|||||||
}
|
}
|
||||||
d.hub.Broadcast("prices", prices)
|
d.hub.Broadcast("prices", prices)
|
||||||
|
|
||||||
// 2. Open positions with live PnL (P3-3) — use safe copy for concurrent read
|
// 2. Open positions with live PnL (P3-3) — read from decoupled snapshot, never blocks trader
|
||||||
positions := d.trader.GetPositionsCopy()
|
positions := d.trader.ReadSnapshot()
|
||||||
posList := make([]map[string]interface{}, 0, len(positions))
|
posList := make([]map[string]interface{}, 0, len(positions))
|
||||||
for _, pos := range positions {
|
for _, pos := range positions {
|
||||||
posEntry := map[string]interface{}{
|
posEntry := map[string]interface{}{
|
||||||
@@ -428,7 +428,7 @@ func (d *Dashboard) handleIndex(w http.ResponseWriter, r *http.Request) {
|
|||||||
|
|
||||||
func (d *Dashboard) handleStatus(w http.ResponseWriter, r *http.Request) {
|
func (d *Dashboard) handleStatus(w http.ResponseWriter, r *http.Request) {
|
||||||
snap := d.store.GetAll()
|
snap := d.store.GetAll()
|
||||||
positions := d.trader.GetPositionsCopy()
|
positions := d.trader.ReadSnapshot()
|
||||||
converged, diverged, flat, total := d.trader.GetClosedStats()
|
converged, diverged, flat, total := d.trader.GetClosedStats()
|
||||||
|
|
||||||
resp := map[string]interface{}{
|
resp := map[string]interface{}{
|
||||||
|
|||||||
@@ -147,8 +147,8 @@ func main() {
|
|||||||
}
|
}
|
||||||
log.Printf("[Status] %d prices / %d coins connected", count, len(snap))
|
log.Printf("[Status] %d prices / %d coins connected", count, len(snap))
|
||||||
|
|
||||||
// Show open positions
|
// Show open positions (read from decoupled snapshot)
|
||||||
if positions := trader.GetOpenPositions(); len(positions) > 0 {
|
if positions := trader.ReadSnapshot(); len(positions) > 0 {
|
||||||
for _, pos := range positions {
|
for _, pos := range positions {
|
||||||
log.Printf(" [Position] %s %s open %d scales $%.0f since %s",
|
log.Printf(" [Position] %s %s open %d scales $%.0f since %s",
|
||||||
pos.Coin, pos.Direction, pos.ScaleLevels, pos.AmountUSD,
|
pos.Coin, pos.Direction, pos.ScaleLevels, pos.AmountUSD,
|
||||||
@@ -162,6 +162,7 @@ func main() {
|
|||||||
|
|
||||||
// Tick the trader (monitor open positions for exit)
|
// Tick the trader (monitor open positions for exit)
|
||||||
trader.Tick(store, notifier)
|
trader.Tick(store, notifier)
|
||||||
|
trader.RefreshSnapshot() // decoupled snapshot for display
|
||||||
t1 := time.Now()
|
t1 := time.Now()
|
||||||
|
|
||||||
// Scan for arbitrage entries using maker fees (limit orders)
|
// Scan for arbitrage entries using maker fees (limit orders)
|
||||||
@@ -190,7 +191,7 @@ func main() {
|
|||||||
// Hourly trade summary — use hour-based tracking (wider window than second-granularity)
|
// Hourly trade summary — use hour-based tracking (wider window than second-granularity)
|
||||||
hour := now.Hour()
|
hour := now.Hour()
|
||||||
if hour != lastHour && now.Minute() < 1 {
|
if hour != lastHour && now.Minute() < 1 {
|
||||||
positions := trader.GetPositionsCopy()
|
positions := trader.ReadSnapshot()
|
||||||
notifier.SendTradeSummary(positions, now.Format("2006-01-02 15:04"))
|
notifier.SendTradeSummary(positions, now.Format("2006-01-02 15:04"))
|
||||||
lastHour = hour
|
lastHour = hour
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -89,6 +89,26 @@ func (t *Trader) GetPositionsCopy() []ArbPosition {
|
|||||||
return r
|
return r
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// RefreshSnapshot takes a trading-lock snapshot of open positions for display use.
|
||||||
|
// Call this after each Tick() from the main loop — never during a trading operation.
|
||||||
|
// The display reads from this snapshot without blocking trading.
|
||||||
|
func (t *Trader) RefreshSnapshot() {
|
||||||
|
copy := t.GetPositionsCopy() // acquires t.mu briefly (not held during Tick call)
|
||||||
|
t.snapMu.Lock()
|
||||||
|
t.positionsSnapshot = copy
|
||||||
|
t.snapMu.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
// ReadSnapshot returns a copy of the last display snapshot — never locks t.mu.
|
||||||
|
// Safe to call from any goroutine without impacting trading latency.
|
||||||
|
func (t *Trader) ReadSnapshot() []ArbPosition {
|
||||||
|
t.snapMu.RLock()
|
||||||
|
defer t.snapMu.RUnlock()
|
||||||
|
r := make([]ArbPosition, len(t.positionsSnapshot))
|
||||||
|
copy(r, t.positionsSnapshot)
|
||||||
|
return r
|
||||||
|
}
|
||||||
|
|
||||||
// Trader handles scalable arbitrage between Bitget and HyperLiquid.
|
// Trader handles scalable arbitrage between Bitget and HyperLiquid.
|
||||||
type Trader struct {
|
type Trader struct {
|
||||||
cfg *Config
|
cfg *Config
|
||||||
@@ -102,6 +122,10 @@ type Trader struct {
|
|||||||
closedTrades []TradeRecord // history of closed trades
|
closedTrades []TradeRecord // history of closed trades
|
||||||
|
|
||||||
OnTradeEvent func(event string, data interface{}) // P3-4: real-time SSE push
|
OnTradeEvent func(event string, data interface{}) // P3-4: real-time SSE push
|
||||||
|
|
||||||
|
// Decoupled snapshot for display — snapMu never contended by trading path
|
||||||
|
snapMu sync.RWMutex
|
||||||
|
positionsSnapshot []ArbPosition
|
||||||
}
|
}
|
||||||
|
|
||||||
// TradeRecord stores a finalized trade for stats tracking.
|
// TradeRecord stores a finalized trade for stats tracking.
|
||||||
|
|||||||
Reference in New Issue
Block a user