package main import ( "fmt" "log" "sync" "time" "exchange-monitor/db" "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 db *db.DB mu sync.Mutex positions map[string]*ArbPosition // coin -> position lastTradeTime map[string]time.Time closedTrades []TradeRecord // history of closed trades OnTradeEvent func(event string, data interface{}) // P3-4: real-time SSE push } // 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, database *db.DB) *Trader { var bt *exchange.BitgetTrade if cfg.BitgetAPIKey != "" { bt = exchange.NewBitgetTrade(cfg.BitgetAPIKey, cfg.BitgetAPISecret, cfg.BitgetPassphrase) } hl, _ := exchange.NewHyperLiquidTrade(cfg.HLPrivateKey, cfg.HLAddress) t := &Trader{ cfg: cfg, db: database, bitget: bt, hyperliquid: hl, positions: make(map[string]*ArbPosition), lastTradeTime: make(map[string]time.Time), } // Restore open positions from DB on restart if database != nil { t.restoreOpenPositions() } return t } 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)) // P3-4: real-time trade event push if t.OnTradeEvent != nil { t.OnTradeEvent("trade_open", map[string]interface{}{ "coin": pos.Coin, "direction": pos.Direction, "entry_spread": diff, "amount_usd": t.cfg.TradeAmountUSD, "time": time.Now().Format("15:04:05"), }) } } // 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() // Persist to SQLite if t.db != nil { go t.persistTrade(pos, diffPct, convergenceLabel, exitReason, netPnl, longPnl, shortPnl, totalFees) } 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) // P3-4: real-time trade event push if t.OnTradeEvent != nil { t.OnTradeEvent("trade_close", map[string]interface{}{ "coin": pos.Coin, "direction": pos.Direction, "entry_spread": pos.EntrySpread, "exit_spread": diffPct, "pnl_pct": netPnl, "convergence": convergenceLabel, "duration": elapsed.Round(time.Second).String(), "time": time.Now().Format("15:04:05"), }) } } 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 } // persistTrade saves a completed trade to SQLite. func (t *Trader) persistTrade(pos *ArbPosition, exitSpread float64, convergence, exitReason string, netPnl, longPnl, shortPnl, totalFees float64) { var entrySpread, fe float64 if pos.LongLeg != nil { entrySpread = pos.EntrySpread } fe = totalFees / 2 // split into entry/exit halves now := time.Now() dbTrade := &db.TradeRecord{ Coin: pos.Coin, Direction: pos.Direction, Status: "closed", EntrySpread: &entrySpread, ExitSpread: &exitSpread, LongExchange: pos.LongLeg.Exchange, ShortExchange: pos.ShortLeg.Exchange, LongEntry: &pos.LongLeg.EntryPrice, LongExit: &pos.LongLeg.ExitPrice, ShortEntry: &pos.ShortLeg.EntryPrice, ShortExit: &pos.ShortLeg.ExitPrice, LongPnl: &longPnl, ShortPnl: &shortPnl, FeeEntry: &fe, FeeExit: &fe, NetPnl: &netPnl, AmountUSD: pos.AmountUSD, ScaleCount: pos.ScaleLevels, ExitReason: &exitReason, Convergence: &convergence, OpenedAt: pos.StartedAt, ClosedAt: &now, } if _, err := t.db.SaveTrade(dbTrade); err != nil { log.Printf("[Trader] Failed to save trade to DB: %v", err) } } // restoreOpenPositions loads open trades from DB and recreates their positions. func (t *Trader) restoreOpenPositions() { openTrades, err := t.db.GetOpenTrades() if err != nil { log.Printf("[Trader] Failed to load open trades: %v", err) return } for i := range openTrades { tr := &openTrades[i] // Recreate position structure from DB record pos := &ArbPosition{ Coin: tr.Coin, Direction: tr.Direction, AmountUSD: tr.AmountUSD, EntrySpread: *tr.EntrySpread, ScaleLevels: tr.ScaleCount, LastScaleAt: tr.OpenedAt, // B#3: prevent immediate scale-in bypass StartedAt: tr.OpenedAt, Status: "open", } if tr.LongEntry != nil { pos.LongLeg = &PositionLeg{ Coin: tr.Coin, Exchange: tr.LongExchange, Side: Long, EntryPrice: *tr.LongEntry, EntryTime: tr.OpenedAt, } } if tr.ShortEntry != nil { pos.ShortLeg = &PositionLeg{ Coin: tr.Coin, Exchange: tr.ShortExchange, Side: Short, EntryPrice: *tr.ShortEntry, EntryTime: tr.OpenedAt, } } t.positions[tr.Coin] = pos // Prevent immediate re-trading of the same coin t.lastTradeTime[tr.Coin] = tr.OpenedAt } if len(openTrades) > 0 { log.Printf("[Trader] Restored %d open positions from DB", len(openTrades)) } }