jack/exchange-monitor-go
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 3: Real-time enhancements

Browse Source 
P3-1: Scan optimization — only BG↔HL (50+ pair combos → 2)
P3-2: Real-time spread chart — spreadHistory ring buffer +
      /api/spread-history endpoint + Chart.js spread chart
P3-3: Live position PnL — positions SSE now includes
      estimated current profit/loss + current spread
P3-4: Real-time trade events — trader.OnTradeEvent callback
      fires SSE 'trade_open' / 'trade_close' immediately
P3-5: Connection status monitoring — tracks last update time
      per exchange, broadcast via stats.connections + /api/connections

Frontend: spread chart card, PnL column in positions,
          connection status dots in stats bar,
          green/red border flash on trade events
This commit is contained in:
jackyu66git
2026-05-03 18:05:19 +08:00
parent da561325d7
commit beb3611778
7 changed files with 528 additions and 316 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dashboard.go
+198 -59
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"io/fs"
"log"
"math"
"net/http"
"sync"
"time"
@@ -19,12 +20,13 @@ import (
type sseClient struct {
ch chan []byte
done chan struct{}
filter string // optional coin filter (empty = all)
filter string
}
type SSEHub struct {
mu sync.RWMutex
clients map[*sseClient]bool
seq uint64
}
func NewSSEHub() *SSEHub {
@@ -71,25 +73,24 @@ func (h *SSEHub) Broadcast(event string, data interface{}) {
select {
case c.ch <- raw:
default:
// Client too slow, skip
}
}
}
// ============================================================
// Price History — ring buffer for charting
// History Ring Buffers
// ============================================================
const maxHistoryPoints = 500
type pricePoint struct {
T int64 `json:"t"` // unix ms
T int64 `json:"t"`
P float64 `json:"p"`
}
type priceHistory struct {
mu sync.RWMutex
buffers map[string]map[string][]pricePoint // coin -> exchange -> points
buffers map[string]map[string][]pricePoint
}
func newPriceHistory() *priceHistory {
@@ -101,7 +102,6 @@ func newPriceHistory() *priceHistory {
func (ph *priceHistory) Record(coin, exchange string, price float64) {
ph.mu.Lock()
defer ph.mu.Unlock()
if ph.buffers[coin] == nil {
ph.buffers[coin] = make(map[string][]pricePoint)
}
@@ -116,58 +116,107 @@ func (ph *priceHistory) Record(coin, exchange string, price float64) {
func (ph *priceHistory) GetHistory(coin, exchange string, limit int) []pricePoint {
ph.mu.RLock()
defer ph.mu.RUnlock()
buf := ph.buffers[coin][exchange]
if len(buf) == 0 {
return nil
}
if limit <= 0 || limit >= len(buf) {
result := make([]pricePoint, len(buf))
copy(result, buf)
return result
r := make([]pricePoint, len(buf))
copy(r, buf)
return r
}
result := make([]pricePoint, limit)
copy(result, buf[len(buf)-limit:])
return result
r := make([]pricePoint, limit)
copy(r, buf[len(buf)-limit:])
return r
}
// ============================================================
// Dashboard — main orchestrator
// Spread History — tracks BG↔HL spread % per coin (P3-2)
// ============================================================
type spreadPoint struct {
T int64 `json:"t"`
Spread float64 `json:"s"` // spread % (positive = BG cheaper than HL for BG->HL direction)
}
type spreadHistory struct {
mu sync.RWMutex
buffers map[string][]spreadPoint // coin -> spread points
}
func newSpreadHistory() *spreadHistory {
return &spreadHistory{
buffers: make(map[string][]spreadPoint),
}
}
func (sh *spreadHistory) Record(coin string, spread float64) {
sh.mu.Lock()
defer sh.mu.Unlock()
sh.buffers[coin] = append(sh.buffers[coin], spreadPoint{T: time.Now().UnixMilli(), Spread: spread})
if len(sh.buffers[coin]) > maxHistoryPoints {
sh.buffers[coin] = sh.buffers[coin][len(sh.buffers[coin])-maxHistoryPoints:]
}
}
func (sh *spreadHistory) GetHistory(coin string, limit int) []spreadPoint {
sh.mu.RLock()
defer sh.mu.RUnlock()
buf := sh.buffers[coin]
if len(buf) == 0 {
return nil
}
if limit <= 0 || limit >= len(buf) {
r := make([]spreadPoint, len(buf))
copy(r, buf)
return r
}
r := make([]spreadPoint, limit)
copy(r, buf[len(buf)-limit:])
return r
}
// ============================================================
// Dashboard
// ============================================================
type Dashboard struct {
hub *SSEHub
history *priceHistory
spreads *spreadHistory
store *PriceStore
trader *Trader
db *db.DB
addr string
// cached arb scan results — updated every tick
mu sync.RWMutex
lastScan []*ArbOpportunity
scanTime time.Time
// cached arb scan results
mu sync.RWMutex
lastScan []*ArbOpportunity
scanTime time.Time
// P3-5: connection status — exchange -> last update time
connMu sync.RWMutex
connMap map[string]time.Time // exchange name -> last price timestamp
}
func NewDashboard(store *PriceStore, trader *Trader, database *db.DB, addr string) *Dashboard {
return &Dashboard{
hub: NewSSEHub(),
history: newPriceHistory(),
spreads: newSpreadHistory(),
store: store,
trader: trader,
db: database,
addr: addr,
connMap: make(map[string]time.Time),
}
}
// Run starts the HTTP server + SSE broadcaster goroutine.
func (d *Dashboard) Run() {
// SSE broadcaster — pushes data every ~1s
go d.broadcastLoop()
mux := http.NewServeMux()
// Static files — embed subdirectory
staticSub, err := fs.Sub(staticFS, "web/static")
if err != nil {
log.Printf("[Web] Failed to create static sub-fs: %v", err)
@@ -175,23 +224,20 @@ func (d *Dashboard) Run() {
mux.Handle("GET /static/", http.StripPrefix("/static/", http.FileServer(http.FS(staticSub))))
}
// Main page
mux.HandleFunc("GET /", d.handleIndex)
// API endpoints
mux.HandleFunc("GET /api/status", d.handleStatus)
mux.HandleFunc("GET /api/history", d.handleHistory)
mux.HandleFunc("GET /api/spread-history", d.handleSpreadHistory) // P3-2
mux.HandleFunc("GET /api/trades", d.handleTrades)
mux.HandleFunc("GET /api/trade/", d.handleTradeDetail)
// SSE
mux.HandleFunc("GET /api/connections", d.handleConnStatus) // P3-5
mux.HandleFunc("GET /events", d.handleSSE)
server := &http.Server{
Addr: d.addr,
Handler: mux,
ReadTimeout: 10 * time.Second,
WriteTimeout: 0, // SSE needs no write timeout
WriteTimeout: 0,
}
log.Printf("[Web] Dashboard listening on http://%s", d.addr)
@@ -211,7 +257,7 @@ func (d *Dashboard) broadcastLoop() {
continue
}
// 1. Prices
// 1. Prices + spreads + connection status
var prices []map[string]interface{}
for _, coin := range TrackedCoins {
exMap := snap[coin.Name]
@@ -224,34 +270,65 @@ func (d *Dashboard) broadcastLoop() {
for ex, p := range exMap {
entry[ex] = p
}
// Add bid-ask spreads
for ex := range exMap {
sp := d.store.GetSpread(coin.Name, ex)
if sp > 0 {
entry[ex+"_spread"] = sp
}
}
// P3-2: Calculate BG↔HL spread and record
bgP := exMap[ExBitget]
hlP := exMap[ExHyperLiquid]
if bgP > 0 && hlP > 0 {
spreadPct := (hlP - bgP) / bgP * 100
entry["bg_hl_spread"] = spreadPct
d.spreads.Record(coin.Name, spreadPct)
}
prices = append(prices, entry)
}
d.hub.Broadcast("prices", prices)
// 2. Open positions
// 2. Open positions with live PnL (P3-3)
positions := d.trader.GetOpenPositions()
if len(positions) > 0 {
posList := make([]map[string]interface{}, 0, len(positions))
for _, pos := range positions {
posList = append(posList, map[string]interface{}{
"coin": pos.Coin,
"direction": pos.Direction,
"amount_usd": pos.AmountUSD,
"entry_spread": pos.EntrySpread,
"scales": pos.ScaleLevels,
"duration": time.Since(pos.StartedAt).Round(time.Second).String(),
"started_at": pos.StartedAt.Format("15:04:05"),
})
posList := make([]map[string]interface{}, 0, len(positions))
for _, pos := range positions {
posEntry := map[string]interface{}{
"coin": pos.Coin,
"direction": pos.Direction,
"amount_usd": pos.AmountUSD,
"entry_spread": pos.EntrySpread,
"scales": pos.ScaleLevels,
"duration": time.Since(pos.StartedAt).Round(time.Second).String(),
"started_at": pos.StartedAt.Format("15:04:05"),
}
d.hub.Broadcast("positions", posList)
// Calculate live PnL from current prices
if exMap := snap[pos.Coin]; exMap != nil {
bgP := exMap[ExBitget]
hlP := exMap[ExHyperLiquid]
if bgP > 0 && hlP > 0 {
var longCurrent, shortCurrent float64
if pos.LongLeg.Exchange == ExBitget {
longCurrent, shortCurrent = bgP, hlP
} else {
longCurrent, shortCurrent = hlP, bgP
}
longPnl := (longCurrent - pos.LongLeg.EntryPrice) / pos.LongLeg.EntryPrice * 100
shortPnl := (pos.ShortLeg.EntryPrice - shortCurrent) / pos.ShortLeg.EntryPrice * 100
totalFees := 2 * (makerFees[ExBitget] + makerFees[ExHyperLiquid])
netPnl := longPnl + shortPnl - totalFees
currentSpread := (hlP - bgP) / bgP * 100
posEntry["current_spread"] = math.Round(currentSpread*10000) / 10000
posEntry["pnl_est"] = math.Round(netPnl*10000) / 10000
}
}
posList = append(posList, posEntry)
}
d.hub.Broadcast("positions", posList)
// 3. Arb scan results
d.mu.RLock()
@@ -275,7 +352,7 @@ func (d *Dashboard) broadcastLoop() {
d.hub.Broadcast("arb", scanList)
}
// 4. Stats
// 4. Stats + connection status (P3-5)
converged, diverged, flat, total := d.trader.GetClosedStats()
stats := map[string]interface{}{
"total_trades": total,
@@ -285,11 +362,31 @@ func (d *Dashboard) broadcastLoop() {
"open_positions": len(positions),
"coins": len(prices),
}
// Connection status
d.connMu.RLock()
connInfo := make(map[string]string)
for ex, lastTime := range d.connMap {
age := time.Since(lastTime)
if age < 10*time.Second {
connInfo[ex] = "online"
} else if age < 30*time.Second {
connInfo[ex] = "stale"
} else {
connInfo[ex] = "offline"
}
}
d.connMu.RUnlock()
stats["connections"] = connInfo
d.hub.Broadcast("stats", stats)
}
}
// UpdateScan caches the latest arb scan results.
// ============================================================
// Public methods called from main.go / trader
// ============================================================
func (d *Dashboard) UpdateScan(opps []*ArbOpportunity) {
d.mu.Lock()
d.lastScan = opps
@@ -297,11 +394,22 @@ func (d *Dashboard) UpdateScan(opps []*ArbOpportunity) {
d.mu.Unlock()
}
// RecordPrice adds a price to the history buffer and optionally broadcasts.
func (d *Dashboard) RecordPrice(coin, exchange string, price float64) {
d.history.Record(coin, exchange, price)
}
// RecordConnStatus updates the last-seen time for an exchange (P3-5).
func (d *Dashboard) RecordConnStatus(exchange string) {
d.connMu.Lock()
d.connMap[exchange] = time.Now()
d.connMu.Unlock()
}
// BroadcastEvent sends an immediate SSE event (P3-4).
func (d *Dashboard) BroadcastEvent(event string, data interface{}) {
d.hub.Broadcast(event, data)
}
// ============================================================
// HTTP Handlers
// ============================================================
@@ -322,11 +430,10 @@ func (d *Dashboard) handleStatus(w http.ResponseWriter, r *http.Request) {
converged, diverged, flat, total := d.trader.GetClosedStats()
resp := map[string]interface{}{
"prices": snap,
"positions": len(positions),
"stats": map[string]int{"total": total, "converged": converged, "diverged": diverged, "flat": flat},
"prices": snap,
"positions": len(positions),
"stats": map[string]int{"total": total, "converged": converged, "diverged": diverged, "flat": flat},
}
writeJSON(w, resp)
}
@@ -334,7 +441,6 @@ func (d *Dashboard) handleHistory(w http.ResponseWriter, r *http.Request) {
coin := r.URL.Query().Get("coin")
exchange := r.URL.Query().Get("exchange")
if coin == "" || exchange == "" {
// Return available coins/exchanges
snap := d.store.GetAll()
coins := make([]string, 0, len(snap))
for c := range snap {
@@ -343,7 +449,6 @@ func (d *Dashboard) handleHistory(w http.ResponseWriter, r *http.Request) {
writeJSON(w, map[string]interface{}{"coins": coins, "exchanges": []string{"Binance", "HyperLiquid", "Bitget", "dYdX"}})
return
}
points := d.history.GetHistory(coin, exchange, 300)
writeJSON(w, map[string]interface{}{
"coin": coin,
@@ -352,22 +457,52 @@ func (d *Dashboard) handleHistory(w http.ResponseWriter, r *http.Request) {
})
}
// handleSpreadHistory returns BG↔HL spread history for a coin (P3-2).
func (d *Dashboard) handleSpreadHistory(w http.ResponseWriter, r *http.Request) {
coin := r.URL.Query().Get("coin")
if coin == "" {
writeJSON(w, map[string]interface{}{"coins": trackedCoinNames()})
return
}
points := d.spreads.GetHistory(coin, 300)
writeJSON(w, map[string]interface{}{
"coin": coin,
"points": points,
})
}
// handleConnStatus returns connection health for all exchanges (P3-5).
func (d *Dashboard) handleConnStatus(w http.ResponseWriter, r *http.Request) {
d.connMu.RLock()
conns := make(map[string]string)
for ex, t := range d.connMap {
age := time.Since(t)
switch {
case age < 10*time.Second:
conns[ex] = "online"
case age < 30*time.Second:
conns[ex] = "stale"
default:
conns[ex] = "offline"
}
}
d.connMu.RUnlock()
writeJSON(w, conns)
}
func (d *Dashboard) handleTrades(w http.ResponseWriter, r *http.Request) {
if d.db == nil {
writeJSON(w, map[string]interface{}{"trades": []interface{}{}, "total": 0})
return
}
page := 1
limit := 20
coin := r.URL.Query().Get("coin")
trades, total, err := d.db.GetTrades(page, limit, coin)
if err != nil {
http.Error(w, err.Error(), 500)
return
}
writeJSON(w, map[string]interface{}{
"trades": trades,
"total": total,
@@ -381,19 +516,16 @@ func (d *Dashboard) handleTradeDetail(w http.ResponseWriter, r *http.Request) {
http.Error(w, "DB not available", 503)
return
}
var id int64
if _, err := fmt.Sscanf(r.URL.Path, "/api/trade/%d", &id); err != nil {
http.Error(w, "Invalid trade ID", 400)
return
}
trade, orders, err := d.db.GetTradeByID(id)
if err != nil {
http.Error(w, err.Error(), 404)
return
}
writeJSON(w, map[string]interface{}{
"trade": trade,
"orders": orders,
@@ -415,7 +547,6 @@ func (d *Dashboard) handleSSE(w http.ResponseWriter, r *http.Request) {
client := d.hub.Subscribe("")
defer d.hub.Unsubscribe(client)
// Send initial heartbeat
fmt.Fprintf(w, "event: connected\ndata: {\"status\":\"ok\"}\n\n")
flusher.Flush()
@@ -437,3 +568,11 @@ func writeJSON(w http.ResponseWriter, v interface{}) {
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(v)
}
func trackedCoinNames() []string {
names := make([]string, len(TrackedCoins))
for i, c := range TrackedCoins {
names[i] = c.Name
}
return names
}