fix: paper trade data pipeline — persist fills, positions, and wire dashboard

TASKS.md

@@ -78,6 +78,8 @@ This document consolidates all implementation tasks from the architecture and de
 | **TB-005** | Observability | HIGH | 4h | Implement MCP metrics (request count, latency, errors) |
 | **TB-006** | Security | HIGH | 3h | Add API key expiration/rotation mechanism |
 
+> **Note (2026-03-20)**: Paper session `initial_capital` changed to $100k — affects new auto-created sessions only (existing sessions are unaffected).
+
 **Total Blocking Effort**: ~18 hours (2-3 days)
 
 ### ⚠️ CRITICAL FOR BETA LAUNCH

cmd/api/handlers_trading.go

@@ -1,17 +1,33 @@
 package main
 
 import (
+	"errors"
+	"fmt"
 	"net/http"
 	"strings"
 	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/rs/zerolog/log"
 
 	"github.com/ajitpratap0/cryptofunk/internal/db"
 )
 
+const (
+	paperSlippageBuy  = 1.001 // 0.1% adverse slippage for market buy orders
+	paperSlippageSell = 0.999 // 0.1% adverse slippage for market sell orders
+	// TODO: make slippage configurable via config.Trading.
+)
+
+// errOppositeSide is a sentinel returned from the WithTx callback when the
+// incoming order is on the opposite side of an existing open position. It is
+// handled by the caller to produce a 422 response without logging as an
+// internal server error.
+var errOppositeSide = errors.New("opposite side trade on existing position")
+
 // Session handlers
 func (s *APIServer) handleListSessions(c *gin.Context) {
 	ctx := c.Request.Context()
@@ -378,6 +394,20 @@ func (s *APIServer) handleCancelOrder(c *gin.Context) {
 	})
 }
 
+// quoteAsset derives the quote asset token from a trading symbol by checking
+// common suffixes. Falls back to "USDT" for unrecognised symbols.
+func quoteAsset(symbol string) string {
+	// Ordered most-specific first: "BUSD" before "BTC" prevents "BTCUSDT" from
+	// matching suffix "BTC" when "USDT" would be the correct quote asset.
+	// TODO: make configurable for non-Binance exchanges (e.g. Kraken uses XBT/USD).
+	for _, suffix := range []string{"USDT", "BUSD", "BTC", "ETH", "BNB"} {
+		if strings.HasSuffix(strings.ToUpper(symbol), suffix) {
+			return suffix
+		}
+	}
+	return "USDT"
+}
+
 // handlePaperTrade executes a paper (simulated) trade order.
 // Market orders are immediately filled; limit orders remain open (NEW status).
 // POST /api/v1/trade
@@ -389,63 +419,278 @@ func (s *APIServer) handlePaperTrade(c *gin.Context) {
 		Quantity float64 `json:"quantity" binding:"required,gt=0"`
 		Price    float64 `json:"price"`
 	}
-
 	if err := c.ShouldBindJSON(&req); err != nil {
-		c.JSON(http.StatusBadRequest, gin.H{
-			"error":   "invalid request body",
-			"details": err.Error(),
-		})
+		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request body", "details": err.Error()})
+		return
+	}
+	isLimit := strings.EqualFold(req.Type, "limit")
+	if isLimit && req.Price <= 0 {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "price is required for limit orders"})
 		return
 	}
 
