package main import ( "encoding/json" "fmt" "io/fs" "log" "math" "net/http" "sync" "time" "exchange-monitor/db" ) // ============================================================ // SSE Hub — manages connected browser clients // ============================================================ type sseClient struct { ch chan []byte done chan struct{} filter string } type SSEHub struct { mu sync.RWMutex clients map[*sseClient]bool seq uint64 } func NewSSEHub() *SSEHub { return &SSEHub{ clients: make(map[*sseClient]bool), } } func (h *SSEHub) Subscribe(filter string) *sseClient { c := &sseClient{ ch: make(chan []byte, 64), done: make(chan struct{}), filter: filter, } h.mu.Lock() h.clients[c] = true h.mu.Unlock() log.Printf("[Web] SSE client connected (clients=%d)", len(h.clients)) return c } func (h *SSEHub) Unsubscribe(c *sseClient) { h.mu.Lock() delete(h.clients, c) count := len(h.clients) h.mu.Unlock() close(c.done) log.Printf("[Web] SSE client disconnected (clients=%d)", count) } func (h *SSEHub) Broadcast(event string, data interface{}) { raw, err := json.Marshal(map[string]interface{}{ "event": event, "data": data, "ts": time.Now().UnixMilli(), }) if err != nil { return } h.mu.RLock() defer h.mu.RUnlock() for c := range h.clients { select { case c.ch <- raw: default: } } } // ============================================================ // History Ring Buffers // ============================================================ const maxHistoryPoints = 500 type pricePoint struct { T int64 `json:"t"` P float64 `json:"p"` } type priceHistory struct { mu sync.RWMutex buffers map[string]map[string][]pricePoint } func newPriceHistory() *priceHistory { return &priceHistory{ buffers: make(map[string]map[string][]pricePoint), } } 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) } buf := ph.buffers[coin][exchange] buf = append(buf, pricePoint{T: time.Now().UnixMilli(), P: price}) if len(buf) > maxHistoryPoints { buf = buf[len(buf)-maxHistoryPoints:] } ph.buffers[coin][exchange] = buf } 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) { r := make([]pricePoint, len(buf)) copy(r, buf) return r } r := make([]pricePoint, limit) copy(r, buf[len(buf)-limit:]) return r } // ============================================================ // 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 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), } } func (d *Dashboard) Run() { go d.broadcastLoop() mux := http.NewServeMux() staticSub, err := fs.Sub(staticFS, "web/static") if err != nil { log.Printf("[Web] Failed to create static sub-fs: %v", err) } else { mux.Handle("GET /static/", http.StripPrefix("/static/", http.FileServer(http.FS(staticSub)))) } mux.HandleFunc("GET /", d.handleIndex) 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) 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, } log.Printf("[Web] Dashboard listening on http://%s", d.addr) if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { log.Printf("[Web] Server error: %v", err) } } // ============================================================ // Stats computation — kept separate from trading logic // ============================================================ // DetailedStats holds aggregated PnL and duration statistics. type DetailedStats struct { TotalTrades int `json:"total_trades"` TotalPnlUSD float64 `json:"total_pnl_usd"` // sum of all trade PnL in USD CapitalPnlPct float64 `json:"capital_pnl_pct"` // TotalPnlUSD / InitialCapital * 100 AvgPnlPct float64 `json:"avg_pnl_pct"` MaxProfitPct float64 `json:"max_profit_pct"` MaxLossPct float64 `json:"max_loss_pct"` AvgDuration string `json:"avg_duration"` TotalDuration string `json:"total_duration"` WinningTrades int `json:"winning_trades"` LosingTrades int `json:"losing_trades"` WinRate float64 `json:"win_rate"` } // calcDetailedStats computes trading statistics from a slice of closed trades. // This is a pure function — no dependency on Trader internals. func calcDetailedStats(trades []TradeRecord, initialCapital float64) DetailedStats { ds := DetailedStats{} if len(trades) == 0 { return ds } var totalDur time.Duration ds.MaxLossPct = 1e9 // sentinel for _, tr := range trades { ds.TotalTrades++ ds.TotalPnlUSD += tr.PnlUSD if tr.PnlPct >= 0 { ds.WinningTrades++ if tr.PnlPct > ds.MaxProfitPct { ds.MaxProfitPct = tr.PnlPct } } else { ds.LosingTrades++ if tr.PnlPct < ds.MaxLossPct { ds.MaxLossPct = tr.PnlPct } } if !tr.ClosedAt.IsZero() && !tr.OpenedAt.IsZero() { totalDur += tr.ClosedAt.Sub(tr.OpenedAt) } } if ds.MaxLossPct == 1e9 { ds.MaxLossPct = 0 } if ds.TotalTrades > 0 { ds.CapitalPnlPct = ds.TotalPnlUSD / initialCapital * 100 ds.AvgPnlPct = ds.TotalPnlUSD / float64(ds.TotalTrades) / initialCapital * 100 ds.WinRate = float64(ds.WinningTrades) / float64(ds.TotalTrades) * 100 } if totalDur > 0 { avgDur := totalDur / time.Duration(ds.TotalTrades) ds.AvgDuration = avgDur.Round(time.Second).String() ds.TotalDuration = totalDur.Round(time.Second).String() } return ds } // broadcastLoop pushes data to SSE clients every 1 second. func (d *Dashboard) broadcastLoop() { tick := time.NewTicker(1 * time.Second) defer tick.Stop() for range tick.C { snap := d.store.GetAll() if len(snap) == 0 { continue } // 1. Prices + spreads + connection status var prices []map[string]interface{} for _, coin := range TrackedCoins { exMap := snap[coin.Name] if exMap == nil { continue } entry := map[string]interface{}{ "coin": coin.Name, } for ex, p := range exMap { entry[ex] = p } 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 with live PnL (P3-3) — read from decoupled snapshot, never blocks trader positions := d.trader.ReadSnapshot() 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"), "started_ts": pos.StartedAt.UnixMilli(), } // 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] if bgP > 0 && hlP > 0 { var longCurrent, shortCurrent float64 if pos.LongLeg.Exchange == ExBitget { longCurrent, shortCurrent = bgP, hlP } else { longCurrent, shortCurrent = hlP, bgP } 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 * (takerFees[ExBitget] + takerFees[ExHyperLiquid]) netPnl := longPnl + shortPnl - totalFees currentSpread := (hlP - bgP) / bgP * 100 if pos.LongLeg.Exchange == ExHyperLiquid { // HL→BG: spread positive when bgP > hlP currentSpread = (bgP - hlP) / hlP * 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() scanCopy := d.lastScan d.mu.RUnlock() if len(scanCopy) > 0 { scanList := make([]map[string]interface{}, 0, len(scanCopy)) for _, opp := range scanCopy { scanList = append(scanList, map[string]interface{}{ "coin": opp.Coin, "direction": opp.Direction, "buy_ex": opp.BuyEx, "sell_ex": opp.SellEx, "buy_price": opp.BuyPrice, "sell_price": opp.SellPrice, "net_profit": opp.NetProfit, "gross": opp.GrossBasis, }) } d.hub.Broadcast("arb", scanList) } // 4. Stats + connection status (P3-5) converged, diverged, flat, total := d.trader.GetClosedStats() detail := calcDetailedStats(d.trader.GetClosedTrades(), d.trader.cfg.InitialCapital) stats := map[string]interface{}{ "total_trades": total, "converged": converged, "diverged": diverged, "flat": flat, "open_positions": len(positions), "coins": len(prices), "capital": d.trader.cfg.InitialCapital, // Detailed PnL & duration stats (session only) "detail": map[string]interface{}{ "total_pnl_usd": math.Round(detail.TotalPnlUSD*100) / 100, "capital_pnl": math.Round(detail.CapitalPnlPct*10000) / 10000, "avg_pnl": detail.AvgPnlPct, "max_profit": detail.MaxProfitPct, "max_loss": detail.MaxLossPct, "avg_dur": detail.AvgDuration, "win_rate": detail.WinRate, "wins": detail.WinningTrades, "losses": detail.LosingTrades, "total_dur": detail.TotalDuration, }, } // 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 // Blacklist — stale spread coins bl := d.trader.GetBlacklist() blList := make([]map[string]interface{}, 0, len(bl)) for coin, t := range bl { if d.trader.cfg.BlacklistDuration > 0 && time.Since(t) >= d.trader.cfg.BlacklistDuration { continue // expired, will be cleaned up on next check } remaining := time.Duration(0) if d.trader.cfg.BlacklistDuration > 0 { remaining = d.trader.cfg.BlacklistDuration - time.Since(t) } blList = append(blList, map[string]interface{}{ "coin": coin, "since": t.Format("15:04:05"), "remaining_sec": int(remaining.Seconds()), }) } stats["blacklist"] = blList d.hub.Broadcast("stats", stats) } } // ============================================================ // Public methods called from main.go / trader // ============================================================ func (d *Dashboard) UpdateScan(opps []*ArbOpportunity) { d.mu.Lock() d.lastScan = opps d.scanTime = time.Now() d.mu.Unlock() } 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 // ============================================================ func (d *Dashboard) handleIndex(w http.ResponseWriter, r *http.Request) { data, err := staticFS.ReadFile("web/static/index.html") if err != nil { http.Error(w, "Not found", 404) return } w.Header().Set("Content-Type", "text/html; charset=utf-8") w.Write(data) } func (d *Dashboard) handleStatus(w http.ResponseWriter, r *http.Request) { snap := d.store.GetAll() positions := d.trader.ReadSnapshot() 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}, } writeJSON(w, resp) } 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 == "" { snap := d.store.GetAll() coins := make([]string, 0, len(snap)) for c := range snap { coins = append(coins, c) } 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, "exchange": exchange, "points": points, }) } // 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, "page": page, "limit": limit, }) } func (d *Dashboard) handleTradeDetail(w http.ResponseWriter, r *http.Request) { if d.db == nil { 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, }) } func (d *Dashboard) handleSSE(w http.ResponseWriter, r *http.Request) { flusher, ok := w.(http.Flusher) if !ok { http.Error(w, "Streaming not supported", 500) return } w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") w.Header().Set("Access-Control-Allow-Origin", "*") client := d.hub.Subscribe("") defer d.hub.Unsubscribe(client) fmt.Fprintf(w, "event: connected\ndata: {\"status\":\"ok\"}\n\n") flusher.Flush() for { select { case <-r.Context().Done(): return case msg, ok := <-client.ch: if !ok { return } fmt.Fprintf(w, "data: %s\n\n", msg) flusher.Flush() } } } 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 }