Merge pull request #1545 from c9s/feat/add-universal-cancel-all-orders

FEATURE: add universal cancel all orders api helper

pkg/bbgo/graceful_shutdown.go

@@ -23,6 +23,9 @@ func (g *GracefulShutdown) Shutdown(shutdownCtx context.Context) {
 	wg.Wait()
 }
 
+// OnShutdown helps you register your shutdown handler
+// the first context object is where you want to register your shutdown handler, where the context has the isolated storage.
+// in your handler, you will get another context for the timeout context.
 func OnShutdown(ctx context.Context, f ShutdownHandler) {
 	isolatedContext := GetIsolationFromContext(ctx)
 	isolatedContext.gracefulShutdown.OnShutdown(f)

pkg/exchange/max/convert.go

@@ -172,7 +172,6 @@ func toGlobalOrder(maxOrder max.Order) (*types.Order, error) {
 	executedVolume := maxOrder.ExecutedVolume
 	remainingVolume := maxOrder.RemainingVolume
 	isMargin := maxOrder.WalletType == max.WalletTypeMargin
-
 	return &types.Order{
 		SubmitOrder: types.SubmitOrder{
 			ClientOrderID: maxOrder.ClientOID,

pkg/strategy/dca2/strategy.go

@@ -8,15 +8,17 @@ import (
 	"sync"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sirupsen/logrus"
+	"go.uber.org/multierr"
+
 	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/exchange/retry"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/c9s/bbgo/pkg/strategy/common"
 	"github.com/c9s/bbgo/pkg/types"
 	"github.com/c9s/bbgo/pkg/util"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/sirupsen/logrus"
-	"go.uber.org/multierr"
+	"github.com/c9s/bbgo/pkg/util/tradingutil"
 )
 
 const ID = "dca2"
@@ -289,30 +291,29 @@ func (s *Strategy) CleanUp(ctx context.Context) error {
 		return fmt.Errorf("Session is nil, please check it")
 	}
 
-	service, support := session.Exchange.(advancedOrderCancelApi)
-	if !support {
-		return fmt.Errorf("advancedOrderCancelApi interface is not implemented, fallback to default graceful cancel, exchange %T", session)
+	// ignore the first cancel error, this skips one open-orders query request
+	if err := tradingutil.UniversalCancelAllOrders(ctx, session.Exchange, nil); err == nil {
+		return nil
 	}
 
+	// if cancel all orders returns error, get the open orders and retry the cancel in each round
 	var werr error
 	for {
 		s.logger.Infof("checking %s open orders...", s.Symbol)
 
 		openOrders, err := retry.QueryOpenOrdersUntilSuccessful(ctx, session.Exchange, s.Symbol)
 		if err != nil {
-			s.logger.WithError(err).Errorf("CancelOrdersByGroupID api call error")
-			werr = multierr.Append(werr, err)
+			s.logger.WithError(err).Errorf("unable to query open orders")
+			continue
 		}
 
+		// all clean up
 		if len(openOrders) == 0 {
 			break
 		}
 
-		s.logger.Infof("found %d open orders left, using cancel all orders api", len(openOrders))
-
-		s.logger.Infof("using cancal all orders api for canceling grid orders...")
-		if err := retry.CancelAllOrdersUntilSuccessful(ctx, service); err != nil {
-			s.logger.WithError(err).Errorf("CancelAllOrders api call error")
+		if err := tradingutil.UniversalCancelAllOrders(ctx, session.Exchange, openOrders); err != nil {
+			s.logger.WithError(err).Errorf("unable to cancel all orders")
 			werr = multierr.Append(werr, err)
 		}
 

pkg/util/tradingutil/cancel.go

@@ -0,0 +1,119 @@
+package tradingutil
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/c9s/bbgo/pkg/exchange/retry"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type CancelAllOrdersService interface {
+	CancelAllOrders(ctx context.Context) ([]types.Order, error)
+}
+
+type CancelAllOrdersBySymbolService interface {
+	CancelOrdersBySymbol(ctx context.Context, symbol string) ([]types.Order, error)
+}
+
+type CancelAllOrdersByGroupIDService interface {
+	CancelOrdersByGroupID(ctx context.Context, groupID uint32) ([]types.Order, error)
+}
+
+// UniversalCancelAllOrders checks if the exchange instance supports the best order cancel strategy
+// it tries the first interface CancelAllOrdersService that does not need any existing order information or symbol information.
+//
+// if CancelAllOrdersService is not supported, then it tries CancelAllOrdersBySymbolService which needs at least one symbol
+// for the cancel api request.
+func UniversalCancelAllOrders(ctx context.Context, exchange types.Exchange, openOrders []types.Order) error {
+	if service, ok := exchange.(CancelAllOrdersService); ok {
+		if _, err := service.CancelAllOrders(ctx); err == nil {
+			return nil
+		} else {
+			log.WithError(err).Errorf("unable to cancel all orders")
+		}
+	}
+
+	if len(openOrders) == 0 {
+		return errors.New("to cancel all orders, openOrders can not be empty")
+	}
+
+	var anyErr error
+	if service, ok := exchange.(CancelAllOrdersBySymbolService); ok {
+		var symbols = CollectOrderSymbols(openOrders)
+		for _, symbol := range symbols {
+			_, err := service.CancelOrdersBySymbol(ctx, symbol)
+			if err != nil {
+				anyErr = err
+			}
+		}
+
+		if anyErr == nil {
+			return nil
+		}
+	}
+
+	if service, ok := exchange.(CancelAllOrdersByGroupIDService); ok {
+		var groupIds = CollectOrderGroupIds(openOrders)
+		for _, groupId := range groupIds {
+			if _, err := service.CancelOrdersByGroupID(ctx, groupId); err != nil {
+				anyErr = err
+			}
+		}
+
+		if anyErr == nil {
+			return nil
+		}
+	}
+
+	if anyErr != nil {
+		return anyErr
+	}
+
+	return fmt.Errorf("unable to cancel all orders, openOrders:%+v", openOrders)
+}
+
+func CollectOrderGroupIds(orders []types.Order) (groupIds []uint32) {
+	groupIdMap := map[uint32]struct{}{}
+	for _, o := range orders {
+		if o.GroupID > 0 {
+			groupIdMap[o.GroupID] = struct{}{}
+		}
+	}
+
+	for id := range groupIdMap {
+		groupIds = append(groupIds, id)
+	}
+
+	return groupIds
+}
+
+func CollectOrderSymbols(orders []types.Order) (symbols []string) {
+	symbolMap := map[string]struct{}{}
+	for _, o := range orders {
+		symbolMap[o.Symbol] = struct{}{}
+	}
+
+	for s := range symbolMap {
+		symbols = append(symbols, s)
+	}
+
+	return symbols
+}
+
+func CollectOpenOrders(ctx context.Context, ex types.Exchange, symbols ...string) ([]types.Order, error) {
+	var collectedOrders []types.Order
+	for _, symbol := range symbols {
+		openOrders, err := retry.QueryOpenOrdersUntilSuccessful(ctx, ex, symbol)
+		if err != nil {
+			return nil, err
+		}
+
+		collectedOrders = append(collectedOrders, openOrders...)
+	}
+
+	return collectedOrders, nil
+}