gctrpc/ordermanager/binance: Add new getManagedOrders command and various improvements (#712)

* first draft of getmanaged orders RPC call

* - ClientIDs for binance, especially spot asset
- applied old ClientOrderId for cancelled orders
- added clientOrderId to GCTRPC

* added tests for Matchfilter and GetManagedOrders

* smaller fixes

* comment fix
added getFilteredOrders to store
changed store mutex to RWMutex
smaller fixes

* fixed bug in Detail Copy and added test

* fixes for Scotts review

* processSubmittedOrder was missing clientOrderId

* changed: TestGetOrdersFiltered expanded
fixed: warning, where variable name collided with package name
fixed: used req.AssetType in binance_wrapper.go

Co-authored-by: Mark Dzulko <81071907+Mark-numus@users.noreply.github.com>

engine/order_manager.go

@@ -379,15 +379,28 @@ func (m *OrderManager) GetOrdersSnapshot(s order.Status) ([]order.Detail, time.T
 			if v[i].LastUpdated.After(latestUpdate) {
 				latestUpdate = v[i].LastUpdated
 			}
-
-			cpy := *v[i]
-			os = append(os, cpy)
+			os = append(os, *v[i])
 		}
 	}
 
 	return os, latestUpdate
 }
 
+// GetOrdersFiltered returns a snapshot of all orders in the order store.
+// Filtering is applied based on the order.Filter unless entries are empty
+func (m *OrderManager) GetOrdersFiltered(f *order.Filter) ([]order.Detail, error) {
+	if m == nil {
+		return nil, fmt.Errorf("order manager %w", ErrNilSubsystem)
+	}
+	if f == nil {
+		return nil, fmt.Errorf("order manager, GetOrdersFiltered: Filter is nil")
+	}
+	if atomic.LoadInt32(&m.started) == 0 {
+		return nil, fmt.Errorf("order manager %w", ErrSubSystemNotStarted)
+	}
+	return m.orderStore.getFilteredOrders(f)
+}
+
 // processSubmittedOrder adds a new order to the manager
 func (m *OrderManager) processSubmittedOrder(newOrder *order.Submit, result order.SubmitResponse) (*OrderSubmitResponse, error) {
 	if !result.IsOrderPlaced {
@@ -439,6 +452,7 @@ func (m *OrderManager) processSubmittedOrder(newOrder *order.Submit, result orde
 		ID:                result.OrderID,
 		AccountID:         newOrder.AccountID,
 		ClientID:          newOrder.ClientID,
+		ClientOrderID:     newOrder.ClientOrderID,
 		WalletAddress:     newOrder.WalletAddress,
 		Type:              newOrder.Type,
 		Side:              newOrder.Side,
@@ -661,8 +675,8 @@ func (s *store) upsert(od *order.Detail) error {
 
 // getByExchange returns orders by exchange
 func (s *store) getByExchange(exchange string) ([]*order.Detail, error) {
-	s.m.Lock()
-	defer s.m.Unlock()
+	s.m.RLock()
+	defer s.m.RUnlock()
 	r, ok := s.Orders[strings.ToLower(exchange)]
 	if !ok {
 		return nil, ErrExchangeNotFound
@@ -673,8 +687,8 @@ func (s *store) getByExchange(exchange string) ([]*order.Detail, error) {
 // getByInternalOrderID will search all orders for our internal orderID
 // and return the order
 func (s *store) getByInternalOrderID(internalOrderID string) (*order.Detail, error) {
-	s.m.Lock()
-	defer s.m.Unlock()
+	s.m.RLock()
+	defer s.m.RUnlock()
 	for _, v := range s.Orders {
 		for x := range v {
 			if v[x].InternalOrderID == internalOrderID {
@@ -690,8 +704,8 @@ func (s *store) exists(det *order.Detail) bool {
 	if det == nil {
 		return false
 	}
-	s.m.Lock()
-	defer s.m.Unlock()
+	s.m.RLock()
+	defer s.m.RUnlock()
 	r, ok := s.Orders[strings.ToLower(det.Exchange)]
 	if !ok {
 		return false
@@ -736,3 +750,35 @@ func (s *store) add(det *order.Detail) error {
 
 	return nil
 }
+
+// getFilteredOrders returns a filtered copy of the orders
+func (s *store) getFilteredOrders(f *order.Filter) ([]order.Detail, error) {
+	if f == nil {
+		return nil, errors.New("filter is nil")
+	}
+	s.m.RLock()
+	defer s.m.RUnlock()
+
+	var os []order.Detail
+	// optimization if Exchange is filtered
+	if f.Exchange != "" {
+		if e, ok := s.Orders[strings.ToLower(f.Exchange)]; ok {
+			for i := range e {
+				if !e[i].MatchFilter(f) {
+					continue
+				}
+				os = append(os, e[i].Copy())
+			}
+		}
+	} else {
+		for _, e := range s.Orders {
+			for i := range e {
+				if !e[i].MatchFilter(f) {
+					continue
+				}
+				os = append(os, e[i].Copy())
+			}
+		}
+	}
+	return os, nil
+}

engine/order_manager_test.go

@@ -559,3 +559,65 @@ func TestProcessOrders(t *testing.T) {
 	m := OrdersSetup(t)
 	m.processOrders()
 }
+
+func TestGetOrdersFiltered(t *testing.T) {
+	m := OrdersSetup(t)
+	_, err := m.GetOrdersFiltered(nil)
+	if err == nil {
+		t.Error("Expected error from nil filter")
+	}
+	orders := []order.Detail{
+		{
+			Exchange: testExchange,
+			ID:       "Test1",
+		},
+		{
+			Exchange: testExchange,
+			ID:       "Test2",
+		},
+	}
+	for i := range orders {
+		if err = m.orderStore.add(&orders[i]); err != nil {
+			t.Error(err)
+		}
+	}
+	res, err := m.GetOrdersFiltered(&order.Filter{ID: "Test2"})
+	if err != nil {
+		t.Error(err)
+	}
+	if len(res) != 1 {
+		t.Errorf("Expected 1 result, got: %d", len(res))
+	}
+}
+
+func Test_getFilteredOrders(t *testing.T) {
+	m := OrdersSetup(t)
+
+	_, err := m.orderStore.getFilteredOrders(nil)
+	if err == nil {
+		t.Error("Error expected when Filter is nil")
+	}
+
+	orders := []order.Detail{
+		{
+			Exchange: testExchange,
+			ID:       "Test1",
+		},
+		{
+			Exchange: testExchange,
+			ID:       "Test2",
+		},
+	}
+	for i := range orders {
+		if err = m.orderStore.add(&orders[i]); err != nil {
+			t.Error(err)
+		}
+	}
+	res, err := m.orderStore.getFilteredOrders(&order.Filter{ID: "Test1"})
+	if err != nil {
+		t.Error(err)
+	}
+	if len(res) != 1 {
+		t.Errorf("Expected 1 result, got: %d", len(res))
+	}
+}

engine/order_manager_types.go

@@ -36,7 +36,7 @@ type orderManagerConfig struct {
 
 // store holds all orders by exchange
 type store struct {
-	m               sync.Mutex
+	m               sync.RWMutex
 	Orders          map[string][]*order.Detail
 	commsManager    iCommsManager
 	exchangeManager iExchangeManager

engine/rpcserver.go

@@ -944,6 +944,90 @@ func (s *RPCServer) GetOrders(_ context.Context, r *gctrpc.GetOrdersRequest) (*g
 	return &gctrpc.GetOrdersResponse{Orders: orders}, nil
 }
 
+// GetManagedOrders returns all orders from the Order Manager for the provided exchange,
+// asset type  and currency pair
+func (s *RPCServer) GetManagedOrders(_ context.Context, r *gctrpc.GetOrdersRequest) (*gctrpc.GetOrdersResponse, error) {
+	if r == nil {
+		return nil, errInvalidArguments
+	}
+
+	a, err := asset.New(r.AssetType)
+	if err != nil {
+		return nil, err
+	}
+
+	if r.Pair == nil {
+		return nil, errCurrencyPairUnset
+	}
+	cp := currency.NewPairWithDelimiter(
+		r.Pair.Base,
+		r.Pair.Quote,
+		r.Pair.Delimiter)
+	exch := s.GetExchangeByName(r.Exchange)
+	err = checkParams(r.Exchange, exch, a, cp)
+	if err != nil {
+		return nil, err
+	}
+
+	var resp []order.Detail
+	filter := order.Filter{
+		Exchange:  exch.GetName(),
+		Pair:      cp,
+		AssetType: a,
+	}
+	resp, err = s.OrderManager.GetOrdersFiltered(&filter)
+	if err != nil {
+		return nil, err
+	}
+
+	var orders []*gctrpc.OrderDetails
+	for x := range resp {
+		var trades []*gctrpc.TradeHistory
+		for i := range resp[x].Trades {
+			t := &gctrpc.TradeHistory{
+				Id:        resp[x].Trades[i].TID,
+				Price:     resp[x].Trades[i].Price,
+				Amount:    resp[x].Trades[i].Amount,
+				Exchange:  r.Exchange,
+				AssetType: a.String(),
+				OrderSide: resp[x].Trades[i].Side.String(),
+				Fee:       resp[x].Trades[i].Fee,
+				Total:     resp[x].Trades[i].Total,
+			}
+			if !resp[x].Trades[i].Timestamp.IsZero() {
+				t.CreationTime = resp[x].Trades[i].Timestamp.Unix()
+			}
+			trades = append(trades, t)
+		}
+		o := &gctrpc.OrderDetails{
+			Exchange:      r.Exchange,
+			Id:            resp[x].ID,
+			ClientOrderId: resp[x].ClientOrderID,
+			BaseCurrency:  resp[x].Pair.Base.String(),
+			QuoteCurrency: resp[x].Pair.Quote.String(),
+			AssetType:     resp[x].AssetType.String(),
+			OrderSide:     resp[x].Side.String(),
+			OrderType:     resp[x].Type.String(),
+			Status:        resp[x].Status.String(),
+			Price:         resp[x].Price,
+			Amount:        resp[x].Amount,
+			OpenVolume:    resp[x].Amount - resp[x].ExecutedAmount,
+			Fee:           resp[x].Fee,
+			Cost:          resp[x].Cost,
+			Trades:        trades,
+		}
+		if !resp[x].Date.IsZero() {
+			o.CreationTime = resp[x].Date.Unix()
+		}
+		if !resp[x].LastUpdated.IsZero() {
+			o.UpdateTime = resp[x].LastUpdated.Unix()
+		}
+		orders = append(orders, o)
+	}
+
+	return &gctrpc.GetOrdersResponse{Orders: orders}, nil
+}
+
 // GetOrder returns order information based on exchange and order ID
 func (s *RPCServer) GetOrder(_ context.Context, r *gctrpc.GetOrderRequest) (*gctrpc.OrderDetails, error) {
 	if r == nil {
@@ -1001,6 +1085,7 @@ func (s *RPCServer) GetOrder(_ context.Context, r *gctrpc.GetOrderRequest) (*gct
 	return &gctrpc.OrderDetails{
 		Exchange:      result.Exchange,
 		Id:            result.ID,
+		ClientOrderId: result.ClientOrderID,
 		BaseCurrency:  result.Pair.Base.String(),
 		QuoteCurrency: result.Pair.Quote.String(),
 		AssetType:     result.AssetType.String(),
@@ -1043,14 +1128,15 @@ func (s *RPCServer) SubmitOrder(_ context.Context, r *gctrpc.SubmitOrderRequest)
 	}
 
 	submission := &order.Submit{
-		Pair:      p,
-		Side:      order.Side(r.Side),
-		Type:      order.Type(r.OrderType),
-		Amount:    r.Amount,
-		Price:     r.Price,
-		ClientID:  r.ClientId,
-		Exchange:  r.Exchange,
-		AssetType: a,
+		Pair:          p,
+		Side:          order.Side(r.Side),
+		Type:          order.Type(r.OrderType),
+		Amount:        r.Amount,
+		Price:         r.Price,
+		ClientID:      r.ClientId,
+		ClientOrderID: r.ClientId,
+		Exchange:      r.Exchange,
+		AssetType:     a,
 	}
 
 	resp, err := s.OrderManager.Submit(submission)
@@ -1196,7 +1282,8 @@ func (s *RPCServer) CancelOrder(_ context.Context, r *gctrpc.CancelOrderRequest)
 		return nil, err
 	}
 
-	err = exch.CancelOrder(&order.Cancel{
+	err = s.OrderManager.Cancel(&order.Cancel{
+		Exchange:      r.Exchange,
 		AccountID:     r.AccountId,
 		ID:            r.OrderId,
 		Side:          order.Side(r.Side),

engine/rpcserver_test.go

@@ -1590,3 +1590,101 @@ func TestGetDataHistoryJobSummary(t *testing.T) {
 		t.Errorf("received %v, expected %v", nil, "result summaries slice")
 	}
 }
+
+func TestGetManagedOrders(t *testing.T) {
+	exchName := "Binance"
+	engerino := &Engine{}
+	em := SetupExchangeManager()
+	exch, err := em.NewExchangeByName(exchName)
+	if err != nil {
+		t.Fatal(err)
+	}
+	exch.SetDefaults()
+	b := exch.GetBase()
+	cp := currency.NewPair(currency.BTC, currency.USDT)
+	b.CurrencyPairs.Pairs = make(map[asset.Item]*currency.PairStore)
+	b.CurrencyPairs.Pairs[asset.Spot] = &currency.PairStore{
+		Available:     currency.Pairs{cp},
+		Enabled:       currency.Pairs{cp},
+		AssetEnabled:  convert.BoolPtr(true),
+		ConfigFormat:  &currency.PairFormat{Uppercase: true},
+		RequestFormat: &currency.PairFormat{Uppercase: true}}
+	em.Add(exch)
+	var wg sync.WaitGroup
+	om, err := SetupOrderManager(em, engerino.CommunicationsManager, &wg, false)
+	if !errors.Is(err, nil) {
+		t.Errorf("received '%v', expected '%v'", err, nil)
+	}
+	err = om.Start()
+	if !errors.Is(err, nil) {
+		t.Errorf("received '%v', expected '%v'", err, nil)
+	}
+	s := RPCServer{Engine: &Engine{ExchangeManager: em, OrderManager: om}}
+
+	p := &gctrpc.CurrencyPair{
+		Delimiter: "-",
+		Base:      currency.BTC.String(),
+		Quote:     currency.USDT.String(),
+	}
+
+	_, err = s.GetManagedOrders(context.Background(), nil)
+	if !errors.Is(err, errInvalidArguments) {
+		t.Errorf("received '%v', expected '%v'", err, errInvalidArguments)
+	}
+
+	_, err = s.GetManagedOrders(context.Background(), &gctrpc.GetOrdersRequest{
+		AssetType: asset.Spot.String(),
+		Pair:      p,
+	})
+	if !errors.Is(err, errExchangeNotLoaded) {
+		t.Errorf("received '%v', expected '%v'", errExchangeNotLoaded, err)
+	}
+
+	_, err = s.GetManagedOrders(context.Background(), &gctrpc.GetOrdersRequest{
+		Exchange:  exchName,
+		AssetType: asset.Spot.String(),
+	})
+	if !errors.Is(err, errCurrencyPairUnset) {
+		t.Errorf("received '%v', expected '%v'", err, errCurrencyPairUnset)
+	}
+
+	_, err = s.GetManagedOrders(context.Background(), &gctrpc.GetOrdersRequest{
+		Exchange: exchName,
+		Pair:     p,
+	})
+	if !errors.Is(err, asset.ErrNotSupported) {
+		t.Errorf("received '%v', expected '%v'", err, asset.ErrNotSupported)
+	}
+
+	o := order.Detail{
+		Price:           100000,
+		Amount:          0.002,
+		Exchange:        "Binance",
+		InternalOrderID: "",
+		ID:              "",
+		ClientOrderID:   "",
+		AccountID:       "",
+		ClientID:        "",
+		WalletAddress:   "",
+		Type:            order.Limit,
+		Side:            "SELL",
+		Status:          order.New,
+		AssetType:       asset.Spot,
+		Pair:            currency.NewPair(currency.BTC, currency.USDT),
+	}
+	err = om.Add(&o)
+	if err != nil {
+		t.Errorf("Error: %v", err)
+	}
+
+	oo, err := s.GetManagedOrders(context.Background(), &gctrpc.GetOrdersRequest{
+		Exchange:  exchName,
+		AssetType: "spot",
+		Pair:      p,
+	})
+	if err != nil {
+		t.Errorf("non expected Error: %v", err)
+	} else if oo == nil || len(oo.GetOrders()) != 1 {
+		t.Errorf("unexpected order result: %v", oo)
+	}
+}