exchange-monitor-go/db/trade_repo.go

package db

import (
	"database/sql"
	"time"
)

// TradeRecord mirrors the database row for trades table.
type TradeRecord struct {
	ID           int64
	Coin         string
	Direction    string
	Status       string // open / closed
	EntrySpread  *float64
	ExitSpread   *float64
	LongExchange string
	ShortExchange string
	LongEntry    *float64
	LongExit     *float64
	ShortEntry   *float64
	ShortExit    *float64
	LongPnl      *float64
	ShortPnl     *float64
	FeeEntry     *float64
	FeeExit      *float64
	NetPnl       *float64
	AmountUSD    float64
	ScaleCount   int
	ExitReason   *string
	Convergence  *string
	OpenedAt     time.Time
	ClosedAt     *time.Time
	PnlLongUSD   *float64 // per-exchange PnL in USD
	PnlShortUSD  *float64
	FeeLongUSD   *float64 // per-exchange fee in USD
	FeeShortUSD  *float64
}

// OrderRecord mirrors the database row for orders table.
type OrderRecord struct {
	ID        int64
	TradeID   int64
	Leg       string // long / short
	Type      string // entry / exit / scale
	Exchange  string
	Side      string // buy / sell
	Price     *float64
	Size      *float64
	Fee       *float64
	OrderID   *string
	Status    *string
	CreatedAt time.Time
}

// SystemOrderRecord represents one system-level arbitrage action (entry/scale/exit).
type SystemOrderRecord struct {
	ID        int64
	TradeID   int64
	Type      string // entry / scale / exit
	Status    string // filled / failed
	Spread    *float64
	LongPrice *float64
	ShortPrice *float64
	LongOrderID  *int64
	ShortOrderID *int64
	CreatedAt time.Time
}

// SaveTrade inserts a new trade and returns its ID.
func (d *DB) SaveTrade(t *TradeRecord) (int64, error) {
	res, err := d.Exec(`INSERT INTO trades (
		coin, direction, status, entry_spread, exit_spread,
		long_exchange, short_exchange, long_entry, long_exit, short_entry, short_exit,
		long_pnl, short_pnl, fee_entry, fee_exit, net_pnl,
		amount_usd, scale_count, exit_reason, convergence, opened_at, closed_at,
		pnl_long_usd, pnl_short_usd, fee_long_usd, fee_short_usd
	) VALUES (?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?, ?,?,?,?,?)`,
		t.Coin, t.Direction, t.Status, t.EntrySpread, t.ExitSpread,
		t.LongExchange, t.ShortExchange, t.LongEntry, t.LongExit, t.ShortEntry, t.ShortExit,
		t.LongPnl, t.ShortPnl, t.FeeEntry, t.FeeExit, t.NetPnl,
		t.AmountUSD, t.ScaleCount, t.ExitReason, t.Convergence, t.OpenedAt, t.ClosedAt,
		t.PnlLongUSD, t.PnlShortUSD, t.FeeLongUSD, t.FeeShortUSD,
	)
	if err != nil {
		return 0, err
	}
	return res.LastInsertId()
}

