exchange-monitor-go/trader.go

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 // "entering", "open", "closed"
	RealizedPnl float64
	ErrorLog    string

	// Exit metadata — saved when close is first attempted; reused by retryClose
	ExitDiffPct     float64 // spread % at exit trigger
	ExitNetPnl      float64 // net PnL % at exit trigger
	ExitLongPnl     float64 // long leg PnL %
	ExitShortPnl    float64 // short leg PnL %
	ExitTotalFees   float64 // total fee %
	ExitConvergence string  // convergence label
	ExitReasonText  string  // reason for exit

	// Track all entry prices for weighted-average PnL across scale-ins (Issue #2)
	LongEntryPrices  []float64 // all long entry prices (initial + scale-ins)
	ShortEntryPrices []float64 // all short entry prices (initial + scale-ins)
}

// DeepCopy returns a copy-safe snapshot of the position (no shared pointers).
func (p *ArbPosition) DeepCopy() ArbPosition {
	c := *p
	if p.LongLeg != nil {
		lc := *p.LongLeg
		c.LongLeg = &lc
	}
	if p.ShortLeg != nil {
		sc := *p.ShortLeg
		c.ShortLeg = &sc
	}
	if p.LongEntryPrices != nil {
		c.LongEntryPrices = make([]float64, len(p.LongEntryPrices))
		copy(c.LongEntryPrices, p.LongEntryPrices)
	}
	if p.ShortEntryPrices != nil {
		c.ShortEntryPrices = make([]float64, len(p.ShortEntryPrices))
		copy(c.ShortEntryPrices, p.ShortEntryPrices)
	}
	return c
}

// GetPositionsCopy returns deep copies of all open positions — safe for concurrent read.
func (t *Trader) GetPositionsCopy() []ArbPosition {
	t.mu.Lock()
	defer t.mu.Unlock()
	r := make([]ArbPosition, 0, len(t.positions))
	for _, p := range t.positions {
		r = append(r, p.DeepCopy())
	}
	return r
}

// RefreshSnapshot takes a trading-lock snapshot of open positions for display use.
// Call this after each Tick() from the main loop — never during a trading operation.
// The display reads from this snapshot without blocking trading.
func (t *Trader) RefreshSnapshot() {
	copy := t.GetPositionsCopy() // acquires t.mu briefly (not held during Tick call)
	t.snapMu.Lock()
	t.positionsSnapshot = copy
	t.snapMu.Unlock()
}

// ReadSnapshot returns a copy of the last display snapshot — never locks t.mu.
// Safe to call from any goroutine without impacting trading latency.
func (t *Trader) ReadSnapshot() []ArbPosition {
	t.snapMu.RLock()
	defer t.snapMu.RUnlock()
	r := make([]ArbPosition, len(t.positionsSnapshot))
	copy(r, t.positionsSnapshot)
	return r
}

// 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
	entering      map[string]bool         // coin -> being entered (async goroutine)
	lastTradeTime map[string]time.Time
	blacklist     map[string]time.Time // coin -> when blacklisted (stale spread)
	closedTrades  []TradeRecord // history of closed trades (current session)

	// Historical stats loaded from DB on startup — combined with session stats in GetClosedStats
	dbConverged, dbDiverged, dbFlat, dbTotal int

	OnTradeEvent func(event string, data interface{}) // P3-4: real-time SSE push

	// Decoupled snapshot for display — snapMu never contended by trading path
	snapMu            sync.RWMutex
	positionsSnapshot []ArbPosition
}

// TradeRecord stores a finalized trade for stats tracking.
type TradeRecord struct {
	Coin         string
	Direction    string
	EntrySpread  float64
	ExitSpread   float64
	PnlPct       float64
	PnlUSD       float64 // absolute PnL in USD
	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),
		entering:       make(map[string]bool),
		lastTradeTime:  make(map[string]time.Time),
		blacklist:      make(map[string]time.Time),
	}

	// Restore open positions from DB on restart
	if database != nil {
		t.restoreOpenPositions()
		// Load historical closed trade stats for convergence display
		if c, d, f, tot, err := database.GetClosedStats(); err == nil {
			t.dbConverged, t.dbDiverged, t.dbFlat, t.dbTotal = c, d, f, tot
		}
	}

	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
		elapsed := time.Since(pos.StartedAt)

		// Retry close for positions that failed to close on previous attempt
		if pos.Status == "close_failed" {
			t.retryClose(pos, bgP, hlP, notifier)
			continue
		}

		// Check scale-in: if spread widened enough, add more
		t.checkScaleIn(pos, bgP, hlP, diffPct, store)

		// Check exit: if spread converged, take profit
		t.checkExit(pos, bgP, hlP, diffPct, notifier)

		// Blacklist: if position still open after 10 minutes without converging,
		// the spread is likely stale data. Add coin to blacklist and force close.
		if pos.Status == "open" && elapsed > 10*time.Minute {
			t.blacklistCoin(pos, notifier)
		}
	}
}

