Merge pull request #1060 from c9s/fix/active-order-book-pending-order-update

fix: process pending order update for active order book

Author: c9s

pkg/bbgo/activeorderbook.go

@@ -25,16 +25,19 @@ type ActiveOrderBook struct {
 	filledCallbacks   []func(o types.Order)
 	canceledCallbacks []func(o types.Order)
 
+	pendingOrderUpdates *types.SyncOrderMap
+
 	// sig is the order update signal
 	// this signal will be emitted when a new order is added or removed.
 	C sigchan.Chan
 }
 
 func NewActiveOrderBook(symbol string) *ActiveOrderBook {
 	return &ActiveOrderBook{
-		Symbol: symbol,
-		orders: types.NewSyncOrderMap(),
-		C:      sigchan.New(1),
+		Symbol:              symbol,
+		orders:              types.NewSyncOrderMap(),
+		pendingOrderUpdates: types.NewSyncOrderMap(),
+		C:                   sigchan.New(1),
 	}
 }
 
@@ -230,6 +233,11 @@ func (b *ActiveOrderBook) orderUpdateHandler(order types.Order) {
 		return
 	}
 
+	if !b.orders.Exists(order.OrderID) {
+		b.pendingOrderUpdates.Add(order)
+		return
+	}
+
 	switch order.Status {
 	case types.OrderStatusFilled:
 		// make sure we have the order and we remove it
@@ -277,25 +285,48 @@ func (b *ActiveOrderBook) Print() {
 func (b *ActiveOrderBook) Update(orders ...types.Order) {
 	hasSymbol := len(b.Symbol) > 0
 	for _, order := range orders {
-		if hasSymbol && b.Symbol == order.Symbol {
-			b.orders.Update(order)
+		if hasSymbol && b.Symbol != order.Symbol {
+			continue
 		}
+
+		b.orders.Update(order)
 	}
 }
 
 func (b *ActiveOrderBook) Add(orders ...types.Order) {
 	hasSymbol := len(b.Symbol) > 0
 	for _, order := range orders {
-		if hasSymbol && b.Symbol == order.Symbol {
+		if hasSymbol && b.Symbol != order.Symbol {
+			continue
+		}
+
+		b.add(order)
+	}
+}
+
+// add the order to the active order book and check the pending order
+func (b *ActiveOrderBook) add(order types.Order) {
+	if pendingOrder, ok := b.pendingOrderUpdates.Get(order.OrderID); ok {
+		if pendingOrder.UpdateTime.Time().After(order.UpdateTime.Time()) {
+			b.orders.Add(pendingOrder)
+		} else {
 			b.orders.Add(order)
 		}
+
+		b.pendingOrderUpdates.Remove(order.OrderID)
+	} else {
+		b.orders.Add(order)
 	}
 }
 
 func (b *ActiveOrderBook) Exists(order types.Order) bool {
 	return b.orders.Exists(order.OrderID)
 }
 
+func (b *ActiveOrderBook) Get(orderID uint64) (types.Order, bool) {
+	return b.orders.Get(orderID)
+}
+
 func (b *ActiveOrderBook) Remove(order types.Order) bool {
 	return b.orders.Remove(order.OrderID)
 }

pkg/bbgo/activeorderbook_test.go

@@ -0,0 +1,57 @@
+package bbgo
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+func TestActiveOrderBook_pendingOrders(t *testing.T) {
+	now := time.Now()
+	ob := NewActiveOrderBook("")
+
+	// if we received filled order first
+	// should be added to pending orders
+	ob.orderUpdateHandler(types.Order{
+		OrderID: 99,
+		SubmitOrder: types.SubmitOrder{
+			Symbol:       "BTCUSDT",
+			Side:         types.SideTypeBuy,
+			Type:         types.OrderTypeLimit,
+			Quantity:     number(0.01),
+			Price:        number(19000.0),
+			AveragePrice: fixedpoint.Zero,
+			StopPrice:    fixedpoint.Zero,
+		},
+		Status:       types.OrderStatusFilled,
+		CreationTime: types.Time(now),
+		UpdateTime:   types.Time(now),
+	})
+
+	assert.Len(t, ob.pendingOrderUpdates.Orders(), 1)
+
+	// should be added to pending orders
+	ob.Add(types.Order{
+		OrderID: 99,
+		SubmitOrder: types.SubmitOrder{
+			Symbol:       "BTCUSDT",
+			Side:         types.SideTypeBuy,
+			Type:         types.OrderTypeLimit,
+			Quantity:     number(0.01),
+			Price:        number(19000.0),
+			AveragePrice: fixedpoint.Zero,
+			StopPrice:    fixedpoint.Zero,
+		},
+		Status:       types.OrderStatusNew,
+		CreationTime: types.Time(now),
+		UpdateTime:   types.Time(now.Add(-time.Second)),
+	})
+
+	o99, ok := ob.Get(99)
+	assert.True(t, ok)
+	assert.Equal(t, types.OrderStatusFilled, o99.Status)
+}

pkg/bbgo/order_execution.go

@@ -62,6 +62,33 @@ func BatchRetryPlaceOrder(ctx context.Context, exchange types.Exchange, errIdx [
 	return createdOrders, err
 }
 
+// BatchPlaceOrderChan post orders with a channel, the created order will be sent to this channel immediately, so that
+// the caller can add the created order to the active order book or the order store to collect trades.
+// this method is used when you have large amount of orders to be sent and most of the orders might be filled as taker order.
+// channel orderC will be closed when all the submit orders are submitted.
+func BatchPlaceOrderChan(ctx context.Context, exchange types.Exchange, orderC chan types.Order, submitOrders ...types.SubmitOrder) (types.OrderSlice, []int, error) {
+	defer close(orderC)
+
+	var createdOrders types.OrderSlice
+	var err error
+	var errIndexes []int
+	for i, submitOrder := range submitOrders {
+		createdOrder, err2 := exchange.SubmitOrder(ctx, submitOrder)
+		if err2 != nil {
+			err = multierr.Append(err, err2)
+			errIndexes = append(errIndexes, i)
+		} else if createdOrder != nil {
+			createdOrder.Tag = submitOrder.Tag
+
+			orderC <- *createdOrder
+
+			createdOrders = append(createdOrders, *createdOrder)
+		}
+	}
+
+	return createdOrders, errIndexes, err
+}
+
 // BatchPlaceOrder
 func BatchPlaceOrder(ctx context.Context, exchange types.Exchange, submitOrders ...types.SubmitOrder) (types.OrderSlice, []int, error) {
 	var createdOrders types.OrderSlice

pkg/types/ordermap.go

@@ -56,6 +56,11 @@ func (m OrderMap) Exists(orderID uint64) bool {
 	return ok
 }
 
+func (m OrderMap) Get(orderID uint64) (Order, bool) {
+	order, ok := m[orderID]
+	return order, ok
+}
+
 func (m OrderMap) FindByStatus(status OrderStatus) (orders OrderSlice) {
 	for _, o := range m {
 		if o.Status == status {
@@ -165,6 +170,13 @@ func (m *SyncOrderMap) Exists(orderID uint64) (exists bool) {
 	return exists
 }
 
+func (m *SyncOrderMap) Get(orderID uint64) (Order, bool) {
+	m.Lock()
+	order, ok := m.orders.Get(orderID)
+	m.Unlock()
+	return order, ok
+}
+
 func (m *SyncOrderMap) Lookup(f func(o Order) bool) *Order {
 	m.Lock()
 	defer m.Unlock()