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
	ExitLongPnlUSD  float64 // per-exchange PnL in USD (for retryClose)
	ExitShortPnlUSD float64
	ExitLongFeeUSD  float64 // per-exchange fee in USD
	ExitShortFeeUSD float64

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

	// DB trade ID — set after first save, used for incremental order/scale/exit persists
	DBTradeID int64
}

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

	// Per-exchange fund tracking
	exchangeFunds map[string]*ExchangeFund

	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

	// Auto-stop after N real trades
	StopCh           chan struct{}
	realTradesTarget int
	realTradesDone   int
	shuttingDown     bool
}

// 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
	PnlLongUSD   float64 // per-exchange PnL in USD
	PnlShortUSD  float64
	FeeLongUSD   float64 // per-exchange total fee in USD (entry+exit)
	FeeShortUSD  float64
}

// ExchangeFund tracks balance and PnL for one exchange.
type ExchangeFund struct {
	Balance  float64 // current available balance
	TotalFee float64 // cumulative fees paid
	TotalPnl float64 // cumulative realized PnL
}

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),
		StopCh:         make(chan struct{}, 1),
		realTradesTarget: 5,
		exchangeFunds: map[string]*ExchangeFund{
			ExBitget:      {Balance: cfg.InitialCapital / 2},
			ExHyperLiquid: {Balance: cfg.InitialCapital / 2},
		},
	}

	// Restore open positions from DB on restart
	if database != nil {
		t.restoreOpenPositions()

		// Load historical closed trades for PnL stats (so total PnL survives restart)
		if closed, err := database.GetAllClosedTrades(); err == nil {
			for i := range closed {
				dbTr := &closed[i]
				pnlPct := safeFloat(dbTr.NetPnl)
				pnlUSD := 2 * dbTr.AmountUSD * pnlPct / 100
				closedAt := time.Time{}
				if dbTr.ClosedAt != nil {
					closedAt = *dbTr.ClosedAt
				}
				record := TradeRecord{
					Coin:        dbTr.Coin,
					Direction:   dbTr.Direction,
					EntrySpread: safeFloat(dbTr.EntrySpread),
					ExitSpread:  safeFloat(dbTr.ExitSpread),
					PnlPct:      pnlPct,
					PnlUSD:      pnlUSD,
					Convergence: safeStr(dbTr.Convergence),
					Reason:      safeStr(dbTr.ExitReason),
					Duration:    closedAt.Sub(dbTr.OpenedAt).Round(time.Second).String(),
					OpenedAt:    dbTr.OpenedAt,
					ClosedAt:    closedAt,
					ScaleLevels: dbTr.ScaleCount,
					AmountUSD:   dbTr.AmountUSD,
				}
				t.closedTrades = append(t.closedTrades, record)
			}
		}
		// 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"
}

// IsShuttingDown returns whether trading is stopped.
func (t *Trader) IsShuttingDown() bool {
	t.mu.Lock()
	defer t.mu.Unlock()
	return t.shuttingDown
}

// Stop sets shuttingDown flag and force-closes all open positions.
func (t *Trader) Stop() {
	t.mu.Lock()
	t.shuttingDown = true
	t.mu.Unlock()
	log.Println("[Trader] ⏹ Trading STOPPED — no new entries, closing positions...")

	// Force-close all open positions immediately
	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 {
		if pos.Status == "open" || pos.Status == "close_failed" {
			t.closeBothLegs(pos)
			pos.Status = "closed"
			pos.ExitedAt = time.Now()
			t.mu.Lock()
			delete(t.positions, pos.Coin)
			t.mu.Unlock()
			log.Printf("[Trader] ⏹ Force-closed %s %s (manual stop)", pos.Coin, pos.Direction)
		}
	}
	log.Println("[Trader] ✅ All positions closed, trading stopped. POST /api/start to resume.")
}

