max: improve max closed order query

pkg/exchange/max/exchange.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
+	"go.uber.org/multierr"
 	"golang.org/x/time/rate"
 
 	maxapi "github.com/c9s/bbgo/pkg/exchange/max/maxapi"
@@ -175,15 +176,96 @@ func (e *Exchange) QueryOpenOrders(ctx context.Context, symbol string) (orders [
 }
 
 // lastOrderID is not supported on MAX
-func (e *Exchange) QueryClosedOrders(ctx context.Context, symbol string, since, until time.Time, lastOrderID uint64) (orders []types.Order, err error) {
+func (e *Exchange) QueryClosedOrders(ctx context.Context, symbol string, since, until time.Time, lastOrderID uint64) ([]types.Order, error) {
-	limit := 1000 // max limit = 1000, default 100
+	log.Warn("!!!MAX EXCHANGE API NOTICE!!!")
-	orderIDs := make(map[uint64]struct{}, limit*2)
+	log.Warn("the since/until conditions will not be effected on closed orders query, max exchange does not support time-range-based query")
-
-	log.Warn("since/until condition will not be effected on closed orders query, max exchange does not support time-range-based query, we will start from the first record")
 	if lastOrderID > 0 {
 		log.Warn("last order id condition will not be effected on max exchange, max exchange does not support last order id query")
 	}
 
+	if v, ok := util.GetEnvVarBool("MAX_QUERY_CLOSED_ORDERS_ALL"); v && ok {
+		log.Warn("MAX_QUERY_CLOSED_ORDERS_ALL is set, we will fetch all closed orders from the first page")
+		return e.queryAllClosedOrders(ctx, symbol, since, until)
+	}
+
+	return e.queryRecentlyClosedOrders(ctx, symbol, since, until)
+}
+
+func (e *Exchange) queryRecentlyClosedOrders(ctx context.Context, symbol string, since time.Time, until time.Time) (orders []types.Order, err error) {
+	limit := 1000 // max limit = 1000, default 100
+	orderIDs := make(map[uint64]struct{}, limit*2)
+	maxPages := 10
+
+	if v, ok := util.GetEnvVarInt("MAX_QUERY_CLOSED_ORDERS_NUM_OF_PAGES"); ok {
+		maxPages = v
+	}
+
+	log.Warnf("fetching recently closed orders, maximum %d pages to fetch", maxPages)
+	log.Warnf("note that, some MAX orders might be missing if you did not sync the closed orders for a while")
+
+queryRecentlyClosedOrders:
+	for page := 1; page < maxPages; page++ {
+		if err := closedOrderQueryLimiter.Wait(ctx); err != nil {
+			return orders, err
+		}
+
+		log.Infof("querying %s closed orders from page %d ~ ", symbol, page)
+		maxOrders, err2 := e.client.OrderService.Closed(toLocalSymbol(symbol), maxapi.QueryOrderOptions{
+			Limit:   limit,
+			Page:    page,
+			OrderBy: "desc",
+		})
+		if err2 != nil {
+			err = multierr.Append(err, err2)
+			break queryRecentlyClosedOrders
+		}
+
+		// no recent orders
+		if len(maxOrders) == 0 {
+			break queryRecentlyClosedOrders
+		}
+
+		log.Debugf("fetched %d orders", len(maxOrders))
+		for _, maxOrder := range maxOrders {
+			if maxOrder.CreatedAtMs.Time().Before(since) {
+				log.Debugf("skip orders with creation time before %s, found %s", since, maxOrder.CreatedAtMs.Time())
+				break queryRecentlyClosedOrders
+			}
+
+			if maxOrder.CreatedAtMs.Time().After(until) {
+				log.Debugf("skip orders with creation time after %s, found %s", until, maxOrder.CreatedAtMs.Time())
+				continue
+			}
+
+			order, err2 := toGlobalOrder(maxOrder)
+			if err2 != nil {
+				err = multierr.Append(err, err2)
+				continue
+			}
+
+			if _, ok := orderIDs[order.OrderID]; ok {
+				log.Debugf("skipping duplicated order: %d", order.OrderID)
+			}
+
+			log.Debugf("max order %d %s %f %s %s", order.OrderID, order.Symbol, order.Price, order.Status, order.CreationTime.Time().Format(time.StampMilli))
+
+			orderIDs[order.OrderID] = struct{}{}
+			orders = append(orders, *order)
+		}
+	}
+
+	// ensure everything is ascending ordered
+	log.Debugf("sorting %d orders", len(orders))
+	sort.Slice(orders, func(i, j int) bool {
+		return orders[i].CreationTime.Time().Before(orders[j].CreationTime.Time())
+	})
+
+	return orders, err
+}
+
+func (e *Exchange) queryAllClosedOrders(ctx context.Context, symbol string, since time.Time, until time.Time) (orders []types.Order, err error) {
+	limit := 1000 // max limit = 1000, default 100
+	orderIDs := make(map[uint64]struct{}, limit*2)
 	page := 1
 	for {
 		if err := closedOrderQueryLimiter.Wait(ctx); err != nil {
@@ -208,7 +290,7 @@ func (e *Exchange) QueryClosedOrders(ctx context.Context, symbol string, since,
 			return maxOrders[i].CreatedAtMs.Time().Before(maxOrders[j].CreatedAtMs.Time())
 		})
 
-		log.Infof("%d orders", len(maxOrders))
+		log.Debugf("%d orders", len(maxOrders))
 		for _, maxOrder := range maxOrders {
 			if maxOrder.CreatedAtMs.Time().Before(since) {
 				log.Debugf("skip orders with creation time before %s, found %s", since, maxOrder.CreatedAtMs.Time())
@@ -230,7 +312,7 @@ func (e *Exchange) QueryClosedOrders(ctx context.Context, symbol string, since,
 
 			orderIDs[order.OrderID] = struct{}{}
 			orders = append(orders, *order)
-			log.Infof("order %+v", order)
+			log.Debugf("order %+v", order)
 		}
 		page++
 	}

pkg/exchange/max/maxapi/order.go

@@ -71,6 +71,7 @@ type QueryOrderOptions struct {
 	Offset  int
 	Limit   int
 	Page    int
+	OrderBy string
 }
 
 // OrderService manages the Order endpoint.
@@ -118,6 +119,9 @@ func (s *OrderService) Closed(market string, options QueryOrderOptions) ([]Order
 	if options.Page > 0 {
 		payload["page"] = options.Page
 	}
+	if len(options.OrderBy) > 0 {
+		payload["order_by"] = options.OrderBy
+	}
 
 	req, err := s.client.newAuthenticatedRequest("GET", "v2/orders", payload, relUrlV2Orders)
 	if err != nil {