feat: add multi-exchange order sync and position tracking

- Add order sync implementations for Hyperliquid, Bybit, OKX, Bitget, Aster
- Add position snapshot functionality for exchange position reset
- Update TraderOrder and TraderFill structures with exchange_type field
- Add exchange sync tests
- Update frontend charts components
- Remove deprecated position_sync.go
This commit is contained in:
tinkle-community
2025-12-27 02:09:48 +08:00
parent 2172b252a5
commit 46922f8c53
19 changed files with 2405 additions and 898 deletions

1
go.mod
View File

@@ -58,6 +58,7 @@ require (
github.com/mailru/easyjson v0.9.1 // indirect
github.com/mattn/go-colorable v0.1.14 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/mattn/go-sqlite3 v1.14.32 // indirect
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
github.com/modern-go/reflect2 v1.0.2 // indirect
github.com/ncruces/go-strftime v0.1.9 // indirect

2
go.sum
View File

@@ -138,6 +138,8 @@ github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWE
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/mattn/go-runewidth v0.0.16 h1:E5ScNMtiwvlvB5paMFdw9p4kSQzbXFikJ5SQO6TULQc=
github.com/mattn/go-runewidth v0.0.16/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
github.com/mattn/go-sqlite3 v1.14.32 h1:JD12Ag3oLy1zQA+BNn74xRgaBbdhbNIDYvQUEuuErjs=
github.com/mattn/go-sqlite3 v1.14.32/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
github.com/minio/sha256-simd v1.0.0 h1:v1ta+49hkWZyvaKwrQB8elexRqm6Y0aMLjCNsrYxo6g=
github.com/minio/sha256-simd v1.0.0/go.mod h1:OuYzVNI5vcoYIAmbIvHPl3N3jUzVedXbKy5RFepssQM=
github.com/mitchellh/mapstructure v1.4.1 h1:CpVNEelQCZBooIPDn+AR3NpivK/TIKU8bDxdASFVQag=

View File

@@ -11,7 +11,6 @@ import (
"nofx/manager"
"nofx/mcp"
"nofx/store"
"nofx/trader"
"os"
"os/signal"
"path/filepath"
@@ -115,11 +114,6 @@ func main() {
logger.Warnf("⚠️ Failed to restore backtest history: %v", err)
}
// Start position sync manager (detects manual closures, TP/SL triggers)
positionSyncManager := trader.NewPositionSyncManager(st, 0) // 0 = use default 10s interval
positionSyncManager.Start()
defer positionSyncManager.Stop()
// Load all traders from database to memory (may auto-start traders with IsRunning=true)
if err := traderManager.LoadTradersFromStore(st); err != nil {
logger.Fatalf("❌ Failed to load traders: %v", err)

View File

@@ -10,12 +10,13 @@ import (
type TraderOrder struct {
ID int64 `json:"id"`
TraderID string `json:"trader_id"`
ExchangeID string `json:"exchange_id"` // 交易所账户UUID
ExchangeOrderID string `json:"exchange_order_id"` // 交易所订单ID
ClientOrderID string `json:"client_order_id"` // 客户端订单ID
Symbol string `json:"symbol"` // 交易对
ExchangeID string `json:"exchange_id"` // Exchange account UUID
ExchangeType string `json:"exchange_type"` // Exchange type (hyperliquid/lighter/binance/etc)
ExchangeOrderID string `json:"exchange_order_id"` // Exchange order ID
ClientOrderID string `json:"client_order_id"` // Client order ID
Symbol string `json:"symbol"` // Trading pair
Side string `json:"side"` // BUY/SELL
PositionSide string `json:"position_side"` // LONG/SHORT (双向持仓模式)
PositionSide string `json:"position_side"` // LONG/SHORT (hedge mode)
Type string `json:"type"` // MARKET/LIMIT/STOP/STOP_MARKET/TAKE_PROFIT/TAKE_PROFIT_MARKET
TimeInForce string `json:"time_in_force"` // GTC/IOC/FOK
Quantity float64 `json:"quantity"` // 订单数量
@@ -38,14 +39,15 @@ type TraderOrder struct {
FilledAt time.Time `json:"filled_at"` // 完全成交时间
}
// TraderFill 成交记录(一个订单可能有多次成交)
// TraderFill trade record (one order may have multiple fills)
type TraderFill struct {
ID int64 `json:"id"`
TraderID string `json:"trader_id"`
ExchangeID string `json:"exchange_id"`
OrderID int64 `json:"order_id"` // 关联的订单ID
ExchangeOrderID string `json:"exchange_order_id"` // 交易所订单ID
ExchangeTradeID string `json:"exchange_trade_id"` // 交易所成交ID
ExchangeID string `json:"exchange_id"` // Exchange account UUID
ExchangeType string `json:"exchange_type"` // Exchange type (hyperliquid/lighter/binance/etc)
OrderID int64 `json:"order_id"` // Related order ID
ExchangeOrderID string `json:"exchange_order_id"` // Exchange order ID
ExchangeTradeID string `json:"exchange_trade_id"` // Exchange trade ID
Symbol string `json:"symbol"`
Side string `json:"side"` // BUY/SELL
Price float64 `json:"price"` // 成交价格
@@ -76,6 +78,7 @@ func (s *OrderStore) InitTables() error {
id INTEGER PRIMARY KEY AUTOINCREMENT,
trader_id TEXT NOT NULL,
exchange_id TEXT NOT NULL DEFAULT '',
exchange_type TEXT NOT NULL DEFAULT '',
exchange_order_id TEXT NOT NULL,
client_order_id TEXT DEFAULT '',
symbol TEXT NOT NULL,
@@ -114,6 +117,7 @@ func (s *OrderStore) InitTables() error {
id INTEGER PRIMARY KEY AUTOINCREMENT,
trader_id TEXT NOT NULL,
exchange_id TEXT NOT NULL DEFAULT '',
exchange_type TEXT NOT NULL DEFAULT '',
order_id INTEGER NOT NULL,
exchange_order_id TEXT NOT NULL,
exchange_trade_id TEXT NOT NULL,
@@ -168,16 +172,16 @@ func (s *OrderStore) CreateOrder(order *TraderOrder) error {
result, err := s.db.Exec(`
INSERT INTO trader_orders (
trader_id, exchange_id, exchange_order_id, client_order_id,
trader_id, exchange_id, exchange_type, exchange_order_id, client_order_id,
symbol, side, position_side, type, time_in_force,
quantity, price, stop_price, status,
filled_quantity, avg_fill_price, commission, commission_asset,
leverage, reduce_only, close_position, working_type, price_protect,
order_action, related_position_id,
created_at, updated_at, filled_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`,
order.TraderID, order.ExchangeID, order.ExchangeOrderID, order.ClientOrderID,
order.TraderID, order.ExchangeID, order.ExchangeType, order.ExchangeOrderID, order.ClientOrderID,
order.Symbol, order.Side, order.PositionSide, order.Type, order.TimeInForce,
order.Quantity, order.Price, order.StopPrice, order.Status,
order.FilledQuantity, order.AvgFillPrice, order.Commission, order.CommissionAsset,
@@ -244,13 +248,13 @@ func (s *OrderStore) CreateFill(fill *TraderFill) error {
result, err := s.db.Exec(`
INSERT INTO trader_fills (
trader_id, exchange_id, order_id, exchange_order_id, exchange_trade_id,
trader_id, exchange_id, exchange_type, order_id, exchange_order_id, exchange_trade_id,
symbol, side, price, quantity, quote_quantity,
commission, commission_asset, realized_pnl, is_maker,
created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`,
fill.TraderID, fill.ExchangeID, fill.OrderID, fill.ExchangeOrderID, fill.ExchangeTradeID,
fill.TraderID, fill.ExchangeID, fill.ExchangeType, fill.OrderID, fill.ExchangeOrderID, fill.ExchangeTradeID,
fill.Symbol, fill.Side, fill.Price, fill.Quantity, fill.QuoteQuantity,
fill.Commission, fill.CommissionAsset, fill.RealizedPnL, fill.IsMaker,
now.Format(time.RFC3339),
@@ -267,7 +271,7 @@ func (s *OrderStore) CreateFill(fill *TraderFill) error {
// GetFillByExchangeTradeID 根据交易所成交ID获取成交记录
func (s *OrderStore) GetFillByExchangeTradeID(exchangeID, exchangeTradeID string) (*TraderFill, error) {
row := s.db.QueryRow(`
SELECT id, trader_id, exchange_id, order_id, exchange_order_id, exchange_trade_id,
SELECT id, trader_id, exchange_id, exchange_type, order_id, exchange_order_id, exchange_trade_id,
symbol, side, price, quantity, quote_quantity,
commission, commission_asset, realized_pnl, is_maker,
created_at
@@ -278,7 +282,7 @@ func (s *OrderStore) GetFillByExchangeTradeID(exchangeID, exchangeTradeID string
var fill TraderFill
var createdAt sql.NullString
err := row.Scan(
&fill.ID, &fill.TraderID, &fill.ExchangeID, &fill.OrderID, &fill.ExchangeOrderID, &fill.ExchangeTradeID,
&fill.ID, &fill.TraderID, &fill.ExchangeID, &fill.ExchangeType, &fill.OrderID, &fill.ExchangeOrderID, &fill.ExchangeTradeID,
&fill.Symbol, &fill.Side, &fill.Price, &fill.Quantity, &fill.QuoteQuantity,
&fill.Commission, &fill.CommissionAsset, &fill.RealizedPnL, &fill.IsMaker,
&createdAt,
@@ -303,7 +307,7 @@ func (s *OrderStore) GetFillByExchangeTradeID(exchangeID, exchangeTradeID string
// GetOrderByExchangeID 根据交易所订单ID获取订单
func (s *OrderStore) GetOrderByExchangeID(exchangeID, exchangeOrderID string) (*TraderOrder, error) {
row := s.db.QueryRow(`
SELECT id, trader_id, exchange_id, exchange_order_id, client_order_id,
SELECT id, trader_id, exchange_id, exchange_type, exchange_order_id, client_order_id,
symbol, side, position_side, type, time_in_force,
quantity, price, stop_price, status,
filled_quantity, avg_fill_price, commission, commission_asset,
@@ -317,7 +321,7 @@ func (s *OrderStore) GetOrderByExchangeID(exchangeID, exchangeOrderID string) (*
var order TraderOrder
var createdAt, updatedAt, filledAt sql.NullString
err := row.Scan(
&order.ID, &order.TraderID, &order.ExchangeID, &order.ExchangeOrderID, &order.ClientOrderID,
&order.ID, &order.TraderID, &order.ExchangeID, &order.ExchangeType, &order.ExchangeOrderID, &order.ClientOrderID,
&order.Symbol, &order.Side, &order.PositionSide, &order.Type, &order.TimeInForce,
&order.Quantity, &order.Price, &order.StopPrice, &order.Status,
&order.FilledQuantity, &order.AvgFillPrice, &order.Commission, &order.CommissionAsset,
@@ -355,7 +359,7 @@ func (s *OrderStore) GetOrderByExchangeID(exchangeID, exchangeOrderID string) (*
// GetTraderOrders 获取trader的订单列表
func (s *OrderStore) GetTraderOrders(traderID string, limit int) ([]*TraderOrder, error) {
rows, err := s.db.Query(`
SELECT id, trader_id, exchange_id, exchange_order_id, client_order_id,
SELECT id, trader_id, exchange_id, exchange_type, exchange_order_id, client_order_id,
symbol, side, position_side, type, time_in_force,
quantity, price, stop_price, status,
filled_quantity, avg_fill_price, commission, commission_asset,
@@ -377,7 +381,7 @@ func (s *OrderStore) GetTraderOrders(traderID string, limit int) ([]*TraderOrder
var order TraderOrder
var createdAt, updatedAt, filledAt sql.NullString
err := rows.Scan(
&order.ID, &order.TraderID, &order.ExchangeID, &order.ExchangeOrderID, &order.ClientOrderID,
&order.ID, &order.TraderID, &order.ExchangeID, &order.ExchangeType, &order.ExchangeOrderID, &order.ClientOrderID,
&order.Symbol, &order.Side, &order.PositionSide, &order.Type, &order.TimeInForce,
&order.Quantity, &order.Price, &order.StopPrice, &order.Status,
&order.FilledQuantity, &order.AvgFillPrice, &order.Commission, &order.CommissionAsset,
@@ -415,7 +419,7 @@ func (s *OrderStore) GetTraderOrders(traderID string, limit int) ([]*TraderOrder
// GetOrderFills 获取订单的成交记录
func (s *OrderStore) GetOrderFills(orderID int64) ([]*TraderFill, error) {
rows, err := s.db.Query(`
SELECT id, trader_id, exchange_id, order_id, exchange_order_id, exchange_trade_id,
SELECT id, trader_id, exchange_id, exchange_type, order_id, exchange_order_id, exchange_trade_id,
symbol, side, price, quantity, quote_quantity,
commission, commission_asset, realized_pnl, is_maker,
created_at
@@ -433,7 +437,7 @@ func (s *OrderStore) GetOrderFills(orderID int64) ([]*TraderFill, error) {
var fill TraderFill
var createdAt sql.NullString
err := rows.Scan(
&fill.ID, &fill.TraderID, &fill.ExchangeID, &fill.OrderID, &fill.ExchangeOrderID, &fill.ExchangeTradeID,
&fill.ID, &fill.TraderID, &fill.ExchangeID, &fill.ExchangeType, &fill.OrderID, &fill.ExchangeOrderID, &fill.ExchangeTradeID,
&fill.Symbol, &fill.Side, &fill.Price, &fill.Quantity, &fill.QuoteQuantity,
&fill.Commission, &fill.CommissionAsset, &fill.RealizedPnL, &fill.IsMaker,
&createdAt,

188
trader/aster_order_sync.go Normal file
View File

@@ -0,0 +1,188 @@
package trader
import (
"fmt"
"nofx/logger"
"nofx/store"
"sort"
"strings"
"time"
)
// SyncOrdersFromAster syncs Aster exchange order history to local database
// Also creates/updates position records to ensure orders/fills/positions data consistency
// exchangeID: Exchange account UUID (from exchanges.id)
// exchangeType: Exchange type ("aster")
func (t *AsterTrader) SyncOrdersFromAster(traderID string, exchangeID string, exchangeType string, st *store.Store) error {
if st == nil {
return fmt.Errorf("store is nil")
}
// Get recent trades (last 24 hours)
startTime := time.Now().Add(-24 * time.Hour)
logger.Infof("🔄 Syncing Aster trades from: %s", startTime.Format(time.RFC3339))
// Use GetTrades method to fetch trade records
trades, err := t.GetTrades(startTime, 500)
if err != nil {
return fmt.Errorf("failed to get trades: %w", err)
}
logger.Infof("📥 Received %d trades from Aster", len(trades))
// Sort trades by time ASC (oldest first) for proper position building
sort.Slice(trades, func(i, j int) bool {
return trades[i].Time.Before(trades[j].Time)
})
// Process trades one by one (no transaction to avoid deadlock)
orderStore := st.Order()
positionStore := st.Position()
posBuilder := store.NewPositionBuilder(positionStore)
syncedCount := 0
for _, trade := range trades {
// Check if trade already exists (use exchangeID which is UUID, not exchange type)
existing, err := orderStore.GetOrderByExchangeID(exchangeID, trade.TradeID)
if err == nil && existing != nil {
continue // Order already exists, skip
}
// Determine order action based on side, positionSide, and realizedPnL
// Aster uses one-way position mode (BOTH), so we need to infer from PnL
// - RealizedPnL != 0 means it's a close trade
// - RealizedPnL == 0 means it's an open trade
orderAction := deriveAsterOrderAction(trade.Side, trade.PositionSide, trade.RealizedPnL)
// Determine position side from order action
positionSide := "LONG"
if strings.Contains(orderAction, "short") {
positionSide = "SHORT"
}
// Normalize side for storage
side := strings.ToUpper(trade.Side)
// Create order record
orderRecord := &store.TraderOrder{
TraderID: traderID,
ExchangeID: exchangeID, // UUID
ExchangeType: exchangeType, // Exchange type
ExchangeOrderID: trade.TradeID,
Symbol: trade.Symbol,
Side: side,
PositionSide: "BOTH", // Aster uses one-way position mode
Type: "LIMIT",
OrderAction: orderAction,
Quantity: trade.Quantity,
Price: trade.Price,
Status: "FILLED",
FilledQuantity: trade.Quantity,
AvgFillPrice: trade.Price,
Commission: trade.Fee,
FilledAt: trade.Time,
CreatedAt: trade.Time,
UpdatedAt: trade.Time,
}
// Insert order record
if err := orderStore.CreateOrder(orderRecord); err != nil {
logger.Infof(" ⚠️ Failed to sync trade %s: %v", trade.TradeID, err)
continue
}
// Create fill record
fillRecord := &store.TraderFill{
TraderID: traderID,
ExchangeID: exchangeID, // UUID
ExchangeType: exchangeType, // Exchange type
OrderID: orderRecord.ID,
ExchangeOrderID: trade.TradeID,
ExchangeTradeID: trade.TradeID,
Symbol: trade.Symbol,
Side: side,
Price: trade.Price,
Quantity: trade.Quantity,
QuoteQuantity: trade.Price * trade.Quantity,
Commission: trade.Fee,
CommissionAsset: "USDT",
RealizedPnL: trade.RealizedPnL,
IsMaker: false,
CreatedAt: trade.Time,
}
if err := orderStore.CreateFill(fillRecord); err != nil {
logger.Infof(" ⚠️ Failed to sync fill for trade %s: %v", trade.TradeID, err)
}
// Create/update position record using PositionBuilder
if err := posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
trade.Symbol, positionSide, orderAction,
trade.Quantity, trade.Price, trade.Fee, trade.RealizedPnL,
trade.Time, trade.TradeID,
); err != nil {
logger.Infof(" ⚠️ Failed to sync position for trade %s: %v", trade.TradeID, err)
} else {
logger.Infof(" 📍 Position updated for trade: %s (action: %s, qty: %.6f)", trade.TradeID, orderAction, trade.Quantity)
}
syncedCount++
logger.Infof(" ✅ Synced trade: %s %s %s qty=%.6f price=%.6f pnl=%.2f fee=%.6f action=%s",
trade.TradeID, trade.Symbol, side, trade.Quantity, trade.Price, trade.RealizedPnL, trade.Fee, orderAction)
}
logger.Infof("✅ Aster order sync completed: %d new trades synced", syncedCount)
return nil
}
// deriveAsterOrderAction determines order action from trade details
// Aster uses one-way position mode (BOTH), so we infer from:
// - Side: BUY or SELL
// - RealizedPnL: non-zero means closing trade
func deriveAsterOrderAction(side, positionSide string, realizedPnL float64) string {
side = strings.ToUpper(side)
positionSide = strings.ToUpper(positionSide)
// Check if this is a closing trade (has realized PnL)
isClose := realizedPnL != 0
if positionSide == "LONG" {
if isClose {
return "close_long"
}
return "open_long"
} else if positionSide == "SHORT" {
if isClose {
return "close_short"
}
return "open_short"
} else {
// BOTH mode - infer from side and PnL
if side == "BUY" {
if isClose {
return "close_short" // Buying to close short
}
return "open_long" // Buying to open long
} else {
if isClose {
return "close_long" // Selling to close long
}
return "open_short" // Selling to open short
}
}
}
// StartOrderSync starts background order sync task for Aster
func (t *AsterTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
ticker := time.NewTicker(interval)
go func() {
for range ticker.C {
if err := t.SyncOrdersFromAster(traderID, exchangeID, exchangeType, st); err != nil {
logger.Infof("⚠️ Aster order sync failed: %v", err)
}
}
}()
logger.Infof("🔄 Aster order sync started (interval: %v)", interval)
}

View File

@@ -363,8 +363,48 @@ func (at *AutoTrader) Run() error {
// Start Lighter order sync if using Lighter exchange
if at.exchange == "lighter" {
if lighterTrader, ok := at.trader.(*LighterTraderV2); ok && at.store != nil {
lighterTrader.StartOrderSync(at.id, at.store.Order(), 30*time.Second)
logger.Infof("🔄 [%s] Lighter order sync enabled (every 30s)", at.name)
lighterTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
logger.Infof("🔄 [%s] Lighter order+position sync enabled (every 30s)", at.name)
}
}
// Start Hyperliquid order sync if using Hyperliquid exchange
if at.exchange == "hyperliquid" {
if hyperliquidTrader, ok := at.trader.(*HyperliquidTrader); ok && at.store != nil {
hyperliquidTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
logger.Infof("🔄 [%s] Hyperliquid order+position sync enabled (every 30s)", at.name)
}
}
// Start Bybit order sync if using Bybit exchange
if at.exchange == "bybit" {
if bybitTrader, ok := at.trader.(*BybitTrader); ok && at.store != nil {
bybitTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
logger.Infof("🔄 [%s] Bybit order+position sync enabled (every 30s)", at.name)
}
}
// Start OKX order sync if using OKX exchange
if at.exchange == "okx" {
if okxTrader, ok := at.trader.(*OKXTrader); ok && at.store != nil {
okxTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
logger.Infof("🔄 [%s] OKX order+position sync enabled (every 30s)", at.name)
}
}
// Start Bitget order sync if using Bitget exchange
if at.exchange == "bitget" {
if bitgetTrader, ok := at.trader.(*BitgetTrader); ok && at.store != nil {
bitgetTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
logger.Infof("🔄 [%s] Bitget order+position sync enabled (every 30s)", at.name)
}
}
// Start Aster order sync if using Aster exchange
if at.exchange == "aster" {
if asterTrader, ok := at.trader.(*AsterTrader); ok && at.store != nil {
asterTrader.StartOrderSync(at.id, at.exchangeID, at.exchange, at.store, 30*time.Second)
logger.Infof("🔄 [%s] Aster order+position sync enabled (every 30s)", at.name)
}
}
@@ -1888,29 +1928,11 @@ func (at *AutoTrader) recordPositionChange(orderID, symbol, side, action string,
return
}
// Calculate P&L
var realizedPnL float64
if side == "LONG" {
realizedPnL = (price - openPos.EntryPrice) * openPos.Quantity
} else {
realizedPnL = (openPos.EntryPrice - price) * openPos.Quantity
}
// Update position record
err = at.store.Position().ClosePosition(
openPos.ID,
price, // exitPrice
orderID, // exitOrderID
realizedPnL,
fee, // fee from exchange API
"ai_decision",
)
if err != nil {
logger.Infof(" ⚠️ Failed to update position: %v", err)
} else {
logger.Infof(" 📊 Position closed [%s] %s %s @ %.4f → %.4f, P&L: %.2f, Fee: %.4f",
at.id[:8], symbol, side, openPos.EntryPrice, price, realizedPnL, fee)
}
// NOTE: Position update removed - Order Sync will handle it automatically
// Order Sync will pick up the fill and update the position through PositionBuilder
// This ensures accurate fee accumulation and PnL calculation
logger.Infof(" ✅ Order placed [%s] %s %s @ %.4f, will be synced by Order Sync",
at.id[:8], symbol, side, price)
}
}

257
trader/bitget_order_sync.go Normal file
View File

@@ -0,0 +1,257 @@
package trader
import (
"encoding/json"
"fmt"
"nofx/logger"
"nofx/store"
"sort"
"strconv"
"strings"
"time"
)
// BitgetTrade represents a trade record from Bitget fill history
type BitgetTrade struct {
Symbol string
TradeID string
OrderID string
Side string // buy or sell
FillPrice float64
FillQty float64
Fee float64
FeeAsset string
ExecTime time.Time
ProfitLoss float64
OrderType string
OrderAction string // open_long, open_short, close_long, close_short
}
// GetTrades retrieves trade/fill records from Bitget
func (t *BitgetTrader) GetTrades(startTime time.Time, limit int) ([]BitgetTrade, error) {
if limit <= 0 {
limit = 100
}
if limit > 100 {
limit = 100 // Bitget max limit is 100
}
params := map[string]interface{}{
"productType": "USDT-FUTURES",
"startTime": fmt.Sprintf("%d", startTime.UnixMilli()),
"limit": fmt.Sprintf("%d", limit),
}
data, err := t.doRequest("GET", "/api/v2/mix/order/fill-history", params)
if err != nil {
return nil, fmt.Errorf("failed to get fill history: %w", err)
}
var resp struct {
FillList []struct {
TradeID string `json:"tradeId"`
Symbol string `json:"symbol"`
OrderID string `json:"orderId"`
Side string `json:"side"` // buy, sell
Price string `json:"price"` // Fill price
BaseVolume string `json:"baseVolume"` // Fill size in base currency
Fee string `json:"fee"` // Fee (negative for cost)
FeeCcy string `json:"feeCcy"` // Fee currency
Profit string `json:"profit"` // Realized PnL
CTime string `json:"cTime"` // Fill time (ms)
TradeSide string `json:"tradeSide"` // open, close
} `json:"fillList"`
}
if err := json.Unmarshal(data, &resp); err != nil {
return nil, fmt.Errorf("failed to parse fills: %w", err)
}
trades := make([]BitgetTrade, 0, len(resp.FillList))
for _, fill := range resp.FillList {
fillPrice, _ := strconv.ParseFloat(fill.Price, 64)
fillQty, _ := strconv.ParseFloat(fill.BaseVolume, 64)
fee, _ := strconv.ParseFloat(fill.Fee, 64)
profit, _ := strconv.ParseFloat(fill.Profit, 64)
cTime, _ := strconv.ParseInt(fill.CTime, 10, 64)
// Determine order action based on side and tradeSide
// Bitget one-way mode:
// - buy + open = open long
// - sell + open = open short
// - sell + close = close long
// - buy + close = close short
orderAction := "open_long"
side := strings.ToLower(fill.Side)
tradeSide := strings.ToLower(fill.TradeSide)
if tradeSide == "open" {
if side == "buy" {
orderAction = "open_long"
} else {
orderAction = "open_short"
}
} else if tradeSide == "close" {
if side == "sell" {
orderAction = "close_long"
} else {
orderAction = "close_short"
}
}
trade := BitgetTrade{
Symbol: fill.Symbol,
TradeID: fill.TradeID,
OrderID: fill.OrderID,
Side: fill.Side,
FillPrice: fillPrice,
FillQty: fillQty,
Fee: -fee, // Bitget returns negative fee
FeeAsset: fill.FeeCcy,
ExecTime: time.UnixMilli(cTime),
ProfitLoss: profit,
OrderType: "MARKET",
OrderAction: orderAction,
}
trades = append(trades, trade)
}
return trades, nil
}
// SyncOrdersFromBitget syncs Bitget exchange order history to local database
// Also creates/updates position records to ensure orders/fills/positions data consistency
// exchangeID: Exchange account UUID (from exchanges.id)
// exchangeType: Exchange type ("bitget")
func (t *BitgetTrader) SyncOrdersFromBitget(traderID string, exchangeID string, exchangeType string, st *store.Store) error {
if st == nil {
return fmt.Errorf("store is nil")
}
// Get recent trades (last 24 hours)
startTime := time.Now().Add(-24 * time.Hour)
logger.Infof("🔄 Syncing Bitget trades from: %s", startTime.Format(time.RFC3339))
// Use GetTrades method to fetch trade records
trades, err := t.GetTrades(startTime, 100)
if err != nil {
return fmt.Errorf("failed to get trades: %w", err)
}
logger.Infof("📥 Received %d trades from Bitget", len(trades))
// Sort trades by time ASC (oldest first) for proper position building
sort.Slice(trades, func(i, j int) bool {
return trades[i].ExecTime.Before(trades[j].ExecTime)
})
// Process trades one by one (no transaction to avoid deadlock)
orderStore := st.Order()
positionStore := st.Position()
posBuilder := store.NewPositionBuilder(positionStore)
syncedCount := 0
for _, trade := range trades {
// Check if trade already exists (use exchangeID which is UUID, not exchange type)
existing, err := orderStore.GetOrderByExchangeID(exchangeID, trade.TradeID)
if err == nil && existing != nil {
continue // Order already exists, skip
}
// Determine position side from order action
positionSide := "LONG"
if strings.Contains(trade.OrderAction, "short") {
positionSide = "SHORT"
}
// Normalize side for storage
side := strings.ToUpper(trade.Side)
// Create order record
orderRecord := &store.TraderOrder{
TraderID: traderID,
ExchangeID: exchangeID, // UUID
ExchangeType: exchangeType, // Exchange type
ExchangeOrderID: trade.TradeID,
Symbol: trade.Symbol,
Side: side,
PositionSide: "BOTH", // Bitget uses one-way position mode
Type: trade.OrderType,
OrderAction: trade.OrderAction,
Quantity: trade.FillQty,
Price: trade.FillPrice,
Status: "FILLED",
FilledQuantity: trade.FillQty,
AvgFillPrice: trade.FillPrice,
Commission: trade.Fee,
FilledAt: trade.ExecTime,
CreatedAt: trade.ExecTime,
UpdatedAt: trade.ExecTime,
}
// Insert order record
if err := orderStore.CreateOrder(orderRecord); err != nil {
logger.Infof(" ⚠️ Failed to sync trade %s: %v", trade.TradeID, err)
continue
}
// Create fill record
fillRecord := &store.TraderFill{
TraderID: traderID,
ExchangeID: exchangeID, // UUID
ExchangeType: exchangeType, // Exchange type
OrderID: orderRecord.ID,
ExchangeOrderID: trade.OrderID,
ExchangeTradeID: trade.TradeID,
Symbol: trade.Symbol,
Side: side,
Price: trade.FillPrice,
Quantity: trade.FillQty,
QuoteQuantity: trade.FillPrice * trade.FillQty,
Commission: trade.Fee,
CommissionAsset: trade.FeeAsset,
RealizedPnL: trade.ProfitLoss,
IsMaker: false,
CreatedAt: trade.ExecTime,
}
if err := orderStore.CreateFill(fillRecord); err != nil {
logger.Infof(" ⚠️ Failed to sync fill for trade %s: %v", trade.TradeID, err)
}
// Create/update position record using PositionBuilder
if err := posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
trade.Symbol, positionSide, trade.OrderAction,
trade.FillQty, trade.FillPrice, trade.Fee, trade.ProfitLoss,
trade.ExecTime, trade.TradeID,
); err != nil {
logger.Infof(" ⚠️ Failed to sync position for trade %s: %v", trade.TradeID, err)
} else {
logger.Infof(" 📍 Position updated for trade: %s (action: %s, qty: %.6f)", trade.TradeID, trade.OrderAction, trade.FillQty)
}
syncedCount++
logger.Infof(" ✅ Synced trade: %s %s %s qty=%.6f price=%.6f pnl=%.2f fee=%.6f action=%s",
trade.TradeID, trade.Symbol, side, trade.FillQty, trade.FillPrice, trade.ProfitLoss, trade.Fee, trade.OrderAction)
}
logger.Infof("✅ Bitget order sync completed: %d new trades synced", syncedCount)
return nil
}
// StartOrderSync starts background order sync task for Bitget
func (t *BitgetTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
ticker := time.NewTicker(interval)
go func() {
for range ticker.C {
if err := t.SyncOrdersFromBitget(traderID, exchangeID, exchangeType, st); err != nil {
logger.Infof("⚠️ Bitget order sync failed: %v", err)
}
}
}()
logger.Infof("🔄 Bitget order sync started (interval: %v)", interval)
}

309
trader/bybit_order_sync.go Normal file
View File

@@ -0,0 +1,309 @@
package trader
import (
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"net/http"
"nofx/logger"
"nofx/store"
"sort"
"strconv"
"strings"
"time"
)
// BybitTrade represents a trade record from Bybit execution list
type BybitTrade struct {
Symbol string
OrderID string
ExecID string
Side string // Buy or Sell
ExecPrice float64
ExecQty float64
ExecFee float64
ExecTime time.Time
IsMaker bool
OrderType string
ClosedSize float64 // For close orders
ClosedPnL float64
OrderAction string // open_long, open_short, close_long, close_short
}
// GetTrades retrieves trade/execution records from Bybit
func (t *BybitTrader) GetTrades(startTime time.Time, limit int) ([]BybitTrade, error) {
return t.getTradesViaHTTP(startTime, limit)
}
// getTradesViaHTTP makes direct HTTP call to Bybit API for execution list
func (t *BybitTrader) getTradesViaHTTP(startTime time.Time, limit int) ([]BybitTrade, error) {
// Build query string
queryParams := fmt.Sprintf("category=linear&startTime=%d&limit=%d", startTime.UnixMilli(), limit)
url := "https://api.bybit.com/v5/execution/list?" + queryParams
// Generate timestamp
timestamp := fmt.Sprintf("%d", time.Now().UnixMilli())
recvWindow := "5000"
// Build signature payload: timestamp + api_key + recv_window + queryString
signPayload := timestamp + t.apiKey + recvWindow + queryParams
// Generate HMAC-SHA256 signature
h := hmac.New(sha256.New, []byte(t.secretKey))
h.Write([]byte(signPayload))
signature := hex.EncodeToString(h.Sum(nil))
// Create request
req, err := http.NewRequest("GET", url, nil)
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
// Add Bybit V5 API headers
req.Header.Set("X-BAPI-API-KEY", t.apiKey)
req.Header.Set("X-BAPI-SIGN", signature)
req.Header.Set("X-BAPI-SIGN-TYPE", "2")
req.Header.Set("X-BAPI-TIMESTAMP", timestamp)
req.Header.Set("X-BAPI-RECV-WINDOW", recvWindow)
req.Header.Set("Content-Type", "application/json")
// Use http.DefaultClient for the request
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to call Bybit API: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response: %w", err)
}
var result struct {
RetCode int `json:"retCode"`
RetMsg string `json:"retMsg"`
Result struct {
List []map[string]interface{} `json:"list"`
} `json:"result"`
}
if err := json.Unmarshal(body, &result); err != nil {
return nil, fmt.Errorf("failed to parse response: %w", err)
}
if result.RetCode != 0 {
return nil, fmt.Errorf("Bybit API error: %s", result.RetMsg)
}
return t.parseTradesResult(result.Result.List)
}
// parseTradesResult parses the execution list result from Bybit API
func (t *BybitTrader) parseTradesResult(list []map[string]interface{}) ([]BybitTrade, error) {
var trades []BybitTrade
for _, item := range list {
symbol, _ := item["symbol"].(string)
orderID, _ := item["orderId"].(string)
execID, _ := item["execId"].(string)
side, _ := item["side"].(string)
orderType, _ := item["orderType"].(string)
isMaker, _ := item["isMaker"].(bool)
execPriceStr, _ := item["execPrice"].(string)
execQtyStr, _ := item["execQty"].(string)
execFeeStr, _ := item["execFee"].(string)
closedSizeStr, _ := item["closedSize"].(string)
closedPnlStr, _ := item["closedPnl"].(string)
execTimeStr, _ := item["execTime"].(string)
execPrice, _ := strconv.ParseFloat(execPriceStr, 64)
execQty, _ := strconv.ParseFloat(execQtyStr, 64)
execFee, _ := strconv.ParseFloat(execFeeStr, 64)
closedSize, _ := strconv.ParseFloat(closedSizeStr, 64)
closedPnl, _ := strconv.ParseFloat(closedPnlStr, 64)
execTimeMs, _ := strconv.ParseInt(execTimeStr, 10, 64)
execTime := time.UnixMilli(execTimeMs)
// Determine order action based on side and closedSize
// If closedSize > 0, it's a close trade
// Side: Buy = long direction, Sell = short direction
orderAction := "open_long"
if closedSize > 0 {
// This is a close trade
if strings.ToLower(side) == "sell" {
orderAction = "close_long" // Selling to close a long
} else {
orderAction = "close_short" // Buying to close a short
}
} else {
// This is an open trade
if strings.ToLower(side) == "buy" {
orderAction = "open_long"
} else {
orderAction = "open_short"
}
}
trade := BybitTrade{
Symbol: symbol,
OrderID: orderID,
ExecID: execID,
Side: side,
ExecPrice: execPrice,
ExecQty: execQty,
ExecFee: execFee,
ExecTime: execTime,
IsMaker: isMaker,
OrderType: orderType,
ClosedSize: closedSize,
ClosedPnL: closedPnl,
OrderAction: orderAction,
}
trades = append(trades, trade)
}
return trades, nil
}
// SyncOrdersFromBybit syncs Bybit exchange order history to local database
// Also creates/updates position records to ensure orders/fills/positions data consistency
// exchangeID: Exchange account UUID (from exchanges.id)
// exchangeType: Exchange type ("bybit")
func (t *BybitTrader) SyncOrdersFromBybit(traderID string, exchangeID string, exchangeType string, st *store.Store) error {
if st == nil {
return fmt.Errorf("store is nil")
}
// Get recent trades (last 24 hours)
startTime := time.Now().Add(-24 * time.Hour)
logger.Infof("🔄 Syncing Bybit trades from: %s", startTime.Format(time.RFC3339))
// Use GetTrades method to fetch trade records
trades, err := t.GetTrades(startTime, 1000)
if err != nil {
return fmt.Errorf("failed to get trades: %w", err)
}
logger.Infof("📥 Received %d trades from Bybit", len(trades))
// Sort trades by time ASC (oldest first) for proper position building
sort.Slice(trades, func(i, j int) bool {
return trades[i].ExecTime.Before(trades[j].ExecTime)
})
// Process trades one by one (no transaction to avoid deadlock)
orderStore := st.Order()
positionStore := st.Position()
posBuilder := store.NewPositionBuilder(positionStore)
syncedCount := 0
for _, trade := range trades {
// Check if trade already exists (use exchangeID which is UUID, not exchange type)
existing, err := orderStore.GetOrderByExchangeID(exchangeID, trade.ExecID)
if err == nil && existing != nil {
continue // Order already exists, skip
}
// Normalize symbol (should already have USDT suffix from Bybit)
symbol := trade.Symbol
// Determine position side from order action
positionSide := "LONG"
if strings.Contains(trade.OrderAction, "short") {
positionSide = "SHORT"
}
// Normalize side for storage
side := strings.ToUpper(trade.Side)
// Create order record
orderRecord := &store.TraderOrder{
TraderID: traderID,
ExchangeID: exchangeID, // UUID
ExchangeType: exchangeType, // Exchange type
ExchangeOrderID: trade.ExecID, // Use ExecID as unique identifier
Symbol: symbol,
Side: side,
PositionSide: "BOTH", // Bybit uses one-way position mode
Type: trade.OrderType,
OrderAction: trade.OrderAction,
Quantity: trade.ExecQty,
Price: trade.ExecPrice,
Status: "FILLED",
FilledQuantity: trade.ExecQty,
AvgFillPrice: trade.ExecPrice,
Commission: trade.ExecFee,
FilledAt: trade.ExecTime,
CreatedAt: trade.ExecTime,
UpdatedAt: trade.ExecTime,
}
// Insert order record
if err := orderStore.CreateOrder(orderRecord); err != nil {
logger.Infof(" ⚠️ Failed to sync trade %s: %v", trade.ExecID, err)
continue
}
// Create fill record
fillRecord := &store.TraderFill{
TraderID: traderID,
ExchangeID: exchangeID, // UUID
ExchangeType: exchangeType, // Exchange type
OrderID: orderRecord.ID,
ExchangeOrderID: trade.OrderID,
ExchangeTradeID: trade.ExecID,
Symbol: symbol,
Side: side,
Price: trade.ExecPrice,
Quantity: trade.ExecQty,
QuoteQuantity: trade.ExecPrice * trade.ExecQty,
Commission: trade.ExecFee,
CommissionAsset: "USDT",
RealizedPnL: trade.ClosedPnL,
IsMaker: trade.IsMaker,
CreatedAt: trade.ExecTime,
}
if err := orderStore.CreateFill(fillRecord); err != nil {
logger.Infof(" ⚠️ Failed to sync fill for trade %s: %v", trade.ExecID, err)
}
// Create/update position record using PositionBuilder
if err := posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
symbol, positionSide, trade.OrderAction,
trade.ExecQty, trade.ExecPrice, trade.ExecFee, trade.ClosedPnL,
trade.ExecTime, trade.ExecID,
); err != nil {
logger.Infof(" ⚠️ Failed to sync position for trade %s: %v", trade.ExecID, err)
} else {
logger.Infof(" 📍 Position updated for trade: %s (action: %s, qty: %.6f)", trade.ExecID, trade.OrderAction, trade.ExecQty)
}
syncedCount++
logger.Infof(" ✅ Synced trade: %s %s %s qty=%.6f price=%.6f pnl=%.2f fee=%.6f action=%s",
trade.ExecID, symbol, side, trade.ExecQty, trade.ExecPrice, trade.ClosedPnL, trade.ExecFee, trade.OrderAction)
}
logger.Infof("✅ Bybit order sync completed: %d new trades synced", syncedCount)
return nil
}
// StartOrderSync starts background order sync task for Bybit
func (t *BybitTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
ticker := time.NewTicker(interval)
go func() {
for range ticker.C {
if err := t.SyncOrdersFromBybit(traderID, exchangeID, exchangeType, st); err != nil {
logger.Infof("⚠️ Bybit order sync failed: %v", err)
}
}
}()
logger.Infof("🔄 Bybit order sync started (interval: %v)", interval)
}

View File

@@ -0,0 +1,337 @@
package trader
import (
"database/sql"
"nofx/store"
"testing"
"time"
_ "github.com/mattn/go-sqlite3"
)
// TestScenario represents a trading scenario to test
type TestScenario struct {
Name string
Trades []TestTrade
ExpectedPos []ExpectedPosition
}
// TestTrade represents a single trade in a test scenario
type TestTrade struct {
Action string // open_long, close_short, etc.
Side string // LONG or SHORT
Symbol string
Quantity float64
Price float64
Fee float64
RealizedPnL float64
}
// ExpectedPosition represents expected position state
type ExpectedPosition struct {
Symbol string
Side string
Quantity float64
Status string // OPEN or CLOSED
}
// Standard test scenarios that all exchanges should pass
func getStandardTestScenarios() []TestScenario {
return []TestScenario{
{
Name: "Simple Open and Close Long",
Trades: []TestTrade{
{Action: "open_long", Side: "LONG", Symbol: "ETHUSDT", Quantity: 0.1, Price: 3500, Fee: 0.5, RealizedPnL: 0},
{Action: "close_long", Side: "LONG", Symbol: "ETHUSDT", Quantity: 0.1, Price: 3600, Fee: 0.5, RealizedPnL: 10},
},
ExpectedPos: []ExpectedPosition{}, // Should be fully closed
},
{
Name: "Simple Open and Close Short",
Trades: []TestTrade{
{Action: "open_short", Side: "SHORT", Symbol: "ETHUSDT", Quantity: 0.1, Price: 3500, Fee: 0.5, RealizedPnL: 0},
{Action: "close_short", Side: "SHORT", Symbol: "ETHUSDT", Quantity: 0.1, Price: 3400, Fee: 0.5, RealizedPnL: 10},
},
ExpectedPos: []ExpectedPosition{},
},
{
Name: "Position Averaging",
Trades: []TestTrade{
{Action: "open_long", Side: "LONG", Symbol: "BTCUSDT", Quantity: 0.01, Price: 50000, Fee: 1.0, RealizedPnL: 0},
{Action: "open_long", Side: "LONG", Symbol: "BTCUSDT", Quantity: 0.01, Price: 51000, Fee: 1.0, RealizedPnL: 0},
{Action: "close_long", Side: "LONG", Symbol: "BTCUSDT", Quantity: 0.02, Price: 52000, Fee: 2.0, RealizedPnL: 30},
},
ExpectedPos: []ExpectedPosition{},
},
{
Name: "Partial Close",
Trades: []TestTrade{
{Action: "open_long", Side: "LONG", Symbol: "SOLUSDT", Quantity: 10, Price: 100, Fee: 2.0, RealizedPnL: 0},
{Action: "close_long", Side: "LONG", Symbol: "SOLUSDT", Quantity: 3, Price: 105, Fee: 0.6, RealizedPnL: 15},
},
ExpectedPos: []ExpectedPosition{
{Symbol: "SOLUSDT", Side: "LONG", Quantity: 7, Status: "OPEN"},
},
},
{
Name: "Multiple Symbols",
Trades: []TestTrade{
{Action: "open_long", Side: "LONG", Symbol: "ETHUSDT", Quantity: 0.1, Price: 3500, Fee: 0.5, RealizedPnL: 0},
{Action: "open_short", Side: "SHORT", Symbol: "BTCUSDT", Quantity: 0.01, Price: 50000, Fee: 1.0, RealizedPnL: 0},
{Action: "close_long", Side: "LONG", Symbol: "ETHUSDT", Quantity: 0.1, Price: 3600, Fee: 0.5, RealizedPnL: 10},
},
ExpectedPos: []ExpectedPosition{
{Symbol: "BTCUSDT", Side: "SHORT", Quantity: 0.01, Status: "OPEN"},
},
},
{
Name: "Bug Scenario - Short then BUY to Close",
Trades: []TestTrade{
// This tests the exact bug we fixed
{Action: "open_short", Side: "SHORT", Symbol: "ETHUSDT", Quantity: 0.0472, Price: 3500, Fee: 0.2, RealizedPnL: 0},
{Action: "close_short", Side: "SHORT", Symbol: "ETHUSDT", Quantity: 0.0472, Price: 3400, Fee: 0.2, RealizedPnL: 4.72},
},
ExpectedPos: []ExpectedPosition{}, // Must be fully closed!
},
{
Name: "Multiple Opens and Closes",
Trades: []TestTrade{
{Action: "open_long", Side: "LONG", Symbol: "ETHUSDT", Quantity: 0.1, Price: 3500, Fee: 0.5, RealizedPnL: 0},
{Action: "close_long", Side: "LONG", Symbol: "ETHUSDT", Quantity: 0.1, Price: 3600, Fee: 0.5, RealizedPnL: 10},
{Action: "open_short", Side: "SHORT", Symbol: "ETHUSDT", Quantity: 0.05, Price: 3600, Fee: 0.3, RealizedPnL: 0},
{Action: "close_short", Side: "SHORT", Symbol: "ETHUSDT", Quantity: 0.05, Price: 3500, Fee: 0.3, RealizedPnL: 5},
{Action: "open_long", Side: "LONG", Symbol: "ETHUSDT", Quantity: 0.2, Price: 3550, Fee: 1.0, RealizedPnL: 0},
},
ExpectedPos: []ExpectedPosition{
{Symbol: "ETHUSDT", Side: "LONG", Quantity: 0.2, Status: "OPEN"},
},
},
}
}
// runStandardTests runs all standard test scenarios
func runStandardTests(t *testing.T, exchangeName string) {
scenarios := getStandardTestScenarios()
for _, scenario := range scenarios {
t.Run(scenario.Name, func(t *testing.T) {
// Setup database
db, err := sql.Open("sqlite3", ":memory:")
if err != nil {
t.Fatalf("Failed to create test database: %v", err)
}
defer db.Close()
positionStore := store.NewPositionStore(db)
if err := positionStore.InitTables(); err != nil {
t.Fatalf("Failed to initialize position tables: %v", err)
}
posBuilder := store.NewPositionBuilder(positionStore)
traderID := "test-trader"
exchangeID := "test-exchange-" + exchangeName
exchangeType := exchangeName
// Process all trades
for i, trade := range scenario.Trades {
err := posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
trade.Symbol, trade.Side, trade.Action,
trade.Quantity, trade.Price, trade.Fee, trade.RealizedPnL,
time.Now().Add(time.Duration(i)*time.Second),
"",
)
if err != nil {
t.Fatalf("Failed to process trade %d (%s): %v", i, trade.Action, err)
}
}
// Verify expected positions
positions, err := positionStore.GetOpenPositions(traderID)
if err != nil {
t.Fatalf("Failed to get positions: %v", err)
}
if len(positions) != len(scenario.ExpectedPos) {
t.Errorf("Expected %d open positions, got %d", len(scenario.ExpectedPos), len(positions))
for _, p := range positions {
t.Errorf(" Got: %s %s qty=%.4f status=%s", p.Symbol, p.Side, p.Quantity, p.Status)
}
return
}
// Verify each expected position
for _, expected := range scenario.ExpectedPos {
found := false
for _, actual := range positions {
if actual.Symbol == expected.Symbol && actual.Side == expected.Side {
found = true
if actual.Quantity != expected.Quantity {
t.Errorf("Position %s %s: expected qty %.4f, got %.4f",
expected.Symbol, expected.Side, expected.Quantity, actual.Quantity)
}
if actual.Status != expected.Status {
t.Errorf("Position %s %s: expected status %s, got %s",
expected.Symbol, expected.Side, expected.Status, actual.Status)
}
break
}
}
if !found {
t.Errorf("Expected position not found: %s %s", expected.Symbol, expected.Side)
}
}
})
}
}
// TestAllExchangesStandardScenarios runs standard scenarios for all exchanges
func TestAllExchangesStandardScenarios(t *testing.T) {
exchanges := []string{"hyperliquid", "binance", "bybit", "okx", "bitget", "aster", "lighter"}
for _, exchange := range exchanges {
t.Run(exchange, func(t *testing.T) {
runStandardTests(t, exchange)
})
}
}
// TestPositionAccumulationBug tests that positions don't accumulate incorrectly
func TestPositionAccumulationBug(t *testing.T) {
db, err := sql.Open("sqlite3", ":memory:")
if err != nil {
t.Fatalf("Failed to create test database: %v", err)
}
defer db.Close()
positionStore := store.NewPositionStore(db)
if err := positionStore.InitTables(); err != nil {
t.Fatalf("Failed to initialize position tables: %v", err)
}
posBuilder := store.NewPositionBuilder(positionStore)
traderID := "test-trader"
exchangeID := "test-exchange"
exchangeType := "hyperliquid"
// Simulate many trades that should cancel out
// This tests that we don't accumulate positions incorrectly
for i := 0; i < 10; i++ {
// Open Long
err := posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
"ETHUSDT", "LONG", "open_long",
0.1, 3500+float64(i*10), 0.5, 0,
time.Now().Add(time.Duration(i*2)*time.Second),
"",
)
if err != nil {
t.Fatalf("Failed to open long %d: %v", i, err)
}
// Close Long
err = posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
"ETHUSDT", "LONG", "close_long",
0.1, 3600+float64(i*10), 0.5, 10,
time.Now().Add(time.Duration(i*2+1)*time.Second),
"",
)
if err != nil {
t.Fatalf("Failed to close long %d: %v", i, err)
}
}
// Should have 0 open positions
positions, err := positionStore.GetOpenPositions(traderID)
if err != nil {
t.Fatalf("Failed to get positions: %v", err)
}
if len(positions) != 0 {
t.Errorf("Expected 0 positions after 10 open/close cycles, got %d", len(positions))
for _, p := range positions {
t.Errorf(" Unexpected: %s %s qty=%.4f", p.Symbol, p.Side, p.Quantity)
}
}
// Should have 10 closed positions with positive PnL
allPositions, err := positionStore.GetClosedPositions(traderID, 100)
if err != nil {
t.Fatalf("Failed to get closed positions: %v", err)
}
closedCount := 0
totalPnL := 0.0
for _, p := range allPositions {
if p.Status == "CLOSED" {
closedCount++
totalPnL += p.RealizedPnL
}
}
if closedCount != 10 {
t.Errorf("Expected 10 closed positions, got %d", closedCount)
}
if totalPnL <= 0 {
t.Errorf("Expected positive total PnL, got %.2f", totalPnL)
}
}
// TestQuantityPrecision tests handling of quantity precision issues
func TestQuantityPrecision(t *testing.T) {
db, err := sql.Open("sqlite3", ":memory:")
if err != nil {
t.Fatalf("Failed to create test database: %v", err)
}
defer db.Close()
positionStore := store.NewPositionStore(db)
if err := positionStore.InitTables(); err != nil {
t.Fatalf("Failed to initialize position tables: %v", err)
}
posBuilder := store.NewPositionBuilder(positionStore)
traderID := "test-trader"
exchangeID := "test-exchange"
exchangeType := "test"
// Open position
err = posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
"BTCUSDT", "LONG", "open_long",
0.01, 50000, 1.0, 0,
time.Now(),
"",
)
if err != nil {
t.Fatalf("Failed to open: %v", err)
}
// Close with slightly different quantity due to precision (0.00999999 vs 0.01)
// Should still close fully within tolerance
err = posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
"BTCUSDT", "LONG", "close_long",
0.00999999, 51000, 1.0, 10,
time.Now().Add(time.Second),
"",
)
if err != nil {
t.Fatalf("Failed to close: %v", err)
}
// Should have 0 open positions (within tolerance)
positions, err := positionStore.GetOpenPositions(traderID)
if err != nil {
t.Fatalf("Failed to get positions: %v", err)
}
if len(positions) != 0 {
t.Errorf("Expected 0 positions (precision tolerance), got %d", len(positions))
}
}

View File

@@ -0,0 +1,150 @@
package trader
import (
"fmt"
"nofx/logger"
"nofx/store"
"sort"
"strings"
"time"
)
// SyncOrdersFromHyperliquid syncs Hyperliquid exchange order history to local database
// Also creates/updates position records to ensure orders/fills/positions data consistency
// exchangeID: Exchange account UUID (from exchanges.id)
// exchangeType: Exchange type ("hyperliquid")
func (t *HyperliquidTrader) SyncOrdersFromHyperliquid(traderID string, exchangeID string, exchangeType string, st *store.Store) error {
if st == nil {
return fmt.Errorf("store is nil")
}
// Get recent trades (last 24 hours)
startTime := time.Now().Add(-24 * time.Hour)
logger.Infof("🔄 Syncing Hyperliquid trades from: %s", startTime.Format(time.RFC3339))
// Use GetTrades method to fetch trade records
trades, err := t.GetTrades(startTime, 1000)
if err != nil {
return fmt.Errorf("failed to get trades: %w", err)
}
logger.Infof("📥 Received %d trades from Hyperliquid", len(trades))
// Sort trades by time ASC (oldest first) for proper position building
sort.Slice(trades, func(i, j int) bool {
return trades[i].Time.Before(trades[j].Time)
})
// Process trades one by one (no transaction to avoid deadlock)
orderStore := st.Order()
positionStore := st.Position()
posBuilder := store.NewPositionBuilder(positionStore)
syncedCount := 0
for _, trade := range trades {
// Check if trade already exists (use exchangeID which is UUID, not exchange type)
existing, err := orderStore.GetOrderByExchangeID(exchangeID, trade.TradeID)
if err == nil && existing != nil {
continue // Order already exists, skip
}
// Normalize symbol (add USDT suffix)
symbol := trade.Symbol
if symbol != "" && !strings.Contains(symbol, "USDT") && !strings.Contains(symbol, "USD") {
symbol = symbol + "USDT"
}
// Use order action from trade (parsed from Hyperliquid Dir field)
// Dir field values: "Open Long", "Open Short", "Close Long", "Close Short"
orderAction := trade.OrderAction
positionSide := "LONG"
if strings.Contains(orderAction, "short") {
positionSide = "SHORT"
}
// Create order record
orderRecord := &store.TraderOrder{
TraderID: traderID,
ExchangeID: exchangeID, // UUID
ExchangeType: exchangeType, // Exchange type
ExchangeOrderID: trade.TradeID,
Symbol: symbol,
Side: trade.Side,
PositionSide: "BOTH", // Hyperliquid uses one-way position mode
Type: "MARKET",
OrderAction: orderAction,
Quantity: trade.Quantity,
Price: trade.Price,
Status: "FILLED",
FilledQuantity: trade.Quantity,
AvgFillPrice: trade.Price,
Commission: trade.Fee,
FilledAt: trade.Time,
CreatedAt: trade.Time,
UpdatedAt: trade.Time,
}
// Insert order record
if err := orderStore.CreateOrder(orderRecord); err != nil {
logger.Infof(" ⚠️ Failed to sync trade %s: %v", trade.TradeID, err)
continue
}
// Create fill record
fillRecord := &store.TraderFill{
TraderID: traderID,
ExchangeID: exchangeID, // UUID
ExchangeType: exchangeType, // Exchange type
OrderID: orderRecord.ID,
ExchangeOrderID: trade.TradeID,
ExchangeTradeID: trade.TradeID,
Symbol: symbol,
Side: trade.Side,
Price: trade.Price,
Quantity: trade.Quantity,
QuoteQuantity: trade.Price * trade.Quantity,
Commission: trade.Fee,
CommissionAsset: "USDT",
RealizedPnL: trade.RealizedPnL,
IsMaker: false, // Hyperliquid GetTrades doesn't provide maker/taker info
CreatedAt: trade.Time,
}
if err := orderStore.CreateFill(fillRecord); err != nil {
logger.Infof(" ⚠️ Failed to sync fill for trade %s: %v", trade.TradeID, err)
}
// Create/update position record using PositionBuilder
if err := posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
symbol, positionSide, orderAction,
trade.Quantity, trade.Price, trade.Fee, trade.RealizedPnL,
trade.Time, trade.TradeID,
); err != nil {
logger.Infof(" ⚠️ Failed to sync position for trade %s: %v", trade.TradeID, err)
} else {
logger.Infof(" 📍 Position updated for trade: %s (action: %s, qty: %.6f)", trade.TradeID, orderAction, trade.Quantity)
}
syncedCount++
logger.Infof(" ✅ Synced trade: %s %s %s qty=%.6f price=%.6f pnl=%.2f fee=%.6f action=%s",
trade.TradeID, symbol, trade.Side, trade.Quantity, trade.Price, trade.RealizedPnL, trade.Fee, orderAction)
}
logger.Infof("✅ Order sync completed: %d new trades synced", syncedCount)
return nil
}
// StartOrderSync starts background order sync task
func (t *HyperliquidTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
ticker := time.NewTicker(interval)
go func() {
for range ticker.C {
if err := t.SyncOrdersFromHyperliquid(traderID, exchangeID, exchangeType, st); err != nil {
logger.Infof("⚠️ Hyperliquid order sync failed: %v", err)
}
}
}()
logger.Infof("🔄 Hyperliquid order sync started (interval: %v)", interval)
}

View File

@@ -0,0 +1,388 @@
package trader
import (
"database/sql"
"nofx/store"
"testing"
"time"
_ "github.com/mattn/go-sqlite3"
)
// TestHyperliquidOrderDirectionParsing tests Dir field parsing
func TestHyperliquidOrderDirectionParsing(t *testing.T) {
tests := []struct {
name string
dirField string
side string
expectedAction string
expectedPosSide string
}{
{
name: "Open Long",
dirField: "Open Long",
side: "BUY",
expectedAction: "open_long",
expectedPosSide: "LONG",
},
{
name: "Open Short",
dirField: "Open Short",
side: "SELL",
expectedAction: "open_short",
expectedPosSide: "SHORT",
},
{
name: "Close Long",
dirField: "Close Long",
side: "SELL",
expectedAction: "close_long",
expectedPosSide: "LONG",
},
{
name: "Close Short",
dirField: "Close Short",
side: "BUY",
expectedAction: "close_short",
expectedPosSide: "SHORT",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Mock fill data structure from Hyperliquid SDK
// We'll test the parsing logic directly
var orderAction string
switch tt.dirField {
case "Open Long":
orderAction = "open_long"
case "Open Short":
orderAction = "open_short"
case "Close Long":
orderAction = "close_long"
case "Close Short":
orderAction = "close_short"
}
if orderAction != tt.expectedAction {
t.Errorf("Expected action %s, got %s", tt.expectedAction, orderAction)
}
})
}
}
// TestHyperliquidPositionBuilding tests the complete flow of position building
func TestHyperliquidPositionBuilding(t *testing.T) {
// Setup in-memory database
db, err := sql.Open("sqlite3", ":memory:")
if err != nil {
t.Fatalf("Failed to create test database: %v", err)
}
defer db.Close()
// Initialize stores
positionStore := store.NewPositionStore(db)
if err := positionStore.InitTables(); err != nil {
t.Fatalf("Failed to initialize position tables: %v", err)
}
posBuilder := store.NewPositionBuilder(positionStore)
traderID := "test-trader"
exchangeID := "test-exchange"
exchangeType := "hyperliquid"
symbol := "ETHUSDT"
// Test Case 1: Open Long → Close Long (should result in 0 position)
t.Run("Open and Close Long", func(t *testing.T) {
// Open Long: BUY 0.1 ETH @ 3500
err := posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
symbol, "LONG", "open_long",
0.1, 3500, 0.5, 0,
time.Now(), "order-1",
)
if err != nil {
t.Fatalf("Failed to process open long: %v", err)
}
// Verify position created
positions, err := positionStore.GetOpenPositions(traderID)
if err != nil {
t.Fatalf("Failed to get positions: %v", err)
}
if len(positions) != 1 {
t.Fatalf("Expected 1 open position, got %d", len(positions))
}
if positions[0].Quantity != 0.1 {
t.Errorf("Expected quantity 0.1, got %f", positions[0].Quantity)
}
// Close Long: SELL 0.1 ETH @ 3600
err = posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
symbol, "LONG", "close_long",
0.1, 3600, 0.5, 10.0, // PnL = (3600-3500)*0.1 = 10
time.Now(), "order-2",
)
if err != nil {
t.Fatalf("Failed to process close long: %v", err)
}
// Verify position closed
positions, err = positionStore.GetOpenPositions(traderID)
if err != nil {
t.Fatalf("Failed to get positions: %v", err)
}
if len(positions) != 0 {
t.Errorf("Expected 0 open positions, got %d", len(positions))
}
})
// Clear positions for next test
db.Exec("DELETE FROM trader_positions")
// Test Case 2: Open Short → Close Short with BUY (the bug scenario!)
t.Run("Open Short then Close with BUY", func(t *testing.T) {
// Open Short: SELL 0.05 ETH @ 3500
err := posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
symbol, "SHORT", "open_short",
0.05, 3500, 0.25, 0,
time.Now(), "order-3",
)
if err != nil {
t.Fatalf("Failed to process open short: %v", err)
}
// Verify SHORT position created
positions, err := positionStore.GetOpenPositions(traderID)
if err != nil {
t.Fatalf("Failed to get positions: %v", err)
}
if len(positions) != 1 {
t.Fatalf("Expected 1 open position, got %d", len(positions))
}
if positions[0].Side != "SHORT" {
t.Errorf("Expected SHORT position, got %s", positions[0].Side)
}
// Close Short: BUY 0.05 ETH @ 3400
// ⚠️ This is the critical test - BUY should close SHORT, not open LONG!
err = posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
symbol, "SHORT", "close_short",
0.05, 3400, 0.25, 5.0, // PnL = (3500-3400)*0.05 = 5
time.Now(), "order-4",
)
if err != nil {
t.Fatalf("Failed to process close short: %v", err)
}
// Verify position CLOSED (not opened a new LONG!)
positions, err = positionStore.GetOpenPositions(traderID)
if err != nil {
t.Fatalf("Failed to get positions: %v", err)
}
if len(positions) != 0 {
t.Errorf("Expected 0 open positions after close, got %d", len(positions))
if len(positions) > 0 {
t.Errorf("Wrong position side: %s (should be closed!)", positions[0].Side)
}
}
})
// Clear positions
db.Exec("DELETE FROM trader_positions")
// Test Case 3: Position Averaging (Open → Add → Close)
t.Run("Position Averaging", func(t *testing.T) {
// Open Long: BUY 0.1 ETH @ 3500
err := posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
symbol, "LONG", "open_long",
0.1, 3500, 0.5, 0,
time.Now(), "order-5",
)
if err != nil {
t.Fatalf("Failed to process first open: %v", err)
}
// Add to Long: BUY 0.1 ETH @ 3600
err = posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
symbol, "LONG", "open_long",
0.1, 3600, 0.5, 0,
time.Now(), "order-6",
)
if err != nil {
t.Fatalf("Failed to process add position: %v", err)
}
// Verify averaged position
positions, err := positionStore.GetOpenPositions(traderID)
if err != nil {
t.Fatalf("Failed to get positions: %v", err)
}
if len(positions) != 1 {
t.Fatalf("Expected 1 position (averaged), got %d", len(positions))
}
if positions[0].Quantity != 0.2 {
t.Errorf("Expected quantity 0.2, got %f", positions[0].Quantity)
}
expectedAvgPrice := (3500*0.1 + 3600*0.1) / 0.2 // = 3550
if positions[0].EntryPrice != expectedAvgPrice {
t.Errorf("Expected avg price %f, got %f", expectedAvgPrice, positions[0].EntryPrice)
}
// Close all: SELL 0.2 ETH @ 3700
err = posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
symbol, "LONG", "close_long",
0.2, 3700, 1.0, 30.0,
time.Now(), "order-7",
)
if err != nil {
t.Fatalf("Failed to process close: %v", err)
}
// Verify fully closed
positions, err = positionStore.GetOpenPositions(traderID)
if err != nil {
t.Fatalf("Failed to get positions: %v", err)
}
if len(positions) != 0 {
t.Errorf("Expected 0 positions, got %d", len(positions))
}
})
// Clear positions
db.Exec("DELETE FROM trader_positions")
// Test Case 4: Partial Close
t.Run("Partial Close", func(t *testing.T) {
// Open Long: BUY 1.0 ETH @ 3500
err := posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
symbol, "LONG", "open_long",
1.0, 3500, 2.0, 0,
time.Now(), "order-8",
)
if err != nil {
t.Fatalf("Failed to process open: %v", err)
}
// Partial Close: SELL 0.3 ETH @ 3600
err = posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
symbol, "LONG", "close_long",
0.3, 3600, 0.6, 30.0,
time.Now(), "order-9",
)
if err != nil {
t.Fatalf("Failed to process partial close: %v", err)
}
// Verify remaining position
positions, err := positionStore.GetOpenPositions(traderID)
if err != nil {
t.Fatalf("Failed to get positions: %v", err)
}
if len(positions) != 1 {
t.Fatalf("Expected 1 position, got %d", len(positions))
}
if positions[0].Quantity != 0.7 {
t.Errorf("Expected remaining quantity 0.7, got %f", positions[0].Quantity)
}
if positions[0].Status != "OPEN" {
t.Errorf("Expected status OPEN, got %s", positions[0].Status)
}
})
}
// TestHyperliquidBugScenario tests the exact bug we fixed
func TestHyperliquidBugScenario(t *testing.T) {
// Setup database
db, err := sql.Open("sqlite3", ":memory:")
if err != nil {
t.Fatalf("Failed to create test database: %v", err)
}
defer db.Close()
positionStore := store.NewPositionStore(db)
if err := positionStore.InitTables(); err != nil {
t.Fatalf("Failed to initialize position tables: %v", err)
}
posBuilder := store.NewPositionBuilder(positionStore)
traderID := "test-trader"
exchangeID := "test-exchange"
exchangeType := "hyperliquid"
// Simulate the exact scenario from the bug report
// Account has 30 USDT, should not be able to hold 1.7 ETH
trades := []struct {
action string
side string
symbol string
qty float64
price float64
fee float64
pnl float64
}{
// Order 853: Open Short
{"open_short", "SHORT", "ETHUSDT", 0.0472, 3500, 0.2, 0},
// Order 854: Close Short (was incorrectly classified as open_long)
{"close_short", "SHORT", "ETHUSDT", 0.0472, 3400, 0.2, 4.72},
// Order 855: Open Long
{"open_long", "LONG", "ETHUSDT", 0.05, 3450, 0.2, 0},
// Order 856: Close Long
{"close_long", "LONG", "ETHUSDT", 0.05, 3550, 0.2, 5.0},
}
for i, trade := range trades {
err := posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
trade.symbol, trade.side, trade.action,
trade.qty, trade.price, trade.fee, trade.pnl,
time.Now().Add(time.Duration(i)*time.Second),
"",
)
if err != nil {
t.Fatalf("Failed to process trade %d: %v", i, err)
}
}
// Verify: Should have 0 open positions
positions, err := positionStore.GetOpenPositions(traderID)
if err != nil {
t.Fatalf("Failed to get positions: %v", err)
}
if len(positions) != 0 {
t.Errorf("Expected 0 open positions, got %d", len(positions))
for _, p := range positions {
t.Errorf(" Unexpected position: %s %s qty=%.4f", p.Symbol, p.Side, p.Quantity)
}
}
// Verify closed positions have correct PnL
allPositions, err := positionStore.GetClosedPositions(traderID, 100)
if err != nil {
t.Fatalf("Failed to get closed positions: %v", err)
}
totalPnL := 0.0
for _, p := range allPositions {
if p.Status == "CLOSED" {
totalPnL += p.RealizedPnL
}
}
expectedTotalPnL := 4.72 + 5.0 // Sum of both close trades
if totalPnL != expectedTotalPnL {
t.Errorf("Expected total PnL %.2f, got %.2f", expectedTotalPnL, totalPnL)
}
}