// TryEntry opens initial position when threshold is met.
// Returns true if entry was accepted (async goroutine will place orders).
// Non-blocking — the main loop is not stalled by REST calls or the 300ms leg delay.
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 t.entering[opp.Coin] {
		t.mu.Unlock()
		return false
	}
	if t.cfg.MaxPositions > 0 && len(t.positions)+len(t.entering) >= t.cfg.MaxPositions {
		t.mu.Unlock()
		return false
	}
	if blTime, bl := t.blacklist[opp.Coin]; bl {
		if t.cfg.BlacklistDuration <= 0 || time.Since(blTime) < t.cfg.BlacklistDuration {
			t.mu.Unlock()
			return false
		}
		// Blacklist expired — remove it and allow re-entry
		delete(t.blacklist, opp.Coin)
	}
	if lastTime, ok := t.lastTradeTime[opp.Coin]; ok && time.Since(lastTime) < time.Duration(t.cfg.TradeCooldownMs)*time.Millisecond {
		t.mu.Unlock()
		return false
	}
	t.entering[opp.Coin] = true
	t.mu.Unlock()

	// Async goroutine — placeOrder calls (REST or mock) don't block the main loop
	go func() {
		t.executeEntry(opp, store, notifier)
		t.mu.Lock()
		delete(t.entering, opp.Coin)
		t.mu.Unlock()
	}()
	return true
}

// executeEntry places both legs using the scan-time prices from ArbOpportunity.
// Synchronous — runs in the scanner tick to avoid WS price movement between
// detection and execution.
func (t *Trader) executeEntry(opp *ArbOpportunity, store *PriceStore, notifier *Notifier) bool {
	// Use scan-time prices directly to avoid WS jitter killing the entry
	bgP, hlP := opp.BuyPrice, opp.SellPrice
	if opp.BuyEx == ExHyperLiquid {
		bgP, hlP = opp.SellPrice, opp.BuyPrice
	}
	if bgP <= 0 || hlP <= 0 {
		return false
	}

	// Quick sanity check: spread direction hasn't completely reversed
	// Use a relaxed check (not full re-read) since WS prices move constantly
	snap := store.GetAll()
	exMap := snap[opp.Coin]
	if exMap != nil {
		currBg := exMap[ExBitget]
		currHl := exMap[ExHyperLiquid]
		if currBg > 0 && currHl > 0 {
			reversalMul := 1 - t.cfg.ReversalTolerancePct/100
			if opp.BuyEx == ExBitget && currHl <= currBg*reversalMul {
				return false // reversed beyond small tolerance
			}
			if opp.BuyEx == ExHyperLiquid && currBg <= currHl*reversalMul {
				return false
			}
		}
	}

	pos := &ArbPosition{
		Coin:        opp.Coin,
		AmountUSD:   t.cfg.TradeAmountUSD,
		StartedAt:   time.Now(),
		Status:      "entering", // prevent checkExit/checkScaleIn during leg placement
		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(),
		}
		pos.LongEntryPrices = []float64{bgP}
		pos.ShortEntryPrices = []float64{hlP}
	} 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(),
		}
		pos.LongEntryPrices = []float64{hlP}
		pos.ShortEntryPrices = []float64{bgP}
	}

	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 false
	}
	time.Sleep(t.cfg.LegDelay)
	if err := t.placeOrder(pos.ShortLeg, "sell", store); err != "" {
		// Leg1 placed successfully, leg2 failed — try to close leg1
		if closeErr := t.closeLeg(pos.LongLeg); closeErr != "" {
			// CRITICAL: leg1 is still open on the exchange!
			// Record the orphan so we don't silently lose tracking
			pos.ErrorLog = fmt.Sprintf("ORPHAN: leg1 %s %s placed OK, leg2 %s %s failed (%s), leg1 close also failed (%s)",
				pos.LongLeg.Exchange, pos.LongLeg.Side,
				pos.ShortLeg.Exchange, pos.ShortLeg.Side,
				err, closeErr)
			log.Printf("[Trader] ⚠️  ORPHAN POSITION on %s: %s", pos.Coin, pos.ErrorLog)
		}
		t.cleanup(pos.Coin)
		return false
	}

	pos.LastScaleAt = time.Now()
	pos.Status = "open" // both legs placed, ready for Tick/exit logic

	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(
		"<b>[开仓]</b>  %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"),
		})
	}
	return true
}

