From 8ccb439f7a9d10db7ed25105af6b3dcd637313ba Mon Sep 17 00:00:00 2001 From: jackyu66git Date: Mon, 4 May 2026 20:34:00 +0800 Subject: [PATCH] refactor: replace REST API with Unix socket IPC + CLI subcommands - Remove insecure HTTP API endpoints (positions, close, close-all, pnl) - Add Unix socket IPC at /tmp/exchange-monitor.sock - Add CLI subcommands: status, close-all, close , stop, start - Bind dashboard HTTP to 127.0.0.1:8888 (localhost only) --- dashboard.go | 112 ---------------------------------------------- ipc.go | 123 +++++++++++++++++++++++++++++++++++++++++++++++++++ main.go | 26 ++++++++++- 3 files changed, 147 insertions(+), 114 deletions(-) create mode 100644 ipc.go diff --git a/dashboard.go b/dashboard.go index 15e89ac..b259cf5 100644 --- a/dashboard.go +++ b/dashboard.go @@ -8,7 +8,6 @@ import ( "math" "net/http" "os" - "strings" "sync" "time" @@ -243,10 +242,6 @@ func (d *Dashboard) Run() { mux.HandleFunc("GET /events", d.handleSSE) mux.HandleFunc("POST /api/stop", d.handleStop) mux.HandleFunc("POST /api/start", d.handleStart) - mux.HandleFunc("GET /api/positions", d.handlePositions) - mux.HandleFunc("GET /api/pnl", d.handlePnL) - mux.HandleFunc("POST /api/close/", d.handleClosePosition) - mux.HandleFunc("POST /api/close-all", d.handleCloseAll) server := &http.Server{ Addr: d.addr, @@ -758,113 +753,6 @@ func (d *Dashboard) handleStart(w http.ResponseWriter, r *http.Request) { writeJSON(w, map[string]string{"status": "started", "message": "Trading resumed"}) } -func (d *Dashboard) handlePositions(w http.ResponseWriter, r *http.Request) { - positions := d.trader.ReadSnapshot() - posList := make([]map[string]interface{}, 0, len(positions)) - for _, pos := range positions { - entry := map[string]interface{}{ - "id": pos.DBTradeID, - "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"), - "status": pos.Status, - "long_exchange": pos.LongLeg.Exchange, - "short_exchange": pos.ShortLeg.Exchange, - "long_entry": pos.LongLeg.EntryPrice, - "short_entry": pos.ShortLeg.EntryPrice, - "long_entry_prices": pos.LongEntryPrices, - "short_entry_prices": pos.ShortEntryPrices, - } - // Live PnL from current prices - snap := d.store.GetAll() - 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) - shortAvg := weightedAvgPrice(pos.ShortEntryPrices, pos.AmountUSD) - longPnl := (longCurrent - longAvg) / longAvg * 100 - shortPnl := (shortAvg - shortCurrent) / shortAvg * 100 - feeEntryUSD := pos.AmountUSD * (takerFees[ExBitget] + takerFees[ExHyperLiquid]) / 100 - feeExitUSD := pos.AmountUSD * (takerFees[ExBitget] + takerFees[ExHyperLiquid]) / 100 - pricePnLUSD := pos.AmountUSD * (longPnl + shortPnl) / 100 - netPnLUSD := pricePnLUSD - feeEntryUSD - feeExitUSD - entry["pnl_pct"] = math.Round((longPnl+shortPnl)*10000) / 10000 - entry["pnl_usd"] = math.Round(netPnLUSD*100) / 100 - entry["long_pnl_pct"] = math.Round(longPnl*10000) / 10000 - entry["short_pnl_pct"] = math.Round(shortPnl*10000) / 10000 - entry["long_current"] = longCurrent - entry["short_current"] = shortCurrent - } - } - posList = append(posList, entry) - } - writeJSON(w, map[string]interface{}{ - "positions": posList, - "count": len(posList), - }) -} - -func (d *Dashboard) handleClosePosition(w http.ResponseWriter, r *http.Request) { - // POST /api/close/{coin} - coin := strings.TrimPrefix(r.URL.Path, "/api/close/") - if coin == "" || coin == r.URL.Path { - http.Error(w, "Missing coin name", 400) - return - } - if err := d.trader.ClosePosition(coin); err != nil { - http.Error(w, err.Error(), 400) - return - } - writeJSON(w, map[string]string{"status": "closed", "coin": coin, "message": "Position closed"}) -} - -func (d *Dashboard) handleCloseAll(w http.ResponseWriter, r *http.Request) { - count := d.trader.CloseAllPositions() - writeJSON(w, map[string]interface{}{ - "status": "closed", - "count": count, - "message": fmt.Sprintf("Closed %d positions", count), - }) -} - -func (d *Dashboard) handlePnL(w http.ResponseWriter, r *http.Request) { - converged, diverged, flat, total := d.trader.GetClosedStats() - trades := d.trader.GetClosedTrades() - detail := calcDetailedStats(trades, d.trader.cfg.InitialCapital) - positions := d.trader.ReadSnapshot() - writeJSON(w, map[string]interface{}{ - "total_trades": total, - "converged": converged, - "diverged": diverged, - "flat": flat, - "open_positions": len(positions), - "capital": d.trader.cfg.InitialCapital, - "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, - }, - }) -} - func writeJSON(w http.ResponseWriter, v interface{}) { w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(v) diff --git a/ipc.go b/ipc.go new file mode 100644 index 0000000..3295462 --- /dev/null +++ b/ipc.go @@ -0,0 +1,123 @@ +package main + +import ( + "encoding/json" + "fmt" + "log" + "net" + "os" + "time" +) + +const sockPath = "/tmp/exchange-monitor.sock" + +// IPCCommand is sent from CLI client to daemon. +type IPCCommand struct { + Action string `json:"action"` // status, close-all, close, stop, start + Coin string `json:"coin,omitempty"` +} + +// IPCResponse is sent back from daemon to CLI client. +type IPCResponse struct { + Success bool `json:"success"` + Data interface{} `json:"data,omitempty"` + Error string `json:"error,omitempty"` +} + +// startIPCServer starts the Unix socket listener for CLI commands. +func (t *Trader) startIPCServer() { + os.Remove(sockPath) // clean up stale socket + + ln, err := net.Listen("unix", sockPath) + if err != nil { + log.Printf("[IPC] Failed to create socket: %v", err) + return + } + log.Printf("[IPC] Listening on %s", sockPath) + + go func() { + defer ln.Close() + for { + conn, err := ln.Accept() + if err != nil { + continue + } + go t.handleIPC(conn) + } + }() +} + +func (t *Trader) handleIPC(conn net.Conn) { + defer conn.Close() + conn.SetDeadline(time.Now().Add(5 * time.Second)) + + var cmd IPCCommand + if err := json.NewDecoder(conn).Decode(&cmd); err != nil { + json.NewEncoder(conn).Encode(IPCResponse{Success: false, Error: "invalid command: " + err.Error()}) + return + } + + var resp IPCResponse + switch cmd.Action { + case "status": + positions := t.ReadSnapshot() + c, d, f, tot := t.GetClosedStats() + resp = IPCResponse{Success: true, Data: map[string]interface{}{ + "positions": positions, + "converged": c, "diverged": d, "flat": f, "total": tot, + }} + case "close-all": + count := t.CloseAllPositions() + resp = IPCResponse{Success: true, Data: map[string]interface{}{ + "closed": count, "message": fmt.Sprintf("Closed %d positions", count), + }} + case "close": + if cmd.Coin == "" { + resp = IPCResponse{Success: false, Error: "missing coin name"} + } else if err := t.ClosePosition(cmd.Coin); err != nil { + resp = IPCResponse{Success: false, Error: err.Error()} + } else { + resp = IPCResponse{Success: true, Data: map[string]string{"closed": cmd.Coin}} + } + case "stop": + t.Stop() + resp = IPCResponse{Success: true, Data: map[string]string{"status": "stopped"}} + case "start": + t.Start() + resp = IPCResponse{Success: true, Data: map[string]string{"status": "started"}} + default: + resp = IPCResponse{Success: false, Error: "unknown action: " + cmd.Action} + } + json.NewEncoder(conn).Encode(resp) +} + +// runIPCClient sends a command to the running daemon and prints the response. +func runIPCClient(action, coin string) { + conn, err := net.DialTimeout("unix", sockPath, 2*time.Second) + if err != nil { + fmt.Fprintf(os.Stderr, "Error: daemon not running? (%v)\n", err) + os.Exit(1) + } + defer conn.Close() + + cmd := IPCCommand{Action: action, Coin: coin} + if err := json.NewEncoder(conn).Encode(cmd); err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + + var resp IPCResponse + if err := json.NewDecoder(conn).Decode(&resp); err != nil { + fmt.Fprintf(os.Stderr, "Error reading response: %v\n", err) + os.Exit(1) + } + + if !resp.Success { + fmt.Fprintf(os.Stderr, "Error: %s\n", resp.Error) + os.Exit(1) + } + + // Pretty-print response + data, _ := json.MarshalIndent(resp.Data, "", " ") + fmt.Println(string(data)) +} diff --git a/main.go b/main.go index 87c265b..0c4e519 100644 --- a/main.go +++ b/main.go @@ -18,6 +18,25 @@ import ( ) func main() { + // CLI subcommand mode: talk to running daemon via IPC + if len(os.Args) > 1 { + switch os.Args[1] { + case "status", "close-all", "stop", "start": + runIPCClient(os.Args[1], "") + case "close": + if len(os.Args) < 3 { + fmt.Fprintln(os.Stderr, "Usage: exchange-monitor close ") + os.Exit(1) + } + runIPCClient("close", os.Args[2]) + default: + fmt.Fprintf(os.Stderr, "Unknown command: %s\n", os.Args[1]) + fmt.Fprintln(os.Stderr, "Commands: status, close-all, close , stop, start") + os.Exit(1) + } + return + } + log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile) // Set up multi-writer: stdout + log file @@ -52,8 +71,11 @@ func main() { // Initialize trader trader := NewTrader(cfg, database) - // Initialize dashboard (web server + SSE) - dashboard := NewDashboard(store, trader, database, ":8888") + // Start Unix socket IPC for CLI commands + trader.startIPCServer() + + // Initialize dashboard (web server + SSE) — localhost only for security + dashboard := NewDashboard(store, trader, database, "127.0.0.1:8888") go dashboard.Run() // Spread window tracker — measures how long spreads stay above threshold