-	if (req.Type == "limit" || req.Type == "LIMIT") && req.Price <= 0 {
-		c.JSON(http.StatusBadRequest, gin.H{
-			"error": "price is required for limit orders",
-		})
+	ctx := c.Request.Context()
+
+	// NOTE: Session lookup and creation happen outside WithTx intentionally.
+	// The session row is long-lived (one per trading mode) and is safe to create
+	// outside the fill transaction. A failed fill transaction does not orphan the
+	// session — the next request simply reuses it.
+
+	// 1. Resolve or create paper session
+	sessions, err := s.db.ListActiveSessions(ctx)
+	if err != nil {
+		log.Error().Err(err).Msg("Failed to list active sessions for paper trade")
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to resolve trading session"})
 		return
 	}
+	var sessionID *uuid.UUID
+	for i := range sessions {
+		if sessions[i].Mode == db.TradingModePaper {
+			id := sessions[i].ID
+			sessionID = &id
+			break
+		}
+	}
+	if sessionID == nil {
+		newSession := &db.TradingSession{
+			ID:   uuid.New(),
+			Mode: db.TradingModePaper,
+			// TODO: Add a session_type or is_multi_asset column to trading_sessions in a follow-up
+			// migration. "PAPER" is a placeholder to distinguish multi-asset paper sessions from
+			// single-symbol sessions (which use the actual symbol, e.g. "BTCUSDT").
+			Symbol:         "PAPER",
+			Exchange:       "paper",
+			InitialCapital: s.config.Trading.InitialCapital,
+			StartedAt:      time.Now(),
+			CreatedAt:      time.Now(),
+			UpdatedAt:      time.Now(),
+		}
+		if err := s.db.CreateSession(ctx, newSession); err != nil {
+			// Only retry on unique constraint violation (PG error code 23505).
+			// A concurrent request may have inserted the same paper session between the
+			// ListActiveSessions call above and this insert (TOCTOU race). In that case
+			// we look up and reuse the existing session.
+			// All other errors (e.g. connection timeout) are returned immediately.
+			var pgErr *pgconn.PgError
+			if errors.As(err, &pgErr) && pgErr.Code == "23505" {
+				log.Warn().Err(err).Msg("Unique constraint on paper session; retrying lookup for concurrent session")
+				sessions2, err2 := s.db.ListActiveSessions(ctx)
+				if err2 == nil {
+					for i := range sessions2 {
+						if sessions2[i].Mode == db.TradingModePaper {
+							id := sessions2[i].ID
+							sessionID = &id
+							break
+						}
+					}
+				}
+				if sessionID == nil {
+					log.Error().Err(err).Msg("Failed to find paper session after unique constraint violation")
+					c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create trading session"})
+					return
+				}
+			} else {
+				log.Error().Err(err).Msg("Failed to create paper session")
+				c.JSON(http.StatusInternalServerError, gin.H{
+					"error": "internal server error",
+				})
+				return
+			}
+		} else {
+			sessionID = &newSession.ID
+		}
+	}
+
+	// 2. Determine execution price
+	refPrice := req.Price
+	if !isLimit && refPrice <= 0 {
+		refPrice = s.exchange.GetMarketPrice(req.Symbol)
+		if refPrice <= 0 {
+			c.JSON(http.StatusBadRequest, gin.H{
+				"error": "no market price configured for symbol; provide a price field",
+			})
+			return
+		}
+	}
+	execPrice := refPrice
+	if !isLimit {
+		if strings.EqualFold(req.Side, "BUY") {
+			execPrice = refPrice * paperSlippageBuy
+		} else {
+			execPrice = refPrice * paperSlippageSell
+		}
+	}
 
-	var price *float64
+	// 3. Build order struct (inserted inside the transaction below)
+	now := time.Now()
+	var pricePtr *float64
 	if req.Price > 0 {
-		price = &req.Price
+		pricePtr = db.PtrFloat64(req.Price)
 	}
+	orderSide := db.ConvertOrderSide(req.Side)
+	orderType := db.ConvertOrderType(req.Type)
 
 	order := &db.Order{
 		ID:        uuid.New(),
+		SessionID: sessionID,
 		Symbol:    req.Symbol,
 		Exchange:  "paper",
-		Side:      db.ConvertOrderSide(req.Side),
-		Type:      db.ConvertOrderType(req.Type),
+		Side:      orderSide,
+		Type:      orderType,
 		Quantity:  req.Quantity,
-		Price:     price,
+		Price:     pricePtr,
 		Status:    db.OrderStatusNew,
-		PlacedAt:  time.Now(),
-		CreatedAt: time.Now(),
-		UpdatedAt: time.Now(),
+		PlacedAt:  now,
+		CreatedAt: now,
+		UpdatedAt: now,
 	}
 
-	ctx := c.Request.Context()
-	if err := s.db.InsertOrder(ctx, order); err != nil {
-		c.JSON(http.StatusInternalServerError, gin.H{
-			"error": "failed to create paper trade order",
-		})
-		return
-	}
-
-	// Simulate immediate fill for market orders
+	// 4. Immediate fill for market orders
 	if order.Type == db.OrderTypeMarket {
-		now := time.Now()
-		execPrice := req.Price
 		execQuoteQty := execPrice * req.Quantity
+		// Use the configured paper trading commission rate (taker fee from trading config).
+		// Falls back to 0.001 (0.1%) if not configured, matching Binance standard tier.
+		commissionRate := s.config.Trading.CommissionRate
+		if commissionRate <= 0 {
+			commissionRate = 0.001
+		}
+		commission := execQuoteQty * commissionRate
 
-		if err := s.db.UpdateOrderStatus(ctx, order.ID, db.OrderStatusFilled, req.Quantity, execQuoteQty, &now, nil, nil); err != nil {
-			log.Warn().Err(err).Str("order_id", order.ID.String()).Msg("Failed to mark paper trade order as filled")
-		} else {
+		posSide := db.PositionSideLong
+		if orderSide == db.OrderSideSell {
+			posSide = db.PositionSideShort
+		}
+
+		// Derive the quote asset from the symbol (e.g. "BTCUSDT" → "USDT").
+		commissionAsset := quoteAsset(req.Symbol)
+
+		// Wrap all fill writes in a single DB transaction so a mid-flight failure
+		// does not leave orphaned rows. The order insert is the first step inside
+		// the transaction so no orphaned order rows can result from a failed fill.
+		// The existingPos lookup is also inside the transaction to eliminate the
+		// TOCTOU race where two concurrent BUY orders could both observe
+		// existingPos == nil and each try to INSERT a new position for the same symbol.
+		// AggregateSessionStats is intentionally kept outside the transaction: it
+		// is a read-then-aggregate UPDATE that can be safely retried.
+		// RepeatableRead + FOR UPDATE on positions prevents concurrent orders from
+		// both observing existingPos == nil and each inserting a duplicate position.
+		// Final safety net: migration 019 adds a UNIQUE partial index on
+		// (session_id, symbol) WHERE exit_time IS NULL. If two concurrent transactions
+		// both observe existingPos == nil and both attempt to INSERT a new open
+		// position, the second INSERT fails with SQLSTATE 23505 (unique_violation),
+		// causing its enclosing transaction to roll back rather than silently creating
+		// a duplicate open position for the same symbol.
+		txErr := s.db.WithTx(ctx, pgx.TxOptions{IsoLevel: pgx.RepeatableRead}, func(tx pgx.Tx) error {
+			// Insert the order as the first step so it is rolled back atomically
+			// with all fill rows if any later step fails.
+			if err := s.db.InsertOrderTx(ctx, tx, order); err != nil {
+				return fmt.Errorf("failed to insert paper trade order: %w", err)
+			}
+
+			// Re-fetch position inside the transaction for a consistent view.
+			existingPos, err := s.db.GetOpenPositionBySymbolTx(ctx, tx, *sessionID, req.Symbol)
+			if err != nil {
+				return fmt.Errorf("failed to look up existing position: %w", err)
+			}
+
+			if existingPos != nil && existingPos.Side != posSide {
+				// Opposite-side trade on an existing open position. Proper close/reduce
+				// logic (netting, realized PnL calculation) is not yet implemented.
+				// Return a sentinel so the outer handler can respond with 422.
+				return errOppositeSide
+			}
+
+			// UpdateOrderStatus inside transaction
+			if err := s.db.UpdateOrderStatusTx(ctx, tx, order.ID, db.OrderStatusFilled, req.Quantity, execQuoteQty, &now, nil, nil); err != nil {
+				return fmt.Errorf("failed to mark paper order filled: %w", err)
+			}
 			order.Status = db.OrderStatusFilled
 			order.ExecutedQuantity = req.Quantity
 			order.ExecutedQuoteQuantity = execQuoteQty
 			order.FilledAt = &now
 			order.UpdatedAt = now
+
+			// InsertTrade inside transaction via DB-layer method.
+			trade := &db.Trade{
+				ID:              uuid.New(),
+				OrderID:         order.ID,
+				ExchangeTradeID: nil,
+				Symbol:          req.Symbol,
+				Exchange:        "paper",
+				Side:            orderSide,
+				Price:           execPrice,
+				Quantity:        req.Quantity,
+				QuoteQuantity:   execQuoteQty,
+				Commission:      commission,
+				CommissionAsset: &commissionAsset,
+				ExecutedAt:      now,
+				IsMaker:         false,
+				Metadata:        nil,
+				CreatedAt:       now,
+			}
+			if err := s.db.InsertTradeTx(ctx, tx, trade); err != nil {
+				return fmt.Errorf("failed to insert paper trade fill row: %w", err)
+			}
+
+			// Create or average into existing position inside transaction via DB-layer methods.
+			if existingPos == nil {
+				entryReason := "paper_trade_api"
+				pos := &db.Position{
+					ID:          uuid.New(),
+					SessionID:   sessionID,
+					Symbol:      req.Symbol,
+					Exchange:    "paper",
+					Side:        posSide,
+					EntryPrice:  execPrice,
+					Quantity:    req.Quantity,
+					EntryTime:   now,
+					EntryReason: &entryReason,
+					CreatedAt:   now,
+					UpdatedAt:   now,
+				}
+				if err := s.db.CreatePositionTx(ctx, tx, pos); err != nil {
+					return fmt.Errorf("failed to create position for paper trade: %w", err)
+				}
+			} else {
+				totalQty := existingPos.Quantity + req.Quantity
+				weightedAvg := (existingPos.Quantity*existingPos.EntryPrice + req.Quantity*execPrice) / totalQty
+				if err := s.db.UpdatePositionAveragingTx(ctx, tx, existingPos.ID, weightedAvg, totalQty, commission); err != nil {
+					return fmt.Errorf("failed to update position for paper trade: %w", err)
+				}
+			}
+			return nil
+		})
+
+		if txErr != nil {
+			if errors.Is(txErr, errOppositeSide) {
+				// Opposite-side trade on an existing open position. Proper close/reduce
+				// logic (netting, realized PnL calculation) is not yet implemented.
+				// Reject the trade rather than silently corrupting position data.
+				log.Warn().
+					Str("symbol", req.Symbol).
+					Str("order_side", string(posSide)).
+					Msg("Opposite-side trade on existing position; position close logic not yet implemented")
+				c.JSON(http.StatusUnprocessableEntity, gin.H{
+					"error": "position close/reduce not yet implemented; opposite-side trade rejected",
+				})
+				return
+			}
+			log.Error().Err(txErr).Msg("Paper trade transaction failed")
+			c.JSON(http.StatusInternalServerError, gin.H{"error": "internal server error"})
+			return
+		}
+
+		// AggregateSessionStats is outside the transaction: it is a safe read-aggregate
+		// UPDATE that can be retried without risk of partial data corruption.
+		if err := s.db.AggregateSessionStats(ctx, *sessionID); err != nil {
+			log.Warn().Err(err).Msg("Failed to aggregate session stats after paper trade")
+		}
+	} else {
+		// Limit orders are not immediately filled; persist the order record in NEW status.
+		if err := s.db.InsertOrder(ctx, order); err != nil {
+			log.Error().Err(err).Msg("Failed to insert paper trade limit order")
+			c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create paper trade order"})
+			return
 		}
 	}
 

cmd/api/main.go

@@ -18,6 +18,7 @@ import (
 	"github.com/ajitpratap0/cryptofunk/internal/api"
 	"github.com/ajitpratap0/cryptofunk/internal/config"
 	"github.com/ajitpratap0/cryptofunk/internal/db"
+	"github.com/ajitpratap0/cryptofunk/internal/exchange"
 	"github.com/ajitpratap0/cryptofunk/internal/safety"
 )
 
@@ -43,6 +44,7 @@ type APIServer struct {
 	orderExecSession   *mcp.ClientSession      // MCP session for order-executor calls
 	mcpClient          *mcp.Client             // MCP client for creating/reconnecting sessions
 	activeSessionID    *uuid.UUID              // Currently active trading session ID (guarded by sessionMu)
+	exchange           exchange.Exchange       // Shared mock exchange instance for paper trading
 }
 
 // HTTP client for orchestrator communication with timeout and connection pooling
@@ -122,6 +124,7 @@ func main() {
 		ctx:                ctx,
 		safetyGuard:        safetyGuard,
 		orderExecutorURL:   getOrderExecutorURL(),
+		exchange:           exchange.NewMockExchange(database),
 	}
 
 	// Initialize MCP client for order-executor (session connects lazily on first order)