refactor: 逐批次实时写入DB，删除persistTrade

db/trade_repo.go

@@ -85,13 +85,13 @@ func (d *DB) SaveTrade(t *TradeRecord) (int64, error) {
 func (d *DB) UpdateTradeStatus(id int64, t *TradeRecord) error {
 	_, err := d.Exec(`UPDATE trades SET
 		status=?, exit_spread=?, long_exit=?, short_exit=?,
-		long_pnl=?, short_pnl=?, net_pnl=?,
+		long_pnl=?, short_pnl=?, fee_entry=?, fee_exit=?, net_pnl=?,
-		scale_count=?, exit_reason=?, convergence=?, closed_at=?
+		amount_usd=?, scale_count=?, exit_reason=?, convergence=?, closed_at=?
 		WHERE id=?`,
 		t.Status, t.ExitSpread,
 		t.LongExit, t.ShortExit,
-		t.LongPnl, t.ShortPnl, t.NetPnl,
+		t.LongPnl, t.ShortPnl, t.FeeEntry, t.FeeExit, t.NetPnl,
-		t.ScaleCount, t.ExitReason, t.Convergence, t.ClosedAt,
+		t.AmountUSD, t.ScaleCount, t.ExitReason, t.Convergence, t.ClosedAt,
 		id,
 	)
 	return err

trader.go

@@ -63,6 +63,9 @@ type ArbPosition struct {
 	// 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).
@@ -410,6 +413,55 @@ func (t *Trader) executeEntry(opp *ArbOpportunity, store *PriceStore, notifier *
 	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)
@@ -495,6 +547,39 @@ func (t *Trader) checkScaleIn(pos *ArbPosition, bgP, hlP, diffPct float64, store
 	pos.LongEntryPrices = append(pos.LongEntryPrices, longPrice)
 	pos.ShortEntryPrices = append(pos.ShortEntryPrices, shortPrice)
 
+	// 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)
 }
@@ -617,9 +702,73 @@ func (t *Trader) checkExit(pos *ArbPosition, bgP, hlP, diffPct float64, notifier
 	t.closedTrades = append(t.closedTrades, record)
 	t.mu.Unlock()
 
-	// Persist to SQLite
+	// Persist exit orders + close trade in DB
-	if t.db != nil {
+	if t.db != nil && pos.DBTradeID > 0 {
-		t.persistTrade(pos, diffPct, convergenceLabel, exitReason, netPnl, longPnl, shortPnl, totalFees)
+		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
+		longExitFee := totalLongShares * pos.LongLeg.ExitPrice * takerFees[pos.LongLeg.Exchange] / 100
+		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: &longExitFee, Status: &status, CreatedAt: now,
+		})
+		shortExitFee := totalShortShares * pos.ShortLeg.ExitPrice * takerFees[pos.ShortLeg.Exchange] / 100
+		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: &shortExitFee, 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: sum fees from in-memory calculation, update status
+		feeEntrySum, feeExitSum := 0.0, 0.0
+		for range pos.LongEntryPrices {
+			feeEntrySum += tradeUnit * takerFees[pos.LongLeg.Exchange] / 100
+		}
+		for range pos.ShortEntryPrices {
+			feeEntrySum += tradeUnit * takerFees[pos.ShortLeg.Exchange] / 100
+		}
+		feeExitSum = longExitFee + shortExitFee
+
+		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,
+		})
 	}
 
 	msg := fmt.Sprintf(
@@ -732,7 +881,7 @@ func (t *Trader) retryClose(pos *ArbPosition, bgP, hlP float64, notifier *Notifi
 
 	closeErr := t.closeBothLegs(pos)
 	if closeErr == "" {
-		// All legs finally closed — record + persist
+		// All legs finally closed — record + update DB
 		pos.Status = "closed"
 		pos.ExitedAt = time.Now()
 
@@ -758,9 +907,68 @@ func (t *Trader) retryClose(pos *ArbPosition, bgP, hlP float64, notifier *Notifi
 		t.closedTrades = append(t.closedTrades, record)
 		t.mu.Unlock()
 
-		if t.db != nil {
+		// Persist exit orders + close trade in DB (only for legs that weren't already closed)
-			t.persistTrade(pos, pos.ExitDiffPct, pos.ExitConvergence, pos.ExitReasonText,
+		if t.db != nil && pos.DBTradeID > 0 {
-				pos.ExitNetPnl, pos.ExitLongPnl, pos.ExitShortPnl, pos.ExitTotalFees)
+			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,
+			})
 		}
 
 		notifier.Send(fmt.Sprintf(
@@ -926,200 +1134,6 @@ func (t *Trader) GetClosedTrades() []TradeRecord {
 }
 
 // persistTrade saves a completed trade to SQLite, with per-leg orders and system_orders.
-func (t *Trader) persistTrade(pos *ArbPosition, exitSpread float64, convergence, exitReason string, netPnl, longPnl, shortPnl, totalFees float64) {
-	var entrySpread float64
-	if pos.LongLeg != nil {
-		entrySpread = pos.EntrySpread
-	}
-
-	now := time.Now()
-	tradeUnit := t.cfg.TradeAmountUSD
-
-	// Pre-calculate all fees BEFORE saving the trade
-	totalFeeEntryUSD := 0.0
-	for range pos.LongEntryPrices {
-		totalFeeEntryUSD += tradeUnit * takerFees[pos.LongLeg.Exchange] / 100
-	}
-	for range pos.ShortEntryPrices {
-		totalFeeEntryUSD += tradeUnit * takerFees[pos.ShortLeg.Exchange] / 100
-	}
-	totalLongShares := 0.0
-	for _, p := range pos.LongEntryPrices {
-		totalLongShares += tradeUnit / p
-	}
-	totalShortShares := 0.0
-	for _, p := range pos.ShortEntryPrices {
-		totalShortShares += tradeUnit / p
-	}
-	totalFeeExitUSD := totalLongShares*pos.LongLeg.ExitPrice*takerFees[pos.LongLeg.Exchange]/100 +
-		totalShortShares*pos.ShortLeg.ExitPrice*takerFees[pos.ShortLeg.Exchange]/100
-
-	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:      &totalFeeEntryUSD,
-		FeeExit:       &totalFeeExitUSD,
-		NetPnl:        &netPnl,
-		AmountUSD:     pos.AmountUSD,
-		ScaleCount:    pos.ScaleLevels,
-		ExitReason:    &exitReason,
-		Convergence:   &convergence,
-		OpenedAt:      pos.StartedAt,
-		ClosedAt:      &now,
-	}
-	tradeID, err := t.db.SaveTrade(dbTrade)
-	if err != nil {
-		log.Printf("[Trader] Failed to save trade to DB: %v", err)
-		return
-	}
-
-	// Save per-leg order records
-	// Long leg: entry (buy), scales (buy), exit (sell)
-	status := "filled"
-	var longEntryOrderIDs []int64
-	for i, p := range pos.LongEntryPrices {
-		shares := tradeUnit / p
-		orderType := "entry"
-		if i > 0 {
-			orderType = "scale"
-		}
-		fee := tradeUnit * takerFees[pos.LongLeg.Exchange] / 100
-		oid, oErr := t.db.SaveOrder(&db.OrderRecord{
-			TradeID:   tradeID,
-			Leg:       "long",
-			Type:      orderType,
-			Exchange:  pos.LongLeg.Exchange,
-			Side:      "buy",
-			Price:     &p,
-			Size:      &shares,
-			Fee:       &fee,
-			Status:    &status,
-			CreatedAt: pos.StartedAt,
-		})
-		if oErr != nil {
-			log.Printf("[Trader] Failed to save long entry order: %v", oErr)
-		} else {
-			longEntryOrderIDs = append(longEntryOrderIDs, oid)
-		}
-	}
-	// Long exit (sell)
-	longExitShares := totalLongShares
-	longExitFee := totalLongShares * pos.LongLeg.ExitPrice * takerFees[pos.LongLeg.Exchange] / 100
-	status = "filled"
-	var longExitOrderID int64
-	if oid, oErr := t.db.SaveOrder(&db.OrderRecord{
-		TradeID:   tradeID,
-		Leg:       "long",
-		Type:      "exit",
-		Exchange:  pos.LongLeg.Exchange,
-		Side:      "sell",
-		Price:     &pos.LongLeg.ExitPrice,
-		Size:      &longExitShares,
-		Fee:       &longExitFee,
-		Status:    &status,
-		CreatedAt: now,
-	}); oErr != nil {
-		log.Printf("[Trader] Failed to save long exit order: %v", oErr)
-	} else {
-		longExitOrderID = oid
-	}
-
-	// Short leg: entry (sell), scales (sell), exit (buy)
-	var shortEntryOrderIDs []int64
-	for i, p := range pos.ShortEntryPrices {
-		shares := tradeUnit / p
-		orderType := "entry"
-		if i > 0 {
-			orderType = "scale"
-		}
-		fee := tradeUnit * takerFees[pos.ShortLeg.Exchange] / 100
-		oid, oErr := t.db.SaveOrder(&db.OrderRecord{
-			TradeID:   tradeID,
-			Leg:       "short",
-			Type:      orderType,
-			Exchange:  pos.ShortLeg.Exchange,
-			Side:      "sell",
-			Price:     &p,
-			Size:      &shares,
-			Fee:       &fee,
-			Status:    &status,
-			CreatedAt: pos.StartedAt,
-		})
-		if oErr != nil {
-			log.Printf("[Trader] Failed to save short entry order: %v", oErr)
-		} else {
-			shortEntryOrderIDs = append(shortEntryOrderIDs, oid)
-		}
-	}
-	// Short exit (buy)
-	shortExitShares := totalShortShares
-	shortExitFee := totalShortShares * pos.ShortLeg.ExitPrice * takerFees[pos.ShortLeg.Exchange] / 100
-	var shortExitOrderID int64
-	if oid, oErr := t.db.SaveOrder(&db.OrderRecord{
-		TradeID:   tradeID,
-		Leg:       "short",
-		Type:      "exit",
-		Exchange:  pos.ShortLeg.Exchange,
-		Side:      "buy",
-		Price:     &pos.ShortLeg.ExitPrice,
-		Size:      &shortExitShares,
-		Fee:       &shortExitFee,
-		Status:    &status,
-		CreatedAt: now,
-	}); oErr != nil {
-		log.Printf("[Trader] Failed to save short exit order: %v", oErr)
-	} else {
-		shortExitOrderID = oid
-	}
-
-	// Save system orders linking long+short legs
-	es := pos.EntrySpread
-	for i := 0; i < len(longEntryOrderIDs) && i < len(shortEntryOrderIDs); i++ {
-		sysType := "entry"
-		if i > 0 {
-			sysType = "scale"
-		}
-		if _, sErr := t.db.SaveSystemOrder(&db.SystemOrderRecord{
-			TradeID:   tradeID,
-			Type:      sysType,
-			Status:    "filled",
-			Spread:    &es,
-			LongPrice:  &pos.LongEntryPrices[i],
-			ShortPrice: &pos.ShortEntryPrices[i],
-			LongOrderID:  &longEntryOrderIDs[i],
-			ShortOrderID: &shortEntryOrderIDs[i],
-			CreatedAt: pos.StartedAt,
-		}); sErr != nil {
-			log.Printf("[Trader] Failed to save entry system order: %v", sErr)
-		}
-	}
-	// Exit system order
-	if _, sErr := t.db.SaveSystemOrder(&db.SystemOrderRecord{
-		TradeID:   tradeID,
-		Type:      "exit",
-		Status:    "filled",
-		Spread:    &exitSpread,
-		LongPrice:  &pos.LongLeg.ExitPrice,
-		ShortPrice: &pos.ShortLeg.ExitPrice,
-		LongOrderID:  &longExitOrderID,
-		ShortOrderID: &shortExitOrderID,
-		CreatedAt: now,
-	}); sErr != nil {
-		log.Printf("[Trader] Failed to save exit system order: %v", sErr)
-	}
-}
-
 // restoreOpenPositions loads open trades from DB and recreates their positions.
 func (t *Trader) restoreOpenPositions() {
 	openTrades, err := t.db.GetOpenTrades()