Merge pull request #1361 from c9s/feature/grid2/recover-preparation-function

FEATURE: prepare query trades funtion for new recover

pkg/strategy/grid2/active_order_recover.go

@@ -2,7 +2,6 @@ package grid2
 
 import (
 	"context"
-	"strconv"
 	"time"
 
 	"github.com/c9s/bbgo/pkg/bbgo"
@@ -142,18 +141,3 @@ func syncActiveOrders(ctx context.Context, opts SyncActiveOrdersOpts) error {
 
 	return errs
 }
-
-func syncActiveOrder(ctx context.Context, activeOrderBook *bbgo.ActiveOrderBook, orderQueryService types.ExchangeOrderQueryService, orderID uint64) error {
-	updatedOrder, err := retry.QueryOrderUntilSuccessful(ctx, orderQueryService, types.OrderQuery{
-		Symbol:  activeOrderBook.Symbol,
-		OrderID: strconv.FormatUint(orderID, 10),
-	})
-
-	if err != nil {
-		return err
-	}
-
-	activeOrderBook.Update(*updatedOrder)
-
-	return nil
-}

pkg/strategy/grid2/active_order_recover_test.go

@@ -174,77 +174,3 @@ func TestSyncActiveOrders(t *testing.T) {
 		assert.Equal(types.OrderStatusNew, activeOrders[0].Status)
 	})
 }
-
-func TestSyncActiveOrder(t *testing.T) {
-	assert := assert.New(t)
-
-	mockCtrl := gomock.NewController(t)
-	defer mockCtrl.Finish()
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	symbol := "ETHUSDT"
-
-	t.Run("sync filled order in active orderbook, active orderbook should remove this order", func(t *testing.T) {
-		mockOrderQueryService := mocks.NewMockExchangeOrderQueryService(mockCtrl)
-		activeOrderbook := bbgo.NewActiveOrderBook(symbol)
-
-		order := types.Order{
-			OrderID: 1,
-			Status:  types.OrderStatusNew,
-			SubmitOrder: types.SubmitOrder{
-				Symbol: symbol,
-			},
-		}
-		activeOrderbook.Add(order)
-
-		updatedOrder := order
-		updatedOrder.Status = types.OrderStatusFilled
-
-		mockOrderQueryService.EXPECT().QueryOrder(ctx, types.OrderQuery{
-			Symbol:  symbol,
-			OrderID: strconv.FormatUint(order.OrderID, 10),
-		}).Return(&updatedOrder, nil)
-
-		if !assert.NoError(syncActiveOrder(ctx, activeOrderbook, mockOrderQueryService, order.OrderID)) {
-			return
-		}
-
-		// verify active orderbook
-		activeOrders := activeOrderbook.Orders()
-		assert.Equal(0, len(activeOrders))
-	})
-
-	t.Run("sync partial-filled order in active orderbook, active orderbook should still keep this order", func(t *testing.T) {
-		mockOrderQueryService := mocks.NewMockExchangeOrderQueryService(mockCtrl)
-		activeOrderbook := bbgo.NewActiveOrderBook(symbol)
-
-		order := types.Order{
-			OrderID: 1,
-			Status:  types.OrderStatusNew,
-			SubmitOrder: types.SubmitOrder{
-				Symbol: symbol,
-			},
-		}
-		activeOrderbook.Add(order)
-
-		updatedOrder := order
-		updatedOrder.Status = types.OrderStatusPartiallyFilled
-
-		mockOrderQueryService.EXPECT().QueryOrder(ctx, types.OrderQuery{
-			Symbol:  symbol,
-			OrderID: strconv.FormatUint(order.OrderID, 10),
-		}).Return(&updatedOrder, nil)
-
-		if !assert.NoError(syncActiveOrder(ctx, activeOrderbook, mockOrderQueryService, order.OrderID)) {
-			return
-		}
-
-		// verify active orderbook
-		activeOrders := activeOrderbook.Orders()
-		assert.Equal(1, len(activeOrders))
-		assert.Equal(order.OrderID, activeOrders[0].OrderID)
-		assert.Equal(updatedOrder.Status, activeOrders[0].Status)
-	})
-}