// checkScaleIn adds more position when spread widens further.
// Issues actual orders on both legs to increase notional exposure (Issue #2).
func (t *Trader) checkScaleIn(pos *ArbPosition, bgP, hlP, diffPct float64, store *PriceStore) {
	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 := t.cfg.ScaleStepPct // add every X% wider
	nextLevel := float64(pos.ScaleLevels+1) * scaleStep
	if diffPct < entryDiff+nextLevel {
		return
	}

	// Cooldown: use configured interval between scales
	if time.Since(pos.LastScaleAt) < t.cfg.ScaleCooldown {
		return
	}

	// Place additional orders on both legs to increase position size
	// Use the current (wider) prices for the new orders
	longPrice := bgP
	shortPrice := hlP
	if pos.LongLeg.Exchange == ExHyperLiquid {
		longPrice, shortPrice = hlP, bgP
	}

	if err := t.placeOrderAt(pos.LongLeg, "buy", store, longPrice); err != "" {
		log.Printf("[Trader] %s: Scale-in long failed: %s", pos.Coin, err)
		return
	}
	time.Sleep(t.cfg.LegDelay)
	if err := t.placeOrderAt(pos.ShortLeg, "sell", store, shortPrice); err != "" {
		log.Printf("[Trader] %s: Scale-in short failed: %s — position partially scaled (long only)", pos.Coin, err)
		// Don't close the long leg — the scale-in long order was placed but the
		// short wasn't. The position has extra long exposure until the next Tick
		// decides what to do. This is a partial fill scenario.
		return
	}

	pos.ScaleLevels++
	pos.LastScaleAt = time.Now()
	pos.AmountUSD += t.cfg.TradeAmountUSD
	pos.LongEntryPrices = append(pos.LongEntryPrices, longPrice)
	pos.ShortEntryPrices = append(pos.ShortEntryPrices, shortPrice)

	log.Printf("[Trader] %s: Scale-in #%d executed | spread=%.4f%% (entry=%.4f%%) | total=$%.0f",
		pos.Coin, pos.ScaleLevels, diffPct, entryDiff, pos.AmountUSD)
}

// checkExit closes position when net profit >= 0.20% (take profit)
// or spread reversed past -0.02% (stop loss) or timeout.
func (t *Trader) checkExit(pos *ArbPosition, bgP, hlP, diffPct float64, notifier *Notifier) {
	if pos.Status != "open" {
		return
	}

	// Current prices for P&L calculation
	var longCurrent, shortCurrent float64
	if pos.LongLeg.Exchange == ExBitget {
		longCurrent, shortCurrent = bgP, hlP
	} else {
		longCurrent, shortCurrent = hlP, bgP
	}

	// Weighted average entry prices across all scale levels
	longAvg := weightedAvgPrice(pos.LongEntryPrices, t.cfg.TradeAmountUSD)
	shortAvg := weightedAvgPrice(pos.ShortEntryPrices, t.cfg.TradeAmountUSD)

	longPnl := (longCurrent - longAvg) / longAvg * 100
	shortPnl := (shortAvg - shortCurrent) / shortAvg * 100
	totalFees := 2 * (takerFees[ExBitget] + takerFees[ExHyperLiquid]) // 开仓 + 平仓手续费
	netPnl := longPnl + shortPnl - totalFees

	elapsed := time.Since(pos.StartedAt)

	shouldExit := false
	exitReason := ""

	// Take profit: net profit >= configured threshold
	if netPnl >= t.cfg.TakeProfitPct {
		shouldExit = true
		exitReason = "利润止盈"
	}

	// Convergence exit: spread narrowed significantly and we're profitable
	// Prevents positions from sitting at near-zero spread waiting for timeout
	if diffPct <= 0.02 && netPnl > 0 {
		shouldExit = true
		exitReason = "价差收敛止盈"
	}

	// Emergency reversal: spread flipped negative — cut losses
	if diffPct < 0 {
		shouldExit = true
		exitReason = "价差反转平仓"
	}

	// Timeout: configured max hold time
	if elapsed > t.cfg.PositionTimeout {
		shouldExit = true
		exitReason = "超时平仓"
	}

	if !shouldExit {
		return
	}

	// 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

	// Save exit metadata for retryClose in case closeBothLegs fails
	pos.ExitDiffPct = diffPct
	pos.ExitNetPnl = netPnl
	pos.ExitLongPnl = longPnl
	pos.ExitShortPnl = shortPnl
	pos.ExitTotalFees = totalFees
	pos.ExitConvergence = convergenceLabel
	pos.ExitReasonText = exitReason

	closeErr := t.closeBothLegs(pos)

	if closeErr != "" {
		// Close failed — keep the position for retry on next Tick
		pos.Status = "close_failed"
		pos.ErrorLog = closeErr
		pos.ExitedAt = time.Now()
		log.Printf("[Trader] ❌ %s: Close failed: %s — will retry on next tick", pos.Coin, closeErr)
		notifier.Send(fmt.Sprintf(
			"<b>[平仓失败]</b>  %s/USDT  %s\n"+
				"  状态: close_failed\n"+
				"  错误: %s\n"+
				"  下一轮将重试关掉剩余的腿\n", pos.Coin, pos.Direction, closeErr))
		return
	}

	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,
		PnlUSD:      pos.AmountUSD * netPnl / 100,
		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 {
		t.persistTrade(pos, diffPct, convergenceLabel, exitReason, netPnl, longPnl, shortPnl, totalFees)
	}

	msg := fmt.Sprintf(
		"<b>[平仓]</b>  %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"+
			"  净收益: <b>%+.4f%%</b>\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,
	)
	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 ""
}

