FEATURE: recollect position before placing the take-profit order

pkg/strategy/dca2/active_order_recover.go

@@ -54,7 +54,7 @@ func (s *Strategy) recoverActiveOrders(ctx context.Context) error {
 	opts := common.SyncActiveOrdersOpts{
 		Logger:            s.logger,
 		Exchange:          s.ExchangeSession.Exchange,
-		OrderQueryService: s.orderQueryService,
+		OrderQueryService: s.roundCollector.queryService,
 		ActiveOrderBook:   activeOrders,
 		OpenOrders:        openOrders,
 	}

pkg/strategy/dca2/recover.go

@@ -24,25 +24,7 @@ type RecoverApiQueryService interface {
 
 func (s *Strategy) recover(ctx context.Context) error {
 	s.logger.Info("[DCA] recover")
-	queryService, ok := s.ExchangeSession.Exchange.(RecoverApiQueryService)
-	if !ok {
-		return fmt.Errorf("[DCA] exchange %s doesn't support queryAPI interface", s.ExchangeSession.ExchangeName)
-	}
-
-	openOrders, err := queryService.QueryOpenOrders(ctx, s.Symbol)
-	if err != nil {
-		return err
-	}
-
-	closedOrders, err := queryService.QueryClosedOrdersDesc(ctx, s.Symbol, recoverSinceLimit, time.Now(), 0)
-	if err != nil {
-		return err
-	}
-
-	currentRound, err := getCurrentRoundOrders(openOrders, closedOrders, s.OrderGroupID)
-	if err != nil {
-		return err
-	}
+	currentRound, err := s.roundCollector.CollectCurrentRound(ctx)
 	debugRoundOrders(s.logger, "current", currentRound)
 
 	// recover profit stats
@@ -59,6 +41,11 @@ func (s *Strategy) recover(ctx context.Context) error {
 	if s.DisablePositionRecover {
 		s.logger.Info("disablePositionRecover is set, skip position recovery")
 	} else {
+		queryService, ok := s.ExchangeSession.Exchange.(RecoverApiQueryService)
+		if !ok {
+			return fmt.Errorf("[DCA] exchange %s doesn't support queryAPI interface", s.ExchangeSession.ExchangeName)
+		}
+
 		if err := recoverPosition(ctx, s.Position, queryService, currentRound); err != nil {
 			return err
 		}

pkg/strategy/dca2/round_collector.go

@@ -2,6 +2,7 @@ package dca2
 
 import (
 	"context"
+	"fmt"
 	"strconv"
 	"time"
 
@@ -19,8 +20,11 @@ type RoundCollector struct {
 	isMax   bool
 
 	// service
+	ex                   types.Exchange
 	historyService       types.ExchangeTradeHistoryService
 	queryService         types.ExchangeOrderQueryService
+	tradeService         types.ExchangeTradeService
+	queryClosedOrderDesc descendingClosedOrderQueryService
 }
 
 func NewRoundCollector(logger *logrus.Entry, symbol string, groupID uint32, ex types.Exchange) *RoundCollector {
@@ -37,16 +41,84 @@ func NewRoundCollector(logger *logrus.Entry, symbol string, groupID uint32, ex t
 		return nil
 	}
 
+	tradeService, ok := ex.(types.ExchangeTradeService)
+	if !ok {
+		logger.Errorf("exchange %s doesn't support ExchangeTradeService", ex.Name())
+		return nil
+	}
+
+	queryClosedOrderDesc, ok := ex.(descendingClosedOrderQueryService)
+	if !ok {
+		logger.Errorf("exchange %s doesn't support query closed orders desc", ex.Name())
+		return nil
+	}
+
 	return &RoundCollector{
 		logger:               logger,
 		symbol:               symbol,
 		groupID:              groupID,
 		isMax:                isMax,
+		ex:                   ex,
 		historyService:       historyService,
 		queryService:         queryService,
+		tradeService:         tradeService,
+		queryClosedOrderDesc: queryClosedOrderDesc,
 	}
 }
 
+func (rc RoundCollector) CollectCurrentRound(ctx context.Context) (Round, error) {
+	openOrders, err := retry.QueryOpenOrdersUntilSuccessful(ctx, rc.ex, rc.symbol)
+	if err != nil {
+		return Round{}, err
+	}
+
+	var closedOrders []types.Order
+	var op = func() (err2 error) {
+		closedOrders, err2 = rc.queryClosedOrderDesc.QueryClosedOrdersDesc(ctx, rc.symbol, recoverSinceLimit, time.Now(), 0)
+		return err2
+	}
+	if err := retry.GeneralBackoff(ctx, op); err != nil {
+		return Round{}, err
+	}
+
+	openPositionSide := types.SideTypeBuy
+	takeProfitSide := types.SideTypeSell
+
+	var allOrders []types.Order
+	allOrders = append(allOrders, openOrders...)
+	allOrders = append(allOrders, closedOrders...)
+
+	types.SortOrdersDescending(allOrders)
+
+	var currentRound Round
+	lastSide := takeProfitSide
+	for _, order := range allOrders {
+		// group id filter is used for debug when local running
+		if order.GroupID != rc.groupID {
+			continue
+		}
+
+		if order.Side == takeProfitSide && lastSide == openPositionSide {
+			break
+		}
+
+		switch order.Side {
+		case openPositionSide:
+			currentRound.OpenPositionOrders = append(currentRound.OpenPositionOrders, order)
+		case takeProfitSide:
+			if currentRound.TakeProfitOrder.OrderID != 0 {
+				return currentRound, fmt.Errorf("there are two take-profit orders in one round, please check it")
+			}
+			currentRound.TakeProfitOrder = order
+		default:
+		}
+
+		lastSide = order.Side
+	}
+
+	return currentRound, nil
+}
+
 func (rc *RoundCollector) CollectFinishRounds(ctx context.Context, fromOrderID uint64) ([]Round, error) {
 	// TODO: pagination for it
 	// query the orders
@@ -96,13 +168,16 @@ func (rc *RoundCollector) CollectRoundTrades(ctx context.Context, round Round) (
 	debugRoundOrders(rc.logger, "collect round trades", round)
 
 	var roundTrades []types.Trade
-
 	var roundOrders []types.Order = round.OpenPositionOrders
+
+	// if the take-profit order's OrderID == 0 -> no take-profit order.
+	if round.TakeProfitOrder.OrderID != 0 {
 		roundOrders = append(roundOrders, round.TakeProfitOrder)
+	}
 
 	for _, order := range roundOrders {
 		rc.logger.Infof("collect trades from order: %s", order.String())
-		if order.ExecutedQuantity.Sign() == 0 {
+		if order.ExecutedQuantity.IsZero() {
 			rc.logger.Info("collect trads from order but no executed quantity ", order.String())
 			continue
 		} else {

pkg/strategy/dca2/strategy.go

@@ -94,7 +94,6 @@ type Strategy struct {
 	nextStateC           chan State
 	state                State
 	roundCollector       *RoundCollector
-	orderQueryService    types.ExchangeOrderQueryService
 	takeProfitPrice      fixedpoint.Value
 	startTimeOfNextRound time.Time
 	nextRoundPaused      bool
@@ -193,13 +192,6 @@ func (s *Strategy) Run(ctx context.Context, _ bbgo.OrderExecutor, session *bbgo.
 		s.OrderGroupID = util.FNV32(instanceID) % math.MaxInt32
 	}
 
-	// orderQueryService
-	if service, ok := s.ExchangeSession.Exchange.(types.ExchangeOrderQueryService); ok {
-		s.orderQueryService = service
-	} else {
-		return fmt.Errorf("exchange %s doesn't support ExchangeOrderQueryService", s.ExchangeSession.ExchangeName)
-	}
-
 	// round collector
 	s.roundCollector = NewRoundCollector(s.logger, s.Symbol, s.OrderGroupID, s.ExchangeSession.Exchange)
 	if s.roundCollector == nil {

pkg/strategy/dca2/take_profit.go

@@ -2,14 +2,38 @@ package dca2
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/c9s/bbgo/pkg/types"
+	"github.com/pkg/errors"
 )
 
 func (s *Strategy) placeTakeProfitOrders(ctx context.Context) error {
 	s.logger.Info("start placing take profit orders")
-	order := generateTakeProfitOrder(s.Market, s.TakeProfitRatio, s.Position, s.OrderGroupID)
+	currentRound, err := s.roundCollector.CollectCurrentRound(ctx)
+	if currentRound.TakeProfitOrder.OrderID != 0 {
+		return fmt.Errorf("there is a take-profit order before placing the take-profit order, please check it")
+	}
+
+	trades, err := s.roundCollector.CollectRoundTrades(ctx, currentRound)
+	if err != nil {
+		return errors.Wrap(err, "failed to place the take-profit order when collecting round trades")
+	}
+
+	position := types.NewPositionFromMarket(s.Market)
+
+	for _, trade := range trades {
+		if trade.FeeProcessing {
+			return fmt.Errorf("failed to place the take-profit order because there is a trade's fee not ready")
+		}
+
+		position.AddTrade(trade)
+	}
+
+	s.logger.Infof("placeTakeProfitOrders: %s", position.String())
+
+	order := generateTakeProfitOrder(s.Market, s.TakeProfitRatio, position, s.OrderGroupID)
 	createdOrders, err := s.OrderExecutor.SubmitOrders(ctx, order)
 	if err != nil {
 		return err