bbgo_origin/pkg/strategy/grid2/recover.go

package grid2

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"github.com/pkg/errors"

	"github.com/c9s/bbgo/pkg/bbgo"
	"github.com/c9s/bbgo/pkg/fixedpoint"
	"github.com/c9s/bbgo/pkg/types"
)

func (s *Strategy) recoverByScanningTrades(ctx context.Context, session *bbgo.ExchangeSession) error {
	defer func() {
		s.updateGridNumOfOrdersMetricsWithLock()
		s.updateOpenOrderPricesMetrics(s.orderExecutor.ActiveMakerOrders().Orders())
	}()

	historyService, implemented := session.Exchange.(types.ExchangeTradeHistoryService)
	// if the exchange doesn't support ExchangeTradeHistoryService, do not run recover
	if !implemented {
		s.logger.Warn("ExchangeTradeHistoryService is not implemented, can not recover grid")
		return nil
	}

	openOrders, err := session.Exchange.QueryOpenOrders(ctx, s.Symbol)
	if err != nil {
		return errors.Wrapf(err, "unable to query open orders when recovering")
	}

	s.logger.Infof("found %d open orders left on the %s order book", len(openOrders), s.Symbol)

	if s.GridProfitStats.InitialOrderID != 0 {
		s.logger.Info("InitialOrderID is already there, need to recover")
	} else if len(openOrders) != 0 {
		s.logger.Info("even though InitialOrderID is 0, there are open orders so need to recover")
	} else {
		s.logger.Info("InitialOrderID is 0 and there is no open orders, query trades to check it")
		// initial order id may be new strategy or lost data in redis, so we need to check trades + open orders
		// if there are open orders or trades, we need to recover
		trades, err := historyService.QueryTrades(ctx, s.Symbol, &types.TradeQueryOptions{
			// from 1, because some API will ignore 0 last trade id
			LastTradeID: 1,
			// if there is any trades, we need to recover.
			Limit: 1,
		})

		if err != nil {
			return errors.Wrapf(err, "unable to query trades when recovering")
		}

		if len(trades) == 0 {
			s.logger.Info("0 trades found, it's a new strategy so no need to recover")
			return nil
		}
	}

	s.logger.Infof("start recovering")
	filledOrders, err := s.getFilledOrdersByScanningTrades(ctx, historyService, s.orderQueryService, openOrders)
	if err != nil {
		return errors.Wrap(err, "grid recover error")
	}
	s.debugOrders("emit filled orders", filledOrders)

	// add open orders into avtive maker orders
	s.addOrdersToActiveOrderBook(openOrders)

	// emit the filled orders
	activeOrderBook := s.orderExecutor.ActiveMakerOrders()
	for _, filledOrder := range filledOrders {
		activeOrderBook.EmitFilled(filledOrder)
	}

	// emit ready after recover
	s.EmitGridReady()

	// debug and send metrics
	// wait for the reverse order to be placed
	time.Sleep(2 * time.Second)
	debugGrid(s.logger, s.grid, s.orderExecutor.ActiveMakerOrders())

	defer bbgo.Sync(ctx, s)

	if s.EnableProfitFixer {
		until := time.Now()
		since := until.Add(-7 * 24 * time.Hour)
		if s.FixProfitSince != nil {
			since = s.FixProfitSince.Time()
		}

		fixer := newProfitFixer(s.grid, s.Symbol, historyService)
		fixer.SetLogger(s.logger)

		// set initial order ID = 0 instead of s.GridProfitStats.InitialOrderID because the order ID could be incorrect
		if err := fixer.Fix(ctx, since, until, 0, s.GridProfitStats); err != nil {
			return err
		}

		s.logger.Infof("fixed profitStats: %#v", s.GridProfitStats)

		s.EmitGridProfit(s.GridProfitStats, nil)
	}

	return nil
}