// retryClose retries closing a position that previously failed.
// Only closes legs not already marked Closed. Notifies periodically.
func (t *Trader) retryClose(pos *ArbPosition, bgP, hlP float64, notifier *Notifier) {
	log.Printf("[Trader] %s: Retrying close (previous err: %s)", pos.Coin, pos.ErrorLog)

	closeErr := t.closeBothLegs(pos)
	if closeErr == "" {
		// All legs finally closed — record + persist
		pos.Status = "closed"
		pos.ExitedAt = time.Now()

		elapsed := time.Since(pos.StartedAt)
		record := TradeRecord{
			Coin:        pos.Coin,
			Direction:   pos.Direction,
			EntrySpread: pos.EntrySpread,
			ExitSpread:  pos.ExitDiffPct,
			PnlPct:      pos.ExitNetPnl,
			Convergence: pos.ExitConvergence,
			Reason:      pos.ExitReasonText,
			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()

		if t.db != nil {
			t.persistTrade(pos, pos.ExitDiffPct, pos.ExitConvergence, pos.ExitReasonText,
				pos.ExitNetPnl, pos.ExitLongPnl, pos.ExitShortPnl, pos.ExitTotalFees)
		}

		notifier.Send(fmt.Sprintf(
			"<b>[平仓重试成功]</b>  %s/USDT  %s\n"+
				"  之前失败: %s\n"+
				"  已成功关掉所有腿 | 盈亏: %+.4f%%\n", pos.Coin, pos.Direction, pos.ErrorLog, pos.ExitNetPnl))
		return
	}

	// Still failing — update log and notify periodically
	pos.ErrorLog = closeErr
	log.Printf("[Trader] ❌ %s: Retry close still failing: %s", pos.Coin, closeErr)
	if time.Since(pos.ExitedAt) > 30*time.Second {
		notifier.Send(fmt.Sprintf(
			"<b>[平仓仍失败]</b>  %s/USDT  %s\n"+
				"  已重试 %s, 仍失败: %s\n"+
				"  请手动检查交易所\n", pos.Coin, pos.Direction,
			time.Since(pos.ExitedAt).Round(time.Second).String(), closeErr))
		pos.ExitedAt = time.Now()
	}
}

// placeOrderAt places an order at a specified price (used for scale-in, Issue #2).
// Unlike placeOrder, this doesn't modify the leg's EntryPrice — it places
// an additional order at the current market price for the same trade amount.
func (t *Trader) placeOrderAt(leg *PositionLeg, side string, store *PriceStore, price float64) string {
	if t.cfg.TestMode {
		// Mock fill using specified price instead of leg's original entry
		origPrice := leg.EntryPrice
		leg.EntryPrice = price
		err := t.mockFill(leg, side, store)
		leg.EntryPrice = origPrice // restore original (entry tracking is per-position, not per-order)
		return err
	}
	if leg.Exchange == ExBitget {
		size := exchange.GetBitgetSize(leg.Coin+"USDT", t.cfg.TradeAmountUSD, price)
		_, err := t.bitget.PlaceMarketOrder(side, leg.Coin+"USDT", size)
		if err != nil {
			return fmt.Sprintf("BG %s error: %v", side, err)
		}
	} else {
		size := exchange.GetHLSize(leg.Coin, t.cfg.TradeAmountUSD, price)
		_, err := t.hyperliquid.PlaceMarketOrder(leg.Coin, side, size)
		if err != nil {
			return fmt.Sprintf("HL %s error: %v", side, err)
		}
	}
	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
}

// weightedAvgPrice computes the weighted average entry price across multiple scale levels.
// Each level trades the same USD amount, so the result is the harmonic mean of prices.
func weightedAvgPrice(prices []float64, amountPerTrade float64) float64 {
	if len(prices) == 0 {
		return 0
	}
	totalShares := 0.0
	totalCost := 0.0
	for _, p := range prices {
		if p <= 0 {
			continue
		}
		totalShares += amountPerTrade / p
		totalCost += amountPerTrade
	}
	if totalShares <= 0 {
		return prices[0] // fallback
	}
	return totalCost / totalShares
}

// GetClosedStats returns convergence stats from all closed trades (DB history + current session).
func (t *Trader) GetClosedStats() (converged, diverged, flat, total int) {
	t.mu.Lock()
	defer t.mu.Unlock()
	// Start with DB historical counts
	converged, diverged, flat, total = t.dbConverged, t.dbDiverged, t.dbFlat, t.dbTotal
	// Add in-memory session trades
	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 {
		if t.cfg.MaxPositions > 0 && len(t.positions) >= t.cfg.MaxPositions {
			log.Printf("[Trader] Skipping restored position %s (max_positions=%d reached)", openTrades[i].Coin, t.cfg.MaxPositions)
			break
		}
		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,
			}
			pos.LongEntryPrices = []float64{*tr.LongEntry}
		}
		if tr.ShortEntry != nil {
			pos.ShortLeg = &PositionLeg{
				Coin: tr.Coin, Exchange: tr.ShortExchange, Side: Short,
				EntryPrice: *tr.ShortEntry, EntryTime: tr.OpenedAt,
			}
			pos.ShortEntryPrices = []float64{*tr.ShortEntry}
		}
		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(t.positions))
	}
}