View File

@@ -1028,12 +1028,42 @@ func (t *HyperliquidTrader) GetTrades(startTime time.Time, limit int) ([]TradeRe
side = "SELL"
}
// Parse Dir field to get order action
// Hyperliquid Dir values: "Open Long", "Open Short", "Close Long", "Close Short"
var orderAction string
switch strings.ToLower(fill.Dir) {
case "open long":
orderAction = "open_long"
case "open short":
orderAction = "open_short"
case "close long":
orderAction = "close_long"
case "close short":
orderAction = "close_short"
default:
// Fallback: use RealizedPnL if Dir is missing/unknown
if pnl != 0 {
if side == "BUY" {
orderAction = "close_short"
} else {
orderAction = "close_long"
}
} else {
if side == "BUY" {
orderAction = "open_long"
} else {
orderAction = "open_short"
}
}
}
// Hyperliquid uses one-way mode, so PositionSide is "BOTH"
trade := TradeRecord{
TradeID: strconv.FormatInt(fill.Tid, 10),
Symbol: fill.Coin,
Side: side,
PositionSide: "BOTH", // Hyperliquid doesn't have hedge mode
OrderAction: orderAction,
Price: price,
Quantity: qty,
RealizedPnL: pnl,

View File

@@ -26,6 +26,7 @@ type TradeRecord struct {
Symbol string // Trading pair (e.g., "BTCUSDT")
Side string // "BUY" or "SELL"
PositionSide string // "LONG", "SHORT", or "BOTH" (for one-way mode)
OrderAction string // "open_long", "open_short", "close_long", "close_short" (from exchange Dir field)
Price float64 // Execution price
Quantity float64 // Executed quantity
RealizedPnL float64 // Realized PnL (non-zero for closing trades)

View File

@@ -7,11 +7,12 @@ import (
"nofx/logger"
"nofx/store"
"net/http"
"sort"
"strings"
"time"
)
// LighterOrderHistory 订单历史记录
// LighterOrderHistory order history record
type LighterOrderHistory struct {
OrderID string `json:"order_id"`
Symbol string `json:"symbol"`
@@ -26,16 +27,23 @@ type LighterOrderHistory struct {
FilledAt int64 `json:"filled_at"`
}
// SyncOrdersFromLighter 同步 Lighter 交易所的订单历史到本地数据库
func (t *LighterTraderV2) SyncOrdersFromLighter(traderID string, orderStore *store.OrderStore) error {
// 确保有 account index
// SyncOrdersFromLighter syncs Lighter exchange order history to local database
// Also creates/updates position records to ensure orders/fills/positions data consistency
// exchangeID: Exchange account UUID (from exchanges.id)
// exchangeType: Exchange type ("lighter")
func (t *LighterTraderV2) SyncOrdersFromLighter(traderID string, exchangeID string, exchangeType string, st *store.Store) error {
if st == nil {
return fmt.Errorf("store is nil")
}
// Ensure we have account index
if t.accountIndex == 0 {
if err := t.initializeAccount(); err != nil {
return fmt.Errorf("failed to get account index: %w", err)
}
}
// 获取最近的订单过去24小时
// Get recent orders (last 24 hours)
startTime := time.Now().Add(-24 * time.Hour).Unix()
endpoint := fmt.Sprintf("%s/api/v1/orders?account_index=%d&start_time=%d&limit=100",
t.baseURL, t.accountIndex, startTime)
@@ -47,7 +55,7 @@ func (t *LighterTraderV2) SyncOrdersFromLighter(traderID string, orderStore *sto
return fmt.Errorf("failed to create request: %w", err)
}
// 添加认证头
// Add authentication header
if err := t.ensureAuthToken(); err != nil {
return fmt.Errorf("failed to get auth token: %w", err)
}
@@ -72,7 +80,7 @@ func (t *LighterTraderV2) SyncOrdersFromLighter(traderID string, orderStore *sto
return fmt.Errorf("API returned status %d", resp.StatusCode)
}
// 解析响应
// Parse response
var apiResp struct {
Code int `json:"code"`
Orders []LighterOrderHistory `json:"orders"`
@@ -89,21 +97,33 @@ func (t *LighterTraderV2) SyncOrdersFromLighter(traderID string, orderStore *sto
logger.Infof("📥 Received %d orders from Lighter", len(apiResp.Orders))
// 同步每个订单
// Sort orders by filled_at ASC (oldest first) for proper position building
sort.Slice(apiResp.Orders, func(i, j int) bool {
return apiResp.Orders[i].FilledAt < apiResp.Orders[j].FilledAt
})
// Process orders one by one (no transaction to avoid deadlock)
orderStore := st.Order()
positionStore := st.Position()
posBuilder := store.NewPositionBuilder(positionStore)
// Get current open positions to help determine action for each order
openPositions, _ := positionStore.GetOpenPositions(traderID)
syncedCount := 0
for _, order := range apiResp.Orders {
// 只同步已成交的订单
// Only sync filled orders
if order.Status != "filled" {
continue
}
// 检查订单是否已存在
existing, err := orderStore.GetOrderByExchangeID("lighter", order.OrderID)
// Check if order already exists (use exchangeID which is UUID, not exchange type)
existing, err := orderStore.GetOrderByExchangeID(exchangeID, order.OrderID)
if err == nil && existing != nil {
continue // 订单已存在,跳过
continue // Order already exists, skip
}
// 解析价格和数量
// Parse price and quantity
price, _ := parseFloat(order.Price)
size, _ := parseFloat(order.Size)
filledSize, _ := parseFloat(order.FilledSize)
@@ -112,24 +132,55 @@ func (t *LighterTraderV2) SyncOrdersFromLighter(traderID string, orderStore *sto
filledSize = size
}
// 确定订单方向和动作
// Determine order action based on existing positions
// Lighter can have both LONG and SHORT positions simultaneously
var positionSide, orderAction, side string
symbol := order.Symbol
if order.Side == "buy" {
side = "BUY"
// 买入可能是开多或平空,这里假设是开多
positionSide = "LONG"
orderAction = "open_long"
// Check if we have an open SHORT position for this symbol
hasShort := false
for _, pos := range openPositions {
if pos.Symbol == symbol && pos.Side == "SHORT" && pos.Status == "OPEN" {
hasShort = true
break
}
}
if hasShort {
positionSide = "SHORT"
orderAction = "close_short"
} else {
positionSide = "LONG"
orderAction = "open_long"
}
} else {
side = "SELL"
// 卖出可能是平多或开空,这里假设是平多
positionSide = "LONG"
orderAction = "close_long"
// Check if we have an open LONG position
hasLong := false
for _, pos := range openPositions {
if pos.Symbol == symbol && pos.Side == "LONG" && pos.Status == "OPEN" {
hasLong = true
break
}
}
if hasLong {
positionSide = "LONG"
orderAction = "close_long"
} else {
positionSide = "SHORT"
orderAction = "open_short"
}
}
// 估算手续费
// Estimate fee
fee := price * filledSize * 0.0004
// 创建订单记录
// Create order record
filledAt := time.Unix(order.FilledAt, 0)
if order.FilledAt == 0 {
filledAt = time.Unix(order.UpdatedAt, 0)
@@ -137,9 +188,10 @@ func (t *LighterTraderV2) SyncOrdersFromLighter(traderID string, orderStore *sto
orderRecord := &store.TraderOrder{
TraderID: traderID,
ExchangeID: "lighter",
ExchangeID: exchangeID, // UUID
ExchangeType: exchangeType, // Exchange type
ExchangeOrderID: order.OrderID,
Symbol: order.Symbol,
Symbol: symbol,
Side: side,
PositionSide: positionSide,
Type: "MARKET",
@@ -155,20 +207,21 @@ func (t *LighterTraderV2) SyncOrdersFromLighter(traderID string, orderStore *sto
UpdatedAt: time.Unix(order.UpdatedAt, 0),
}
// 插入订单记录
// Insert order record
if err := orderStore.CreateOrder(orderRecord); err != nil {
logger.Infof(" ⚠️ Failed to sync order %s: %v", order.OrderID, err)
continue
}
// 创建成交记录
// Create fill record
fillRecord := &store.TraderFill{
TraderID: traderID,
ExchangeID: "lighter",
ExchangeID: exchangeID, // UUID
ExchangeType: exchangeType, // Exchange type
OrderID: orderRecord.ID,
ExchangeOrderID: order.OrderID,
ExchangeTradeID: fmt.Sprintf("%s-%d", order.OrderID, time.Now().UnixNano()),
Symbol: order.Symbol,
Symbol: symbol,
Side: side,
Price: price,
Quantity: filledSize,
@@ -184,20 +237,63 @@ func (t *LighterTraderV2) SyncOrdersFromLighter(traderID string, orderStore *sto
logger.Infof(" ⚠️ Failed to sync fill for order %s: %v", order.OrderID, err)
}
// Calculate PnL for close orders
var realizedPnL float64
if strings.HasPrefix(orderAction, "close_") {
// Get the open position to calculate PnL
openPos, _ := positionStore.GetOpenPositionBySymbol(traderID, symbol, positionSide)
if openPos != nil {
if positionSide == "LONG" {
realizedPnL = (price - openPos.EntryPrice) * filledSize
} else {
realizedPnL = (openPos.EntryPrice - price) * filledSize
}
realizedPnL -= fee
}
}
// Create/update position record using PositionBuilder
if err := posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
symbol, positionSide, orderAction,
filledSize, price, fee, realizedPnL,
filledAt, order.OrderID,
); err != nil {
logger.Infof(" ⚠️ Failed to sync position for order %s: %v", order.OrderID, err)
}
// Update openPositions list dynamically
if strings.HasPrefix(orderAction, "open_") {
// Add to openPositions (approximate)
openPositions = append(openPositions, &store.TraderPosition{
Symbol: symbol,
Side: positionSide,
Status: "OPEN",
})
} else if strings.HasPrefix(orderAction, "close_") {
// Remove from openPositions (approximate)
for i, pos := range openPositions {
if pos.Symbol == symbol && pos.Side == positionSide && pos.Status == "OPEN" {
openPositions = append(openPositions[:i], openPositions[i+1:]...)
break
}
}
}
syncedCount++
logger.Infof(" ✅ Synced order: %s %s %s qty=%.6f price=%.6f", order.OrderID, order.Symbol, side, filledSize, price)
logger.Infof(" ✅ Synced order: %s %s %s qty=%.6f price=%.6f", order.OrderID, symbol, side, filledSize, price)
}
logger.Infof("✅ Order sync completed: %d new orders synced", syncedCount)
return nil
}
// StartOrderSync 启动订单同步后台任务
func (t *LighterTraderV2) StartOrderSync(traderID string, orderStore *store.OrderStore, interval time.Duration) {
// StartOrderSync starts background order sync task
func (t *LighterTraderV2) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
ticker := time.NewTicker(interval)
go func() {
for range ticker.C {
if err := t.SyncOrdersFromLighter(traderID, orderStore); err != nil {
if err := t.SyncOrdersFromLighter(traderID, exchangeID, exchangeType, st); err != nil {
// Only log non-404 errors to reduce log spam
if !strings.Contains(err.Error(), "status 404") {
logger.Infof("⚠️ Order sync failed: %v", err)
@@ -205,5 +301,5 @@ func (t *LighterTraderV2) StartOrderSync(traderID string, orderStore *store.Orde
}
}
}()
logger.Infof("🔄 Lighter order sync started (interval: %v)", interval)
logger.Infof("🔄 Lighter order+position sync started (interval: %v)", interval)
}

280
trader/okx_order_sync.go Normal file
View File

@@ -0,0 +1,280 @@
package trader
import (
"encoding/json"
"fmt"
"nofx/logger"
"nofx/store"
"sort"
"strconv"
"strings"
"time"
)
// OKXTrade represents a trade record from OKX fills history
type OKXTrade struct {
InstID string
Symbol string
TradeID string
OrderID string
Side string // buy or sell
PosSide string // long or short
FillPrice float64
FillQty float64 // In contracts
FillQtyBase float64 // In base asset (BTC, ETH, etc)
Fee float64
FeeAsset string
ExecTime time.Time
IsMaker bool
OrderType string
OrderAction string // open_long, open_short, close_long, close_short
}
// GetTrades retrieves trade/fill records from OKX
func (t *OKXTrader) GetTrades(startTime time.Time, limit int) ([]OKXTrade, error) {
if limit <= 0 {
limit = 100
}
if limit > 100 {
limit = 100 // OKX max limit is 100
}
// Build query path
// OKX fills-history endpoint for historical fills
path := fmt.Sprintf("/api/v5/trade/fills-history?instType=SWAP&limit=%d", limit)
if !startTime.IsZero() {
path += fmt.Sprintf("&begin=%d", startTime.UnixMilli())
}
data, err := t.doRequest("GET", path, nil)
if err != nil {
return nil, fmt.Errorf("failed to get fills history: %w", err)
}
var fills []struct {
InstID string `json:"instId"` // e.g., "BTC-USDT-SWAP"
TradeID string `json:"tradeId"` // Trade ID
OrdID string `json:"ordId"` // Order ID
BillID string `json:"billId"` // Bill ID
Side string `json:"side"` // buy or sell
PosSide string `json:"posSide"` // long, short, or net
FillPx string `json:"fillPx"` // Fill price
FillSz string `json:"fillSz"` // Fill size (contracts)
Fee string `json:"fee"` // Fee (negative for cost)
FeeCcy string `json:"feeCcy"` // Fee currency
Ts string `json:"ts"` // Trade timestamp (ms)
ExecType string `json:"execType"` // T: taker, M: maker
Tag string `json:"tag"` // Order tag
}
if err := json.Unmarshal(data, &fills); err != nil {
return nil, fmt.Errorf("failed to parse fills: %w", err)
}
trades := make([]OKXTrade, 0, len(fills))
for _, fill := range fills {
fillPrice, _ := strconv.ParseFloat(fill.FillPx, 64)
fillSz, _ := strconv.ParseFloat(fill.FillSz, 64)
fee, _ := strconv.ParseFloat(fill.Fee, 64)
ts, _ := strconv.ParseInt(fill.Ts, 10, 64)
// Convert symbol: BTC-USDT-SWAP -> BTCUSDT
symbol := t.convertSymbolBack(fill.InstID)
// Convert contract count to base asset quantity
fillQtyBase := fillSz
inst, err := t.getInstrument(symbol)
if err == nil && inst.CtVal > 0 {
fillQtyBase = fillSz * inst.CtVal
}
// Determine order action based on side and posSide
// OKX uses dual position mode:
// - buy + long = open long
// - sell + long = close long
// - sell + short = open short
// - buy + short = close short
orderAction := "open_long"
posSide := strings.ToLower(fill.PosSide)
side := strings.ToLower(fill.Side)
if posSide == "long" {
if side == "buy" {
orderAction = "open_long"
} else {
orderAction = "close_long"
}
} else if posSide == "short" {
if side == "sell" {
orderAction = "open_short"
} else {
orderAction = "close_short"
}
} else {
// One-way mode (net position)
if side == "buy" {
orderAction = "open_long"
} else {
orderAction = "open_short"
}
}
trade := OKXTrade{
InstID: fill.InstID,
Symbol: symbol,
TradeID: fill.TradeID,
OrderID: fill.OrdID,
Side: fill.Side,
PosSide: fill.PosSide,
FillPrice: fillPrice,
FillQty: fillSz,
FillQtyBase: fillQtyBase,
Fee: -fee, // OKX returns negative fee
FeeAsset: fill.FeeCcy,
ExecTime: time.UnixMilli(ts),
IsMaker: fill.ExecType == "M",
OrderType: "MARKET",
OrderAction: orderAction,
}
trades = append(trades, trade)
}
return trades, nil
}
// SyncOrdersFromOKX syncs OKX exchange order history to local database
// Also creates/updates position records to ensure orders/fills/positions data consistency
// exchangeID: Exchange account UUID (from exchanges.id)
// exchangeType: Exchange type ("okx")
func (t *OKXTrader) SyncOrdersFromOKX(traderID string, exchangeID string, exchangeType string, st *store.Store) error {
if st == nil {
return fmt.Errorf("store is nil")
}
// Get recent trades (last 24 hours)
startTime := time.Now().Add(-24 * time.Hour)
logger.Infof("🔄 Syncing OKX trades from: %s", startTime.Format(time.RFC3339))
// Use GetTrades method to fetch trade records
trades, err := t.GetTrades(startTime, 100)
if err != nil {
return fmt.Errorf("failed to get trades: %w", err)
}
logger.Infof("📥 Received %d trades from OKX", len(trades))
// Sort trades by time ASC (oldest first) for proper position building
sort.Slice(trades, func(i, j int) bool {
return trades[i].ExecTime.Before(trades[j].ExecTime)
})
// Process trades one by one (no transaction to avoid deadlock)
orderStore := st.Order()
positionStore := st.Position()
posBuilder := store.NewPositionBuilder(positionStore)
syncedCount := 0
for _, trade := range trades {
// Check if trade already exists (use exchangeID which is UUID, not exchange type)
existing, err := orderStore.GetOrderByExchangeID(exchangeID, trade.TradeID)
if err == nil && existing != nil {
continue // Order already exists, skip
}
// Determine position side from order action
positionSide := "LONG"
if strings.Contains(trade.OrderAction, "short") {
positionSide = "SHORT"
}
// Normalize side for storage
side := strings.ToUpper(trade.Side)
// Create order record
orderRecord := &store.TraderOrder{
TraderID: traderID,
ExchangeID: exchangeID, // UUID
ExchangeType: exchangeType, // Exchange type
ExchangeOrderID: trade.TradeID,
Symbol: trade.Symbol,
Side: side,
PositionSide: positionSide,
Type: trade.OrderType,
OrderAction: trade.OrderAction,
Quantity: trade.FillQtyBase,
Price: trade.FillPrice,
Status: "FILLED",
FilledQuantity: trade.FillQtyBase,
AvgFillPrice: trade.FillPrice,
Commission: trade.Fee,
FilledAt: trade.ExecTime,
CreatedAt: trade.ExecTime,
UpdatedAt: trade.ExecTime,
}
// Insert order record
if err := orderStore.CreateOrder(orderRecord); err != nil {
logger.Infof(" ⚠️ Failed to sync trade %s: %v", trade.TradeID, err)
continue
}
// Create fill record
fillRecord := &store.TraderFill{
TraderID: traderID,
ExchangeID: exchangeID, // UUID
ExchangeType: exchangeType, // Exchange type
OrderID: orderRecord.ID,
ExchangeOrderID: trade.OrderID,
ExchangeTradeID: trade.TradeID,
Symbol: trade.Symbol,
Side: side,
Price: trade.FillPrice,
Quantity: trade.FillQtyBase,
QuoteQuantity: trade.FillPrice * trade.FillQtyBase,
Commission: trade.Fee,
CommissionAsset: trade.FeeAsset,
RealizedPnL: 0, // OKX fills don't include PnL per trade
IsMaker: trade.IsMaker,
CreatedAt: trade.ExecTime,
}
if err := orderStore.CreateFill(fillRecord); err != nil {
logger.Infof(" ⚠️ Failed to sync fill for trade %s: %v", trade.TradeID, err)
}
// Create/update position record using PositionBuilder
if err := posBuilder.ProcessTrade(
traderID, exchangeID, exchangeType,
trade.Symbol, positionSide, trade.OrderAction,
trade.FillQtyBase, trade.FillPrice, trade.Fee, 0, // No per-trade PnL from OKX
trade.ExecTime, trade.TradeID,
); err != nil {
logger.Infof(" ⚠️ Failed to sync position for trade %s: %v", trade.TradeID, err)
} else {
logger.Infof(" 📍 Position updated for trade: %s (action: %s, qty: %.6f)", trade.TradeID, trade.OrderAction, trade.FillQtyBase)
}
syncedCount++
logger.Infof(" ✅ Synced trade: %s %s %s qty=%.6f price=%.6f fee=%.6f action=%s",
trade.TradeID, trade.Symbol, side, trade.FillQtyBase, trade.FillPrice, trade.Fee, trade.OrderAction)
}
logger.Infof("✅ OKX order sync completed: %d new trades synced", syncedCount)
return nil
}
// StartOrderSync starts background order sync task for OKX
func (t *OKXTrader) StartOrderSync(traderID string, exchangeID string, exchangeType string, st *store.Store, interval time.Duration) {
ticker := time.NewTicker(interval)
go func() {
for range ticker.C {
if err := t.SyncOrdersFromOKX(traderID, exchangeID, exchangeType, st); err != nil {
logger.Infof("⚠️ OKX order sync failed: %v", err)
}
}
}()
logger.Infof("🔄 OKX order sync started (interval: %v)", interval)
}

101
trader/position_snapshot.go Normal file
View File

@@ -0,0 +1,101 @@
package trader
import (
"fmt"
"nofx/logger"
"nofx/store"
"time"
)
// CreatePositionSnapshot gets current real positions from exchange and creates snapshot positions
// This function will:
// 1. Delete all OPEN old positions from database
// 2. Get current real positions from exchange
// 3. Create a "snapshot" record for each real position
func CreatePositionSnapshot(traderID, exchangeID, exchangeType string, trader Trader, st *store.Store) error {
logger.Infof("📸 Creating position snapshot for trader %s (%s)...", traderID, exchangeType)
positionStore := st.Position()
// Step 1: Delete all OPEN positions
logger.Infof("🗑️ Deleting all OPEN positions from database...")
if err := positionStore.DeleteAllOpenPositions(traderID); err != nil {
return fmt.Errorf("failed to delete open positions: %w", err)
}
logger.Infof("✅ Deleted all OPEN positions")
// Step 2: Get current positions from exchange
logger.Infof("📡 Fetching current positions from exchange...")
positions, err := trader.GetPositions()
if err != nil {
return fmt.Errorf("failed to get positions from exchange: %w", err)
}
if len(positions) == 0 {
logger.Infof("✅ No open positions on exchange, snapshot complete")
return nil
}
logger.Infof("📥 Found %d positions on exchange", len(positions))
// Step 3: Create snapshot record for each position
now := time.Now()
createdCount := 0
for _, posMap := range positions {
// Parse position data
symbol, _ := posMap["symbol"].(string)
sideStr, _ := posMap["side"].(string)
positionAmt, _ := posMap["positionAmt"].(float64)
entryPrice, _ := posMap["entryPrice"].(float64)
markPrice, _ := posMap["markPrice"].(float64)
leverage, _ := posMap["leverage"].(float64)
// Skip positions with 0 quantity
if positionAmt == 0 {
continue
}
// Determine position side
side := "LONG"
if sideStr == "short" {
side = "SHORT"
}
// Use current mark price as entry price (approximation)
// If entryPrice is 0, use markPrice
if entryPrice == 0 {
entryPrice = markPrice
}
snapshotPosition := &store.TraderPosition{
TraderID: traderID,
ExchangeID: exchangeID,
ExchangeType: exchangeType,
ExchangePositionID: fmt.Sprintf("snapshot_%s_%s_%d", symbol, side, now.UnixMilli()),
Symbol: symbol,
Side: side,
Quantity: positionAmt,
EntryPrice: entryPrice,
EntryOrderID: "snapshot", // Mark as snapshot
EntryTime: now,
Leverage: int(leverage),
Status: "OPEN",
Source: "snapshot", // Mark source as snapshot
CreatedAt: now,
UpdatedAt: now,
}
if err := positionStore.CreateOpenPosition(snapshotPosition); err != nil {
logger.Infof(" ⚠️ Failed to create snapshot position for %s %s: %v", symbol, side, err)
continue
}
logger.Infof(" ✅ Created snapshot: %s %s %.6f @ %.2f (leverage: %dx)",
symbol, side, positionAmt, entryPrice, int(leverage))
createdCount++
}
logger.Infof("✅ Position snapshot complete: %d positions created", createdCount)
return nil
}

View File

@@ -1,809 +0,0 @@
package trader
import (
"fmt"
"nofx/logger"
"nofx/store"
"strings"
"sync"
"time"
)
// PositionSyncManager Position status synchronization manager
// Responsible for periodically synchronizing exchange positions, detecting manual closures and other changes
type PositionSyncManager struct {
store *store.Store
interval time.Duration
historySyncInterval time.Duration // Interval for full history sync
stopCh chan struct{}
wg sync.WaitGroup
traderCache map[string]Trader // trader_id -> Trader instance cache
configCache map[string]*store.TraderFullConfig // trader_id -> config cache
cacheMutex sync.RWMutex
lastHistorySync map[string]time.Time // trader_id -> last history sync time
lastHistorySyncMutex sync.RWMutex
}
// NewPositionSyncManager Create position synchronization manager
func NewPositionSyncManager(st *store.Store, interval time.Duration) *PositionSyncManager {
if interval == 0 {
interval = 10 * time.Second
}
return &PositionSyncManager{
store: st,
interval: interval,
historySyncInterval: 5 * time.Minute, // Sync closed positions every 5 minutes
stopCh: make(chan struct{}),
traderCache: make(map[string]Trader),
configCache: make(map[string]*store.TraderFullConfig),
lastHistorySync: make(map[string]time.Time),
}
}
// Start Start position synchronization service
func (m *PositionSyncManager) Start() {
m.wg.Add(1)
go m.run()
logger.Info("📊 Position sync manager started")
// Run startup sync in background
go m.startupSync()
}
// Stop Stop position synchronization service
func (m *PositionSyncManager) Stop() {
close(m.stopCh)
m.wg.Wait()
// Clear cache
m.cacheMutex.Lock()
m.traderCache = make(map[string]Trader)
m.configCache = make(map[string]*store.TraderFullConfig)
m.cacheMutex.Unlock()
logger.Info("📊 Position sync manager stopped")
}
// run Main loop
func (m *PositionSyncManager) run() {
defer m.wg.Done()
// Execute immediately on startup
m.syncPositions()
ticker := time.NewTicker(m.interval)
defer ticker.Stop()
for {
select {
case <-m.stopCh:
return
case <-ticker.C:
m.syncPositions()
}
}
}
// syncPositions Synchronize all position statuses
func (m *PositionSyncManager) syncPositions() {
// Get all OPEN status positions
localPositions, err := m.store.Position().GetAllOpenPositions()
if err != nil {
logger.Infof("⚠️ Failed to get local positions: %v", err)
return
}
if len(localPositions) == 0 {
return
}
// Group by trader_id
positionsByTrader := make(map[string][]*store.TraderPosition)
for _, pos := range localPositions {
positionsByTrader[pos.TraderID] = append(positionsByTrader[pos.TraderID], pos)
}
// Process each trader
for traderID, traderPositions := range positionsByTrader {
m.syncTraderPositions(traderID, traderPositions)
}
}
// syncTraderPositions Synchronize positions for a single trader
func (m *PositionSyncManager) syncTraderPositions(traderID string, localPositions []*store.TraderPosition) {
// Get or create trader instance
trader, err := m.getOrCreateTrader(traderID)
if err != nil {
logger.Infof("⚠️ Failed to get trader instance (ID: %s): %v", traderID, err)
return
}
// Get exchange info for history sync
config, _ := m.getTraderConfig(traderID)
exchangeID := ""
exchangeType := ""
if config != nil {
exchangeID = config.Exchange.ID // UUID for database association
exchangeType = config.Exchange.ExchangeType // "binance", "bybit" etc for trader creation
}
// Maybe run periodic history sync
if exchangeID != "" && exchangeType != "" {
m.maybeRunHistorySync(traderID, exchangeID, exchangeType, trader)
}
// Get current exchange positions
exchangePositions, err := trader.GetPositions()
if err != nil {
logger.Infof("⚠️ Failed to get exchange positions (ID: %s): %v", traderID, err)
return
}
// Build exchange position map: symbol_side -> position
// Note: Exchange returns side as "long"/"short" (lowercase), database stores "LONG"/"SHORT" (uppercase)
exchangeMap := make(map[string]map[string]interface{})
for _, pos := range exchangePositions {
symbol, _ := pos["symbol"].(string)
side, _ := pos["side"].(string) // Note: use "side" not "positionSide"
if symbol == "" || side == "" {
continue
}
// Normalize side to uppercase for matching with database
normalizedSide := strings.ToUpper(side)
key := fmt.Sprintf("%s_%s", symbol, normalizedSide)
exchangeMap[key] = pos
}
// Compare local and exchange positions
for _, localPos := range localPositions {
key := fmt.Sprintf("%s_%s", localPos.Symbol, localPos.Side)
exchangePos, exists := exchangeMap[key]
if !exists {
// Exchange doesn't have this position → it has been closed
m.closeLocalPosition(localPos, trader, "manual")
continue
}
// Check if quantity is 0 or very small
qty := getFloatFromMap(exchangePos, "positionAmt")
if qty < 0 {
qty = -qty // Short position quantity is negative
}
if qty < 0.0000001 {
// Quantity is 0, position closed
m.closeLocalPosition(localPos, trader, "manual")
}
}
}
// closeLocalPosition Mark local position as closed
func (m *PositionSyncManager) closeLocalPosition(pos *store.TraderPosition, trader Trader, reason string) {
// Try to get accurate closure data from exchange first
closedPnLRecord := m.findClosedPnLRecord(trader, pos)
var exitPrice, realizedPnL, fee float64
var closeReason, exitOrderID string
if closedPnLRecord != nil {
// Use accurate data from exchange
exitPrice = closedPnLRecord.ExitPrice
realizedPnL = closedPnLRecord.RealizedPnL
fee = closedPnLRecord.Fee
closeReason = closedPnLRecord.CloseType
exitOrderID = closedPnLRecord.OrderID
logger.Infof("📊 Found accurate closure data from exchange for %s %s", pos.Symbol, pos.Side)
} else {
// Fallback: use market price and calculate PnL
exitPrice = pos.EntryPrice // Default to entry price
if price, err := trader.GetMarketPrice(pos.Symbol); err == nil && price > 0 {
exitPrice = price
}
// Calculate PnL
if pos.Side == "LONG" {
realizedPnL = (exitPrice - pos.EntryPrice) * pos.Quantity
} else {
realizedPnL = (pos.EntryPrice - exitPrice) * pos.Quantity
}
closeReason = reason
fee = 0
exitOrderID = ""
logger.Infof("⚠️ Using market price for closure (no exchange data): %s %s", pos.Symbol, pos.Side)
}
// Update database
err := m.store.Position().ClosePosition(
pos.ID,
exitPrice,
exitOrderID,
realizedPnL,
fee,
closeReason,
)
if err != nil {
logger.Infof("⚠️ Failed to update position status: %v", err)
} else {
logger.Infof("📊 Position closed [%s] %s %s @ %.4f → %.4f, PnL: %.2f, Fee: %.4f (%s)",
pos.TraderID[:8], pos.Symbol, pos.Side, pos.EntryPrice, exitPrice, realizedPnL, fee, closeReason)
}
}
// findClosedPnLRecord Try to find matching ClosedPnL record from exchange
// For Binance, directly query trades for the specific symbol (more reliable than Income API)
func (m *PositionSyncManager) findClosedPnLRecord(trader Trader, pos *store.TraderPosition) *ClosedPnLRecord {
// Try to get trades directly for this symbol (Binance-specific, more reliable)
if binanceTrader, ok := trader.(*FuturesTrader); ok {
return m.findClosedPnLFromBinanceTrades(binanceTrader, pos)
}
// Fallback: use GetClosedPnL for other exchanges
startTime := time.Now().Add(-24 * time.Hour)
records, err := trader.GetClosedPnL(startTime, 100)
if err != nil {
logger.Infof("⚠️ Failed to get closed PnL records: %v", err)
return nil
}
return m.aggregateClosedRecords(records, pos)
}
// findClosedPnLFromBinanceTrades queries Binance directly for trades of a specific symbol
func (m *PositionSyncManager) findClosedPnLFromBinanceTrades(trader *FuturesTrader, pos *store.TraderPosition) *ClosedPnLRecord {
// Query trades for this specific symbol from the last hour
startTime := time.Now().Add(-1 * time.Hour)
trades, err := trader.GetTradesForSymbol(pos.Symbol, startTime, 100)
if err != nil {
logger.Infof("⚠️ Failed to get trades for %s: %v", pos.Symbol, err)
return nil
}
if len(trades) == 0 {
logger.Infof("⚠️ No trades found for %s in the last hour", pos.Symbol)
return nil
}
// Find all closing trades (realizedPnl != 0) that match this position
var totalQty, totalPnL, totalFee float64
var weightedExitPrice float64
var latestExitTime time.Time
var latestTradeID string
matchCount := 0
posSide := strings.ToLower(pos.Side)
for _, trade := range trades {
// Skip opening trades
if trade.RealizedPnL == 0 {
continue
}
// Determine if this trade closes our position
// For LONG position: SELL closes it
// For SHORT position: BUY closes it
isClosingTrade := false
tradeSide := strings.ToUpper(trade.Side)
positionSide := strings.ToUpper(trade.PositionSide)
if positionSide == "LONG" && posSide == "long" {
isClosingTrade = true
} else if positionSide == "SHORT" && posSide == "short" {
isClosingTrade = true
} else if positionSide == "BOTH" || positionSide == "" {
// One-way mode
if tradeSide == "SELL" && posSide == "long" {
isClosingTrade = true
} else if tradeSide == "BUY" && posSide == "short" {
isClosingTrade = true
}
}
if !isClosingTrade {
continue
}
// Aggregate this trade
totalQty += trade.Quantity
totalPnL += trade.RealizedPnL
totalFee += trade.Fee
weightedExitPrice += trade.Price * trade.Quantity
matchCount++
if trade.Time.After(latestExitTime) {
latestExitTime = trade.Time
latestTradeID = trade.TradeID
}
}
if matchCount == 0 {
logger.Infof("⚠️ No closing trades found for %s %s", pos.Symbol, pos.Side)
return nil
}
avgExitPrice := weightedExitPrice / totalQty
logger.Infof("📊 Found %d closing trades for %s %s: qty=%.4f, exitPrice=%.6f, pnl=%.4f, fee=%.4f",
matchCount, pos.Symbol, pos.Side, totalQty, avgExitPrice, totalPnL, totalFee)
return &ClosedPnLRecord{
Symbol: pos.Symbol,
Side: posSide,
EntryPrice: pos.EntryPrice,
ExitPrice: avgExitPrice,
Quantity: totalQty,
RealizedPnL: totalPnL,
Fee: totalFee,
ExitTime: latestExitTime,
EntryTime: pos.EntryTime,
OrderID: latestTradeID,
ExchangeID: latestTradeID,
CloseType: "unknown",
}
}
// aggregateClosedRecords aggregates closed PnL records for a position
func (m *PositionSyncManager) aggregateClosedRecords(records []ClosedPnLRecord, pos *store.TraderPosition) *ClosedPnLRecord {
if len(records) == 0 {
return nil
}
posSide := strings.ToLower(pos.Side)
var matchingRecords []ClosedPnLRecord
for i := range records {
record := &records[i]
if record.Symbol != pos.Symbol {
continue
}
recordSide := strings.ToLower(record.Side)
if recordSide != posSide {
continue
}
matchingRecords = append(matchingRecords, *record)
}
if len(matchingRecords) == 0 {
return nil
}
var totalQty, totalPnL, totalFee float64
var weightedExitPrice float64
var latestExitTime time.Time
var latestOrderID, latestExchangeID string
for _, rec := range matchingRecords {
totalQty += rec.Quantity
totalPnL += rec.RealizedPnL
totalFee += rec.Fee
weightedExitPrice += rec.ExitPrice * rec.Quantity
if rec.ExitTime.After(latestExitTime) {
latestExitTime = rec.ExitTime
latestOrderID = rec.OrderID
latestExchangeID = rec.ExchangeID
}
}
avgExitPrice := weightedExitPrice / totalQty
logger.Infof("📊 Aggregated %d closing trades for %s %s: qty=%.4f, pnl=%.4f, fee=%.4f",
len(matchingRecords), pos.Symbol, pos.Side, totalQty, totalPnL, totalFee)
return &ClosedPnLRecord{
Symbol: pos.Symbol,
Side: posSide,
EntryPrice: pos.EntryPrice,
ExitPrice: avgExitPrice,
Quantity: totalQty,
RealizedPnL: totalPnL,
Fee: totalFee,
ExitTime: latestExitTime,
EntryTime: pos.EntryTime,
OrderID: latestOrderID,
ExchangeID: latestExchangeID,
CloseType: "unknown",
}
}
// abs returns absolute value of float64
func abs(x float64) float64 {
if x < 0 {
return -x
}
return x
}
// getOrCreateTrader Get or create trader instance
func (m *PositionSyncManager) getOrCreateTrader(traderID string) (Trader, error) {
m.cacheMutex.RLock()
trader, exists := m.traderCache[traderID]
m.cacheMutex.RUnlock()
if exists && trader != nil {
return trader, nil
}
// Need to create new trader instance
config, err := m.getTraderConfig(traderID)
if err != nil {
return nil, fmt.Errorf("failed to get trader config: %w", err)
}
trader, err = m.createTrader(config)
if err != nil {
return nil, fmt.Errorf("failed to create trader instance: %w", err)
}
m.cacheMutex.Lock()
m.traderCache[traderID] = trader
m.cacheMutex.Unlock()
return trader, nil
}
// getTraderConfig Get trader configuration
func (m *PositionSyncManager) getTraderConfig(traderID string) (*store.TraderFullConfig, error) {
m.cacheMutex.RLock()
config, exists := m.configCache[traderID]
m.cacheMutex.RUnlock()
if exists {
return config, nil
}
// Get from database
traders, err := m.store.Trader().ListAll()
if err != nil {
return nil, fmt.Errorf("failed to get trader list: %w", err)
}
var userID string
for _, t := range traders {
if t.ID == traderID {
userID = t.UserID
break
}
}
if userID == "" {
return nil, fmt.Errorf("trader not found: %s", traderID)
}
config, err = m.store.Trader().GetFullConfig(userID, traderID)
if err != nil {
return nil, err
}
m.cacheMutex.Lock()
m.configCache[traderID] = config
m.cacheMutex.Unlock()
return config, nil
}
// createTrader Create trader instance based on configuration
func (m *PositionSyncManager) createTrader(config *store.TraderFullConfig) (Trader, error) {
exchange := config.Exchange
// Use exchange.ExchangeType to determine specific exchange, not exchange.ID (UUID) or exchange.Type (cex/dex)
switch exchange.ExchangeType {
case "binance":
return NewFuturesTrader(exchange.APIKey, exchange.SecretKey, config.Trader.UserID), nil
case "bybit":
return NewBybitTrader(exchange.APIKey, exchange.SecretKey), nil
case "okx":
return NewOKXTrader(exchange.APIKey, exchange.SecretKey, exchange.Passphrase), nil
case "bitget":
return NewBitgetTrader(exchange.APIKey, exchange.SecretKey, exchange.Passphrase), nil
case "hyperliquid":
return NewHyperliquidTrader(exchange.SecretKey, exchange.HyperliquidWalletAddr, exchange.Testnet)
case "aster":
return NewAsterTrader(exchange.AsterUser, exchange.AsterSigner, exchange.AsterPrivateKey)
case "lighter":
if exchange.LighterWalletAddr == "" || exchange.LighterAPIKeyPrivateKey == "" {
return nil, fmt.Errorf("Lighter requires wallet address and API Key private key")
}
// Lighter only supports mainnet
return NewLighterTraderV2(
exchange.LighterWalletAddr,
exchange.LighterAPIKeyPrivateKey,
exchange.LighterAPIKeyIndex,
false, // Always use mainnet for Lighter
)
default:
return nil, fmt.Errorf("unsupported exchange type: %s", exchange.ExchangeType)
}
}
// InvalidateCache Invalidate cache
func (m *PositionSyncManager) InvalidateCache(traderID string) {
m.cacheMutex.Lock()
defer m.cacheMutex.Unlock()
delete(m.traderCache, traderID)
delete(m.configCache, traderID)
}
// getFloatFromMap Get float64 value from map
func getFloatFromMap(m map[string]interface{}, key string) float64 {
if v, ok := m[key]; ok {
switch val := v.(type) {
case float64:
return val
case int64:
return float64(val)
case int:
return float64(val)
case string:
var f float64
fmt.Sscanf(val, "%f", &f)
return f
}
}
return 0
}
// =============================================================================
// Startup and History Sync Methods
// =============================================================================
// startupSync performs initial sync on startup
// 1. Sync existing positions from exchange (to detect external positions)
// 2. Sync closed positions history from exchange
func (m *PositionSyncManager) startupSync() {
logger.Info("📊 Starting startup sync...")
// Get all traders
traders, err := m.store.Trader().ListAll()
if err != nil {
logger.Infof("⚠️ Failed to get traders for startup sync: %v", err)
return
}
for _, traderInfo := range traders {
traderID := traderInfo.ID
// Get trader instance
trader, err := m.getOrCreateTrader(traderID)
if err != nil {
logger.Infof("⚠️ Failed to get trader instance for startup sync (ID: %s): %v", traderID, err)
continue
}
// Get exchange info
config, err := m.getTraderConfig(traderID)
if err != nil {
logger.Infof("⚠️ Failed to get trader config for startup sync (ID: %s): %v", traderID, err)
continue
}
exchangeID := config.Exchange.ID // UUID
exchangeType := config.Exchange.ExchangeType // "binance", "bybit" etc
// 1. Sync current open positions from exchange
m.syncExternalPositions(traderID, exchangeID, exchangeType, trader)
// 2. Sync closed positions history from exchange
m.syncClosedPositionsHistory(traderID, exchangeID, exchangeType, trader)
}
logger.Info("📊 Startup sync completed")
}
// syncExternalPositions syncs positions that exist on exchange but not locally
// These could be positions opened manually or from other systems
func (m *PositionSyncManager) syncExternalPositions(traderID, exchangeID, exchangeType string, trader Trader) {
// Get current positions from exchange
exchangePositions, err := trader.GetPositions()
if err != nil {
logger.Infof("⚠️ Failed to get exchange positions for external sync (ID: %s): %v", traderID, err)
return
}
// Get local open positions
localPositions, err := m.store.Position().GetOpenPositions(traderID)
if err != nil {
logger.Infof("⚠️ Failed to get local positions for external sync (ID: %s): %v", traderID, err)
return
}
// Build local position map: symbol_side -> position
localMap := make(map[string]*store.TraderPosition)
for _, pos := range localPositions {
key := fmt.Sprintf("%s_%s", pos.Symbol, pos.Side)
localMap[key] = pos
}
// Find positions that exist on exchange but not locally
for _, pos := range exchangePositions {
symbol, _ := pos["symbol"].(string)
side, _ := pos["side"].(string)
if symbol == "" || side == "" {
continue
}
// Normalize side
normalizedSide := side
if side == "Buy" || side == "LONG" || side == "long" {
normalizedSide = "LONG"
} else if side == "Sell" || side == "SHORT" || side == "short" {
normalizedSide = "SHORT"
}
key := fmt.Sprintf("%s_%s", symbol, normalizedSide)
// Check if we already have this position locally
if _, exists := localMap[key]; exists {
continue // Already tracking this position
}
// This is an external position - create local record
qty := getFloatFromMap(pos, "positionAmt")
if qty < 0 {
qty = -qty
}
if qty < 0.0000001 {
continue // No actual position
}
entryPrice := getFloatFromMap(pos, "entryPrice")
leverage := int(getFloatFromMap(pos, "leverage"))
if leverage == 0 {
leverage = 1
}
// Get entry time if available
createdTime := getFloatFromMap(pos, "createdTime")
var entryTime time.Time
if createdTime > 0 {
entryTime = time.UnixMilli(int64(createdTime))
} else {
entryTime = time.Now() // Use current time as fallback
}
// Generate unique exchange position ID
exchangePositionID := fmt.Sprintf("%s_%s_%d", symbol, normalizedSide, entryTime.UnixMilli())
newPos := &store.TraderPosition{
TraderID: traderID,
ExchangeID: exchangeID,
ExchangeType: exchangeType,
ExchangePositionID: exchangePositionID,
Symbol: symbol,
Side: normalizedSide,
Quantity: qty,
EntryPrice: entryPrice,
EntryTime: entryTime,
Leverage: leverage,
Source: "sync", // Mark as synced from exchange
}
if err := m.store.Position().CreateOpenPosition(newPos); err != nil {
logger.Infof("⚠️ Failed to create external position record: %v", err)
} else {
logger.Infof("📊 Synced external position: [%s] %s %s @ %.4f (qty: %.4f)",
traderID[:8], symbol, normalizedSide, entryPrice, qty)
}
}
}
// syncClosedPositionsHistory syncs closed positions from exchange history
// IMPORTANT: Only exchanges with position-level history API should sync history:
// - Bybit: /v5/position/closed-pnl (accurate position records)
// - OKX: /api/v5/account/positions-history (accurate position records)
// Other exchanges (Binance, Hyperliquid, Lighter, Aster) only have trade-level data,
// which cannot accurately reconstruct positions. They should NOT sync historical positions.
func (m *PositionSyncManager) syncClosedPositionsHistory(traderID, exchangeID, exchangeType string, trader Trader) {
// Only sync history for exchanges with position-level API
// Binance/Hyperliquid/Lighter/Aster only have trade-level data, skip history sync
switch exchangeType {
case "bybit", "okx":
// These exchanges have position-level history API, proceed with sync
default:
// Other exchanges don't have accurate position history API
// Their GetClosedPnL only returns recent trades for closure detection, not for history sync
return
}
// Get last sync time from database
lastSyncTime, err := m.store.Position().GetLastClosedPositionTime(traderID)
if err != nil {
logger.Infof("⚠️ Failed to get last closed position time (ID: %s): %v", traderID, err)
// First sync: go back 90 days to get more history
lastSyncTime = time.Now().Add(-90 * 24 * time.Hour)
}
// Subtract a small buffer to avoid missing positions at the boundary
startTime := lastSyncTime.Add(-1 * time.Minute)
// Pagination loop to get all records
const batchSize = 500
totalCreated := 0
totalSkipped := 0
for {
// Get closed positions from exchange
closedRecords, err := trader.GetClosedPnL(startTime, batchSize)
if err != nil {
logger.Infof("⚠️ Failed to get closed PnL records (ID: %s): %v", traderID, err)
break
}
if len(closedRecords) == 0 {
break
}
// Convert to store.ClosedPnLRecord and sync
storeRecords := make([]store.ClosedPnLRecord, len(closedRecords))
var latestExitTime time.Time
for i, rec := range closedRecords {
storeRecords[i] = store.ClosedPnLRecord{
Symbol: rec.Symbol,
Side: rec.Side,
EntryPrice: rec.EntryPrice,
ExitPrice: rec.ExitPrice,
Quantity: rec.Quantity,
RealizedPnL: rec.RealizedPnL,
Fee: rec.Fee,
Leverage: rec.Leverage,
EntryTime: rec.EntryTime,
ExitTime: rec.ExitTime,
OrderID: rec.OrderID,
CloseType: rec.CloseType,
ExchangeID: rec.ExchangeID,
}
// Track latest exit time for pagination
if rec.ExitTime.After(latestExitTime) {
latestExitTime = rec.ExitTime
}
}
created, skipped, err := m.store.Position().SyncClosedPositions(traderID, exchangeID, exchangeType, storeRecords)
if err != nil {
logger.Infof("⚠️ Failed to sync closed positions (ID: %s): %v", traderID, err)
break
}
totalCreated += created
totalSkipped += skipped
// If we got fewer records than batch size, we've reached the end
if len(closedRecords) < batchSize {
break
}
// Move start time forward for next batch (add 1ms to avoid duplicate)
startTime = latestExitTime.Add(time.Millisecond)
}
if totalCreated > 0 {
logger.Infof("📊 Synced %d new closed positions for trader %s (skipped %d duplicates)",
totalCreated, traderID[:8], totalSkipped)
}
// Update last history sync time
m.lastHistorySyncMutex.Lock()
m.lastHistorySync[traderID] = time.Now()
m.lastHistorySyncMutex.Unlock()
}
// maybeRunHistorySync checks if it's time to run history sync for a trader
func (m *PositionSyncManager) maybeRunHistorySync(traderID, exchangeID, exchangeType string, trader Trader) {
m.lastHistorySyncMutex.RLock()
lastSync, exists := m.lastHistorySync[traderID]
m.lastHistorySyncMutex.RUnlock()
if !exists || time.Since(lastSync) >= m.historySyncInterval {
m.syncClosedPositionsHistory(traderID, exchangeID, exchangeType, trader)
}
}

View File

@@ -83,6 +83,8 @@ export function AdvancedChart({
const [showIndicatorPanel, setShowIndicatorPanel] = useState(false)
const [showOrderMarkers, setShowOrderMarkers] = useState(true) // 订单标记显示开关,默认显示
const isInitialLoadRef = useRef(true) // 跟踪是否为初始加载
const [tooltipData, setTooltipData] = useState<any>(null)
const tooltipRef = useRef<HTMLDivElement>(null)
// 指标配置
const [indicators, setIndicators] = useState<IndicatorConfig[]>([
@@ -355,6 +357,31 @@ export function AdvancedChart({
window.addEventListener('resize', handleResize)
// 监听鼠标移动,显示 OHLC 信息
chart.subscribeCrosshairMove((param) => {
if (!param.time || !param.point || !candlestickSeriesRef.current) {
setTooltipData(null)
return
}
const data = param.seriesData.get(candlestickSeriesRef.current as any)
if (!data) {
setTooltipData(null)
return
}
const candleData = data as any
setTooltipData({
time: param.time,
open: candleData.open,
high: candleData.high,
low: candleData.low,
close: candleData.close,
x: param.point.x,
y: param.point.y,
})
})
return () => {
window.removeEventListener('resize', handleResize)
chart.remove()
@@ -741,6 +768,56 @@ export function AdvancedChart({
<div style={{ position: 'relative' }}>
<div ref={chartContainerRef} />
{/* OHLC Tooltip */}
{tooltipData && (
<div
ref={tooltipRef}
style={{
position: 'absolute',
left: '10px',
top: '10px',
padding: '8px 12px',
background: 'rgba(15, 18, 21, 0.95)',
border: '1px solid rgba(240, 185, 11, 0.3)',
borderRadius: '6px',
color: '#EAECEF',
fontSize: '12px',
fontFamily: 'monospace',
pointerEvents: 'none',
zIndex: 10,
backdropFilter: 'blur(10px)',
boxShadow: '0 4px 12px rgba(0, 0, 0, 0.5)',
}}
>
<div style={{ marginBottom: '6px', color: '#F0B90B', fontWeight: 'bold', fontSize: '11px' }}>
{new Date((tooltipData.time as number) * 1000).toLocaleString(language === 'zh' ? 'zh-CN' : 'en-US', {
month: 'short',
day: 'numeric',
hour: '2-digit',
minute: '2-digit',
})}
</div>
<div style={{ display: 'grid', gridTemplateColumns: 'auto 1fr', gap: '4px 12px', fontSize: '11px' }}>
<span style={{ color: '#848E9C' }}>O:</span>
<span style={{ color: '#EAECEF', fontWeight: '500' }}>{tooltipData.open?.toFixed(2)}</span>
<span style={{ color: '#848E9C' }}>H:</span>
<span style={{ color: '#0ECB81', fontWeight: '500' }}>{tooltipData.high?.toFixed(2)}</span>
<span style={{ color: '#848E9C' }}>L:</span>
<span style={{ color: '#F6465D', fontWeight: '500' }}>{tooltipData.low?.toFixed(2)}</span>
<span style={{ color: '#848E9C' }}>C:</span>
<span style={{
color: tooltipData.close >= tooltipData.open ? '#0ECB81' : '#F6465D',
fontWeight: 'bold'
}}>
{tooltipData.close?.toFixed(2)}
</span>
</div>
</div>
)}
{/* NOFX 水印 */}
<div
style={{

View File

@@ -53,6 +53,8 @@ export function ChartWithOrders({
const seriesMarkersRef = useRef<any>(null) // Markers primitive for v5
const [loading, setLoading] = useState(true)
const [error, setError] = useState<string | null>(null)
const [tooltipData, setTooltipData] = useState<any>(null)
const tooltipRef = useRef<HTMLDivElement>(null)
// 解析时间:支持 Unix 时间戳(数字)或字符串格式
const parseCustomTime = (time: any): number => {
@@ -243,6 +245,31 @@ export function ChartWithOrders({
window.addEventListener('resize', handleResize)
// 监听鼠标移动,显示 OHLC 信息
chart.subscribeCrosshairMove((param) => {
if (!param.time || !param.point || !candlestickSeriesRef.current) {
setTooltipData(null)
return
}
const data = param.seriesData.get(candlestickSeriesRef.current as any)
if (!data) {
setTooltipData(null)
return
}
const candleData = data as any
setTooltipData({
time: param.time,
open: candleData.open,
high: candleData.high,
low: candleData.low,
close: candleData.close,
x: param.point.x,
y: param.point.y,
})
})
return () => {
window.removeEventListener('resize', handleResize)
chart.remove()
@@ -370,7 +397,59 @@ export function ChartWithOrders({
</div>
{/* 图表容器 */}
<div ref={chartContainerRef} style={{ position: 'relative' }} />
<div style={{ position: 'relative' }}>
<div ref={chartContainerRef} />
{/* OHLC Tooltip */}
{tooltipData && (
<div
ref={tooltipRef}
style={{
position: 'absolute',
left: '10px',
top: '10px',
padding: '8px 12px',
background: 'rgba(15, 18, 21, 0.95)',
border: '1px solid rgba(240, 185, 11, 0.3)',
borderRadius: '6px',
color: '#EAECEF',
fontSize: '12px',
fontFamily: 'monospace',
pointerEvents: 'none',
zIndex: 10,
backdropFilter: 'blur(10px)',
boxShadow: '0 4px 12px rgba(0, 0, 0, 0.5)',
}}
>
<div style={{ marginBottom: '6px', color: '#F0B90B', fontWeight: 'bold', fontSize: '11px' }}>
{new Date((tooltipData.time as number) * 1000).toLocaleString(language === 'zh' ? 'zh-CN' : 'en-US', {
month: 'short',
day: 'numeric',
hour: '2-digit',
minute: '2-digit',
})}
</div>
<div style={{ display: 'grid', gridTemplateColumns: 'auto 1fr', gap: '4px 12px', fontSize: '11px' }}>
<span style={{ color: '#848E9C' }}>O:</span>
<span style={{ color: '#EAECEF', fontWeight: '500' }}>{tooltipData.open?.toFixed(2)}</span>
<span style={{ color: '#848E9C' }}>H:</span>
<span style={{ color: '#0ECB81', fontWeight: '500' }}>{tooltipData.high?.toFixed(2)}</span>
<span style={{ color: '#848E9C' }}>L:</span>
<span style={{ color: '#F6465D', fontWeight: '500' }}>{tooltipData.low?.toFixed(2)}</span>
<span style={{ color: '#848E9C' }}>C:</span>
<span style={{
color: tooltipData.close >= tooltipData.open ? '#0ECB81' : '#F6465D',
fontWeight: 'bold'
}}>
{tooltipData.close?.toFixed(2)}
</span>
</div>
</div>
)}
</div>
{/* 错误提示 */}
{error && (