liuyu/nofx
1
0
Fork 0
mirror of https://github.com/NoFxAiOS/nofx.git synced 2026-07-12 15:26:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: use actual fill price from exchange API for position records

Browse Source 
- Remove trader_orders table and OrderSyncManager (never worked correctly)
- Poll GetOrderStatus to get actual avgPrice, executedQty, and commission
- Get entry price from exchange GetPositions API when closing positions
- Pass fee to trader_positions table on close
- Move TraderStats type to position.go
This commit is contained in:
tinkle-community
2025-12-08 12:15:41 +08:00
parent f39fc8af23
commit 8a5744e0a0
9 changed files with 135 additions and 904 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
trader/auto_trader.go
View File

@@ -572,14 +572,32 @@ func (at *AutoTrader) buildTradingContext() (*decision.Context, error) {
// Calculate P&L percentage (based on margin, considering leverage)
pnlPct := calculatePnLPercentage(unrealizedPnl, marginUsed)
// Track position first seen time
// Get position open time from exchange (preferred) or fallback to local tracking
posKey := symbol + "_" + side
currentPositionKeys[posKey] = true
if _, exists := at.positionFirstSeenTime[posKey]; !exists {
// New position, record current time
at.positionFirstSeenTime[posKey] = time.Now().UnixMilli()
var updateTime int64
// Priority 1: Get from database (trader_positions table) - most accurate
if at.store != nil {
if dbPos, err := at.store.Position().GetOpenPositionBySymbol(at.id, symbol, side); err == nil && dbPos != nil {
if !dbPos.EntryTime.IsZero() {
updateTime = dbPos.EntryTime.UnixMilli()
}
}
}
// Priority 2: Get from exchange API (Bybit: createdTime, OKX: createdTime)
if updateTime == 0 {
if createdTime, ok := pos["createdTime"].(int64); ok && createdTime > 0 {
updateTime = createdTime
}
}
// Priority 3: Fallback to local tracking
if updateTime == 0 {
if _, exists := at.positionFirstSeenTime[posKey]; !exists {
at.positionFirstSeenTime[posKey] = time.Now().UnixMilli()
}
updateTime = at.positionFirstSeenTime[posKey]
}
updateTime := at.positionFirstSeenTime[posKey]
// Get peak profit rate for this position
at.peakPnLCacheMutex.RLock()
@@ -910,13 +928,21 @@ func (at *AutoTrader) executeCloseLongWithRecord(decision *decision.Decision, ac
}
actionRecord.Price = marketData.CurrentPrice
// Get entry price (for P&L calculation)
// Get entry price and quantity from exchange API (most accurate)
var entryPrice float64
var quantity float64
if at.store != nil {
if openOrder, err := at.store.Order().GetLatestOpenOrder(at.id, decision.Symbol, "long"); err == nil {
entryPrice = openOrder.AvgPrice
quantity = openOrder.ExecutedQty
positions, err := at.trader.GetPositions()
if err == nil {
for _, pos := range positions {
if pos["symbol"] == decision.Symbol && pos["side"] == "long" {
if ep, ok := pos["entryPrice"].(float64); ok {
entryPrice = ep
}
if amt, ok := pos["positionAmt"].(float64); ok && amt > 0 {
quantity = amt
}
break
}
}
}
@@ -949,13 +975,21 @@ func (at *AutoTrader) executeCloseShortWithRecord(decision *decision.Decision, a
}
actionRecord.Price = marketData.CurrentPrice
// Get entry price (for P&L calculation)
// Get entry price and quantity from exchange API (most accurate)
var entryPrice float64
var quantity float64
if at.store != nil {
if openOrder, err := at.store.Order().GetLatestOpenOrder(at.id, decision.Symbol, "short"); err == nil {
entryPrice = openOrder.AvgPrice
quantity = openOrder.ExecutedQty
positions, err := at.trader.GetPositions()
if err == nil {
for _, pos := range positions {
if pos["symbol"] == decision.Symbol && pos["side"] == "short" {
if ep, ok := pos["entryPrice"].(float64); ok {
entryPrice = ep
}
if amt, ok := pos["positionAmt"].(float64); ok {
quantity = -amt // positionAmt is negative for short
}
break
}
}
}
@@ -1435,7 +1469,7 @@ func (at *AutoTrader) ClearPeakPnLCache(symbol, side string) {
delete(at.peakPnLCache, posKey)
}
// recordAndConfirmOrder records order and polls for confirmation status
// recordAndConfirmOrder polls order status for actual fill data and records position
// action: open_long, open_short, close_long, close_short
// entryPrice: entry price when closing (0 when opening)
func (at *AutoTrader) recordAndConfirmOrder(orderResult map[string]interface{}, symbol, action string, quantity float64, price float64, leverage int, entryPrice float64) {
@@ -1461,53 +1495,58 @@ func (at *AutoTrader) recordAndConfirmOrder(orderResult map[string]interface{},
return
}
// Determine side and positionSide
var side, positionSide string
// Determine positionSide
var positionSide string
switch action {
case "open_long":
side = "BUY"
case "open_long", "close_long":
positionSide = "LONG"
case "close_long":
side = "SELL"
positionSide = "LONG"
case "open_short":
side = "SELL"
positionSide = "SHORT"
case "close_short":
side = "BUY"
case "open_short", "close_short":
positionSide = "SHORT"
}
// Create order record
order := &store.TraderOrder{
TraderID: at.id,
OrderID: orderID,
Symbol: symbol,
Side: side,
PositionSide: positionSide,
Action: action,
OrderType: "MARKET",
Quantity: quantity,
Price: price,
Leverage: leverage,
Status: "NEW",
EntryPrice: entryPrice,
// Poll order status to get actual fill price, quantity and fee
var actualPrice = price // fallback to market price
var actualQty = quantity // fallback to requested quantity
var fee float64
// Wait for order to be filled and get actual fill data
time.Sleep(500 * time.Millisecond)
for i := 0; i < 5; i++ {
status, err := at.trader.GetOrderStatus(symbol, orderID)
if err == nil {
statusStr, _ := status["status"].(string)
if statusStr == "FILLED" {
// Get actual fill price
if avgPrice, ok := status["avgPrice"].(float64); ok && avgPrice > 0 {
actualPrice = avgPrice
}
// Get actual executed quantity
if execQty, ok := status["executedQty"].(float64); ok && execQty > 0 {
actualQty = execQty
}
// Get commission/fee
if commission, ok := status["commission"].(float64); ok {
fee = commission
}
logger.Infof(" ✅ Order filled: avgPrice=%.6f, qty=%.6f, fee=%.6f", actualPrice, actualQty, fee)
break
} else if statusStr == "CANCELED" || statusStr == "EXPIRED" || statusStr == "REJECTED" {
logger.Infof(" ⚠️ Order %s, skipping position record", statusStr)
return
}
}
time.Sleep(500 * time.Millisecond)
}
// Save to database
if err := at.store.Order().Create(order); err != nil {
logger.Infof(" ⚠️ Failed to record order: %v", err)
return
}
logger.Infof(" 📝 Recording position (ID: %s, action: %s, price: %.6f, qty: %.6f, fee: %.4f)",
orderID, action, actualPrice, actualQty, fee)
logger.Infof(" 📝 Order recorded (ID: %s, action: %s)", orderID, action)
// Record position change
at.recordPositionChange(orderID, symbol, positionSide, action, quantity, price, leverage, entryPrice)
// Record position change with actual fill data
at.recordPositionChange(orderID, symbol, positionSide, action, actualQty, actualPrice, leverage, entryPrice, fee)
}
// recordPositionChange records position change (create record on open, update record on close)
func (at *AutoTrader) recordPositionChange(orderID, symbol, side, action string, quantity, price float64, leverage int, entryPrice float64) {
func (at *AutoTrader) recordPositionChange(orderID, symbol, side, action string, quantity, price float64, leverage int, entryPrice float64, fee float64) {
if at.store == nil {
return
}
@@ -1555,14 +1594,14 @@ func (at *AutoTrader) recordPositionChange(orderID, symbol, side, action string,
price, // exitPrice
orderID, // exitOrderID
realizedPnL,
0, // fee (not calculated yet)
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",
at.id[:8], symbol, side, openPos.EntryPrice, price, realizedPnL)
logger.Infof(" 📊 Position closed [%s] %s %s @ %.4f → %.4f, P&L: %.2f, Fee: %.4f",
at.id[:8], symbol, side, openPos.EntryPrice, price, realizedPnL, fee)
}
}
}

1
trader/binance_futures.go
View File

@@ -195,6 +195,7 @@ func (t *FuturesTrader) GetPositions() ([]map[string]interface{}, error) {
posMap["unRealizedProfit"], _ = strconv.ParseFloat(pos.UnRealizedProfit, 64)
posMap["leverage"], _ = strconv.ParseFloat(pos.Leverage, 64)
posMap["liquidationPrice"], _ = strconv.ParseFloat(pos.LiquidationPrice, 64)
// Note: Binance SDK doesn't expose updateTime field, will fallback to local tracking
// Determine direction
if posAmt > 0 {

8
trader/bybit_trader.go
View File

@@ -220,6 +220,12 @@ func (t *BybitTrader) GetPositions() ([]map[string]interface{}, error) {
liqPriceStr, _ := pos["liqPrice"].(string)
liqPrice, _ := strconv.ParseFloat(liqPriceStr, 64)
// Position created/updated time (milliseconds timestamp)
createdTimeStr, _ := pos["createdTime"].(string)
createdTime, _ := strconv.ParseInt(createdTimeStr, 10, 64)
updatedTimeStr, _ := pos["updatedTime"].(string)
updatedTime, _ := strconv.ParseInt(updatedTimeStr, 10, 64)
positionSide, _ := pos["side"].(string) // Buy = LONG, Sell = SHORT
// Convert to unified format
@@ -240,6 +246,8 @@ func (t *BybitTrader) GetPositions() ([]map[string]interface{}, error) {
"unrealizedPnL": unrealisedPnl,
"liquidationPrice": liqPrice,
"leverage": leverage,
"createdTime": createdTime, // Position open time (ms)
"updatedTime": updatedTime, // Position last update time (ms)
}
positions = append(positions, position)

8
trader/okx_trader.go
View File

@@ -312,6 +312,8 @@ func (t *OKXTrader) GetPositions() ([]map[string]interface{}, error) {
Lever string `json:"lever"`
LiqPx string `json:"liqPx"`
Margin string `json:"margin"`
CTime string `json:"cTime"` // Position created time (ms)
UTime string `json:"uTime"` // Position last update time (ms)
}
if err := json.Unmarshal(data, &positions); err != nil {
@@ -344,6 +346,10 @@ func (t *OKXTrader) GetPositions() ([]map[string]interface{}, error) {
posAmt = -posAmt
}
// Parse timestamps
cTime, _ := strconv.ParseInt(pos.CTime, 10, 64)
uTime, _ := strconv.ParseInt(pos.UTime, 10, 64)
posMap := map[string]interface{}{
"symbol": symbol,
"positionAmt": posAmt,
@@ -353,6 +359,8 @@ func (t *OKXTrader) GetPositions() ([]map[string]interface{}, error) {
"leverage": leverage,
"liquidationPrice": liqPrice,
"side": side,
"createdTime": cTime, // Position open time (ms)
"updatedTime": uTime, // Position last update time (ms)
}
result = append(result, posMap)
}

313
trader/order_sync.go
View File

@@ -1,313 +0,0 @@
package trader
import (
"fmt"
"nofx/logger"
"nofx/store"
"sync"
"time"
)
// OrderSyncManager Order status synchronization manager
// Responsible for periodically scanning all NEW status orders and updating their status
type OrderSyncManager struct {
store *store.Store
interval time.Duration
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
}
// NewOrderSyncManager Create order synchronization manager
func NewOrderSyncManager(st *store.Store, interval time.Duration) *OrderSyncManager {
if interval == 0 {
interval = 10 * time.Second
}
return &OrderSyncManager{
store: st,
interval: interval,
stopCh: make(chan struct{}),
traderCache: make(map[string]Trader),
configCache: make(map[string]*store.TraderFullConfig),
}
}
// Start Start order synchronization service
func (m *OrderSyncManager) Start() {
m.wg.Add(1)
go m.run()
logger.Info("📦 Order sync manager started")
}
// Stop Stop order synchronization service
func (m *OrderSyncManager) 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("📦 Order sync manager stopped")
}
// run Main loop
func (m *OrderSyncManager) run() {
defer m.wg.Done()
// Execute immediately on startup
m.syncOrders()
ticker := time.NewTicker(m.interval)
defer ticker.Stop()
for {
select {
case <-m.stopCh:
return
case <-ticker.C:
m.syncOrders()
}
}
}
// syncOrders Synchronize all pending orders
func (m *OrderSyncManager) syncOrders() {
// Get all NEW status orders
orders, err := m.store.Order().GetAllPendingOrders()
if err != nil {
logger.Infof("⚠️ Failed to get pending orders: %v", err)
return
}
if len(orders) == 0 {
return
}
logger.Infof("📦 Starting to sync %d pending orders...", len(orders))
// Group by trader_id
ordersByTrader := make(map[string][]*store.TraderOrder)
for _, order := range orders {
ordersByTrader[order.TraderID] = append(ordersByTrader[order.TraderID], order)
}
// Process each trader
for traderID, traderOrders := range ordersByTrader {
m.syncTraderOrders(traderID, traderOrders)
}
}
// syncTraderOrders Synchronize orders for a single trader
func (m *OrderSyncManager) syncTraderOrders(traderID string, orders []*store.TraderOrder) {
// 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
}
for _, order := range orders {
m.syncSingleOrder(trader, order)
}
}
// syncSingleOrder Synchronize single order status
func (m *OrderSyncManager) syncSingleOrder(trader Trader, order *store.TraderOrder) {
status, err := trader.GetOrderStatus(order.Symbol, order.OrderID)
if err != nil {
// Query failed, check order creation time, assume filled after certain time
if time.Since(order.CreatedAt) > 5*time.Minute {
logger.Infof("⚠️ Order query timeout, assuming filled (ID: %s)", order.OrderID)
m.markOrderFilled(order, 0, 0, 0)
}
return
}
statusStr, _ := status["status"].(string)
switch statusStr {
case "FILLED":
avgPrice, _ := status["avgPrice"].(float64)
executedQty, _ := status["executedQty"].(float64)
commission, _ := status["commission"].(float64)
// If API doesn't return quantity, use original quantity
if executedQty == 0 {
executedQty = order.Quantity
}
m.markOrderFilled(order, avgPrice, executedQty, commission)
case "CANCELED", "EXPIRED":
order.Status = statusStr
if err := m.store.Order().Update(order); err != nil {
logger.Infof("⚠️ Failed to update order status: %v", err)
} else {
logger.Infof("📦 Order status updated: %s (ID: %s)", statusStr, order.OrderID)
}
}
}
// markOrderFilled Mark order as filled
func (m *OrderSyncManager) markOrderFilled(order *store.TraderOrder, avgPrice, executedQty, commission float64) {
// If avgPrice is 0, use order price
if avgPrice == 0 {
avgPrice = order.Price
}
if executedQty == 0 {
executedQty = order.Quantity
}
// Calculate realized PnL (only for closing orders)
var realizedPnL float64
if (order.Action == "close_long" || order.Action == "close_short") && order.EntryPrice > 0 && avgPrice > 0 {
if order.Action == "close_long" {
// Long close PnL = (close price - entry price) * quantity
realizedPnL = (avgPrice - order.EntryPrice) * executedQty
} else {
// Short close PnL = (entry price - close price) * quantity
realizedPnL = (order.EntryPrice - avgPrice) * executedQty
}
}
order.AvgPrice = avgPrice
order.ExecutedQty = executedQty
order.Status = "FILLED"
order.Fee = commission
order.RealizedPnL = realizedPnL
order.FilledAt = time.Now()
if err := m.store.Order().Update(order); err != nil {
logger.Infof("⚠️ Failed to update order status: %v", err)
} else {
if realizedPnL != 0 {
logger.Infof("✅ Order filled (ID: %s, avgPrice: %.4f, qty: %.4f, PnL: %.2f)",
order.OrderID, avgPrice, executedQty, realizedPnL)
} else {
logger.Infof("✅ Order filled (ID: %s, avgPrice: %.4f, qty: %.4f)",
order.OrderID, avgPrice, executedQty)
}
}
}
// getOrCreateTrader Get or create trader instance
func (m *OrderSyncManager) 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
// First get trader config
config, err := m.getTraderConfig(traderID)
if err != nil {
return nil, fmt.Errorf("failed to get trader config: %w", err)
}
// Create trader based on exchange type
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 *OrderSyncManager) 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 - need to find trader's corresponding userID
// First query all traders to find corresponding userID
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 *OrderSyncManager) createTrader(config *store.TraderFullConfig) (Trader, error) {
exchange := config.Exchange
// Use exchange.ID to determine specific exchange, not exchange.Type (cex/dex)
switch exchange.ID {
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 "hyperliquid":
return NewHyperliquidTrader(exchange.SecretKey, exchange.HyperliquidWalletAddr, exchange.Testnet)
case "aster":
return NewAsterTrader(exchange.AsterUser, exchange.AsterSigner, exchange.AsterPrivateKey)
case "lighter":
if exchange.LighterAPIKeyPrivateKey != "" {
return NewLighterTraderV2(
exchange.LighterPrivateKey,
exchange.LighterWalletAddr,
exchange.LighterAPIKeyPrivateKey,
exchange.Testnet,
)
}
return NewLighterTrader(exchange.LighterPrivateKey, exchange.LighterWalletAddr, exchange.Testnet)
default:
return nil, fmt.Errorf("unsupported exchange: %s", exchange.ID)
}
}
// InvalidateCache Invalidate cache (call when configuration changes)
func (m *OrderSyncManager) InvalidateCache(traderID string) {
m.cacheMutex.Lock()
defer m.cacheMutex.Unlock()
delete(m.traderCache, traderID)
delete(m.configCache, traderID)
}