// Start clears shuttingDown flag and resumes trading.
func (t *Trader) Start() {
	t.mu.Lock()
	t.shuttingDown = false
	t.mu.Unlock()
	log.Println("[Trader] ▶ Trading RESUMED")
}

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

	// Force-close remaining positions when shutting down
	if t.shuttingDown && len(positions) > 0 {
		t.mu.Unlock()
		for _, pos := range positions {
			if pos.Status == "open" || pos.Status == "close_failed" {
				t.closeBothLegs(pos)
				pos.Status = "closed"
				pos.ExitedAt = time.Now()
				delete(t.positions, pos.Coin)
				log.Printf("[Trader] ⏹ Force-closed %s %s (shutdown)", pos.Coin, pos.Direction)
			}
		}
		// All force-closed — signal stop
		select {
		case t.StopCh <- struct{}{}:
		default:
		}
		return
	}
	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, bgP, hlP, diffPct, 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 t.shuttingDown {
		t.mu.Unlock()
		return false
	}
	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
	}

	// Margin check: verify both exchanges have sufficient funds
	reqAmt := t.cfg.TradeAmountUSD * (1 + takerFees[opp.BuyEx]/100 + takerFees[opp.SellEx]/100)
	if t.exchangeFunds[opp.BuyEx].Balance < reqAmt {
		t.mu.Unlock()
		return false
	}
	if t.exchangeFunds[opp.SellEx].Balance < reqAmt {
		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
		pos.Status = "failed"
		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

	// Persist entry to DB immediately (incremental — not batch at close)
	if t.db != nil {
		now := time.Now()
		status := "filled"
		tradeUnit := t.cfg.TradeAmountUSD
		es := pos.EntrySpread

		dbTrade := &db.TradeRecord{
			Coin:          pos.Coin,
			Direction:     pos.Direction,
			Status:        "open",
			EntrySpread:   &es,
			LongExchange:  pos.LongLeg.Exchange,
			ShortExchange: pos.ShortLeg.Exchange,
			LongEntry:     &pos.LongLeg.EntryPrice,
			ShortEntry:    &pos.ShortLeg.EntryPrice,
			AmountUSD:     t.cfg.TradeAmountUSD,
			OpenedAt:      now,
		}
		if tradeID, err := t.db.SaveTrade(dbTrade); err == nil {
			pos.DBTradeID = tradeID

			longFee := tradeUnit * takerFees[pos.LongLeg.Exchange] / 100
			shortFee := tradeUnit * takerFees[pos.ShortLeg.Exchange] / 100
			longShares := tradeUnit / pos.LongLeg.EntryPrice
			shortShares := tradeUnit / pos.ShortLeg.EntryPrice

			longOID, _ := t.db.SaveOrder(&db.OrderRecord{
				TradeID: tradeID, Leg: "long", Type: "entry",
				Exchange: pos.LongLeg.Exchange, Side: "buy",
				Price: &pos.LongLeg.EntryPrice, Size: &longShares,
				Fee: &longFee, Status: &status, CreatedAt: now,
			})
			shortOID, _ := t.db.SaveOrder(&db.OrderRecord{
				TradeID: tradeID, Leg: "short", Type: "entry",
				Exchange: pos.ShortLeg.Exchange, Side: "sell",
				Price: &pos.ShortLeg.EntryPrice, Size: &shortShares,
				Fee: &shortFee, Status: &status, CreatedAt: now,
			})
			t.db.SaveSystemOrder(&db.SystemOrderRecord{
				TradeID: tradeID, Type: "entry", Status: "filled",
				Spread: &es,
				LongPrice: &pos.LongLeg.EntryPrice, ShortPrice: &pos.ShortLeg.EntryPrice,
				LongOrderID: &longOID, ShortOrderID: &shortOID,
				CreatedAt: 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(
		"<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)

	// Update leg EntryPrice to reflect weighted average across all scale levels
	pos.LongLeg.EntryPrice = weightedAvgPrice(pos.LongEntryPrices, t.cfg.TradeAmountUSD)
	pos.ShortLeg.EntryPrice = weightedAvgPrice(pos.ShortEntryPrices, t.cfg.TradeAmountUSD)

	// Persist scale orders to DB immediately
	if t.db != nil && pos.DBTradeID > 0 {
		now := time.Now()
		status := "filled"
		tradeUnit := t.cfg.TradeAmountUSD
		es := pos.EntrySpread

		longFee := tradeUnit * takerFees[pos.LongLeg.Exchange] / 100
		shortFee := tradeUnit * takerFees[pos.ShortLeg.Exchange] / 100
		longShares := tradeUnit / longPrice
		shortShares := tradeUnit / shortPrice

		longOID, _ := t.db.SaveOrder(&db.OrderRecord{
			TradeID: pos.DBTradeID, Leg: "long", Type: "scale",
			Exchange: pos.LongLeg.Exchange, Side: "buy",
			Price: &longPrice, Size: &longShares,
			Fee: &longFee, Status: &status, CreatedAt: now,
		})
		shortOID, _ := t.db.SaveOrder(&db.OrderRecord{
			TradeID: pos.DBTradeID, Leg: "short", Type: "scale",
			Exchange: pos.ShortLeg.Exchange, Side: "sell",
			Price: &shortPrice, Size: &shortShares,
			Fee: &shortFee, Status: &status, CreatedAt: now,
		})
		t.db.SaveSystemOrder(&db.SystemOrderRecord{
			TradeID: pos.DBTradeID, Type: "scale", Status: "filled",
			Spread: &es,
			LongPrice: &longPrice, ShortPrice: &shortPrice,
			LongOrderID: &longOID, ShortOrderID: &shortOID,
			CreatedAt: now,
		})
	}

	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
	netPnl, totalFees := calcArbPnL(longPnl, shortPnl, pos.ScaleLevels, t.cfg.TradeAmountUSD) // 净利为总资本的百分比

	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 to ≤ 0.02% (includes reversal)
	if diffPct <= 0.02 {
		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
	// Pre-compute per-exchange PnL/fees for retryClose
	numBatchesRetry := 1 + pos.ScaleLevels
	pos.ExitLongPnlUSD = longPnl / 100 * float64(numBatchesRetry) * t.cfg.TradeAmountUSD
	pos.ExitShortPnlUSD = shortPnl / 100 * float64(numBatchesRetry) * t.cfg.TradeAmountUSD
	{
		totalLongSharesRetry := 0.0
		for _, p := range pos.LongEntryPrices {
			totalLongSharesRetry += t.cfg.TradeAmountUSD / p
		}
		totalShortSharesRetry := 0.0
		for _, p := range pos.ShortEntryPrices {
			totalShortSharesRetry += t.cfg.TradeAmountUSD / p
		}
		pos.ExitLongFeeUSD = float64(numBatchesRetry)*t.cfg.TradeAmountUSD*takerFees[pos.LongLeg.Exchange]/100 +
			totalLongSharesRetry*longCurrent*takerFees[pos.LongLeg.Exchange]/100
		pos.ExitShortFeeUSD = float64(numBatchesRetry)*t.cfg.TradeAmountUSD*takerFees[pos.ShortLeg.Exchange]/100 +
			totalShortSharesRetry*shortCurrent*takerFees[pos.ShortLeg.Exchange]/100
	}

	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"

	// Compute per-leg PnL and fees in USD
	numBatches := 1 + pos.ScaleLevels
	legCapital := t.cfg.TradeAmountUSD
	longPnlUSD := longPnl / 100 * float64(numBatches) * legCapital
	shortPnlUSD := shortPnl / 100 * float64(numBatches) * legCapital

	totalLongShares := 0.0
	for _, p := range pos.LongEntryPrices {
		totalLongShares += legCapital / p
	}
	totalShortShares := 0.0
	for _, p := range pos.ShortEntryPrices {
		totalShortShares += legCapital / p
	}

	longEntryFeeSum := float64(numBatches) * legCapital * takerFees[pos.LongLeg.Exchange] / 100
	shortEntryFeeSum := float64(numBatches) * legCapital * takerFees[pos.ShortLeg.Exchange] / 100
	longExitFeeAmt := totalLongShares * pos.LongLeg.ExitPrice * takerFees[pos.LongLeg.Exchange] / 100
	shortExitFeeAmt := totalShortShares * pos.ShortLeg.ExitPrice * takerFees[pos.ShortLeg.Exchange] / 100
	longFeeUSD := longEntryFeeSum + longExitFeeAmt
	shortFeeUSD := shortEntryFeeSum + shortExitFeeAmt

	// Update per-exchange fund tracking
	t.mu.Lock()
	if ef, ok := t.exchangeFunds[pos.LongLeg.Exchange]; ok {
		ef.Balance -= longFeeUSD
		ef.Balance += longPnlUSD
		ef.TotalFee += longFeeUSD
		ef.TotalPnl += longPnlUSD
	}
	if ef, ok := t.exchangeFunds[pos.ShortLeg.Exchange]; ok {
		ef.Balance -= shortFeeUSD
		ef.Balance += shortPnlUSD
		ef.TotalFee += shortFeeUSD
		ef.TotalPnl += shortPnlUSD
	}
	t.mu.Unlock()

	// Save trade record for stats
	record := TradeRecord{
		Coin:        pos.Coin,
		Direction:   pos.Direction,
		EntrySpread: pos.EntrySpread,
		ExitSpread:  diffPct,
		PnlPct:      netPnl,
		PnlUSD:      2 * 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,
		PnlLongUSD:  longPnlUSD,
		PnlShortUSD: shortPnlUSD,
		FeeLongUSD:  longFeeUSD,
		FeeShortUSD: shortFeeUSD,
	}

	t.mu.Lock()
	delete(t.positions, pos.Coin)
	t.lastTradeTime[pos.Coin] = time.Now()
	t.closedTrades = append(t.closedTrades, record)
	t.realTradesDone++
	t.mu.Unlock()

	// Auto-stop: after 5 real trades, signal shutdown
	if t.realTradesTarget > 0 && t.realTradesDone >= t.realTradesTarget {
		log.Printf("[Trader] ✅ %d real trades completed — shutting down...", t.realTradesDone)
		t.shuttingDown = true
		select {
		case t.StopCh <- struct{}{}:
		default:
		}
	}

	// Persist exit orders + close trade in DB
	if t.db != nil && pos.DBTradeID > 0 {
		now := time.Now()
		status := "filled"

		// Save exit orders
		longExitShares := totalLongShares
		longOID, _ := t.db.SaveOrder(&db.OrderRecord{
			TradeID: pos.DBTradeID, Leg: "long", Type: "exit",
			Exchange: pos.LongLeg.Exchange, Side: "sell",
			Price: &pos.LongLeg.ExitPrice, Size: &longExitShares,
			Fee: &longExitFeeAmt, Status: &status, CreatedAt: now,
		})
		shortExitShares := totalShortShares
		shortOID, _ := t.db.SaveOrder(&db.OrderRecord{
			TradeID: pos.DBTradeID, Leg: "short", Type: "exit",
			Exchange: pos.ShortLeg.Exchange, Side: "buy",
			Price: &pos.ShortLeg.ExitPrice, Size: &shortExitShares,
			Fee: &shortExitFeeAmt, Status: &status, CreatedAt: now,
		})
		// Save exit system order
		t.db.SaveSystemOrder(&db.SystemOrderRecord{
			TradeID: pos.DBTradeID, Type: "exit", Status: "filled",
			Spread: &diffPct,
			LongPrice: &pos.LongLeg.ExitPrice, ShortPrice: &pos.ShortLeg.ExitPrice,
			LongOrderID: &longOID, ShortOrderID: &shortOID,
			CreatedAt: now,
		})

		// Close trade with per-exchange fee/pnl
		feeEntrySum := longEntryFeeSum + shortEntryFeeSum
		feeExitSum := longExitFeeAmt + shortExitFeeAmt

		t.db.UpdateTradeStatus(pos.DBTradeID, &db.TradeRecord{
			Status:      "closed",
			ExitSpread:  &diffPct,
			LongExit:    &pos.LongLeg.ExitPrice,
			ShortExit:   &pos.ShortLeg.ExitPrice,
			LongPnl:     &longPnl,
			ShortPnl:    &shortPnl,
			FeeEntry:    &feeEntrySum,
			FeeExit:     &feeExitSum,
			NetPnl:      &netPnl,
			AmountUSD:   pos.AmountUSD,
			ScaleCount:  pos.ScaleLevels,
			ExitReason:  &exitReason,
			Convergence: &convergenceLabel,
			ClosedAt:    &now,
			PnlLongUSD:  &longPnlUSD,
			PnlShortUSD: &shortPnlUSD,
			FeeLongUSD:  &longFeeUSD,
			FeeShortUSD: &shortFeeUSD,
		})
	}

	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
		log.Printf("[ExRes] BG %s %s: size=%s → response=%s", side, leg.Coin+"USDT", size, 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
		log.Printf("[ExRes] HL %s %s: size=%s → response=%s", side, leg.Coin, size, 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 ""
	}

	if leg.Exchange == ExBitget {
		resp, err := t.bitget.PlaceMarketOrder(side, leg.Coin+"USDT", leg.Size)
		if err != nil {
			return fmt.Sprintf("%v", err)
		}
		log.Printf("[ExRes] BG close %s %s: size=%s → response=%s", side, leg.Coin+"USDT", leg.Size, resp)
	} else {
		resp, err := t.hyperliquid.PlaceMarketOrder(leg.Coin, side, leg.Size)
		if err != nil {
			return fmt.Sprintf("%v", err)
		}
		log.Printf("[ExRes] HL close %s %s: size=%s → response=%s", side, leg.Coin, leg.Size, resp)
	}
	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 + update DB
		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,
			PnlUSD:      2 * pos.AmountUSD * pos.ExitNetPnl / 100,
			Convergence: pos.ExitConvergence,
			Reason:      pos.ExitReasonText,
			Duration:    elapsed.Round(time.Second).String(),
			OpenedAt:    pos.StartedAt,
			ClosedAt:    pos.ExitedAt,
			ScaleLevels: pos.ScaleLevels,
			AmountUSD:   pos.AmountUSD,
			PnlLongUSD:  pos.ExitLongPnlUSD,
			PnlShortUSD: pos.ExitShortPnlUSD,
			FeeLongUSD:  pos.ExitLongFeeUSD,
			FeeShortUSD: pos.ExitShortFeeUSD,
		}

		t.mu.Lock()
		delete(t.positions, pos.Coin)
		t.lastTradeTime[pos.Coin] = time.Now()
		t.closedTrades = append(t.closedTrades, record)
		// Update exchange funds
		if ef, ok := t.exchangeFunds[pos.LongLeg.Exchange]; ok {
			ef.Balance -= pos.ExitLongFeeUSD
			ef.Balance += pos.ExitLongPnlUSD
			ef.TotalFee += pos.ExitLongFeeUSD
			ef.TotalPnl += pos.ExitLongPnlUSD
		}
		if ef, ok := t.exchangeFunds[pos.ShortLeg.Exchange]; ok {
			ef.Balance -= pos.ExitShortFeeUSD
			ef.Balance += pos.ExitShortPnlUSD
			ef.TotalFee += pos.ExitShortFeeUSD
			ef.TotalPnl += pos.ExitShortPnlUSD
		}
		t.mu.Unlock()

		// Persist exit orders + close trade in DB (only for legs that weren't already closed)
		if t.db != nil && pos.DBTradeID > 0 {
			now := time.Now()
			status := "filled"
			tradeUnit := t.cfg.TradeAmountUSD

			totalLongShares := 0.0
			for _, p := range pos.LongEntryPrices {
				totalLongShares += tradeUnit / p
			}
			totalShortShares := 0.0
			for _, p := range pos.ShortEntryPrices {
				totalShortShares += tradeUnit / p
			}

			// Save exit orders for legs that were just now closed
			if pos.LongLeg.Closed {
				longExitFee := totalLongShares * pos.LongLeg.ExitPrice * takerFees[pos.LongLeg.Exchange] / 100
				longExitShares := totalLongShares
				_, _ = t.db.SaveOrder(&db.OrderRecord{
					TradeID: pos.DBTradeID, Leg: "long", Type: "exit",
					Exchange: pos.LongLeg.Exchange, Side: "sell",
					Price: &pos.LongLeg.ExitPrice, Size: &longExitShares,
					Fee: &longExitFee, Status: &status, CreatedAt: now,
				})
			}
			if pos.ShortLeg.Closed {
				shortExitFee := totalShortShares * pos.ShortLeg.ExitPrice * takerFees[pos.ShortLeg.Exchange] / 100
				shortExitShares := totalShortShares
				_, _ = t.db.SaveOrder(&db.OrderRecord{
					TradeID: pos.DBTradeID, Leg: "short", Type: "exit",
					Exchange: pos.ShortLeg.Exchange, Side: "buy",
					Price: &pos.ShortLeg.ExitPrice, Size: &shortExitShares,
					Fee: &shortExitFee, Status: &status, CreatedAt: now,
				})
			}
			// Save exit system order (idempotent-safe since we always overwrite on retry)
			t.db.SaveSystemOrder(&db.SystemOrderRecord{
				TradeID: pos.DBTradeID, Type: "exit", Status: "filled",
				Spread: &pos.ExitDiffPct,
				LongPrice: &pos.LongLeg.ExitPrice, ShortPrice: &pos.ShortLeg.ExitPrice,
				CreatedAt: now,
			})

			// Close trade using previously saved exit metadata
			feePct := pos.ExitTotalFees
			t.db.UpdateTradeStatus(pos.DBTradeID, &db.TradeRecord{
				Status:      "closed",
				ExitSpread:  &pos.ExitDiffPct,
				LongExit:    &pos.LongLeg.ExitPrice,
				ShortExit:   &pos.ShortLeg.ExitPrice,
				LongPnl:     &pos.ExitLongPnl,
				ShortPnl:    &pos.ExitShortPnl,
				FeeEntry:    &feePct,
				FeeExit:     &feePct,
				NetPnl:      &pos.ExitNetPnl,
				AmountUSD:   pos.AmountUSD,
				ScaleCount:  pos.ScaleLevels,
				ExitReason:  &pos.ExitReasonText,
				Convergence: &pos.ExitConvergence,
				ClosedAt:    &now,
				PnlLongUSD:  &pos.ExitLongPnlUSD,
				PnlShortUSD: &pos.ExitShortPnlUSD,
				FeeLongUSD:  &pos.ExitLongFeeUSD,
				FeeShortUSD: &pos.ExitShortFeeUSD,
			})
		}

		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
}

// calcArbPnL computes net PnL and total fees in USD, then normalizes to % of total deployed capital.
// This correctly handles scale-ins where the old formula (longPnl+shortPnl - (2+N)*0.105)
// double-counted fees because it didn't divide by (1+N) batches.
func calcArbPnL(longPnl, shortPnl float64, scaleLevels int, tradeAmountUSD float64) (netPnlPct, feePct float64) {
	numBatches := 1 + scaleLevels
	legCapital := tradeAmountUSD
	totalCapital := float64(numBatches) * 2 * legCapital

	// Gross PnL in USD
	longPnlUSD := longPnl / 100 * float64(numBatches) * legCapital
	shortPnlUSD := shortPnl / 100 * float64(numBatches) * legCapital
	grossPnLUSD := longPnlUSD + shortPnlUSD

	// Fee in USD (entry+exit per order-pair)
	feeUSD := float64(2+scaleLevels) * legCapital * (takerFees[ExBitget] + takerFees[ExHyperLiquid]) / 100

	netPnlPct = (grossPnLUSD - feeUSD) / totalCapital * 100
	feePct = feeUSD / totalCapital * 100
	return
}

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

// GetExchangeFunds returns a copy of per-exchange fund states.
func (t *Trader) GetExchangeFunds() map[string]ExchangeFund {
	t.mu.Lock()
	defer t.mu.Unlock()
	r := make(map[string]ExchangeFund, len(t.exchangeFunds))
	for ex, ef := range t.exchangeFunds {
		r[ex] = *ef
	}
	return r
}

// persistTrade saves a completed trade to SQLite, with per-leg orders and system_orders.
// 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}
		}

		// Restore scale-in prices from orders table for correct weighted average
		scaleLong, scaleShort, err := t.db.GetScalePrices(tr.ID)
		if err == nil {
			pos.LongEntryPrices = append(pos.LongEntryPrices, scaleLong...)
			pos.ShortEntryPrices = append(pos.ShortEntryPrices, scaleShort...)
			// Refresh leg EntryPrice to reflect all scale levels
			if len(pos.LongEntryPrices) > 1 {
				pos.LongLeg.EntryPrice = weightedAvgPrice(pos.LongEntryPrices, t.cfg.TradeAmountUSD)
			}
			if len(pos.ShortEntryPrices) > 1 {
				pos.ShortLeg.EntryPrice = weightedAvgPrice(pos.ShortEntryPrices, t.cfg.TradeAmountUSD)
			}
		}
		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.
// Calculates exit PnL fields so retryClose writes correct data to DB.
func (t *Trader) blacklistCoin(pos *ArbPosition, bgP, hlP, diffPct float64, notifier *Notifier) {
	// Compute exit PnL the same way checkExit does
	var longCurrent, shortCurrent float64
	if pos.LongLeg.Exchange == ExBitget {
		longCurrent, shortCurrent = bgP, hlP
	} else {
		longCurrent, shortCurrent = hlP, bgP
	}
	longAvg := weightedAvgPrice(pos.LongEntryPrices, t.cfg.TradeAmountUSD)
	shortAvg := weightedAvgPrice(pos.ShortEntryPrices, t.cfg.TradeAmountUSD)
	longPnl := (longCurrent - longAvg) / longAvg * 100
	shortPnl := (shortAvg - shortCurrent) / shortAvg * 100
	netPnl, totalFees := calcArbPnL(longPnl, shortPnl, pos.ScaleLevels, t.cfg.TradeAmountUSD)

	pos.ExitDiffPct = diffPct
	pos.ExitNetPnl = netPnl
	pos.ExitLongPnl = longPnl
	pos.ExitShortPnl = shortPnl
	pos.ExitTotalFees = totalFees
	pos.LongLeg.ExitPrice = longCurrent
	pos.ShortLeg.ExitPrice = shortCurrent
	pos.ExitReasonText = "黑名单强平"

	// Convergence label
	convergedPct := (pos.EntrySpread - diffPct) / pos.EntrySpread * 100
	if convergedPct < -10 {
		pos.ExitConvergence = "价差发散"
	} else if convergedPct < 10 {
		pos.ExitConvergence = "价差持平"
	} else {
		pos.ExitConvergence = "价差收敛"
	}

	t.mu.Lock()
	t.blacklist[pos.Coin] = time.Now()
	t.mu.Unlock()

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

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

// safeFloat returns 0 for nil float64 pointers (DB nullable fields).
func safeFloat(f *float64) float64 {
	if f == nil {
		return 0
	}
	return *f
}

// safeStr returns empty string for nil string pointers (DB nullable fields).
func safeStr(s *string) string {
	if s == nil {
		return ""
	}
	return *s
}