Merge pull request #1679 from c9s/c9s/xalign/notification

FIX: [xalign] fix max withdraw history api query

pkg/exchange/max/maxapi/account.go

@@ -201,14 +201,14 @@ type Withdraw struct {
 	Notes     string                     `json:"notes"`
 }
 
-//go:generate GetRequest -url "v2/withdrawals" -type GetWithdrawHistoryRequest -responseType []Withdraw
+//go:generate GetRequest -url "v3/withdrawals" -type GetWithdrawHistoryRequest -responseType []Withdraw
 type GetWithdrawHistoryRequest struct {
 	client requestgen.AuthenticatedAPIClient
 
-	currency string     `param:"currency"`
+	currency *string    `param:"currency"`
+	state    *string    `param:"state"`        // submitting, submitted, rejected, accepted, checking, refunded, canceled, suspect
 	from     *time.Time `param:"from,seconds"` // seconds
 	to       *time.Time `param:"to,seconds"`   // seconds
-	state    *string    `param:"state"`        // submitting, submitted, rejected, accepted, checking, refunded, canceled, suspect
 	limit    *int       `param:"limit"`
 }
 

pkg/exchange/max/maxapi/get_account_request_requestgen.go

@@ -119,13 +119,21 @@ func (g *GetAccountRequest) GetSlugsMap() (map[string]string, error) {
 	return slugs, nil
 }
 