func (s *Strategy) getFilledOrdersByScanningTrades(ctx context.Context, queryTradesService types.ExchangeTradeHistoryService, queryOrderService types.ExchangeOrderQueryService, openOrdersOnGrid []types.Order) ([]types.Order, error) {
	// set grid
	grid := s.newGrid()
	s.setGrid(grid)

	expectedNumOfOrders := s.GridNum - 1
	numGridOpenOrders := int64(len(openOrdersOnGrid))
	s.debugLog("open orders nums: %d, expected nums: %d", numGridOpenOrders, expectedNumOfOrders)
	if expectedNumOfOrders == numGridOpenOrders {
		// no need to recover, only need to add open orders back to active order book
		return nil, nil
	} else if expectedNumOfOrders < numGridOpenOrders {
		return nil, fmt.Errorf("amount of grid's open orders should not > amount of expected grid's orders")
	}

	// 1. build twin-order map
	twinOrdersOpen, err := s.buildTwinOrderMap(grid.Pins, openOrdersOnGrid)
	if err != nil {
		return nil, errors.Wrapf(err, "failed to build pin order map with open orders")
	}

	// 2. build the filled twin-order map by querying trades
	expectedFilledNum := int(expectedNumOfOrders - numGridOpenOrders)
	twinOrdersFilled, err := s.buildFilledTwinOrderMapFromTrades(ctx, queryTradesService, queryOrderService, twinOrdersOpen, expectedFilledNum)
	if err != nil {
		return nil, errors.Wrapf(err, "failed to build filled pin order map")
	}

	// 3. get the filled orders from twin-order map
	filledOrders := twinOrdersFilled.AscendingOrders()

	// 4. verify the grid
	if err := s.verifyFilledTwinGrid(s.grid.Pins, twinOrdersOpen, filledOrders); err != nil {
		return nil, errors.Wrapf(err, "verify grid with error")
	}

	return filledOrders, nil
}

func (s *Strategy) verifyFilledTwinGrid(pins []Pin, twinOrders TwinOrderMap, filledOrders []types.Order) error {
	s.debugLog("verifying filled grid - pins: %+v", pins)
	s.debugOrders("verifying filled grid - filled orders", filledOrders)
	s.debugLog("verifying filled grid - open twin orders:\n%s", twinOrders.String())

	if err := s.addOrdersIntoTwinOrderMap(twinOrders, filledOrders); err != nil {
		return errors.Wrapf(err, "verifying filled grid error when add orders into twin order map")
	}

	s.debugLog("verifying filled grid - filled twin orders:\n%+v", twinOrders.String())

	for i, pin := range pins {
		// we use twinOrderMap to make sure there are no duplicated order at one grid, and we use the sell price as key so we skip the pins[0] which is only for buy price
		if i == 0 {
			continue
		}

		twin, exist := twinOrders[fixedpoint.Value(pin)]
		if !exist {
			return fmt.Errorf("there is no order at price (%+v)", pin)
		}

		if !twin.Exist() {
			return fmt.Errorf("all the price need a twin")
		}

		if !twin.IsValid() {
			return fmt.Errorf("all the twins need to be valid")
		}
	}

	return nil
}

// buildTwinOrderMap build the pin-order map with grid and open orders.
// The keys of this map contains all required pins of this grid.
// If the Order of the pin is empty types.Order (OrderID == 0), it means there is no open orders at this pin.
func (s *Strategy) buildTwinOrderMap(pins []Pin, openOrders []types.Order) (TwinOrderMap, error) {
	twinOrderMap := make(TwinOrderMap)

	for i, pin := range pins {
		// twin order map only use sell price as key, so skip 0
		if i == 0 {
			continue
		}

		twinOrderMap[fixedpoint.Value(pin)] = TwinOrder{}
	}

	for _, openOrder := range openOrders {
		twinKey, err := findTwinOrderMapKey(s.grid, openOrder)
		if err != nil {
			return nil, errors.Wrapf(err, "failed to build twin order map")
		}

		twinOrder, exist := twinOrderMap[twinKey]
		if !exist {
			return nil, fmt.Errorf("the price of the openOrder (id: %d) is not in pins", openOrder.OrderID)
		}

		if twinOrder.Exist() {
			return nil, fmt.Errorf("there are multiple order in a twin")
		}

		twinOrder.SetOrder(openOrder)
		twinOrderMap[twinKey] = twinOrder
	}

	return twinOrderMap, nil
}