// blacklistCoin adds a coin to the blacklist and force-closes its position.
func (t *Trader) blacklistCoin(pos *ArbPosition, notifier *Notifier) {
	t.mu.Lock()
	t.blacklist[pos.Coin] = time.Now()
	t.mu.Unlock()

	log.Printf("[Trader] ⛔ %s: Blacklisted — position open %.0fm without convergence", pos.Coin, time.Since(pos.StartedAt).Minutes())
	notifier.Send(fmt.Sprintf(
		"<b>[黑名单]</b>  %s/USDT\n"+
			"  开仓 %.0f 分钟未收敛\n"+
			"  已加入黑名单观察\n",
		pos.Coin, time.Since(pos.StartedAt).Minutes()))

	// Force-close the position immediately
	pos.Status = "close_failed" // triggers retryClose on next tick
}

// GetBlacklist returns a copy of the current blacklist (coin -> blacklisted at).
func (t *Trader) GetBlacklist() map[string]time.Time {
	t.mu.Lock()
	defer t.mu.Unlock()
	r := make(map[string]time.Time, len(t.blacklist))
	for k, v := range t.blacklist {
		r[k] = v
	}
	return r
}

// IsBlacklisted checks if a coin is currently blacklisted (within duration).
func (t *Trader) IsBlacklisted(coin string) bool {
	t.mu.Lock()
	defer t.mu.Unlock()
	blTime, exists := t.blacklist[coin]
	if !exists {
		return false
	}
	if t.cfg.BlacklistDuration > 0 && time.Since(blTime) >= t.cfg.BlacklistDuration {
		delete(t.blacklist, coin)
		return false
	}
	return true
}

// RemoveBlacklist removes a coin from the blacklist manually.
func (t *Trader) RemoveBlacklist(coin string) {
	t.mu.Lock()
	defer t.mu.Unlock()
	delete(t.blacklist, coin)
	log.Printf("[Trader] ✅ %s: Removed from blacklist", coin)
}