pkg/strategy/grid2/recover.go

@@ -0,0 +1,143 @@
+package grid2
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/c9s/bbgo/pkg/bbgo"
+	"github.com/c9s/bbgo/pkg/exchange/retry"
+	"github.com/c9s/bbgo/pkg/types"
+	"github.com/pkg/errors"
+)
+
+/*
+  	Background knowledge
+  	1. active orderbook add orders only when receive new order event or call Add/Update method manually
+  	2. active orderbook remove orders only when receive filled/cancelled event or call Remove/Update method manually
+  	As a result
+  	1. at the same twin-order-price, there is order in open orders but not in active orderbook
+  		- not receive new order event
+		  	=> add order into active orderbook
+  	2. at the same twin-order-price, there is order in active orderbook but not in open orders
+  		- not receive filled event
+			=> query the filled order and call Update method
+  	3. at the same twin-order-price, there is no order in open orders and no order in active orderbook
+		- failed to create the order
+			=> query the last order from trades to emit filled, and it will submit again
+		- not receive new order event and the order filled before we find it.
+			=> query the untracked order (also is the last order) from trades to emit filled and it will submit the reversed order
+  	4. at the same twin-order-price, there are different orders in open orders and active orderbook
+	  	- should not happen !!!
+		  	=> log error
+	5. at the same twin-order-price, there is the same order in open orders and active orderbook
+		- normal case
+			=> no need to do anything
+	After killing pod, active orderbook must be empty. we can think it is the same as not receive new event.
+	Process
+	1. build twin orderbook with pins and open orders.
+	2. build twin orderbook with pins and active orders.
+	3. compare above twin orderbooks to add open orders into active orderbook and update active orders.
+	4. run grid recover to make sure all the twin price has its order.
+*/
+
+func buildTwinOrderBook(pins []Pin, orders []types.Order) (*TwinOrderBook, error) {
+	book := newTwinOrderBook(pins)
+
+	for _, order := range orders {
+		if err := book.AddOrder(order); err != nil {
+			return nil, err
+		}
+	}
+
+	return book, nil
+}
+
+func syncActiveOrder(ctx context.Context, activeOrderBook *bbgo.ActiveOrderBook, orderQueryService types.ExchangeOrderQueryService, orderID uint64) error {
+	updatedOrder, err := retry.QueryOrderUntilSuccessful(ctx, orderQueryService, types.OrderQuery{
+		Symbol:  activeOrderBook.Symbol,
+		OrderID: strconv.FormatUint(orderID, 10),
+	})
+
+	if err != nil {
+		return err
+	}
+
+	activeOrderBook.Update(*updatedOrder)
+
+	return nil
+}
+
+func queryTradesToUpdateTwinOrderBook(
+	ctx context.Context,
+	symbol string,
+	twinOrderBook *TwinOrderBook,
+	queryTradesService types.ExchangeTradeHistoryService,
+	queryOrderService types.ExchangeOrderQueryService,
+	existedOrders *types.SyncOrderMap,
+	since, until time.Time,
+	logger func(format string, args ...interface{})) error {
+	if twinOrderBook == nil {
+		return fmt.Errorf("twin orderbook should not be nil, please check it")
+	}
+
+	var fromTradeID uint64 = 0
+	var limit int64 = 1000
+	for {
+		trades, err := queryTradesService.QueryTrades(ctx, symbol, &types.TradeQueryOptions{
+			StartTime:   &since,
+			EndTime:     &until,
+			LastTradeID: fromTradeID,
+			Limit:       limit,
+		})
+
+		if err != nil {
+			return errors.Wrapf(err, "failed to query trades to recover the grid")
+		}
+
+		if logger != nil {
+			logger("QueryTrades from %s <-> %s (from: %d) return %d trades", since, until, fromTradeID, len(trades))
+		}
+
+		for _, trade := range trades {
+			if trade.Time.After(until) {
+				return nil
+			}
+
+			if logger != nil {
+				logger(trade.String())
+			}
+
+			if existedOrders.Exists(trade.OrderID) {
+				// already queries, skip
+				continue
+			}
+			order, err := retry.QueryOrderUntilSuccessful(ctx, queryOrderService, types.OrderQuery{
+				Symbol:  trade.Symbol,
+				OrderID: strconv.FormatUint(trade.OrderID, 10),
+			})
+
+			if err != nil {
+				return errors.Wrapf(err, "failed to query order by trade (trade id: %d, order id: %d)", trade.ID, trade.OrderID)
+			}
+
+			if logger != nil {
+				logger(order.String())
+			}
+			// avoid query this order again
+			existedOrders.Add(*order)
+			// add 1 to avoid duplicate
+			fromTradeID = trade.ID + 1
+
+			if err := twinOrderBook.AddOrder(*order); err != nil {
+				return errors.Wrapf(err, "failed to add queried order into twin orderbook")
+			}
+		}
+
+		// stop condition
+		if int64(len(trades)) < limit {
+			return nil
+		}
+	}
+}