// buildFilledTwinOrderMapFromTrades will query the trades from last 24 hour and use them to build a pin order map
// It will skip the orders on pins at which open orders are already
func (s *Strategy) buildFilledTwinOrderMapFromTrades(ctx context.Context, queryTradesService types.ExchangeTradeHistoryService, queryOrderService types.ExchangeOrderQueryService, twinOrdersOpen TwinOrderMap, expectedFillNum int) (TwinOrderMap, error) {
	twinOrdersFilled := make(TwinOrderMap)

	// existedOrders is used to avoid re-query the same orders
	existedOrders := twinOrdersOpen.SyncOrderMap()

	// get the filled orders when bbgo is down in order from trades
	until := time.Now()
	// the first query only query the last 1 hour, because mostly shutdown and recovery happens within 1 hour
	since := until.Add(-1 * time.Hour)
	// hard limit for recover
	recoverSinceLimit := time.Date(2023, time.March, 10, 0, 0, 0, 0, time.UTC)

	if s.RecoverGridWithin != 0 && until.Add(-1*s.RecoverGridWithin).After(recoverSinceLimit) {
		recoverSinceLimit = until.Add(-1 * s.RecoverGridWithin)
	}

	for {
		if err := s.queryTradesToUpdateTwinOrdersMap(ctx, queryTradesService, queryOrderService, twinOrdersOpen, twinOrdersFilled, existedOrders, since, until); err != nil {
			return nil, errors.Wrapf(err, "failed to query trades to update twin orders map")
		}

		until = since
		since = until.Add(-6 * time.Hour)

		if len(twinOrdersFilled) >= expectedFillNum {
			s.logger.Infof("stop querying trades because twin orders filled (%d) >= expected filled nums (%d)", len(twinOrdersFilled), expectedFillNum)
			break
		}

		if s.GridProfitStats != nil && s.GridProfitStats.Since != nil && until.Before(*s.GridProfitStats.Since) {
			s.logger.Infof("stop querying trades because the time range is out of the strategy's since (%s)", *s.GridProfitStats.Since)
			break
		}

		if until.Before(recoverSinceLimit) {
			s.logger.Infof("stop querying trades because the time range is out of the limit (%s)", recoverSinceLimit)
			break
		}
	}

	return twinOrdersFilled, nil
}

func (s *Strategy) queryTradesToUpdateTwinOrdersMap(ctx context.Context, queryTradesService types.ExchangeTradeHistoryService, queryOrderService types.ExchangeOrderQueryService, twinOrdersOpen, twinOrdersFilled TwinOrderMap, existedOrders *types.SyncOrderMap, since, until time.Time) error {
	var fromTradeID uint64 = 0
	var limit int64 = 1000
	for {
		trades, err := queryTradesService.QueryTrades(ctx, s.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 with open orders")
		}

		s.debugLog("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
			}

			s.debugLog(trade.String())

			if existedOrders.Exists(trade.OrderID) {
				// already queries, skip
				continue
			}
			order, err := queryOrderService.QueryOrder(ctx, types.OrderQuery{
				OrderID: strconv.FormatUint(trade.OrderID, 10),
			})

			if err != nil {
				return errors.Wrapf(err, "failed to query order by trade")
			}

			s.debugLog(order.String())
			// avoid query this order again
			existedOrders.Add(*order)
			// add 1 to avoid duplicate
			fromTradeID = trade.ID + 1

			twinOrderKey, err := findTwinOrderMapKey(s.grid, *order)
			if err != nil {
				return errors.Wrapf(err, "failed to find grid order map's key when recover")
			}

			twinOrderOpen, exist := twinOrdersOpen[twinOrderKey]
			if !exist {
				return fmt.Errorf("the price of the order with the same GroupID is not in pins")
			}

			if twinOrderOpen.Exist() {
				continue
			}

			if twinOrder, exist := twinOrdersFilled[twinOrderKey]; exist {
				to := twinOrder.GetOrder()
				if to.UpdateTime.Time().After(order.UpdateTime.Time()) {
					s.logger.Infof("twinOrder's update time (%s) should not be after order's update time (%s)", to.UpdateTime, order.UpdateTime)
					continue
				}
			}

			twinOrder := TwinOrder{}
			twinOrder.SetOrder(*order)
			twinOrdersFilled[twinOrderKey] = twinOrder
		}

		// stop condition
		if int64(len(trades)) < limit {
			return nil
		}
	}
}

func (s *Strategy) addOrdersIntoTwinOrderMap(twinOrders TwinOrderMap, orders []types.Order) error {
	for _, order := range orders {
		k, err := findTwinOrderMapKey(s.grid, order)
		if err != nil {
			return errors.Wrap(err, "failed to add orders into twin order map")
		}

		if v, exist := twinOrders[k]; !exist {
			return fmt.Errorf("the price (%+v) is not in pins", k)
		} else if v.Exist() {
			return fmt.Errorf("there is already a twin order at this price (%+v)", k)
		} else {
			twin := TwinOrder{}
			twin.SetOrder(order)
			twinOrders[k] = twin
		}
	}

	return nil
}