FEATURE: [dca2] emit position after recovery and refactor

pkg/strategy/dca2/background_runner.go

@@ -1,35 +0,0 @@
-package dca2
-
-import (
-	"context"
-	"time"
-
-	"github.com/c9s/bbgo/pkg/bbgo"
-	"github.com/c9s/bbgo/pkg/util"
-)
-
-func (s *Strategy) runBackgroundTask(ctx context.Context) {
-	s.logger.Info("run background task")
-
-	// recover active orders
-	recoverActiveOrdersInterval := util.MillisecondsJitter(10*time.Minute, 5*60*1000)
-	recoverActiveOrdersTicker := time.NewTicker(recoverActiveOrdersInterval)
-	defer recoverActiveOrdersTicker.Stop()
-
-	// sync strategy
-	syncPersistenceTicker := time.NewTicker(1 * time.Hour)
-	defer syncPersistenceTicker.Stop()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-syncPersistenceTicker.C:
-			bbgo.Sync(ctx, s)
-		case <-recoverActiveOrdersTicker.C:
-			if err := s.recoverActiveOrders(ctx); err != nil {
-				s.logger.WithError(err).Warn(err, "failed to recover active orders")
-			}
-		}
-	}
-}

pkg/strategy/dca2/strategy.go

@@ -326,6 +326,9 @@ func (s *Strategy) Run(ctx context.Context, _ bbgo.OrderExecutor, session *bbgo.
 				s.logger.Infof("profit stats %s", s.ProfitStats.String())
 				s.logger.Infof("startTimeOfNextRound %s", s.startTimeOfNextRound)
 
+				// emit position after recovery
+				s.OrderExecutor.TradeCollector().EmitPositionUpdate(s.Position)
+
 				s.updateTakeProfitPrice()
 
 				// store persistence
@@ -340,7 +343,7 @@ func (s *Strategy) Run(ctx context.Context, _ bbgo.OrderExecutor, session *bbgo.
 		})
 	})
 
-	go s.runBackgroundTask(ctx)
+	go s.syncPeriodically(ctx)
 
 	bbgo.OnShutdown(ctx, func(ctx context.Context, wg *sync.WaitGroup) {
 		defer wg.Done()

pkg/strategy/dca2/sync.go

@@ -4,30 +4,38 @@ import (
 	"context"
 	"time"
 
+	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/exchange/retry"
 	"github.com/c9s/bbgo/pkg/strategy/common"
 	"github.com/c9s/bbgo/pkg/util"
 )
 
-func (s *Strategy) recoverPeriodically(ctx context.Context) {
-	s.logger.Info("monitor and recover periodically")
-	interval := util.MillisecondsJitter(10*time.Minute, 5*60*1000)
-	ticker := time.NewTicker(interval)
-	defer ticker.Stop()
+func (s *Strategy) syncPeriodically(ctx context.Context) {
+	s.logger.Info("sync periodically")
+
+	// sync persistence
+	syncPersistenceTicker := time.NewTicker(1 * time.Hour)
+	defer syncPersistenceTicker.Stop()
+
+	// sync active orders
+	syncActiveOrdersTicker := time.NewTicker(util.MillisecondsJitter(10*time.Minute, 5*60*1000))
+	defer syncActiveOrdersTicker.Stop()
 
 	for {
 		select {
 		case <-ctx.Done():
 			return
-		case <-ticker.C:
-			if err := s.recoverActiveOrders(ctx); err != nil {
-				s.logger.WithError(err).Warn(err, "failed to recover active orders")
+		case <-syncPersistenceTicker.C:
+			bbgo.Sync(ctx, s)
+		case <-syncActiveOrdersTicker.C:
+			if err := s.syncActiveOrders(ctx); err != nil {
+				s.logger.WithError(err).Warn(err, "failed to sync active orders")
 			}
 		}
 	}
 }
 
-func (s *Strategy) recoverActiveOrders(ctx context.Context) error {
+func (s *Strategy) syncActiveOrders(ctx context.Context) error {
 	s.logger.Info("recover active orders...")
 	openOrders, err := retry.QueryOpenOrdersUntilSuccessfulLite(ctx, s.ExchangeSession.Exchange, s.Symbol)
 	if err != nil {

pkg/strategy/dca2/take_profit.go

@@ -12,6 +12,10 @@ import (
 func (s *Strategy) placeTakeProfitOrders(ctx context.Context) error {
 	s.logger.Info("start placing take profit orders")
 	currentRound, err := s.collector.CollectCurrentRound(ctx)
+	if err != nil {
+		return errors.Wrap(err, "failed to place the take-profit order when collecting current round")
+	}
+
 	if currentRound.TakeProfitOrder.OrderID != 0 {
 		return fmt.Errorf("there is a take-profit order before placing the take-profit order, please check it")
 	}