// UpdateTradeStatus updates an existing trade's close data.
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=?, fee_entry=?, fee_exit=?, net_pnl=?,
		amount_usd=?, scale_count=?, exit_reason=?, convergence=?, closed_at=?,
		pnl_long_usd=?, pnl_short_usd=?, fee_long_usd=?, fee_short_usd=?
		WHERE id=?`,
		t.Status, t.ExitSpread,
		t.LongExit, t.ShortExit,
		t.LongPnl, t.ShortPnl, t.FeeEntry, t.FeeExit, t.NetPnl,
		t.AmountUSD, t.ScaleCount, t.ExitReason, t.Convergence, t.ClosedAt,
		t.PnlLongUSD, t.PnlShortUSD, t.FeeLongUSD, t.FeeShortUSD,
		id,
	)
	return err
}

// GetOpenTrades returns all trades with status='open'.
func (d *DB) GetOpenTrades() ([]TradeRecord, error) {
	rows, err := d.Query(`SELECT id, coin, direction, status, entry_spread, exit_spread,
		long_exchange, short_exchange, long_entry, long_exit, short_entry, short_exit,
		long_pnl, short_pnl, fee_entry, fee_exit, net_pnl,
		amount_usd, scale_count, exit_reason, convergence, opened_at, closed_at,
		pnl_long_usd, pnl_short_usd, fee_long_usd, fee_short_usd
		FROM trades WHERE status='open'`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	return scanTrades(rows)
}

// GetTrades returns paginated closed trades.
func (d *DB) GetTrades(page, limit int, coin string) ([]TradeRecord, int, error) {
	// Count total
	var total int
	countSQL := "SELECT COUNT(*) FROM trades WHERE status='closed'"
	args := []interface{}{}
	if coin != "" {
		countSQL += " AND coin=?"
		args = append(args, coin)
	}
	if err := d.QueryRow(countSQL, args...).Scan(&total); err != nil {
		return nil, 0, err
	}

	// Fetch page
	offset := (page - 1) * limit
	query := `SELECT id, coin, direction, status, entry_spread, exit_spread,
		long_exchange, short_exchange, long_entry, long_exit, short_entry, short_exit,
		long_pnl, short_pnl, fee_entry, fee_exit, net_pnl,
		amount_usd, scale_count, exit_reason, convergence, opened_at, closed_at,
		pnl_long_usd, pnl_short_usd, fee_long_usd, fee_short_usd
		FROM trades WHERE status='closed'`
	if coin != "" {
		query += " AND coin=?"
	}
	query += " ORDER BY closed_at DESC LIMIT ? OFFSET ?"

	allArgs := args
	allArgs = append(allArgs, limit, offset)

	rows, err := d.Query(query, allArgs...)
	if err != nil {
		return nil, 0, err
	}
	defer rows.Close()

	trades, err := scanTrades(rows)
	return trades, total, err
}

// SaveOrder inserts an order record.
func (d *DB) SaveOrder(o *OrderRecord) (int64, error) {
	res, err := d.Exec(`INSERT INTO orders
		(trade_id, leg, type, exchange, side, price, size, fee, order_id, status, created_at)
		VALUES (?,?,?,?,?, ?,?,?,?,?, ?)`,
		o.TradeID, o.Leg, o.Type, o.Exchange, o.Side,
		o.Price, o.Size, o.Fee, o.OrderID, o.Status, o.CreatedAt,
	)
	if err != nil {
		return 0, err
	}
	return res.LastInsertId()
}

// SaveSystemOrder inserts a system-level order record.
func (d *DB) SaveSystemOrder(o *SystemOrderRecord) (int64, error) {
	res, err := d.Exec(`INSERT INTO system_orders
		(trade_id, type, status, spread, long_price, short_price, long_order_id, short_order_id, created_at)
		VALUES (?,?,?,?,?, ?,?,?,?)`,
		o.TradeID, o.Type, o.Status, o.Spread,
		o.LongPrice, o.ShortPrice, o.LongOrderID, o.ShortOrderID, o.CreatedAt,
	)
	if err != nil {
		return 0, err
	}
	return res.LastInsertId()
}

// GetTradeByID returns a single trade with its orders.
func (d *DB) GetTradeByID(id int64) (*TradeRecord, []OrderRecord, error) {
	row := d.QueryRow(`SELECT id, coin, direction, status, entry_spread, exit_spread,
		long_exchange, short_exchange, long_entry, long_exit, short_entry, short_exit,
		long_pnl, short_pnl, fee_entry, fee_exit, net_pnl,
		amount_usd, scale_count, exit_reason, convergence, opened_at, closed_at,
		pnl_long_usd, pnl_short_usd, fee_long_usd, fee_short_usd
		FROM trades WHERE id=?`, id)

	var t TradeRecord
	err := row.Scan(
		&t.ID, &t.Coin, &t.Direction, &t.Status, &t.EntrySpread, &t.ExitSpread,
		&t.LongExchange, &t.ShortExchange, &t.LongEntry, &t.LongExit, &t.ShortEntry, &t.ShortExit,
		&t.LongPnl, &t.ShortPnl, &t.FeeEntry, &t.FeeExit, &t.NetPnl,
		&t.AmountUSD, &t.ScaleCount, &t.ExitReason, &t.Convergence, &t.OpenedAt, &t.ClosedAt,
		&t.PnlLongUSD, &t.PnlShortUSD, &t.FeeLongUSD, &t.FeeShortUSD,
	)
	if err != nil {
		return nil, nil, err
	}

	// Fetch orders
	oRows, err := d.Query(`SELECT id, trade_id, leg, type, exchange, side,
		price, size, fee, order_id, status, created_at
		FROM orders WHERE trade_id=? ORDER BY id`, id)
	if err != nil {
		return nil, nil, err
	}
	defer oRows.Close()

	var orders []OrderRecord
	for oRows.Next() {
		var o OrderRecord
		if err := oRows.Scan(&o.ID, &o.TradeID, &o.Leg, &o.Type, &o.Exchange, &o.Side,
			&o.Price, &o.Size, &o.Fee, &o.OrderID, &o.Status, &o.CreatedAt); err != nil {
			return nil, nil, err
		}
		orders = append(orders, o)
	}
	return &t, orders, nil
}

func scanTrades(rows *sql.Rows) ([]TradeRecord, error) {
	var trades []TradeRecord
	for rows.Next() {
		var t TradeRecord
		if err := rows.Scan(
			&t.ID, &t.Coin, &t.Direction, &t.Status, &t.EntrySpread, &t.ExitSpread,
			&t.LongExchange, &t.ShortExchange, &t.LongEntry, &t.LongExit, &t.ShortEntry, &t.ShortExit,
			&t.LongPnl, &t.ShortPnl, &t.FeeEntry, &t.FeeExit, &t.NetPnl,
			&t.AmountUSD, &t.ScaleCount, &t.ExitReason, &t.Convergence, &t.OpenedAt, &t.ClosedAt,
			&t.PnlLongUSD, &t.PnlShortUSD, &t.FeeLongUSD, &t.FeeShortUSD,
		); err != nil {
			return nil, err
		}
		trades = append(trades, t)
	}
	return trades, rows.Err()
}

// GetScalePrices returns scale-in order prices for a trade, grouped by leg.
func (d *DB) GetScalePrices(tradeID int64) (longPrices, shortPrices []float64, err error) {
	rows, err := d.Query(`SELECT leg, price FROM orders
		WHERE trade_id=? AND type='scale' AND price IS NOT NULL
		ORDER BY id`, tradeID)
	if err != nil {
		return nil, nil, err
	}
	defer rows.Close()
	for rows.Next() {
		var leg string
		var price float64
		if err := rows.Scan(&leg, &price); err != nil {
			return nil, nil, err
		}
		switch leg {
		case "long":
			longPrices = append(longPrices, price)
		case "short":
			shortPrices = append(shortPrices, price)
		}
	}
	return longPrices, shortPrices, rows.Err()
}

// GetAllClosedTrades returns all closed trades for PnL history restoration.
func (d *DB) GetAllClosedTrades() ([]TradeRecord, error) {
	rows, err := d.Query(`SELECT id, coin, direction, status, entry_spread, exit_spread,
		long_exchange, short_exchange, long_entry, long_exit, short_entry, short_exit,
		long_pnl, short_pnl, fee_entry, fee_exit, net_pnl,
		amount_usd, scale_count, exit_reason, convergence, opened_at, closed_at,
		pnl_long_usd, pnl_short_usd, fee_long_usd, fee_short_usd
		FROM trades WHERE status='closed' ORDER BY id`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	return scanTrades(rows)
}

// GetClosedStats returns convergence counts from the database.
func (d *DB) GetClosedStats() (converged, diverged, flat, total int, err error) {
	if err = d.QueryRow("SELECT COUNT(*) FROM trades WHERE status='closed'").Scan(&total); err != nil {
		return
	}
	if err = d.QueryRow("SELECT COUNT(*) FROM trades WHERE status='closed' AND convergence='价差收敛'").Scan(&converged); err != nil {
		return
	}
	if err = d.QueryRow("SELECT COUNT(*) FROM trades WHERE status='closed' AND convergence='价差发散'").Scan(&diverged); err != nil {
		return
	}
	if err = d.QueryRow("SELECT COUNT(*) FROM trades WHERE status='closed' AND (convergence IS NULL OR convergence NOT IN ('价差收敛','价差发散'))").Scan(&flat); err != nil {
		return
	}
	return
}