+// GetPath returns the request path of the API
+func (g *GetAccountRequest) GetPath() string {
+	return "v2/members/accounts/:currency"
+}
+
+// Do generates the request object and send the request object to the API endpoint
 func (g *GetAccountRequest) Do(ctx context.Context) (*Account, error) {
 
 	// no body params
 	var params interface{}
 	query := url.Values{}
 
-	apiURL := "v2/members/accounts/:currency"
+	var apiURL string
+
+	apiURL = g.GetPath()
 	slugs, err := g.GetSlugsMap()
 	if err != nil {
 		return nil, err
@@ -144,8 +152,32 @@ func (g *GetAccountRequest) Do(ctx context.Context) (*Account, error) {
 	}
 
 	var apiResponse Account
-	if err := response.DecodeJSON(&apiResponse); err != nil {
-		return nil, err
+
+	type responseUnmarshaler interface {
+		Unmarshal(data []byte) error
+	}
+
+	if unmarshaler, ok := interface{}(&apiResponse).(responseUnmarshaler); ok {
+		if err := unmarshaler.Unmarshal(response.Body); err != nil {
+			return nil, err
+		}
+	} else {
+		// The line below checks the content type, however, some API server might not send the correct content type header,
+		// Hence, this is commented for backward compatibility
+		// response.IsJSON()
+		if err := response.DecodeJSON(&apiResponse); err != nil {
+			return nil, err
+		}
+	}
+
+	type responseValidator interface {
+		Validate() error
+	}
+
+	if validator, ok := interface{}(&apiResponse).(responseValidator); ok {
+		if err := validator.Validate(); err != nil {
+			return nil, err
+		}
 	}
 	return &apiResponse, nil
 }

pkg/exchange/max/maxapi/get_accounts_request_requestgen.go

@@ -109,13 +109,21 @@ func (g *GetAccountsRequest) GetSlugsMap() (map[string]string, error) {
 	return slugs, nil
 }
 
+// GetPath returns the request path of the API
+func (g *GetAccountsRequest) GetPath() string {
+	return "v2/members/accounts"
+}
+
+// Do generates the request object and send the request object to the API endpoint
 func (g *GetAccountsRequest) Do(ctx context.Context) ([]Account, error) {
 
 	// no body params
 	var params interface{}
 	query := url.Values{}
 
-	apiURL := "v2/members/accounts"
+	var apiURL string
+
+	apiURL = g.GetPath()
 
 	req, err := g.client.NewAuthenticatedRequest(ctx, "GET", apiURL, query, params)
 	if err != nil {
@@ -128,8 +136,32 @@ func (g *GetAccountsRequest) Do(ctx context.Context) ([]Account, error) {
 	}
 
 	var apiResponse []Account
-	if err := response.DecodeJSON(&apiResponse); err != nil {
-		return nil, err
+
+	type responseUnmarshaler interface {
+		Unmarshal(data []byte) error
+	}
+
+	if unmarshaler, ok := interface{}(&apiResponse).(responseUnmarshaler); ok {
+		if err := unmarshaler.Unmarshal(response.Body); err != nil {
+			return nil, err
+		}
+	} else {
+		// The line below checks the content type, however, some API server might not send the correct content type header,
+		// Hence, this is commented for backward compatibility
+		// response.IsJSON()
+		if err := response.DecodeJSON(&apiResponse); err != nil {
+			return nil, err
+		}
+	}
+
+	type responseValidator interface {
+		Validate() error
+	}
+
+	if validator, ok := interface{}(&apiResponse).(responseValidator); ok {
+		if err := validator.Validate(); err != nil {
+			return nil, err
+		}
 	}
 	return apiResponse, nil
 }

pkg/exchange/max/maxapi/get_deposit_history_request_requestgen.go

@@ -178,6 +178,12 @@ func (g *GetDepositHistoryRequest) GetSlugsMap() (map[string]string, error) {
 	return slugs, nil
 }
 
+// GetPath returns the request path of the API
+func (g *GetDepositHistoryRequest) GetPath() string {
+	return "v2/deposits"
+}
+
+// Do generates the request object and send the request object to the API endpoint
 func (g *GetDepositHistoryRequest) Do(ctx context.Context) ([]Deposit, error) {
 
 	// empty params for GET operation
@@ -187,7 +193,9 @@ func (g *GetDepositHistoryRequest) Do(ctx context.Context) ([]Deposit, error) {
 		return nil, err
 	}
 
-	apiURL := "v2/deposits"
+	var apiURL string
+
+	apiURL = g.GetPath()
 
 	req, err := g.client.NewAuthenticatedRequest(ctx, "GET", apiURL, query, params)
 	if err != nil {
@@ -200,8 +208,32 @@ func (g *GetDepositHistoryRequest) Do(ctx context.Context) ([]Deposit, error) {
 	}
 
 	var apiResponse []Deposit
-	if err := response.DecodeJSON(&apiResponse); err != nil {
-		return nil, err
+
+	type responseUnmarshaler interface {
+		Unmarshal(data []byte) error
+	}
+
+	if unmarshaler, ok := interface{}(&apiResponse).(responseUnmarshaler); ok {
+		if err := unmarshaler.Unmarshal(response.Body); err != nil {
+			return nil, err
+		}
+	} else {
+		// The line below checks the content type, however, some API server might not send the correct content type header,
+		// Hence, this is commented for backward compatibility
+		// response.IsJSON()
+		if err := response.DecodeJSON(&apiResponse); err != nil {
+			return nil, err
+		}
+	}
+
+	type responseValidator interface {
+		Validate() error
+	}
+
+	if validator, ok := interface{}(&apiResponse).(responseValidator); ok {
+		if err := validator.Validate(); err != nil {
+			return nil, err
+		}
 	}
 	return apiResponse, nil
 }

pkg/exchange/max/maxapi/get_vip_level_request_requestgen.go

@@ -109,13 +109,21 @@ func (g *GetVipLevelRequest) GetSlugsMap() (map[string]string, error) {
 	return slugs, nil
 }
 
+// GetPath returns the request path of the API
+func (g *GetVipLevelRequest) GetPath() string {
+	return "v2/members/vip_level"
+}
+
+// Do generates the request object and send the request object to the API endpoint
 func (g *GetVipLevelRequest) Do(ctx context.Context) (*VipLevel, error) {
 
 	// no body params
 	var params interface{}
 	query := url.Values{}
 
-	apiURL := "v2/members/vip_level"
+	var apiURL string
+
+	apiURL = g.GetPath()
 
 	req, err := g.client.NewAuthenticatedRequest(ctx, "GET", apiURL, query, params)
 	if err != nil {
@@ -128,8 +136,32 @@ func (g *GetVipLevelRequest) Do(ctx context.Context) (*VipLevel, error) {
 	}
 
 	var apiResponse VipLevel
-	if err := response.DecodeJSON(&apiResponse); err != nil {
-		return nil, err
+
+	type responseUnmarshaler interface {
+		Unmarshal(data []byte) error
+	}
+
+	if unmarshaler, ok := interface{}(&apiResponse).(responseUnmarshaler); ok {
+		if err := unmarshaler.Unmarshal(response.Body); err != nil {
+			return nil, err
+		}
+	} else {
+		// The line below checks the content type, however, some API server might not send the correct content type header,
+		// Hence, this is commented for backward compatibility
+		// response.IsJSON()
+		if err := response.DecodeJSON(&apiResponse); err != nil {
+			return nil, err
+		}
+	}
+
+	type responseValidator interface {
+		Validate() error
+	}
+
+	if validator, ok := interface{}(&apiResponse).(responseValidator); ok {
+		if err := validator.Validate(); err != nil {
+			return nil, err
+		}
 	}
 	return &apiResponse, nil
 }

pkg/exchange/max/maxapi/get_withdraw_history_request_requestgen.go

@@ -1,4 +1,4 @@
-// Code generated by "requestgen -method GET -url v2/withdrawals -type GetWithdrawHistoryRequest -responseType []Withdraw"; DO NOT EDIT.
+// Code generated by "requestgen -method GET -url v3/withdrawals -type GetWithdrawHistoryRequest -responseType []Withdraw"; DO NOT EDIT.
 
 package max
 
@@ -14,7 +14,12 @@ import (
 )
 
 func (g *GetWithdrawHistoryRequest) Currency(currency string) *GetWithdrawHistoryRequest {
-	g.currency = currency
+	g.currency = &currency
+	return g
+}
+
+func (g *GetWithdrawHistoryRequest) State(state string) *GetWithdrawHistoryRequest {
+	g.state = &state
 	return g
 }
 
@@ -28,11 +33,6 @@ func (g *GetWithdrawHistoryRequest) To(to time.Time) *GetWithdrawHistoryRequest
 	return g
 }
 
-func (g *GetWithdrawHistoryRequest) State(state string) *GetWithdrawHistoryRequest {
-	g.state = &state
-	return g
-}
-
 func (g *GetWithdrawHistoryRequest) Limit(limit int) *GetWithdrawHistoryRequest {
 	g.limit = &limit
 	return g
@@ -54,10 +54,21 @@ func (g *GetWithdrawHistoryRequest) GetQueryParameters() (url.Values, error) {
 func (g *GetWithdrawHistoryRequest) GetParameters() (map[string]interface{}, error) {
 	var params = map[string]interface{}{}
 	// check currency field -> json key currency
-	currency := g.currency
+	if g.currency != nil {
+		currency := *g.currency
 
-	// assign parameter of currency
-	params["currency"] = currency
+		// assign parameter of currency
+		params["currency"] = currency
+	} else {
+	}
+	// check state field -> json key state
+	if g.state != nil {
+		state := *g.state
+
+		// assign parameter of state
+		params["state"] = state
+	} else {
+	}
 	// check from field -> json key from
 	if g.from != nil {
 		from := *g.from
@@ -76,14 +87,6 @@ func (g *GetWithdrawHistoryRequest) GetParameters() (map[string]interface{}, err
 		params["to"] = strconv.FormatInt(to.Unix(), 10)
 	} else {
 	}
-	// check state field -> json key state
-	if g.state != nil {
-		state := *g.state
-
-		// assign parameter of state
-		params["state"] = state
-	} else {
-	}
 	// check limit field -> json key limit
 	if g.limit != nil {
 		limit := *g.limit
@@ -175,6 +178,12 @@ func (g *GetWithdrawHistoryRequest) GetSlugsMap() (map[string]string, error) {
 	return slugs, nil
 }
 
+// GetPath returns the request path of the API
+func (g *GetWithdrawHistoryRequest) GetPath() string {
+	return "v3/withdrawals"
+}
+
+// Do generates the request object and send the request object to the API endpoint
 func (g *GetWithdrawHistoryRequest) Do(ctx context.Context) ([]Withdraw, error) {
 
 	// empty params for GET operation
@@ -184,7 +193,9 @@ func (g *GetWithdrawHistoryRequest) Do(ctx context.Context) ([]Withdraw, error)
 		return nil, err
 	}
 
-	apiURL := "v2/withdrawals"
+	var apiURL string
+
+	apiURL = g.GetPath()
 
 	req, err := g.client.NewAuthenticatedRequest(ctx, "GET", apiURL, query, params)
 	if err != nil {
@@ -197,8 +208,32 @@ func (g *GetWithdrawHistoryRequest) Do(ctx context.Context) ([]Withdraw, error)
 	}
 
 	var apiResponse []Withdraw
-	if err := response.DecodeJSON(&apiResponse); err != nil {
-		return nil, err
+
+	type responseUnmarshaler interface {
+		Unmarshal(data []byte) error
+	}
+
+	if unmarshaler, ok := interface{}(&apiResponse).(responseUnmarshaler); ok {
+		if err := unmarshaler.Unmarshal(response.Body); err != nil {
+			return nil, err
+		}
+	} else {
+		// The line below checks the content type, however, some API server might not send the correct content type header,
+		// Hence, this is commented for backward compatibility
+		// response.IsJSON()
+		if err := response.DecodeJSON(&apiResponse); err != nil {
+			return nil, err
+		}
+	}
+
+	type responseValidator interface {
+		Validate() error
+	}
+
+	if validator, ok := interface{}(&apiResponse).(responseValidator); ok {
+		if err := validator.Validate(); err != nil {
+			return nil, err
+		}
 	}
 	return apiResponse, nil
 }

pkg/priceresolver/simple.go

@@ -0,0 +1,119 @@
+package priceresolver
+
+import (
+	"sync"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+// SimplePriceResolver implements a map-structure-based price index
+type SimplePriceResolver struct {
+	// symbolPrices stores the latest trade price by mapping symbol to price
+	symbolPrices map[string]fixedpoint.Value
+	markets      types.MarketMap
+
+	// pricesByBase stores the prices by currency names as a 2-level map
+	// BTC -> USDT -> 48000.0
+	// BTC -> TWD -> 1536000
+	pricesByBase map[string]map[string]fixedpoint.Value
+
+	// pricesByQuote is for reversed pairs, like USDT/TWD or BNB/BTC
+	// the reason that we don't store the reverse pricing in the same map is:
+	// expression like (1/price) could produce precision issue since the data type is fixed-point, only 8 fraction numbers are supported.
+	pricesByQuote map[string]map[string]fixedpoint.Value
+
+	mu sync.Mutex
+}
+
+func NewSimplePriceResolver(markets types.MarketMap) *SimplePriceResolver {
+	return &SimplePriceResolver{
+		markets:       markets,
+		symbolPrices:  make(map[string]fixedpoint.Value),
+		pricesByBase:  make(map[string]map[string]fixedpoint.Value),
+		pricesByQuote: make(map[string]map[string]fixedpoint.Value),
+	}
+}
+
+func (m *SimplePriceResolver) Update(symbol string, price fixedpoint.Value) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	m.symbolPrices[symbol] = price
+	market, ok := m.markets[symbol]
+	if !ok {
+		log.Warnf("market info %s not found, unable to update price", symbol)
+		return
+	}
+
+	quoteMap, ok2 := m.pricesByBase[market.BaseCurrency]
+	if !ok2 {
+		quoteMap = make(map[string]fixedpoint.Value)
+		m.pricesByBase[market.BaseCurrency] = quoteMap
+	}
+
+	quoteMap[market.QuoteCurrency] = price
+
+	baseMap, ok3 := m.pricesByQuote[market.QuoteCurrency]
+	if !ok3 {
+		baseMap = make(map[string]fixedpoint.Value)
+		m.pricesByQuote[market.QuoteCurrency] = baseMap
+	}
+
+	baseMap[market.BaseCurrency] = price
+}
+
+func (m *SimplePriceResolver) UpdateFromTrade(trade types.Trade) {
+	m.Update(trade.Symbol, trade.Price)
+}
+
+func (m *SimplePriceResolver) inferencePrice(asset string, assetPrice fixedpoint.Value, preferredFiats ...string) (fixedpoint.Value, bool) {
+	// log.Infof("inferencePrice %s = %f", asset, assetPrice.Float64())
+	quotePrices, ok := m.pricesByBase[asset]
+	if ok {
+		for quote, price := range quotePrices {
+			for _, fiat := range preferredFiats {
+				if quote == fiat {
+					return price.Mul(assetPrice), true
+				}
+			}
+		}
+
+		for quote, price := range quotePrices {
+			if infPrice, ok := m.inferencePrice(quote, price.Mul(assetPrice), preferredFiats...); ok {
+				return infPrice, true
+			}
+		}
+	}
+
+	// for example, quote = TWD here, we can get a price map with:
+	// USDT: 32.0 (for USDT/TWD at 32.0)
+	basePrices, ok := m.pricesByQuote[asset]
+	if ok {
+		for base, basePrice := range basePrices {
+			// log.Infof("base %s @ %s", base, basePrice.String())
+			for _, fiat := range preferredFiats {
+				if base == fiat {
+					// log.Infof("ret %f / %f = %f", assetPrice.Float64(), basePrice.Float64(), assetPrice.Div(basePrice).Float64())
+					return assetPrice.Div(basePrice), true
+				}
+			}
+		}
+
+		for base, basePrice := range basePrices {
+			if infPrice, ok2 := m.inferencePrice(base, assetPrice.Div(basePrice), preferredFiats...); ok2 {
+				return infPrice, true
+			}
+		}
+	}
+
+	return fixedpoint.Zero, false
+}
+
+func (m *SimplePriceResolver) ResolvePrice(asset string, preferredFiats ...string) (fixedpoint.Value, bool) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.inferencePrice(asset, fixedpoint.One, preferredFiats...)
+}

pkg/priceresolver/simple_test.go

@@ -0,0 +1,146 @@
+package priceresolver
+
+import (
+	"testing"
+
+	. "github.com/c9s/bbgo/pkg/testing/testhelper"
+	"github.com/c9s/bbgo/pkg/types"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSimplePriceResolver(t *testing.T) {
+	markets := types.MarketMap{
+		"BTCUSDT": types.Market{
+			BaseCurrency:  "BTC",
+			QuoteCurrency: "USDT",
+		},
+		"ETHUSDT": types.Market{
+			BaseCurrency:  "ETH",
+			QuoteCurrency: "USDT",
+		},
+		"BTCTWD": types.Market{
+			BaseCurrency:  "BTC",
+			QuoteCurrency: "TWD",
+		},
+		"ETHTWD": types.Market{
+			BaseCurrency:  "ETH",
+			QuoteCurrency: "TWD",
+		},
+		"USDTTWD": types.Market{
+			BaseCurrency:  "USDT",
+			QuoteCurrency: "TWD",
+		},
+		"ETHBTC": types.Market{
+			BaseCurrency:  "ETH",
+			QuoteCurrency: "BTC",
+		},
+	}
+
+	t.Run("direct reference", func(t *testing.T) {
+		pm := NewSimplePriceResolver(markets)
+		pm.UpdateFromTrade(types.Trade{
+			Symbol: "BTCUSDT",
+			Price:  Number(48000.0),
+		})
+		pm.UpdateFromTrade(types.Trade{
+			Symbol: "ETHUSDT",
+			Price:  Number(2800.0),
+		})
+		pm.UpdateFromTrade(types.Trade{
+			Symbol: "USDTTWD",
+			Price:  Number(32.0),
+		})
+
+		finalPrice, ok := pm.ResolvePrice("BTC", "USDT")
+		if assert.True(t, ok) {
+			assert.Equal(t, "48000", finalPrice.String())
+		}
+
+		finalPrice, ok = pm.ResolvePrice("ETH", "USDT")
+		if assert.True(t, ok) {
+			assert.Equal(t, "2800", finalPrice.String())
+		}
+
+		finalPrice, ok = pm.ResolvePrice("USDT", "TWD")
+		if assert.True(t, ok) {
+			assert.Equal(t, "32", finalPrice.String())
+		}
+	})
+
+	t.Run("simple reference", func(t *testing.T) {
+		pm := NewSimplePriceResolver(markets)
+		pm.UpdateFromTrade(types.Trade{
+			Symbol: "BTCUSDT",
+			Price:  Number(48000.0),
+		})
+		pm.UpdateFromTrade(types.Trade{
+			Symbol: "ETHUSDT",
+			Price:  Number(2800.0),
+		})
+		pm.UpdateFromTrade(types.Trade{
+			Symbol: "USDTTWD",
+			Price:  Number(32.0),
+		})
+
+		finalPrice, ok := pm.ResolvePrice("BTC", "TWD")
+		if assert.True(t, ok) {
+			assert.Equal(t, "1536000", finalPrice.String())
+		}
+	})
+
+	t.Run("crypto reference", func(t *testing.T) {
+		pm := NewSimplePriceResolver(markets)
+		pm.UpdateFromTrade(types.Trade{
+			Symbol: "BTCUSDT",
+			Price:  Number(52000.0),
+		})
+		pm.UpdateFromTrade(types.Trade{
+			Symbol: "ETHBTC",
+			Price:  Number(0.055),
+		})
+		pm.UpdateFromTrade(types.Trade{
+			Symbol: "USDTTWD",
+			Price:  Number(32.0),
+		})
+
+		finalPrice, ok := pm.ResolvePrice("ETH", "USDT")
+		if assert.True(t, ok) {
+			assert.Equal(t, "2860", finalPrice.String())
+		}
+	})
+
+	t.Run("inverse reference", func(t *testing.T) {
+		pm := NewSimplePriceResolver(markets)
+		pm.UpdateFromTrade(types.Trade{
+			Symbol: "BTCTWD",
+			Price:  Number(1536000.0),
+		})
+		pm.UpdateFromTrade(types.Trade{
+			Symbol: "USDTTWD",
+			Price:  Number(32.0),
+		})
+
+		finalPrice, ok := pm.ResolvePrice("BTC", "USDT")
+		if assert.True(t, ok) {
+			assert.Equal(t, "48000", finalPrice.String())
+		}
+	})
+
+	t.Run("inverse reference", func(t *testing.T) {
+		pm := NewSimplePriceResolver(markets)
+		pm.UpdateFromTrade(types.Trade{
+			Symbol: "BTCTWD",
+			Price:  Number(1536000.0),
+		})
+		pm.UpdateFromTrade(types.Trade{
+			Symbol: "USDTTWD",
+			Price:  Number(32.0),
+		})
+
+		finalPrice, ok := pm.ResolvePrice("TWD", "USDT")
+		if assert.True(t, ok) {
+			assert.InDelta(t, 0.03125, finalPrice.Float64(), 0.0001)
+		}
+	})
+}

pkg/strategy/xalign/strategy.go

@@ -9,10 +9,12 @@ import (
 	"time"
 
 	"github.com/sirupsen/logrus"
+	"golang.org/x/time/rate"
 
 	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/core"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/priceresolver"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
@@ -20,6 +22,8 @@ const ID = "xalign"
 
 var log = logrus.WithField("strategy", ID)
 
+var activeTransferNotificationLimiter = rate.NewLimiter(rate.Every(5*time.Minute), 1)
+
 func init() {
 	bbgo.RegisterStrategy(ID, &Strategy{})
 }
@@ -47,8 +51,14 @@ type Strategy struct {
 	Duration                 types.Duration              `json:"for"`
 	MaxAmounts               map[string]fixedpoint.Value `json:"maxAmounts"`
 
+	SlackNotify                bool             `json:"slackNotify"`
+	SlackNotifyMentions        []string         `json:"slackNotifyMentions"`
+	SlackNotifyThresholdAmount fixedpoint.Value `json:"slackNotifyThresholdAmount,omitempty"`
+
 	faultBalanceRecords map[string][]TimeBalance
 
+	priceResolver *priceresolver.SimplePriceResolver
+
 	sessions   map[string]*bbgo.ExchangeSession
 	orderBooks map[string]*bbgo.ActiveOrderBook
 
@@ -114,7 +124,10 @@ func (s *Strategy) aggregateBalances(
 	return totalBalances, sessionBalances
 }
 
-func (s *Strategy) detectActiveTransfers(ctx context.Context, sessions map[string]*bbgo.ExchangeSession) (bool, error) {
+func (s *Strategy) detectActiveWithdraw(
+	ctx context.Context,
+	sessions map[string]*bbgo.ExchangeSession,
+) (*types.Withdraw, error) {
 	var err2 error
 	until := time.Now()
 	since := until.Add(-time.Hour * 24)
@@ -134,12 +147,12 @@ func (s *Strategy) detectActiveTransfers(ctx context.Context, sessions map[strin
 		for _, withdraw := range withdraws {
 			switch withdraw.Status {
 			case types.WithdrawStatusProcessing, types.WithdrawStatusSent, types.WithdrawStatusAwaitingApproval:
-				return true, nil
+				return &withdraw, nil
 			}
 		}
 	}
 
-	return false, err2
+	return nil, err2
 }
 
 func (s *Strategy) selectSessionForCurrency(
@@ -340,6 +353,7 @@ func (s *Strategy) CrossRun(ctx context.Context, _ bbgo.OrderExecutionRouter, se
 
 	s.orderStore = core.NewOrderStore("")
 
+	markets := types.MarketMap{}
 	for _, sessionName := range s.PreferredSessions {
 		session, ok := sessions[sessionName]
 		if !ok {
@@ -353,8 +367,12 @@ func (s *Strategy) CrossRun(ctx context.Context, _ bbgo.OrderExecutionRouter, se
 		s.orderBooks[sessionName] = orderBook
 
 		s.sessions[sessionName] = session
+
+		// session.Market(symbol)
 	}
 
+	s.priceResolver = priceresolver.NewSimplePriceResolver(markets)
+
 	bbgo.OnShutdown(ctx, func(ctx context.Context, wg *sync.WaitGroup) {
 		defer wg.Done()
 		for n, session := range s.sessions {
@@ -405,7 +423,6 @@ func (s *Strategy) recordBalance(totalBalances types.BalanceMap) {
 }
 
 func (s *Strategy) align(ctx context.Context, sessions map[string]*bbgo.ExchangeSession) {
-
 	for sessionName, session := range sessions {
 		ob, ok := s.orderBooks[sessionName]
 		if !ok {
@@ -414,16 +431,20 @@ func (s *Strategy) align(ctx context.Context, sessions map[string]*bbgo.Exchange
 		}
 		if ok {
 			if err := ob.GracefulCancel(ctx, session.Exchange); err != nil {
-				log.WithError(err).Errorf("can not cancel order")
+				log.WithError(err).Errorf("unable to cancel order")
 			}
 		}
 	}
 
-	foundActiveTransfer, err := s.detectActiveTransfers(ctx, sessions)
+	pendingWithdraw, err := s.detectActiveWithdraw(ctx, sessions)
 	if err != nil {
 		log.WithError(err).Errorf("unable to check active transfers")
-	} else if foundActiveTransfer {
+	} else if pendingWithdraw != nil {
 		log.Warnf("found active transfer, skip balance align check")
+
+		if activeTransferNotificationLimiter.Allow() {
+			bbgo.Notify("Found active withdraw, skip balance align", pendingWithdraw)
+		}
 		return
 	}