pkg/strategy/grid2/recover_test.go

@@ -0,0 +1,237 @@
+package grid2
+
+import (
+	"context"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/c9s/bbgo/pkg/bbgo"
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+	"github.com/c9s/bbgo/pkg/types/mocks"
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+)
+
+func newStrategy(t *TestData) *Strategy {
+	s := t.Strategy
+	s.Debug = true
+	s.Initialize()
+	s.Market = t.Market
+	s.Position = types.NewPositionFromMarket(t.Market)
+	s.orderExecutor = bbgo.NewGeneralOrderExecutor(&bbgo.ExchangeSession{}, t.Market.Symbol, ID, s.InstanceID(), s.Position)
+	return &s
+}
+
+func TestBuildTwinOrderBook(t *testing.T) {
+	assert := assert.New(t)
+
+	pins := []Pin{
+		Pin(fixedpoint.NewFromInt(200)),
+		Pin(fixedpoint.NewFromInt(300)),
+		Pin(fixedpoint.NewFromInt(500)),
+		Pin(fixedpoint.NewFromInt(400)),
+		Pin(fixedpoint.NewFromInt(100)),
+	}
+	t.Run("build twin orderbook with no order", func(t *testing.T) {
+		b, err := buildTwinOrderBook(pins, nil)
+		if !assert.NoError(err) {
+			return
+		}
+
+		assert.Equal(0, b.Size())
+		assert.Nil(b.GetTwinOrder(fixedpoint.NewFromInt(100)))
+		assert.False(b.GetTwinOrder(fixedpoint.NewFromInt(200)).Exist())
+		assert.False(b.GetTwinOrder(fixedpoint.NewFromInt(300)).Exist())
+		assert.False(b.GetTwinOrder(fixedpoint.NewFromInt(400)).Exist())
+		assert.False(b.GetTwinOrder(fixedpoint.NewFromInt(500)).Exist())
+	})
+
+	t.Run("build twin orderbook with some valid orders", func(t *testing.T) {
+		orders := []types.Order{
+			{
+				OrderID: 1,
+				SubmitOrder: types.SubmitOrder{
+					Side:  types.SideTypeBuy,
+					Price: fixedpoint.NewFromInt(100),
+				},
+			},
+			{
+				OrderID: 5,
+				SubmitOrder: types.SubmitOrder{
+					Side:  types.SideTypeSell,
+					Price: fixedpoint.NewFromInt(500),
+				},
+			},
+		}
+		b, err := buildTwinOrderBook(pins, orders)
+		if !assert.NoError(err) {
+			return
+		}
+
+		assert.Equal(2, b.Size())
+		assert.Equal(2, b.EmptyTwinOrderSize())
+		assert.Nil(b.GetTwinOrder(fixedpoint.NewFromInt(100)))
+		assert.True(b.GetTwinOrder(fixedpoint.NewFromInt(200)).Exist())
+		assert.False(b.GetTwinOrder(fixedpoint.NewFromInt(300)).Exist())
+		assert.False(b.GetTwinOrder(fixedpoint.NewFromInt(400)).Exist())
+		assert.True(b.GetTwinOrder(fixedpoint.NewFromInt(500)).Exist())
+	})
+
+	t.Run("build twin orderbook with invalid orders", func(t *testing.T) {})
+}
+
+func TestSyncActiveOrder(t *testing.T) {
+	assert := assert.New(t)
+
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	symbol := "ETHUSDT"
+
+	t.Run("sync filled order in active orderbook, active orderbook should remove this order", func(t *testing.T) {
+		mockOrderQueryService := mocks.NewMockExchangeOrderQueryService(mockCtrl)
+		activeOrderbook := bbgo.NewActiveOrderBook(symbol)
+
+		order := types.Order{
+			OrderID: 1,
+			Status:  types.OrderStatusNew,
+			SubmitOrder: types.SubmitOrder{
+				Symbol: symbol,
+			},
+		}
+		activeOrderbook.Add(order)
+
+		updatedOrder := order
+		updatedOrder.Status = types.OrderStatusFilled
+
+		mockOrderQueryService.EXPECT().QueryOrder(ctx, types.OrderQuery{
+			Symbol:  symbol,
+			OrderID: strconv.FormatUint(order.OrderID, 10),
+		}).Return(&updatedOrder, nil)
+
+		if !assert.NoError(syncActiveOrder(ctx, activeOrderbook, mockOrderQueryService, order.OrderID)) {
+			return
+		}
+
+		// verify active orderbook
+		activeOrders := activeOrderbook.Orders()
+		assert.Equal(0, len(activeOrders))
+	})
+
+	t.Run("sync partial-filled order in active orderbook, active orderbook should still keep this order", func(t *testing.T) {
+		mockOrderQueryService := mocks.NewMockExchangeOrderQueryService(mockCtrl)
+		activeOrderbook := bbgo.NewActiveOrderBook(symbol)
+
+		order := types.Order{
+			OrderID: 1,
+			Status:  types.OrderStatusNew,
+			SubmitOrder: types.SubmitOrder{
+				Symbol: symbol,
+			},
+		}
+		activeOrderbook.Add(order)
+
+		updatedOrder := order
+		updatedOrder.Status = types.OrderStatusPartiallyFilled
+
+		mockOrderQueryService.EXPECT().QueryOrder(ctx, types.OrderQuery{
+			Symbol:  symbol,
+			OrderID: strconv.FormatUint(order.OrderID, 10),
+		}).Return(&updatedOrder, nil)
+
+		if !assert.NoError(syncActiveOrder(ctx, activeOrderbook, mockOrderQueryService, order.OrderID)) {
+			return
+		}
+
+		// verify active orderbook
+		activeOrders := activeOrderbook.Orders()
+		assert.Equal(1, len(activeOrders))
+		assert.Equal(order.OrderID, activeOrders[0].OrderID)
+		assert.Equal(updatedOrder.Status, activeOrders[0].Status)
+	})
+}
+
+func TestQueryTradesToUpdateTwinOrderBook(t *testing.T) {
+	assert := assert.New(t)
+
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	symbol := "ETHUSDT"
+	pins := []Pin{
+		Pin(fixedpoint.NewFromInt(100)),
+		Pin(fixedpoint.NewFromInt(200)),
+		Pin(fixedpoint.NewFromInt(300)),
+		Pin(fixedpoint.NewFromInt(400)),
+		Pin(fixedpoint.NewFromInt(500)),
+	}
+
+	t.Run("query trades and update twin orderbook successfully in one page", func(t *testing.T) {
+		book := newTwinOrderBook(pins)
+		mockTradeHistoryService := mocks.NewMockExchangeTradeHistoryService(mockCtrl)
+		mockOrderQueryService := mocks.NewMockExchangeOrderQueryService(mockCtrl)
+
+		trades := []types.Trade{
+			{
+				ID:      1,
+				OrderID: 1,
+				Symbol:  symbol,
+				Time:    types.Time(time.Now().Add(-2 * time.Hour)),
+			},
+			{
+				ID:      2,
+				OrderID: 2,
+				Symbol:  symbol,
+				Time:    types.Time(time.Now().Add(-1 * time.Hour)),
+			},
+		}
+		orders := []types.Order{
+			{
+				OrderID: 1,
+				Status:  types.OrderStatusNew,
+				SubmitOrder: types.SubmitOrder{
+					Symbol: symbol,
+					Side:   types.SideTypeBuy,
+					Price:  fixedpoint.NewFromInt(100),
+				},
+			},
+			{
+				OrderID: 2,
+				Status:  types.OrderStatusFilled,
+				SubmitOrder: types.SubmitOrder{
+					Symbol: symbol,
+					Side:   types.SideTypeSell,
+					Price:  fixedpoint.NewFromInt(500),
+				},
+			},
+		}
+		mockTradeHistoryService.EXPECT().QueryTrades(gomock.Any(), gomock.Any(), gomock.Any()).Return(trades, nil).Times(1)
+		mockOrderQueryService.EXPECT().QueryOrder(gomock.Any(), types.OrderQuery{
+			Symbol:  symbol,
+			OrderID: "1",
+		}).Return(&orders[0], nil)
+		mockOrderQueryService.EXPECT().QueryOrder(gomock.Any(), types.OrderQuery{
+			Symbol:  symbol,
+			OrderID: "2",
+		}).Return(&orders[1], nil)
+
+		assert.Equal(0, book.Size())
+		if !assert.NoError(queryTradesToUpdateTwinOrderBook(ctx, symbol, book, mockTradeHistoryService, mockOrderQueryService, book.SyncOrderMap(), time.Now().Add(-24*time.Hour), time.Now(), nil)) {
+			return
+		}
+
+		assert.Equal(2, book.Size())
+		assert.True(book.GetTwinOrder(fixedpoint.NewFromInt(200)).Exist())
+		assert.Equal(orders[0].OrderID, book.GetTwinOrder(fixedpoint.NewFromInt(200)).GetOrder().OrderID)
+		assert.True(book.GetTwinOrder(fixedpoint.NewFromInt(500)).Exist())
+		assert.Equal(orders[1].OrderID, book.GetTwinOrder(fixedpoint.NewFromInt(500)).GetOrder().OrderID)
+	})
+}

pkg/strategy/grid2/twin_order.go

@@ -154,7 +154,10 @@ func newTwinOrderBook(pins []Pin) *TwinOrderBook {
 	pinIdx := make(map[fixedpoint.Value]int)
 	m := make(map[fixedpoint.Value]*TwinOrder)
 	for i, pin := range v {
+		// we use sell price for twin orderbook's price, so we skip the first pin as price
+		if i > 0 {
 			m[pin] = &TwinOrder{}
+		}
 		pinIdx[pin] = i
 	}
 

pkg/strategy/grid2/twin_order_test.go

@@ -23,6 +23,11 @@ func TestTwinOrderBook(t *testing.T) {
 	assert.Equal(4, book.EmptyTwinOrderSize())
 	for _, pin := range pins {
 		twinOrder := book.GetTwinOrder(fixedpoint.Value(pin))
+		if fixedpoint.NewFromInt(1) == fixedpoint.Value(pin) {
+			assert.Nil(twinOrder)
+			continue
+		}
+
 		if !assert.NotNil(twinOrder) {
 			continue
 		}