xmaker: sync order status after sending hedge order from the executor

Author: c9s

pkg/core/orderstore.go

@@ -111,7 +111,9 @@ func (s *OrderStore) Add(orders ...types.Order) {
 	for _, o := range orders {
 		if o.OrderID == 0 {
 			logrus.WithFields(o.LogFields()).Errorf("[orderstore] adding order %+v with OrderID 0", o)
+			continue
 		}
+
 		old, ok := s.orders[o.OrderID]
 		if ok && o.Tag == "" && old.Tag != "" {
 			o.Tag = old.Tag
@@ -126,6 +128,7 @@ func (s *OrderStore) Remove(o types.Order) {
 
 	if o.OrderID == 0 {
 		logrus.WithFields(o.LogFields()).Errorf("[orderstore-Remove] given order %+v with OrderID 0", o)
+		return
 	}
 
 	delete(s.orders, o.OrderID)

pkg/core/tradecollector.go

@@ -302,7 +302,6 @@ func (c *TradeCollector) processTrade(trade types.Trade) bool {
 
 	c.mu.Lock()
 
-	// if it's already done, remove the trade from the trade store
 	if _, done := c.doneTrades[key]; done {
 		c.mu.Unlock()
 		return false

pkg/strategy/xmaker/hedgeexecutor.go

@@ -122,6 +122,44 @@ func (m *MarketOrderHedgeExecutor) Hedge(
 	}
 
 	m.logger.Infof("hedge order created: %+v", hedgeOrder)
+
+	if hedgeOrder != nil {
+		updatedOrder, err := m.syncOrder(ctx, *hedgeOrder)
+		if err != nil {
+			m.logger.WithError(err).WithFields(hedgeOrder.LogFields()).Errorf("failed to sync order: %+v", hedgeOrder)
+		} else if updatedOrder != nil {
+			orderLogger := m.logger.WithFields(updatedOrder.LogFields())
+			// Compare the order quantity, if the order quantity is changed from the server side
+			// we should uncover the difference
+			if !updatedOrder.Quantity.Eq(hedgeOrder.Quantity) {
+				// if we sent sell 1 BTC (-1 BTC), but the order was adjusted to sell 0.1 BTC (-0.1 BTC)
+				// we should uncover +0.9 BTC to the position exposure (meaning -0.9 BTC the the covered position)
+
+				// Net Position: 1.0, Pending: 0.0
+				// Submit Order { Quantity: 1, Side: Sell }
+				// Net Position: 1.0, Pending: 1.0
+				// Order Updated { Quantity: 0.1, Side: Sell }
+				// to quantityToDelta = -(-0.9) (Sell) = +0.9
+				// to coverDeltaDiff = +0.9
+				// to uncover = 0.9 = cover -0.9
+				// Net Position: 1.0, Pending: 0.1
+				// Covered Position: 0.1
+				// Uncover: 0.9
+				quantityDiff := updatedOrder.Quantity.Sub(hedgeOrder.Quantity)
+
+				orderLogger.Infof("hedge order quantity changed from %s to %s, adjusting covered position by %s",
+					hedgeOrder.Quantity.String(), updatedOrder.Quantity.String(), quantityDiff.String())
+
+				if !quantityDiff.IsZero() {
+					coverDeltaDiff := quantityToDelta(quantityDiff, side)
+					m.positionExposure.Uncover(coverDeltaDiff)
+					orderLogger.Infof("hedge order quantity changed from %s to %s, adjusting covered position by %s",
+						hedgeOrder.Quantity.String(), updatedOrder.Quantity.String(), coverDeltaDiff.String())
+				}
+			}
+		}
+	}
+
 	return nil
 }
 

pkg/strategy/xmaker/hedgemarket.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/core"
+	"github.com/c9s/bbgo/pkg/exchange/retry"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/c9s/bbgo/pkg/service"
 	"github.com/c9s/bbgo/pkg/strategy/xmaker/pricer"
@@ -133,6 +134,8 @@ type HedgeMarket struct {
 	market  types.Market
 	stream  types.Stream
 
+	orderQueryService types.ExchangeOrderQueryService
+
 	connectivity *types.Connectivity
 
 	book      *types.StreamOrderBook
@@ -210,6 +213,8 @@ func NewHedgeMarket(
 	}
 	logger := log.WithFields(logFields)
 
+	orderQueryService, _ := session.Exchange.(types.ExchangeOrderQueryService)
+
 	m := &HedgeMarket{
 		HedgeMarketConfig: config,
 		session:           session,
@@ -218,6 +223,8 @@ func NewHedgeMarket(
 		book:              book,
 		depthBook:         depthBook,
 
+		orderQueryService: orderQueryService,
+
 		connectivity: connectivity,
 
 		positionExposure: NewPositionExposure(symbol),
@@ -268,6 +275,46 @@ func (m *HedgeMarket) SetLogger(logger logrus.FieldLogger) {
 	})
 }
 
+func (m *HedgeMarket) syncOrder(ctx context.Context, order types.Order) (*types.Order, error) {
+	updatedOrder, err := retry.QueryOrderUntilSuccessful(ctx, m.orderQueryService, order.AsQuery())
+	if err != nil {
+		return nil, fmt.Errorf("unable to query order #%d: %w", order.OrderID, err)
+	} else if updatedOrder != nil {
+		if updatedOrder.Quantity.Compare(order.Quantity) != 0 {
+			m.logger.WithFields(order.LogFields()).
+				Warnf("order quantity changed from %s to %s for order #%d",
+					order.Quantity.String(), updatedOrder.Quantity.String(), order.OrderID)
+		}
+	}
+
+	orderTrades, err := retry.QueryOrderTradesUntilSuccessful(ctx, m.orderQueryService, order.AsQuery())
+	if err != nil {
+		return updatedOrder, fmt.Errorf("unable to query order #%d trades: %w", order.OrderID, err)
+	}
+
+	for _, trade := range orderTrades {
+		m.tradeCollector.ProcessTrade(trade)
+	}
+
+	return updatedOrder, nil
+}
+
+func (m *HedgeMarket) syncHistoryOrder(ctx context.Context) {
+	historyOrders := m.orderStore.Orders()
+	m.logger.Infof("loaded %d historical orders", len(historyOrders))
+
+	for _, order := range historyOrders {
+		updatedOrder, err := m.syncOrder(ctx, order)
+		if err != nil {
+			m.logger.WithError(err).WithFields(order.LogFields()).Warnf("failed to sync order")
+		} else if updatedOrder != nil {
+			m.orderStore.Update(*updatedOrder)
+		}
+	}
+
+	m.orderStore.Prune(8 * time.Hour)
+}
+
 // SetPriceFeeMode sets the fee mode used when computing hedge quote prices.
 // This method is safe to call at runtime.
 func (m *HedgeMarket) SetPriceFeeMode(mode FeeMode) {
@@ -646,6 +693,17 @@ func (m *HedgeMarket) calculateDebtQuota(totalValue, debtValue, minMarginLevel,
 	return debtQuota
 }
 
+// Send sends the position delta to the hedge market.
+func (m *HedgeMarket) Send(delta fixedpoint.Value) {
+	select {
+	case m.positionDeltaC <- delta:
+	case <-time.After(30 * time.Second):
+		m.logger.Warnf("position delta channel is full, dropping delta: %f", delta.Float64())
+	}
+}
+
+// hedge executes the hedge logic to adjust the position exposure to zero.
+// you should pass the position delta to the positionDeltaC channel
 func (m *HedgeMarket) hedge(ctx context.Context) error {
 	if err := m.hedgeExecutor.Clear(ctx); err != nil {
 		return fmt.Errorf("failed to clear hedge executor: %w", err)