From 719f0a061db2b39393f72d0797853d21328e0e09 Mon Sep 17 00:00:00 2001 From: jackyu66git Date: Sun, 3 May 2026 16:54:36 +0800 Subject: [PATCH] Initial commit --- .env.example | 39 +++ .gitignore | 18 ++ config.go | 83 +++++ exchange/aevo.go | 129 ++++++++ exchange/binance.go | 79 +++++ exchange/bitget.go | 95 ++++++ exchange/bitget_trade.go | 138 +++++++++ exchange/connector.go | 168 ++++++++++ exchange/dydx.go | 174 +++++++++++ exchange/helpers.go | 42 +++ exchange/hyperliquid.go | 66 ++++ exchange/hyperliquid_trade.go | 188 ++++++++++++ exchange/ping.go | 16 + go.mod | 5 + go.sum | 2 + main.go | 210 +++++++++++++ morning_report.sh | 35 +++ notifier.go | 92 ++++++ run_test.sh | 4 + scanner.go | 175 +++++++++++ trader.go | 557 ++++++++++++++++++++++++++++++++++ types.go | 121 ++++++++ 22 files changed, 2436 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 config.go create mode 100644 exchange/aevo.go create mode 100644 exchange/binance.go create mode 100644 exchange/bitget.go create mode 100644 exchange/bitget_trade.go create mode 100644 exchange/connector.go create mode 100644 exchange/dydx.go create mode 100644 exchange/helpers.go create mode 100644 exchange/hyperliquid.go create mode 100644 exchange/hyperliquid_trade.go create mode 100644 exchange/ping.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 main.go create mode 100644 morning_report.sh create mode 100644 notifier.go create mode 100755 run_test.sh create mode 100644 scanner.go create mode 100644 trader.go create mode 100644 types.go diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..ad68996 --- /dev/null +++ b/.env.example @@ -0,0 +1,39 @@ +# 交易所监控 + 自动套利 +# 复制为 .env 并填入实际值 + +# Telegram 推送 +TELEGRAM_BOT_TOKEN=*** +TELEGRAM_CHAT_ID=你的聊天ID + +# ============================================ +# 自动交易开关 (设置为 1 启用) +TRADE_ENABLED=0 + +# 交易参数 +TRADE_THRESHOLD=0.15 # 最低套利利润率 (%) +TRADE_AMOUNT_USD=10 # 每腿金额 (USDT) +TRADE_COOLDOWN_MS=30000 # 同一币种套利冷却 (毫秒) + +# Bitget API (需开通合约API) +BITGET_API_KEY=*** +BITGET_API_SECRET=*** +BITGET_PASSPHRASE=你的密码短语 + +# HyperLiquid API (钱包私钥) +HL_PRIVATE_KEY=你的ed25519私钥(hex) +HL_ADDRESS=你的钱包地址 + +# ============================================ +# 以下为 ema-monitor 使用的参数 (保持不变) +DATA_API_BASE=http://localhost:80 +SYMBOL=BTC/USDT:USDT +FETCH_LIMIT_BASE=8000 +POLL_INTERVAL=10 +PROXIMITY_THRESHOLD_PCT=0.15 +ALERT_COOLDOWN=3600 + +# ============================================ +# 测试模式 (模拟交易，不需要真实 API Key) +# TEST_MODE=true 时，TRADE_ENABLED 被忽略 +TEST_MODE=false +MOCK_SLIPPAGE_PCT=0.005 # 每腿模拟滑点 (%) diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0ee7f89 --- /dev/null +++ b/.gitignore @@ -0,0 +1,18 @@ +# Binaries +exchange-monitor +*.test +*.out + +# Env files with secrets +.env + +# Log files +*.log +trade_stats.txt + +# Test files +test_*.go + +# IDE +.vscode/ +.idea/ diff --git a/config.go b/config.go new file mode 100644 index 0000000..0630d28 --- /dev/null +++ b/config.go @@ -0,0 +1,83 @@ +package main + +import ( + "os" + "strconv" +) + +type Config struct { + TelegramBotToken string + TelegramChatID string + AlertCooldownSec int // seconds between alerts for same coin + ArbThreshold float64 // minimum net profit % to trigger alert + ScanIntervalMs int // how often scanner runs (milliseconds) + + // Automated trading + TradeEnabled bool + TradeThreshold float64 // minimum profit % to execute trade (>0.15%) + TradeAmountUSD float64 // amount per trade in USDT + TradeCooldownMs int // ms between trades of same coin + + // Test mode (no real API keys needed) + TestMode bool + MockSlippagePct float64 // simulated slippage per order (e.g. 0.01 = 0.01%) + + // Bitget API + BitgetAPIKey string + BitgetAPISecret string + BitgetPassphrase string + + // HyperLiquid API + HLPrivateKey string // ed25519 private key hex + HLAddress string // wallet address +} + +func LoadConfig() *Config { + getEnv := func(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def + } + getFloat := func(key string, def float64) float64 { + v := os.Getenv(key) + if v == "" { + return def + } + f, err := strconv.ParseFloat(v, 64) + if err != nil { + return def + } + return f + } + getBool := func(key string, def bool) bool { + v := os.Getenv(key) + if v == "" { + return def + } + return v == "1" || v == "true" || v == "yes" + } + + return &Config{ + TelegramBotToken: getEnv("TELEGRAM_BOT_TOKEN", ""), + TelegramChatID: getEnv("TELEGRAM_CHAT_ID", ""), + AlertCooldownSec: 300, + ArbThreshold: 0.03, + ScanIntervalMs: 500, + + TradeEnabled: getBool("TRADE_ENABLED", false), + TradeThreshold: getFloat("TRADE_THRESHOLD", 0.15), + TradeAmountUSD: getFloat("TRADE_AMOUNT_USD", 10), + TradeCooldownMs: 30000, + + TestMode: getBool("TEST_MODE", false), + MockSlippagePct: getFloat("MOCK_SLIPPAGE_PCT", 0.005), + + BitgetAPIKey: getEnv("BITGET_API_KEY", ""), + BitgetAPISecret: getEnv("BITGET_API_SECRET", ""), + BitgetPassphrase: getEnv("BITGET_PASSPHRASE", ""), + + HLPrivateKey: getEnv("HL_PRIVATE_KEY", ""), + HLAddress: getEnv("HL_ADDRESS", ""), + } +} diff --git a/exchange/aevo.go b/exchange/aevo.go new file mode 100644 index 0000000..aaa7bb0 --- /dev/null +++ b/exchange/aevo.go @@ -0,0 +1,129 @@ +package exchange + +import ( + "encoding/json" + "log" + "time" +) + +// AevoWS connects to Aevo WebSocket for ticker data. +type AevoWS struct { + Conn *PriceConnector + Tracked []TrackedSymbol // coin + instrument name pairs +} + +type TrackedSymbol struct { + Coin string // "BTC" + InstrumentID string // "BTC-PERP" +} + +type aevoTickerMsg struct { + Op string `json:"op"` + Data json.RawMessage `json:"data"` +} + +type aevoTickerData struct { + Timestamp string `json:"timestamp"` + Tickers []aevoInstrument `json:"tickers"` +} + +type aevoInstrument struct { + InstrumentName string `json:"instrument_name"` + Mark *aevoPriceObj `json:"mark,omitempty"` + LastPrice string `json:"last_price,omitempty"` +} + +type aevoPriceObj struct { + Price string `json:"price"` +} + +func NewAevoWS(tracked []TrackedSymbol) *AevoWS { + ae := &AevoWS{ + Tracked: tracked, + Conn: NewPriceConnector("wss://ws.aevo.xyz", "Aevo", 120*time.Second, 30*time.Second), + } + ae.Conn.PingInterval = 45 * time.Second + return ae +} + +// Run connects to Aevo WS and streams ticker data. +func (a *AevoWS) Run(updateFn func(coin string, price float64)) error { + a.Conn.OnConnect = func() { + log.Printf("[Aevo WS] Connected, subscribing to %d tickers", len(a.Tracked)) + + for _, t := range a.Tracked { + sub := map[string]interface{}{ + "op": "subscribe", + "data": []string{"ticker:" + t.Coin}, + } + if err := a.Conn.SendJSON(sub); err != nil { + log.Printf("[Aevo WS] Subscribe %s error: %v", t.InstrumentID, err) + } + } + } + + a.Conn.OnMessage = func(msg []byte) { + var raw map[string]json.RawMessage + if err := json.Unmarshal(msg, &raw); err != nil { + return + } + + // Check for error + if errMsg, hasErr := raw["error"]; hasErr { + var errStr string + json.Unmarshal(errMsg, &errStr) + if errStr != "" { + // Log once, skip errors + return + } + } + + // Parse ticker data + op, hasOp := raw["op"] + if !hasOp { + return + } + var opStr string + if err := json.Unmarshal(op, &opStr); err != nil || opStr != "ticker" { + return + } + + dataRaw, hasData := raw["data"] + if !hasData { + return + } + + var data aevoTickerData + if err := json.Unmarshal(dataRaw, &data); err != nil { + return + } + + for _, ticker := range data.Tickers { + // Find the coin for this instrument + coin := "" + for _, t := range a.Tracked { + if t.InstrumentID == ticker.InstrumentName { + coin = t.Coin + break + } + } + if coin == "" { + continue + } + + // Try mark price first, then last_price + var price float64 + if ticker.Mark != nil && ticker.Mark.Price != "" { + price = parseFloat(ticker.Mark.Price) + } else if ticker.LastPrice != "" { + price = parseFloat(ticker.LastPrice) + } + + if price > 0 { + updateFn(coin, price) + } + } + } + + return a.Conn.Run() +} diff --git a/exchange/binance.go b/exchange/binance.go new file mode 100644 index 0000000..77dbce2 --- /dev/null +++ b/exchange/binance.go @@ -0,0 +1,79 @@ +package exchange + +import ( + "encoding/json" + "fmt" + "log" + "strconv" + "strings" + "time" +) + +type BinanceWS struct { + Tracked []string +} + +func NewBinanceWS(tracked []string) *BinanceWS { + return &BinanceWS{Tracked: tracked} +} +// Run connects to Binance WS and streams bookTicker data. +func (b *BinanceWS) Run(updateFn func(coin string, price, bid, ask float64)) error { + streams := "" + for i, sym := range b.Tracked { + if i > 0 { + streams += "/" + } + streams += fmt.Sprintf("%s@bookTicker", strings.ToLower(sym)) + } + url := fmt.Sprintf("wss://fstream.binance.com/stream?streams=%s", streams) + + conn := NewPriceConnector(url, "Binance", 120*time.Second, 30*time.Second) + conn.PingInterval = 45 * time.Second + + conn.OnConnect = func() { + log.Printf("[Binance WS] Connected") + } + + conn.OnMessage = func(msg []byte) { + // Combined stream: {"stream":"...","data":{...}} + // Navigate through "data" using map to avoid field name conflicts + var raw map[string]json.RawMessage + if err := json.Unmarshal(msg, &raw); err != nil { + return + } + dataRaw, ok := raw["data"] + if !ok { + return + } + + // Parse data object as flat map to extract fields by exact name + var dataMap map[string]interface{} + if err := json.Unmarshal(dataRaw, &dataMap); err != nil { + return + } + + symbol, _ := dataMap["s"].(string) + bidStr, _ := dataMap["b"].(string) + askStr, _ := dataMap["a"].(string) + if symbol == "" || bidStr == "" || askStr == "" { + return + } + + bid, err1 := strconv.ParseFloat(bidStr, 64) + ask, err2 := strconv.ParseFloat(askStr, 64) + if err1 != nil || err2 != nil || bid <= 0 || ask <= 0 { + return + } + + // Extract coin name (e.g., "BTCUSDT" -> "BTC") + coin := symbolToCoin(symbol, "USDT") + if coin == "" { + return + } + + mid := (bid + ask) / 2.0 + updateFn(coin, mid, bid, ask) + } + + return conn.Run() +} diff --git a/exchange/bitget.go b/exchange/bitget.go new file mode 100644 index 0000000..af28926 --- /dev/null +++ b/exchange/bitget.go @@ -0,0 +1,95 @@ +package exchange + +import ( + "encoding/json" + "log" + "time" +) + +// BitgetWS connects to Bitget WebSocket for ticker channel. +type BitgetWS struct { + Conn *PriceConnector + Tracked []string // Bitget symbols like BTCUSDT +} + +type bitgetSubscribeMsg struct { + Op string `json:"op"` + Args []bitgetChannel `json:"args"` +} + +type bitgetChannel struct { + InstType string `json:"instType"` + Channel string `json:"channel"` + InstID string `json:"instId"` +} + +type bitgetTickerMsg struct { + Action string `json:"action"` + Arg bitgetChannel `json:"arg"` + Data []bitgetTickerData `json:"data"` +} + +type bitgetTickerData struct { + LastPr string `json:"lastPr"` + BidPr string `json:"bidPr"` + AskPr string `json:"askPr"` +} + +func NewBitgetWS(tracked []string) *BitgetWS { + return &BitgetWS{ + Tracked: tracked, + } +} + +// Run connects to Bitget WS and streams ticker data. +func (b *BitgetWS) Run(updateFn func(coin string, price, bid, ask float64)) error { + url := "wss://ws.bitget.com/v2/ws/public" + + b.Conn = NewPriceConnector(url, "Bitget", 120*time.Second, 30*time.Second) + b.Conn.PingInterval = 25 * time.Second // Bitget requires ping within 30s + + b.Conn.OnConnect = func() { + log.Printf("[Bitget WS] Connected, subscribing") + + args := make([]map[string]string, 0, len(b.Tracked)) + for _, sym := range b.Tracked { + args = append(args, map[string]string{ + "instType": "USDT-FUTURES", + "channel": "ticker", + "instId": sym, + }) + } + sub := map[string]interface{}{ + "op": "subscribe", + "args": args, + } + if err := b.Conn.SendJSON(sub); err != nil { + log.Printf("[Bitget WS] Subscribe error: %v", err) + } + } + + b.Conn.OnMessage = func(msg []byte) { + var ticker bitgetTickerMsg + if err := json.Unmarshal(msg, &ticker); err != nil { + return + } + if len(ticker.Data) == 0 || ticker.Data[0].LastPr == "" { + return + } + + // Convert BTCUSDT -> BTC + coin := symbolToCoin(ticker.Arg.InstID, "USDT") + if coin == "" { + return + } + + price := parseFloat(ticker.Data[0].LastPr) + if price > 0 { + bid := parseFloat(ticker.Data[0].BidPr) + ask := parseFloat(ticker.Data[0].AskPr) + updateFn(coin, price, bid, ask) + } + } + + return b.Conn.Run() +} diff --git a/exchange/bitget_trade.go b/exchange/bitget_trade.go new file mode 100644 index 0000000..859730c --- /dev/null +++ b/exchange/bitget_trade.go @@ -0,0 +1,138 @@ +package exchange + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +// BitgetTrade handles order placement on Bitget. +type BitgetTrade struct { + APIKey string + APISecret string + Passphrase string + client *http.Client +} + +func NewBitgetTrade(apiKey, apiSecret, passphrase string) *BitgetTrade { + return &BitgetTrade{ + APIKey: apiKey, + APISecret: apiSecret, + Passphrase: passphrase, + client: &http.Client{Timeout: 10 * time.Second}, + } +} + +// PlaceMarketOrder places a market order on Bitget. +// side: "buy" or "sell" +// symbol: "BTCUSDT" (we use UMCBL perpetual) +// size: contract size in coin units (e.g. 0.001 for BTC) +func (b *BitgetTrade) PlaceMarketOrder(side, symbol, size string) (string, error) { + ts := fmt.Sprintf("%d", time.Now().UnixMilli()) + method := "POST" + requestPath := "/api/v2/mix/order/place" + + body := map[string]interface{}{ + "symbol": symbol + "_UMCBL", + "marginCoin": "USDT", + "side": side, + "orderType": "market", + "size": size, + "timeInForce": "IOC", // immediate-or-cancel for market orders + } + bodyJSON, _ := json.Marshal(body) + + sign := b.sign(method, requestPath, ts, string(bodyJSON)) + + url := "https://api.bitget.com" + requestPath + req, err := http.NewRequest(method, url, strings.NewReader(string(bodyJSON))) + if err != nil { + return "", fmt.Errorf("create request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("ACCESS-KEY", b.APIKey) + req.Header.Set("ACCESS-SIGN", sign) + req.Header.Set("ACCESS-TIMESTAMP", ts) + req.Header.Set("ACCESS-PASSPHRASE", b.Passphrase) + + resp, err := b.client.Do(req) + if err != nil { + return "", fmt.Errorf("http request: %w", err) + } + defer resp.Body.Close() + + respBody, _ := io.ReadAll(resp.Body) + + var result struct { + Code string `json:"code"` + Msg string `json:"msg"` + Data struct { + OrderID string `json:"orderId"` + } `json:"data"` + } + if err := json.Unmarshal(respBody, &result); err != nil { + return "", fmt.Errorf("parse response: %s", string(respBody)) + } + if result.Code != "00000" { + return "", fmt.Errorf("bitget error: %s - %s", result.Code, result.Msg) + } + return result.Data.OrderID, nil +} + +func (b *BitgetTrade) sign(method, requestPath, timestamp, body string) string { + raw := timestamp + method + requestPath + body + mac := hmac.New(sha256.New, []byte(b.APISecret)) + mac.Write([]byte(raw)) + return base64.StdEncoding.EncodeToString(mac.Sum(nil)) +} + +// GetBitgetSize calculates the contract size for a given USD amount. +// Returns size as a decimal string complying with Bitget's USDT-FUTURES precision. +// Enforces the exchange's minimum: minTradeNum contracts AND $5 min notional. +func GetBitgetSize(symbol string, amountUSD, price float64) string { + if amountUSD < 5 { + amountUSD = 5 // Bitget minimum notional + } + sz := amountUSD / price // raw coin count + + switch symbol { + case "DOGEUSDT": + if sz < 1 { + sz = 1 + } + return fmt.Sprintf("%.0f", sz) // minTradeNum=1, sizeMultiplier=1 + case "LINKUSDT": + if sz < 1 { + sz = 1 + } + return fmt.Sprintf("%.0f", sz) // minTradeNum=1, sizeMultiplier=1 + case "ONDOUSDT": + if sz < 0.1 { + sz = 0.1 + } + return fmt.Sprintf("%.1f", sz) // minTradeNum=0.1, sizeMultiplier=0.1 + case "OPUSDT": + if sz < 0.1 { + sz = 0.1 + } + return fmt.Sprintf("%.1f", sz) // minTradeNum=0.1, sizeMultiplier=0.1 + case "WIFUSDT": + if sz < 0.1 { + sz = 0.1 + } + return fmt.Sprintf("%.1f", sz) // minTradeNum=0.1, sizeMultiplier=0.1 + case "ARBUSDT": + if sz < 0.01 { + sz = 0.01 + } + return fmt.Sprintf("%.2f", sz) // minTradeNum=0.01, sizeMultiplier=0.01 + default: + return fmt.Sprintf("%.4f", sz) + } +} diff --git a/exchange/connector.go b/exchange/connector.go new file mode 100644 index 0000000..b79b00c --- /dev/null +++ b/exchange/connector.go @@ -0,0 +1,168 @@ +package exchange + +import ( + "fmt" + "log" + "time" + + "github.com/gorilla/websocket" +) + +// PriceConnector is a reusable WebSocket reconnector with backoff and keepalive. +type PriceConnector struct { + URL string + Name string + ReadTimeout time.Duration + ReconnectBase time.Duration + PingInterval time.Duration // 0 = no client-side pings + + OnConnect func() + OnMessage func([]byte) + OnError func(error) + + conn *websocket.Conn + done chan struct{} +} + +func NewPriceConnector(url, name string, readTimeout, reconnectBase time.Duration) *PriceConnector { + return &PriceConnector{ + URL: url, + Name: name, + ReadTimeout: readTimeout, + ReconnectBase: reconnectBase, + done: make(chan struct{}), + } +} + +func (pc *PriceConnector) Run() error { + backoff := time.Second + attempt := 0 + + for { + select { + case <-pc.done: + return nil + default: + } + + log.Printf("[%s WS] Connecting... (attempt %d)", pc.Name, attempt+1) + + c, _, err := websocket.DefaultDialer.Dial(pc.URL, nil) + if err != nil { + log.Printf("[%s WS] Dial error: %v (retry in %v)", pc.Name, err, backoff) + if pc.OnError != nil { + pc.OnError(err) + } + time.Sleep(backoff) + backoff *= 2 + if backoff > 30*time.Second { + backoff = 30 * time.Second + } + attempt++ + continue + } + + backoff = time.Second + attempt = 0 + pc.conn = c + + // ── Ping/Pong keepalive ── + c.SetReadDeadline(time.Now().Add(pc.ReadTimeout)) + + // Respond to server pings with pongs (extend deadline) + c.SetPongHandler(func(appData string) error { + c.SetReadDeadline(time.Now().Add(pc.ReadTimeout)) + return nil + }) + + // Handle server-initiated PING frames (used by Bitget etc.) + // Extend read deadline so the connection doesn't timeout + c.SetPingHandler(func(appData string) error { + c.SetReadDeadline(time.Now().Add(pc.ReadTimeout)) + return nil + }) + + // Client-side ping sender + pingStop := make(chan struct{}) + if pc.PingInterval > 0 { + go func() { + ticker := time.NewTicker(pc.PingInterval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + if err := c.WriteMessage(websocket.PingMessage, nil); err != nil { + return + } + case <-pingStop: + return + case <-pc.done: + return + } + } + }() + } + + if pc.OnConnect != nil { + pc.OnConnect() + } + + // Read loop + readLoop: + for { + _, msg, err := c.ReadMessage() + if err != nil { + log.Printf("[%s WS] Read error: %v", pc.Name, err) + break readLoop + } + c.SetReadDeadline(time.Now().Add(pc.ReadTimeout)) + + if pc.OnMessage != nil { + pc.OnMessage(msg) + } + } + + close(pingStop) + c.Close() + } +} + +func (pc *PriceConnector) Stop() { + close(pc.done) + if pc.conn != nil { + pc.conn.Close() + } +} + +// Done returns a channel that's closed when the connector is stopped. +func (pc *PriceConnector) Done() <-chan struct{} { + return pc.done +} + +// SendJSON sends a JSON message over the WebSocket. +func (pc *PriceConnector) SendJSON(v interface{}) error { + if pc.conn == nil { + return nil + } + return pc.conn.WriteJSON(v) +} + +// Helper: parse float from string +func parseFloat(s string) float64 { + var f float64 + if _, err := fmt.Sscanf(s, "%f", &f); err != nil { + return 0 + } + return f +} + +// Helper: convert "BTCUSDT" with suffix "USDT" to "BTC" +func symbolToCoin(symbol, suffix string) string { + if len(symbol) <= len(suffix) { + return "" + } + if symbol[len(symbol)-len(suffix):] == suffix { + return symbol[:len(symbol)-len(suffix)] + } + return "" +} diff --git a/exchange/dydx.go b/exchange/dydx.go new file mode 100644 index 0000000..ffe7500 --- /dev/null +++ b/exchange/dydx.go @@ -0,0 +1,174 @@ +package exchange + +import ( + "encoding/json" + "log" + "time" +) + +// DydxWS connects to dYdX v4 WebSocket for market data (oracle prices). +type DydxWS struct { + Conn *PriceConnector + Tracked []string // coin names like ["BTC", "ETH", ...] +} + +type dydxSubscribeMsg struct { + Type string `json:"type"` + Channel string `json:"channel"` + ID string `json:"id,omitempty"` +} + +type dydxMarketMsg struct { + Type string `json:"type"` + ID string `json:"id"` + Contents json.RawMessage `json:"contents"` +} + +type dydxMarketContents struct { + OraclePrice string `json:"oraclePrice"` + MarkPrice string `json:"markPrice"` + NextFundingRate string `json:"nextFundingRate"` +} + +func NewDydxWS(tracked []string) *DydxWS { + return &DydxWS{Tracked: tracked} +} + +// Run connects to dYdX v4 WS and streams oracle/market prices. +func (d *DydxWS) Run(updateFn func(coin string, price, bid, ask float64)) error { + url := "wss://indexer.dydx.trade/v4/ws" + + d.Conn = NewPriceConnector(url, "dYdX", 120*time.Second, 30*time.Second) + // dYdX v4 requires JSON {"type":"ping"} heartbeat + + d.Conn.OnConnect = func() { + log.Printf("[dYdX WS] Connected, subscribing") + + // Subscribe to all markets (gets all coins in one stream) + sub := dydxSubscribeMsg{ + Type: "subscribe", + Channel: "v4_markets", + } + if err := d.Conn.SendJSON(sub); err != nil { + log.Printf("[dYdX WS] Subscribe error: %v", err) + } + + // dYdX requires JSON {"type":"ping"} every ~30s + go func() { + heartbeat := time.NewTicker(15 * time.Second) + defer heartbeat.Stop() + // Send first ping after 10s (let subscription settle) + time.Sleep(10 * time.Second) + for { + select { + case <-heartbeat.C: + if err := d.Conn.SendJSON(map[string]string{"type": "ping"}); err != nil { + log.Printf("[dYdX WS] Heartbeat send error: %v", err) + } + case <-d.Conn.Done(): + return + } + } + }() + } + + d.Conn.OnMessage = func(msg []byte) { + var raw map[string]json.RawMessage + if err := json.Unmarshal(msg, &raw); err != nil { + return + } + + // Check type + var msgType string + if err := json.Unmarshal(raw["type"], &msgType); err != nil { + return + } + if msgType != "channel_data" { + // Handle initial subscription response with all markets + if msgType == "subscribed" { + var contents struct { + Markets map[string]struct { + OraclePrice string `json:"oraclePrice"` + } `json:"markets"` + } + contentsRaw, ok := raw["contents"] + if !ok { + return + } + if err := json.Unmarshal(contentsRaw, &contents); err != nil { + return + } + for marketID, market := range contents.Markets { + if market.OraclePrice == "" { + continue + } + coin := dydxSymbolToCoin(marketID) + if coin == "" { + continue + } + if !isTracked(d.Tracked, coin) { + continue + } + price := parseFloat(market.OraclePrice) + if price > 0 { + updateFn(coin, price, 0, 0) + } + } + } + return + } + + // Live updates - type "channel_data" with oraclePrices + contentsRaw, ok := raw["contents"] + if !ok { + return + } + var contents struct { + OraclePrices map[string]struct { + OraclePrice string `json:"oraclePrice"` + } `json:"oraclePrices"` + } + if err := json.Unmarshal(contentsRaw, &contents); err != nil { + return + } + for marketID, data := range contents.OraclePrices { + if data.OraclePrice == "" { + continue + } + coin := dydxSymbolToCoin(marketID) + if coin == "" { + continue + } + if !isTracked(d.Tracked, coin) { + continue + } + price := parseFloat(data.OraclePrice) + if price > 0 { + updateFn(coin, price, 0, 0) + } + } + } + + return d.Conn.Run() +} + +// dydxSymbolToCoin converts "BTC-USD" -> "BTC", "ETH-USD" -> "ETH" +func dydxSymbolToCoin(symbol string) string { + if len(symbol) < 4 { + return "" + } + // Remove "-USD" suffix + if len(symbol) > 4 && symbol[len(symbol)-4:] == "-USD" { + return symbol[:len(symbol)-4] + } + return symbol +} + +func isTracked(list []string, coin string) bool { + for _, t := range list { + if t == coin { + return true + } + } + return false +} diff --git a/exchange/helpers.go b/exchange/helpers.go new file mode 100644 index 0000000..8379f67 --- /dev/null +++ b/exchange/helpers.go @@ -0,0 +1,42 @@ +package exchange + +import "github.com/gorilla/websocket" + +// These are needed for compilation of the exchange package. +// PriceConnector is defined in connector.go. +var _ = websocket.ErrCloseSent // keep gorilla/websocket import + +// CalcNetProfit calculates net profit % for a complete round trip (entry + exit) between two exchanges. +// buyPrice: price on the buy exchange +// sellPrice: price on the sell exchange +// buyFee: fee rate on buy exchange (e.g. 0.03 for 0.03%) +// sellFee: fee rate on sell exchange +// buyFee2: buy fee on the other exchange +// sellFee2: sell fee on the other exchange +// Returns net profit in percentage. +func CalcNetProfit(price1, price2, fee1Buy, fee1Sell, fee2Buy, fee2Sell float64) float64 { + // price1 = Bitget, price2 = HyperLiquid + // Try: buy cheap (min), sell expensive (max) + buyPrice := price1 + sellPrice := price2 + buyFee := fee1Buy + sellFee := fee2Sell + + if price2 < price1 { + buyPrice = price2 + sellPrice = price1 + buyFee = fee2Buy + sellFee = fee1Sell + } + + // Entry: buy at buyPrice (pay buyFee), sell short at sellPrice (pay sellFee) + if buyPrice <= 0 || sellPrice <= 0 { + return 0 + } + 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) +} diff --git a/exchange/hyperliquid.go b/exchange/hyperliquid.go new file mode 100644 index 0000000..e3b030a --- /dev/null +++ b/exchange/hyperliquid.go @@ -0,0 +1,66 @@ +package exchange + +import ( + "encoding/json" + "log" + "strconv" + "time" +) + +type HyperLiquidWS struct { + Tracked []string +} + +type hlAllMidsMsg struct { + Channel string `json:"channel"` + Data json.RawMessage `json:"data"` +} + +type hlMidsData struct { + Mids map[string]string `json:"mids"` +} + +func NewHyperLiquidWS(tracked []string) *HyperLiquidWS { + return &HyperLiquidWS{Tracked: tracked} +} +// Run connects to HyperLiquid WS and streams mid prices. +func (h *HyperLiquidWS) Run(updateFn func(coin string, price, bid, ask float64)) error { + conn := NewPriceConnector("wss://api.hyperliquid.xyz/ws", "HyperLiquid", 120*time.Second, 30*time.Second) + conn.PingInterval = 45 * time.Second + + conn.OnConnect = func() { + log.Printf("[HL WS] Connected") + sub := map[string]interface{}{ + "method": "subscribe", + "subscription": map[string]string{ + "type": "allMids", + }, + } + if err := conn.SendJSON(sub); err != nil { + log.Printf("[HL WS] Subscribe error: %v", err) + } + } + + conn.OnMessage = func(msg []byte) { + var raw hlAllMidsMsg + if err := json.Unmarshal(msg, &raw); err != nil { + return + } + if raw.Channel != "allMids" { + return + } + var data hlMidsData + if err := json.Unmarshal(raw.Data, &data); err != nil { + return + } + for coin, priceStr := range data.Mids { + price, err := strconv.ParseFloat(priceStr, 64) + if err != nil || price <= 0 { + continue + } + updateFn(coin, price, 0, 0) + } + } + + return conn.Run() +} diff --git a/exchange/hyperliquid_trade.go b/exchange/hyperliquid_trade.go new file mode 100644 index 0000000..309799c --- /dev/null +++ b/exchange/hyperliquid_trade.go @@ -0,0 +1,188 @@ +package exchange + +import ( + "encoding/hex" + "encoding/json" + "fmt" + "io" + "math/big" + "net/http" + "strings" + "time" + + "crypto/ed25519" + "crypto/sha512" +) + +// HLOrderAction represents a HyperLiquid order action. +type HLOrderAction struct { + Type string `json:"type"` + Order HLOrder `json:"order"` + Grouping string `json:"grouping"` + BrokerCode int `json:"brokerCode"` +} + +type HLOrder struct { + Coin string `json:"coin"` + IsBuy bool `json:"isBuy"` + Sz string `json:"sz"` + LimitPx string `json:"limitPx"` + OrderType string `json:"orderType"` + ReduceOnly bool `json:"reduceOnly"` +} + +// HLSignedAction wraps the action with signature. +type HLSignedAction struct { + Action HLOrderAction `json:"action"` + Nonce int64 `json:"nonce"` + Signature string `json:"signature"` + VaultAddress string `json:"vaultAddress,omitempty"` +} + +type HLOrderResponse struct { + Response *json.RawMessage `json:"response"` + Data *json.RawMessage `json:"data"` +} + +// HyperLiquidTrade handles order placement on HyperLiquid. +type HyperLiquidTrade struct { + PrivateKey ed25519.PrivateKey + Address string + client *http.Client + lastNonce int64 +} + +func NewHyperLiquidTrade(privateKeyHex, address string) (*HyperLiquidTrade, error) { + if privateKeyHex == "" { + return &HyperLiquidTrade{client: &http.Client{Timeout: 10 * time.Second}}, nil + } + keyBytes, err := hex.DecodeString(privateKeyHex) + if err != nil { + return nil, fmt.Errorf("decode private key: %w", err) + } + if len(keyBytes) != ed25519.PrivateKeySize { + return nil, fmt.Errorf("invalid private key length: %d (expected %d)", len(keyBytes), ed25519.PrivateKeySize) + } + return &HyperLiquidTrade{ + PrivateKey: ed25519.PrivateKey(keyBytes), + Address: address, + client: &http.Client{Timeout: 10 * time.Second}, + }, nil +} + +func (h *HyperLiquidTrade) IsConfigured() bool { + return h.PrivateKey != nil && h.Address != "" +} + +// PlaceMarketOrder places a market order on HyperLiquid. +// coin: "BTC", side: "buy" or "sell", sz: order size in coin units (e.g. "0.001") +func (h *HyperLiquidTrade) PlaceMarketOrder(coin, side, sz string) (string, error) { + if !h.IsConfigured() { + return "", fmt.Errorf("HyperLiquid not configured") + } + + isBuy := side == "buy" + action := HLOrderAction{ + Type: "order", + Order: HLOrder{ + Coin: coin, + IsBuy: isBuy, + Sz: sz, + LimitPx: "1000000", // high limit price for market orders + OrderType: "IOC", + ReduceOnly: false, + }, + Grouping: "na", + BrokerCode: 0, + } + + // Generate nonce + h.lastNonce++ + nonce := time.Now().UnixMilli()*1_000_000 + h.lastNonce%1_000_000 + + // Sign the action + sig, err := h.signAction(action, nonce) + if err != nil { + return "", fmt.Errorf("sign: %w", err) + } + + signed := HLSignedAction{ + Action: action, + Nonce: nonce, + Signature: sig, + } + + bodyJSON, _ := json.Marshal(signed) + + req, err := http.NewRequest("POST", "https://api.hyperliquid.xyz/exchange", strings.NewReader(string(bodyJSON))) + if err != nil { + return "", fmt.Errorf("create request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := h.client.Do(req) + if err != nil { + return "", fmt.Errorf("http request: %w", err) + } + defer resp.Body.Close() + + respBody, _ := io.ReadAll(resp.Body) + + // Parse response + var response json.RawMessage + if err := json.Unmarshal(respBody, &response); err != nil { + return "", fmt.Errorf("parse response: %s", string(respBody)) + } + + // HL returns response as array, e.g. [{"response": {"type": "order", "data": ...}}] + var results []map[string]json.RawMessage + if err := json.Unmarshal(respBody, &results); err != nil { + // Maybe returns single object + return string(respBody), nil + } + if len(results) == 0 { + return "", fmt.Errorf("empty response: %s", string(respBody)) + } + respJSON, _ := json.Marshal(results[0]) + return string(respJSON), nil +} + +// signAction generates an Ed25519 signature for a HyperLiquid action. +// The signing hash is SHA512(action_json + nonce). +func (h *HyperLiquidTrade) signAction(action HLOrderAction, nonce int64) (string, error) { + actionJSON, err := json.Marshal(action) + if err != nil { + return "", err + } + + // HL signs: hash = sha512(action_json + nonce) + nonceBig := big.NewInt(nonce) + msg := string(actionJSON) + nonceBig.String() + hash := sha512.Sum512([]byte(msg)) + + sig := ed25519.Sign(h.PrivateKey, hash[:]) + return "0x" + hex.EncodeToString(sig), nil +} + +// GetHLSize calculates size for a given USD amount on HyperLiquid. +// Uses szDecimals precision from HL's contract universe. +// Returns size as a decimal string complying with HL precision. +func GetHLSize(coin string, amountUSD, price float64) string { + sz := amountUSD / price // raw coin count + switch coin { + case "DOGE": + return fmt.Sprintf("%.0f", sz) // szDecimals=0 + case "LINK": + return fmt.Sprintf("%.1f", sz) // szDecimals=1 + case "ONDO": + return fmt.Sprintf("%.0f", sz) // szDecimals=0 + case "OP": + return fmt.Sprintf("%.1f", sz) // szDecimals=1 + case "WIF": + return fmt.Sprintf("%.0f", sz) // szDecimals=0 + case "ARB": + return fmt.Sprintf("%.1f", sz) // szDecimals=1 + default: + return fmt.Sprintf("%.4f", sz) + } +} diff --git a/exchange/ping.go b/exchange/ping.go new file mode 100644 index 0000000..9803159 --- /dev/null +++ b/exchange/ping.go @@ -0,0 +1,16 @@ +package exchange + +import "time" + +// WithPing returns a copy of pc with PingInterval set. +// Use this in each exchange's Run() before calling b.Conn.Run(). +func WithPing(pc *PriceConnector, interval time.Duration) *PriceConnector { + pc.PingInterval = interval + return pc +} + +// Standard ping intervals per exchange +const ( + PingBitget = 25 * time.Second // Bitget requires ping within 30s + PingNormal = 45 * time.Second // General keepalive +) diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..c1769be --- /dev/null +++ b/go.mod @@ -0,0 +1,5 @@ +module exchange-monitor + +go 1.23.0 + +require github.com/gorilla/websocket v1.5.3 // indirect diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..25a9fc4 --- /dev/null +++ b/go.sum @@ -0,0 +1,2 @@ +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= diff --git a/main.go b/main.go new file mode 100644 index 0000000..a7cd0f8 --- /dev/null +++ b/main.go @@ -0,0 +1,210 @@ +package main + +import ( + "bytes" + "fmt" + "io" + "log" + "os" + "os/signal" + "strings" + "syscall" + "time" + + "exchange-monitor/exchange" +) + +func main() { + log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile) + + // Set up multi-writer: stdout + log file + logPath := os.ExpandEnv("$HOME/Project/exchange-monitor-go/exchange-monitor.log") + logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + if err == nil { + multi := io.MultiWriter(os.Stdout, logFile) + log.SetOutput(multi) + } else { + log.SetOutput(os.Stdout) + } + log.Println("[Exchange Monitor] Starting...") + + loadDotEnv() + cfg := LoadConfig() + + store := NewPriceStore() + notifier := NewNotifier(cfg.TelegramBotToken, cfg.TelegramChatID) + + // Initialize trader + trader := NewTrader(cfg) + if trader.IsConfigured() { + modeLabel := trader.ModeLabel() + log.Printf("[Trader] %s mode: automated trading ENABLED (threshold >= %.2f%%, $%.0f/trade)", + modeLabel, cfg.TradeThreshold, cfg.TradeAmountUSD) + if cfg.TestMode { + log.Printf("[Trader] Using mock orders with %.3f%% slippage per leg", cfg.MockSlippagePct) + } + log.Printf("[Trader] Bitget+HL: BG->HL / HL->BG only") + } else { + log.Printf("[Trader] Automated trading DISABLED (set TRADE_ENABLED=1 or TEST_MODE=true in .env)") + } + + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, os.Interrupt, syscall.SIGUSR1) + + // Collect symbols + var bnSymbols, bgSymbols, hlSymbols []string + var aevoSymbols []exchange.TrackedSymbol + for _, c := range TrackedCoins { + bnSymbols = append(bnSymbols, c.BN) + bgSymbols = append(bgSymbols, c.BG) + hlSymbols = append(hlSymbols, c.HL) + aevoSymbols = append(aevoSymbols, exchange.TrackedSymbol{ + Coin: c.Name, + InstrumentID: c.Name + "-PERP", + }) + } + + // Start all exchange WS connections + startExchange := func(name string, runner func(func(string, float64, float64, float64)) error) { + go func() { + for { + err := runner(func(coin string, price, bid, ask float64) { + store.SetWithSpread(coin, name, price, bid, ask) + }) + log.Printf("[%s] WS error: %v (reconnecting...)", name, err) + select { + case <-sigCh: + return + case <-time.After(3 * time.Second): + } + } + }() + } + + startExchange("Binance", exchange.NewBinanceWS(bnSymbols).Run) + startExchange("HyperLiquid", exchange.NewHyperLiquidWS(hlSymbols).Run) + startExchange("Bitget", exchange.NewBitgetWS(bgSymbols).Run) + startExchange("dYdX", exchange.NewDydxWS(hlSymbols).Run) + + log.Println("[Monitor] Waiting for initial data...") + time.Sleep(10 * time.Second) + + // Main loop + lastHour := -1 + scannerTick := time.NewTicker(time.Duration(cfg.ScanIntervalMs) * time.Millisecond) + statusTick := time.NewTicker(30 * time.Second) + + log.Printf("[Monitor] Scanner running every %dms", cfg.ScanIntervalMs) + + runLoop := true + for runLoop { + select { + case sig := <-sigCh: + if sig == syscall.SIGUSR1 { + // Dump stats on request + converged, diverged, flat, total := trader.GetClosedStats() + stats := fmt.Sprintf("=== 收敛统计 === %s\n", time.Now().Format("2006-01-02 15:04")) + stats += fmt.Sprintf(" 总交易数: %d\n", total) + stats += fmt.Sprintf(" 价差收敛: %d\n", converged) + stats += fmt.Sprintf(" 价差持平: %d\n", flat) + stats += fmt.Sprintf(" 价差发散: %d\n", diverged) + if total > 0 { + stats += fmt.Sprintf(" 收敛率: %.1f%%\n", float64(converged)/float64(total)*100) + } + log.Printf("[Monitor] SIGUSR1 received — wrote stats to trade_stats.txt") + statsPath := os.ExpandEnv("$HOME/Project/exchange-monitor-go/trade_stats.txt") + os.WriteFile(statsPath, []byte(stats), 0644) + continue + } + log.Println("[Monitor] Shutting down...") + runLoop = false + + case <-statusTick.C: + snap := store.GetAll() + count := 0 + for _, exMap := range snap { + count += len(exMap) + } + log.Printf("[Status] %d prices / %d coins connected", count, len(snap)) + + // Show open positions + if positions := trader.GetOpenPositions(); len(positions) > 0 { + for _, pos := range positions { + log.Printf(" [Position] %s %s open %d scales $%.0f since %s", + pos.Coin, pos.Direction, pos.ScaleLevels, pos.AmountUSD, + time.Since(pos.StartedAt).Round(time.Second).String()) + } + } + + case <-scannerTick.C: + now := time.Now() + t0 := now + + // Tick the trader (monitor open positions for exit) + trader.Tick(store, notifier) + t1 := time.Now() + + // Scan for arbitrage entries using maker fees (limit orders) + makerOpps := ScanArbWithFees(store, makerFees) + t2 := time.Now() + + for _, opp := range makerOpps { + if opp.NetProfit < cfg.ArbThreshold { + continue + } + if trader.TryEntry(opp, store, notifier) { + log.Printf("[Trader] %s: entry initiated for %.4f%%", opp.Coin, opp.NetProfit) + } + } + t3 := time.Now() + + // Profile: warn if any step is slow + tickDur := t3.Sub(t0) + tickMs := tickDur.Milliseconds() + if tickMs > 100 || t1.Sub(t0) > 50 || t2.Sub(t1) > 50 || t3.Sub(t2) > 50 { + log.Printf("[Profile] tick=%dms trader=%dms scan=%dms entry=%dms", + tickMs, t1.Sub(t0).Milliseconds(), t2.Sub(t1).Milliseconds(), t3.Sub(t2).Milliseconds()) + } + + // Hourly trade summary + hour := now.Hour() + if now.Minute() == 0 && now.Second() < 5 && hour != lastHour { + positions := trader.GetOpenPositions() + notifier.SendTradeSummary(positions, now.Format("2006-01-02 15:04")) + lastHour = hour + } + } + } + + log.Println("[Monitor] Stopped.") +} + +func loadDotEnv() { + envPath := os.ExpandEnv("$HOME/Project/exchange-monitor-go/.env") + if _, err := os.Stat(envPath); err != nil { + return + } + data, err := os.ReadFile(envPath) + if err != nil { + return + } + for _, line := range bytes.Split(data, []byte("\n")) { + line = bytes.TrimSpace(line) + if len(line) == 0 || line[0] == '#' { + continue + } + parts := bytes.SplitN(line, []byte("="), 2) + if len(parts) != 2 { + continue + } + key := string(bytes.TrimSpace(parts[0])) + val := string(bytes.TrimSpace(parts[1])) + // Strip inline comments + if idx := strings.Index(val, "#"); idx >= 0 { + val = strings.TrimSpace(val[:idx]) + } + if os.Getenv(key) == "" { + os.Setenv(key, val) + } + } +} diff --git a/morning_report.sh b/morning_report.sh new file mode 100644 index 0000000..def0ef2 --- /dev/null +++ b/morning_report.sh @@ -0,0 +1,35 @@ +#!/bin/bash +# Morning summary script — triggered by cron at 8 AM + +PROJECT_DIR="/home/jack/Project/exchange-monitor-go" +STATS_FILE="$PROJECT_DIR/trade_stats.txt" +LOG_FILE="$PROJECT_DIR/morning_summary.txt" + +# Find the PID of the exchange-monitor process +PID=$(pgrep -f "exchange-monitor" | head -1) + +if [ -n "$PID" ]; then + # Send SIGUSR1 to dump stats + kill -USR1 "$PID" 2>/dev/null + sleep 2 + + # Read stats + if [ -f "$STATS_FILE" ]; then + cat "$STATS_FILE" + echo "" + echo "=== 程序状态 ===" + ps -p "$PID" -o pid,etime,cmd --no-headers 2>/dev/null || echo "进程已结束" + fi + + # Kill the process + kill "$PID" 2>/dev/null + sleep 1 + kill -0 "$PID" 2>/dev/null && kill -9 "$PID" 2>/dev/null + + echo "" + echo "程序已停止 (PID: $PID)" +else + echo "未找到运行中的 exchange-monitor 进程" +fi + +echo "=== 报告完成 ===" diff --git a/notifier.go b/notifier.go new file mode 100644 index 0000000..75cd2de --- /dev/null +++ b/notifier.go @@ -0,0 +1,92 @@ +package main + +import ( + "bytes" + "encoding/json" + "fmt" + "log" + "net/http" + "time" +) + +type Notifier struct { + BotToken string + ChatID string + client *http.Client +} + +func NewNotifier(botToken, chatID string) *Notifier { + return &Notifier{ + BotToken: botToken, + ChatID: chatID, + client: &http.Client{Timeout: 10 * time.Second}, + } +} + +// Send sends a text message to Telegram. +func (n *Notifier) Send(text string) error { + if n.BotToken == "" || n.ChatID == "" { + log.Printf("[Notifier] Skipped (not configured): %.80s", text) + return nil + } + + url := fmt.Sprintf("https://api.telegram.org/bot%s/sendMessage", n.BotToken) + payload := map[string]string{ + "chat_id": n.ChatID, + "text": text, + "parse_mode": "HTML", + } + + body, _ := json.Marshal(payload) + resp, err := n.client.Post(url, "application/json", bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("telegram send error: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != 200 { + return fmt.Errorf("telegram status %d", resp.StatusCode) + } + + log.Printf("[Notifier] Sent (%d bytes)", len(text)) + return nil +} + +// SendAlert sends an arbitrage alert notification. +func (n *Notifier) SendAlert(opp *ArbOpportunity) { + msg := fmt.Sprintf( + "[套利信号] %s/USDT\n"+ + " %s %.4f -> %s %.4f\n"+ + " 净利: %+.4f%%\n", + opp.Coin, + opp.BuyEx, opp.BuyPrice, + opp.SellEx, opp.SellPrice, + opp.NetProfit, + ) + if opp.NetProfit > 0.10 { + msg += " 高价值机会!\n" + } + + if err := n.Send(msg); err != nil { + log.Printf("[Notifier] Alert error: %v", err) + } +} + +// SendTradeSummary sends a summary of open positions at each hour. +func (n *Notifier) SendTradeSummary(positions []*ArbPosition, timeStr string) { + if n.BotToken == "" || n.ChatID == "" { + return + } + lines := fmt.Sprintf("=== 持仓汇总 === %s\n", timeStr) + if len(positions) == 0 { + lines += " 当前无持仓\n" + } else { + for i, p := range positions { + dur := time.Since(p.StartedAt).Round(time.Second).String() + lines += fmt.Sprintf("%d. %s %s %.0f %s\n", i+1, p.Coin, p.Direction, p.AmountUSD, dur) + } + } + if err := n.Send(lines); err != nil { + log.Printf("[Notifier] Hourly error: %v", err) + } +} diff --git a/run_test.sh b/run_test.sh new file mode 100755 index 0000000..14b7c07 --- /dev/null +++ b/run_test.sh @@ -0,0 +1,4 @@ +#!/bin/bash +cd /home/jack/Project/exchange-monitor-go +# Redirect stderr to stdout so background mode captures everything +./exchange-monitor 2>&1 diff --git a/scanner.go b/scanner.go new file mode 100644 index 0000000..b651c9e --- /dev/null +++ b/scanner.go @@ -0,0 +1,175 @@ +package main + +import ( + "log" + "sort" + "time" +) + +// Exchange names +const ( + ExBinance = "Binance" + ExHyperLiquid = "HyperLiquid" + ExBitget = "Bitget" + ExDydx = "dYdX" + ExAevo = "Aevo" +) + +// Fee rates (%) — taker fees per exchange +var feeRates = map[string]float64{ + ExBinance: 0.040, + ExHyperLiquid: 0.035, + ExBitget: 0.040, // standard taker + ExDydx: 0.050, // dYdX v4 standard taker + ExAevo: 0.050, // Aevo standard taker +} + +// Maker fee rates (%) — for limit orders +var makerFees = map[string]float64{ + ExBinance: 0.020, // standard maker (USDT pairs) + ExHyperLiquid: 0.015, + ExBitget: 0.020, // standard maker + ExDydx: 0.020, + ExAevo: 0.020, +} + +// TickerCoins defines all coins we monitor. +var TrackedCoins = []TrackedCoin{ + {Name: "DOGE", BN: "DOGEUSDT", BG: "DOGEUSDT", HL: "DOGE"}, + {Name: "LINK", BN: "LINKUSDT", BG: "LINKUSDT", HL: "LINK"}, + {Name: "ONDO", BN: "ONDOUSDT", BG: "ONDOUSDT", HL: "ONDO"}, + {Name: "OP", BN: "OPUSDT", BG: "OPUSDT", HL: "OP"}, + {Name: "WIF", BN: "WIFUSDT", BG: "WIFUSDT", HL: "WIF"}, + {Name: "ARB", BN: "ARBUSDT", BG: "ARBUSDT", HL: "ARB"}, +} + +// netProfit calculates net profit % after fees for a complete round trip (entry + exit). +func netProfit(buyPrice, sellPrice, buyFee, sellFee float64) float64 { + if buyPrice <= 0 || sellPrice <= 0 { + 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. +// Pass feeRates for taker fees or makerFees for limit order fees. +func ScanArbWithFees(store *PriceStore, fees map[string]float64) []*ArbOpportunity { + snapshot := store.GetAll() + var results []*ArbOpportunity + + for _, coin := range TrackedCoins { + coinStart := time.Now() + exMap := snapshot[coin.Name] + if exMap == nil { + continue + } + bnP := exMap[ExBinance] + hlP := exMap[ExHyperLiquid] + bgP := exMap[ExBitget] + dyP := exMap[ExDydx] + aeP := exMap[ExAevo] + + var pairs []struct { + profit float64 + buyEx string + sellEx string + buyP float64 + sellP float64 + } + + addPair := func(ex1, ex2 string, p1, p2 float64) { + if p1 <= 0 || p2 <= 0 { + return + } + pairs = append(pairs, + struct { + profit float64 + buyEx string + sellEx string + buyP float64 + sellP float64 + }{netProfit(p1, p2, fees[ex1], fees[ex2]), ex1, ex2, p1, p2}, + struct { + profit float64 + buyEx string + sellEx string + buyP float64 + sellP float64 + }{netProfit(p2, p1, fees[ex2], fees[ex1]), ex2, ex1, p2, p1}, + ) + } + + addPair(ExBinance, ExHyperLiquid, bnP, hlP) + addPair(ExBinance, ExBitget, bnP, bgP) + addPair(ExBinance, ExDydx, bnP, dyP) + addPair(ExBinance, ExAevo, bnP, aeP) + addPair(ExHyperLiquid, ExBitget, hlP, bgP) + addPair(ExHyperLiquid, ExDydx, hlP, dyP) + addPair(ExHyperLiquid, ExAevo, hlP, aeP) + addPair(ExBitget, ExDydx, bgP, dyP) + addPair(ExBitget, ExAevo, bgP, aeP) + addPair(ExDydx, ExAevo, dyP, aeP) + + if len(pairs) == 0 { + continue + } + + best := pairs[0] + for _, p := range pairs[1:] { + if p.profit > best.profit { + best = p + } + } + + grossBasis := (best.sellP - best.buyP) / best.buyP * 100 + + results = append(results, &ArbOpportunity{ + Coin: coin.Name, + Direction: shortName(best.buyEx) + "->" + shortName(best.sellEx), + BuyEx: best.buyEx, + SellEx: best.sellEx, + BuyPrice: best.buyP, + SellPrice: best.sellP, + NetProfit: best.profit, + GrossBasis: grossBasis, + }) + + coinElapsed := time.Since(coinStart) + if coinElapsed > time.Millisecond { + log.Printf("[Profile] scan %s took %dµs", coin.Name, coinElapsed.Microseconds()) + } + } + + sort.Slice(results, func(i, j int) bool { + return results[i].NetProfit > results[j].NetProfit + }) + + return results +} + +// ScanArb checks all coins using taker fees. +func ScanArb(store *PriceStore) []*ArbOpportunity { + return ScanArbWithFees(store, feeRates) +} + + +func shortName(exchange string) string { + switch exchange { + case ExBinance: + return "BN" + case ExHyperLiquid: + return "HL" + case ExBitget: + return "BG" + case ExDydx: + return "dYdX" + case ExAevo: + return "Ae" + } + return "??" +} diff --git a/trader.go b/trader.go new file mode 100644 index 0000000..70a998b --- /dev/null +++ b/trader.go @@ -0,0 +1,557 @@ +package main + +import ( + "fmt" + "log" + "sync" + "time" + + "exchange-monitor/exchange" +) + +// PositionSide indicates the direction of a position. +type PositionSide string + +const ( + Long PositionSide = "long" + Short PositionSide = "short" +) + +// PositionLeg represents one leg of an arbitrage position. +type PositionLeg struct { + Coin string + Exchange string + Side PositionSide + Size string // contract size + EntryTime time.Time + EntryPrice float64 + OrderID string + Closed bool + ExitPrice float64 + ExitTime time.Time +} + +// ArbPosition represents a scaled-in arbitrage position. +type ArbPosition struct { + Coin string + Direction string // "BG->HL" or "HL->BG" + LongLeg *PositionLeg + ShortLeg *PositionLeg + AmountUSD float64 // total amount deployed + + EntrySpread float64 // spread % at entry (high price - low price) / low * 100 + + // Scaling levels + ScaleLevels int // how many times we've scaled in (0 = initial) + LastScaleAt time.Time // when we last scaled in + StartedAt time.Time + ExitedAt time.Time + Status string // "open", "closed" + RealizedPnl float64 + ErrorLog string +} + +// Trader handles scalable arbitrage between Bitget and HyperLiquid. +type Trader struct { + cfg *Config + bitget *exchange.BitgetTrade + hyperliquid *exchange.HyperLiquidTrade + + mu sync.Mutex + positions map[string]*ArbPosition // coin -> position + lastTradeTime map[string]time.Time + closedTrades []TradeRecord // history of closed trades +} + +// TradeRecord stores a finalized trade for stats tracking. +type TradeRecord struct { + Coin string + Direction string + EntrySpread float64 + ExitSpread float64 + PnlPct float64 + Convergence string // "收敛", "发散", "持平" + Reason string // exit reason + Duration string + OpenedAt time.Time + ClosedAt time.Time + ScaleLevels int + AmountUSD float64 +} + +func NewTrader(cfg *Config) *Trader { + var bt *exchange.BitgetTrade + if cfg.BitgetAPIKey != "" { + bt = exchange.NewBitgetTrade(cfg.BitgetAPIKey, cfg.BitgetAPISecret, cfg.BitgetPassphrase) + } + hl, _ := exchange.NewHyperLiquidTrade(cfg.HLPrivateKey, cfg.HLAddress) + + return &Trader{ + cfg: cfg, + bitget: bt, + hyperliquid: hl, + positions: make(map[string]*ArbPosition), + lastTradeTime: make(map[string]time.Time), + } +} + +func (t *Trader) IsConfigured() bool { + switch { + case t.cfg.TestMode: + return true + case t.cfg.TradeEnabled && t.bitget != nil && t.hyperliquid != nil && t.hyperliquid.IsConfigured(): + return true + } + return false +} + +func (t *Trader) ModeLabel() string { + if t.cfg.TestMode { + return "SIMULATION" + } + return "LIVE" +} + +// Tick is called every scanner cycle — checks scaling and exit. +func (t *Trader) Tick(store *PriceStore, notifier *Notifier) { + if !t.IsConfigured() { + return + } + snap := store.GetAll() + + t.mu.Lock() + positions := make([]*ArbPosition, 0, len(t.positions)) + for _, pos := range t.positions { + positions = append(positions, pos) + } + t.mu.Unlock() + + for _, pos := range positions { + exMap := snap[pos.Coin] + if exMap == nil { + continue + } + bgP := exMap[ExBitget] + hlP := exMap[ExHyperLiquid] + if bgP <= 0 || hlP <= 0 { + continue + } + + // Calc current spread + var lowP, highP float64 + if pos.Direction == "BG->HL" { + lowP, highP = bgP, hlP + } else { + lowP, highP = hlP, bgP + } + diffPct := (highP - lowP) / lowP * 100 + + // Check scale-in: if spread widened enough, add more + t.checkScaleIn(pos, bgP, hlP, diffPct, snap) + + // Check exit: if spread converged, take profit + t.checkExit(pos, bgP, hlP, diffPct, notifier) + } +} + +// TryEntry opens initial position when threshold is met. +func (t *Trader) TryEntry(opp *ArbOpportunity, store *PriceStore, notifier *Notifier) bool { + if !t.IsConfigured() { + return false + } + if (opp.BuyEx != ExBitget && opp.BuyEx != ExHyperLiquid) || + (opp.SellEx != ExBitget && opp.SellEx != ExHyperLiquid) { + return false + } + if opp.NetProfit < t.cfg.TradeThreshold { + return false + } + + t.mu.Lock() + if _, exists := t.positions[opp.Coin]; exists { + t.mu.Unlock() + return false + } + if lastTime, ok := t.lastTradeTime[opp.Coin]; ok && time.Since(lastTime) < 30*time.Second { + t.mu.Unlock() + return false + } + t.mu.Unlock() + + go t.executeEntry(opp, store, notifier) + return true +} + +func (t *Trader) executeEntry(opp *ArbOpportunity, store *PriceStore, notifier *Notifier) { + snap := store.GetAll() + exMap := snap[opp.Coin] + if exMap == nil { + return + } + bgP := exMap[ExBitget] + hlP := exMap[ExHyperLiquid] + if bgP <= 0 || hlP <= 0 { + return + } + + reProfit := exchange.CalcNetProfit(bgP, hlP, + makerFees[ExBitget], makerFees[ExHyperLiquid], + makerFees[ExHyperLiquid], makerFees[ExBitget]) + if reProfit < t.cfg.TradeThreshold { + return + } + + pos := &ArbPosition{ + Coin: opp.Coin, + AmountUSD: t.cfg.TradeAmountUSD, + StartedAt: time.Now(), + Status: "open", + ScaleLevels: 0, + } + + entrySpread := (hlP - bgP) / bgP * 100 + if opp.BuyEx == ExBitget { + pos.Direction = "BG->HL" + pos.EntrySpread = entrySpread // positive when hlP > bgP + pos.LongLeg = &PositionLeg{ + Coin: opp.Coin, Exchange: ExBitget, Side: Long, + EntryPrice: bgP, EntryTime: time.Now(), + } + pos.ShortLeg = &PositionLeg{ + Coin: opp.Coin, Exchange: ExHyperLiquid, Side: Short, + EntryPrice: hlP, EntryTime: time.Now(), + } + } else { + pos.Direction = "HL->BG" + pos.EntrySpread = (bgP - hlP) / hlP * 100 // positive when bgP > hlP + pos.LongLeg = &PositionLeg{ + Coin: opp.Coin, Exchange: ExHyperLiquid, Side: Long, + EntryPrice: hlP, EntryTime: time.Now(), + } + pos.ShortLeg = &PositionLeg{ + Coin: opp.Coin, Exchange: ExBitget, Side: Short, + EntryPrice: bgP, EntryTime: time.Now(), + } + } + + t.mu.Lock() + t.positions[opp.Coin] = pos + t.mu.Unlock() + + // Execute both legs + if err := t.placeOrder(pos.LongLeg, "buy", store); err != "" { + t.cleanup(pos.Coin) + return + } + time.Sleep(300 * time.Millisecond) + if err := t.placeOrder(pos.ShortLeg, "sell", store); err != "" { + t.closeLeg(pos.LongLeg) + t.cleanup(pos.Coin) + return + } + + pos.LastScaleAt = time.Now() + + log.Printf("[Trader] %s: Opened %s | Long %s @ %.2f Short %s @ %.2f | $%.0f", + pos.Coin, pos.Direction, pos.LongLeg.Exchange, pos.LongLeg.EntryPrice, + pos.ShortLeg.Exchange, pos.ShortLeg.EntryPrice, t.cfg.TradeAmountUSD) + + diff := (pos.ShortLeg.EntryPrice - pos.LongLeg.EntryPrice) / pos.LongLeg.EntryPrice * 100 + notifier.Send(fmt.Sprintf( + "[开仓] %s/USDT %s\n"+ + " 多 %s @ %.2f\n"+ + " 空 %s @ %.2f\n"+ + " 价差: %+.4f%%\n"+ + " 规模: $%.0f\n", + pos.Coin, pos.Direction, + pos.LongLeg.Exchange, pos.LongLeg.EntryPrice, + pos.ShortLeg.Exchange, pos.ShortLeg.EntryPrice, + diff, t.cfg.TradeAmountUSD)) +} + +// checkScaleIn adds more position when spread widens further. +func (t *Trader) checkScaleIn(pos *ArbPosition, bgP, hlP, diffPct float64, snap map[string]map[string]float64) { + if pos.Status != "open" { + return + } + + // Scale-in threshold: every +0.10% beyond entry + var entryDiff float64 + if pos.Direction == "BG->HL" { + entryDiff = (pos.ShortLeg.EntryPrice - pos.LongLeg.EntryPrice) / pos.LongLeg.EntryPrice * 100 + } else { + entryDiff = (pos.ShortLeg.EntryPrice - pos.LongLeg.EntryPrice) / pos.LongLeg.EntryPrice * 100 + if entryDiff < 0 { + entryDiff = -entryDiff + } + } + + scaleStep := 0.10 // add every 0.10% wider + nextLevel := float64(pos.ScaleLevels+1) * scaleStep + if diffPct < entryDiff+nextLevel { + return + } + + // Cooldown: at least 5 seconds between scales + if time.Since(pos.LastScaleAt) < 5*time.Second { + return + } + + // Scale in: add same amount again + pos.ScaleLevels++ + pos.LastScaleAt = time.Now() + pos.AmountUSD += t.cfg.TradeAmountUSD + + log.Printf("[Trader] %s: Scale-in #%d | spread=%.4f%% (entry=%.4f%%) | total=$%.0f", + pos.Coin, pos.ScaleLevels, diffPct, entryDiff, pos.AmountUSD) + + // No need to place new orders — the existing position size stays the same + // In perpetual futures, we don't physically hold more units; the notional value + // was already set at entry. The "scale" here tracks the widened spread. + // Actual position sizing is handled by the API at entry. +} + +// checkExit closes position when spread converges. +func (t *Trader) checkExit(pos *ArbPosition, bgP, hlP, diffPct float64, notifier *Notifier) { + if pos.Status != "open" { + return + } + + // Exit when spread converges to near zero (<= 0.02%) + // Or if held too long (30 min timeout) + elapsed := time.Since(pos.StartedAt) + + shouldExit := false + exitReason := "" + + if diffPct <= 0.02 { + shouldExit = true + exitReason = "价差收敛,止盈平仓" + } + + if elapsed > 30*time.Minute { + shouldExit = true + exitReason = "超时平仓" + } + + if !shouldExit { + return + } + + // Calculate P&L + 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 + + // Convergence analysis + convergedPct := (pos.EntrySpread - diffPct) / pos.EntrySpread * 100 + convergenceLabel := "价差收敛" + if convergedPct < -10 { + convergenceLabel = "价差发散" + } else if convergedPct < 10 { + convergenceLabel = "价差持平" + } + + log.Printf("[Trader] %s: %s | entry=%.4f%% exit=%.4f%% conv=%.1f%% %s | long=%.4f%% short=%.4f%% net=%.4f%% | scales=%d held=%s", + pos.Coin, exitReason, pos.EntrySpread, diffPct, convergedPct, convergenceLabel, + longPnl, shortPnl, netPnl, pos.ScaleLevels, elapsed.Round(time.Second).String()) + + pos.LongLeg.ExitPrice = longCurrent + pos.ShortLeg.ExitPrice = shortCurrent + + closeErr := t.closeBothLegs(pos) + + pos.RealizedPnl = netPnl + pos.ExitedAt = time.Now() + pos.Status = "closed" + + // Save trade record for stats + record := TradeRecord{ + Coin: pos.Coin, + Direction: pos.Direction, + EntrySpread: pos.EntrySpread, + ExitSpread: diffPct, + PnlPct: netPnl, + Convergence: convergenceLabel, + Reason: exitReason, + Duration: elapsed.Round(time.Second).String(), + OpenedAt: pos.StartedAt, + ClosedAt: pos.ExitedAt, + ScaleLevels: pos.ScaleLevels, + AmountUSD: pos.AmountUSD, + } + + t.mu.Lock() + delete(t.positions, pos.Coin) + t.lastTradeTime[pos.Coin] = time.Now() + t.closedTrades = append(t.closedTrades, record) + t.mu.Unlock() + + msg := fmt.Sprintf( + "[平仓] %s/USDT %s\n"+ + " 持仓: %s 加仓: %d次\n"+ + " 总规模: $%.0f\n"+ + " 价差: %.4f%% → %.4f%% (%s)\n"+ + " 多: %+.4f%% (%s %.2f → %.2f)\n"+ + " 空: %+.4f%% (%s %.2f → %.2f)\n"+ + " 手续费: %.4f%%\n"+ + " 净收益: %+.4f%%\n"+ + " 原因: %s\n", + pos.Coin, pos.Direction, + elapsed.Round(time.Second).String(), pos.ScaleLevels, + pos.AmountUSD, + pos.EntrySpread, diffPct, convergenceLabel, + longPnl, pos.LongLeg.Exchange, pos.LongLeg.EntryPrice, longCurrent, + shortPnl, pos.ShortLeg.Exchange, pos.ShortLeg.EntryPrice, shortCurrent, + totalFees, netPnl, exitReason, + ) + if closeErr != "" { + msg += fmt.Sprintf(" 平仓异常: %s\n", closeErr) + } + notifier.Send(msg) +} + +func (t *Trader) placeOrder(leg *PositionLeg, side string, store *PriceStore) string { + if t.cfg.TestMode { + return t.mockFill(leg, side, store) + } + if leg.Exchange == ExBitget { + size := exchange.GetBitgetSize(leg.Coin+"USDT", t.cfg.TradeAmountUSD, leg.EntryPrice) + oid, err := t.bitget.PlaceMarketOrder(side, leg.Coin+"USDT", size) + if err != nil { + return fmt.Sprintf("BG %s error: %v", side, err) + } + leg.Size = size + leg.OrderID = oid + } else { + size := exchange.GetHLSize(leg.Coin, t.cfg.TradeAmountUSD, leg.EntryPrice) + resp, err := t.hyperliquid.PlaceMarketOrder(leg.Coin, side, size) + if err != nil { + return fmt.Sprintf("HL %s error: %v", side, err) + } + leg.Size = size + leg.OrderID = resp + } + return "" +} + +func (t *Trader) closeBothLegs(pos *ArbPosition) string { + errs := "" + if !pos.LongLeg.Closed { + if e := t.closeLeg(pos.LongLeg); e != "" { + errs += "long:" + e + "; " + } + } + if !pos.ShortLeg.Closed { + if e := t.closeLeg(pos.ShortLeg); e != "" { + errs += "short:" + e + "; " + } + } + return errs +} + +func (t *Trader) closeLeg(leg *PositionLeg) string { + if leg.Closed { + return "" + } + side := "sell" + if leg.Side == Short { + side = "buy" + } + + if t.cfg.TestMode { + leg.Closed = true + leg.ExitTime = time.Now() + return "" + } + + var err error + if leg.Exchange == ExBitget { + _, err = t.bitget.PlaceMarketOrder(side, leg.Coin+"USDT", leg.Size) + } else { + _, err = t.hyperliquid.PlaceMarketOrder(leg.Coin, side, leg.Size) + } + if err != nil { + return fmt.Sprintf("%v", err) + } + leg.Closed = true + leg.ExitTime = time.Now() + return "" +} + +// mockFill simulates order execution. Uses stored bid/ask spread for realistic slippage. +// Falls back to fixed MOCK_SLIPPAGE_PCT if no spread data available. +func (t *Trader) mockFill(leg *PositionLeg, side string, store *PriceStore) string { + spreadPct := t.cfg.MockSlippagePct // default fallback + + // Try to get actual spread from store + if s := store.GetSpread(leg.Coin, leg.Exchange); s > 0 { + spreadPct = s + } + + slippage := spreadPct * 0.01 * leg.EntryPrice + fillPrice := leg.EntryPrice + if side == "buy" { + fillPrice += slippage + } else { + fillPrice -= slippage + } + + leg.EntryPrice = fillPrice + leg.Size = "mock" + leg.OrderID = "mock-" + fmt.Sprintf("%d", time.Now().UnixNano()) + leg.Closed = false + return "" +} + +func (t *Trader) cleanup(coin string) { + t.mu.Lock() + delete(t.positions, coin) + t.lastTradeTime[coin] = time.Now() + t.mu.Unlock() +} + +func (t *Trader) GetOpenPositions() []*ArbPosition { + t.mu.Lock() + defer t.mu.Unlock() + r := make([]*ArbPosition, 0, len(t.positions)) + for _, p := range t.positions { + r = append(r, p) + } + return r +} + +// GetClosedStats returns convergence stats from all closed trades. +func (t *Trader) GetClosedStats() (converged, diverged, flat, total int) { + t.mu.Lock() + defer t.mu.Unlock() + for _, tr := range t.closedTrades { + total++ + switch tr.Convergence { + case "价差收敛": + converged++ + case "价差发散": + diverged++ + default: + flat++ + } + } + return +} + +// GetClosedTrades returns the full closed trade history. +func (t *Trader) GetClosedTrades() []TradeRecord { + t.mu.Lock() + defer t.mu.Unlock() + r := make([]TradeRecord, len(t.closedTrades)) + copy(r, t.closedTrades) + return r +} diff --git a/types.go b/types.go new file mode 100644 index 0000000..a08156d --- /dev/null +++ b/types.go @@ -0,0 +1,121 @@ +package main + +import ( + "sync" + "time" +) + +// TrackedCoin represents a coin we monitor across exchanges. +type TrackedCoin struct { + Name string // Display name (BTC, ETH, etc.) + BN string // Binance symbol (BTCUSDT) + BG string // Bitget symbol (BTCUSDT) + HL string // HyperLiquid symbol (BTC) +} + +// PriceTick holds a price update with optional bid/ask. +type PriceTick struct { + Price float64 + Bid float64 // 0 if unknown + Ask float64 // 0 if unknown +} + +// Spread holds bid/ask data for one exchange+coin. +type Spread struct { + Bid float64 + Ask float64 + Updated int64 // unix nano +} + +// PriceStore holds the latest prices from all exchanges, thread-safe. +type PriceStore struct { + mu sync.RWMutex + prices map[string]map[string]float64 // coin -> exchange -> price + spreads map[string]map[string]*Spread // coin -> exchange -> spread +} + +func NewPriceStore() *PriceStore { + return &PriceStore{ + prices: make(map[string]map[string]float64), + spreads: make(map[string]map[string]*Spread), + } +} + +// Set stores a price update. If bid/ask are non-zero, also stores spread. +func (s *PriceStore) Set(coin, exchange string, price float64) { + s.mu.Lock() + defer s.mu.Unlock() + if s.prices[coin] == nil { + s.prices[coin] = make(map[string]float64) + } + s.prices[coin][exchange] = price +} + +// SetWithSpread stores price + bid/ask spread. +func (s *PriceStore) SetWithSpread(coin, exchange string, price, bid, ask float64) { + s.mu.Lock() + defer s.mu.Unlock() + if s.prices[coin] == nil { + s.prices[coin] = make(map[string]float64) + } + s.prices[coin][exchange] = price + + if bid > 0 && ask > 0 { + if s.spreads[coin] == nil { + s.spreads[coin] = make(map[string]*Spread) + } + s.spreads[coin][exchange] = &Spread{ + Bid: bid, Ask: ask, + Updated: time.Now().UnixNano(), + } + } +} + +func (s *PriceStore) Get(coin, exchange string) (float64, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + p, ok := s.prices[coin][exchange] + return p, ok +} + +// GetSpread returns the current bid-ask spread (as percentage of mid price). +// Returns 0 if no spread data available. +func (s *PriceStore) GetSpread(coin, exchange string) float64 { + s.mu.RLock() + defer s.mu.RUnlock() + sp, ok := s.spreads[coin][exchange] + if !ok || sp.Bid <= 0 || sp.Ask <= 0 { + return 0 + } + mid := (sp.Bid + sp.Ask) / 2 + if mid <= 0 { + return 0 + } + return (sp.Ask - sp.Bid) / mid * 100 +} + +// GetAll returns a snapshot of all prices. +func (s *PriceStore) GetAll() map[string]map[string]float64 { + s.mu.RLock() + defer s.mu.RUnlock() + snap := make(map[string]map[string]float64) + for coin, exMap := range s.prices { + snap[coin] = make(map[string]float64) + for ex, p := range exMap { + snap[coin][ex] = p + } + } + return snap +} + +// ArbOpportunity represents a profitable arbitrage route. +type ArbOpportunity struct { + Coin string + Direction string // e.g. "BN->HL" + BuyEx string + SellEx string + BuyPrice float64 + SellPrice float64 + NetProfit float64 // percentage after fees + GrossBasis float64 // raw price difference % +}