all: refactor NewAccountValueCalculator

pkg/bbgo/account_value_calc.go

@@ -39,7 +39,6 @@ func NewAccountValueCalculator(
 		priceSolver:   priceSolver,
 		session:       session,
 		quoteCurrency: quoteCurrency,
-		prices:        make(map[string]fixedpoint.Value),
 		tickers:       make(map[string]types.Ticker),
 	}
 }
@@ -62,99 +61,69 @@ func (c *AccountValueCalculator) UpdatePrices(ctx context.Context) error {
 	return c.priceSolver.UpdateFromTickers(ctx, c.session.Exchange, symbols...)
 }
 
-func (c *AccountValueCalculator) DebtValue(ctx context.Context) (fixedpoint.Value, error) {
+func (c *AccountValueCalculator) DebtValue() fixedpoint.Value {
-	debtValue := fixedpoint.Zero
-
-	if len(c.prices) == 0 {
-		if err := c.UpdatePrices(ctx); err != nil {
-			return debtValue, err
-		}
-	}
-
 	balances := c.session.Account.Balances()
-	for _, b := range balances {
+	return totalValueInQuote(balances, c.priceSolver, c.quoteCurrency, func(
-		symbol := b.Currency + c.quoteCurrency
+		prev fixedpoint.Value, b types.Balance, price fixedpoint.Value,
-		price, ok := c.prices[symbol]
+	) fixedpoint.Value {
-		if !ok {
+		return prev.Add(b.Debt().Mul(price))
-			continue
+	})
-		}
-
-		debtValue = debtValue.Add(b.Debt().Mul(price))
-	}
-
-	return debtValue, nil
 }
 
-func (c *AccountValueCalculator) MarketValue(ctx context.Context) (fixedpoint.Value, error) {
+func (c *AccountValueCalculator) MarketValue() fixedpoint.Value {
-	marketValue := fixedpoint.Zero
-
-	if len(c.prices) == 0 {
-		if err := c.UpdatePrices(ctx); err != nil {
-			return marketValue, err
-		}
-	}
-
 	balances := c.session.Account.Balances()
-	for _, b := range balances {
+	return totalValueInQuote(balances, c.priceSolver, c.quoteCurrency, func(
-		if b.Currency == c.quoteCurrency {
+		prev fixedpoint.Value, b types.Balance, price fixedpoint.Value,
-			marketValue = marketValue.Add(b.Total())
+	) fixedpoint.Value {
-			continue
+		return prev.Add(b.Total().Mul(price))
-		}
+	})
 
-		if c.priceSolver != nil {
-			if price, ok := c.priceSolver.ResolvePrice(b.Currency, c.quoteCurrency); ok {
-				marketValue = marketValue.Add(b.Total().Mul(price))
-			}
-		} else {
-			symbol := b.Currency + c.quoteCurrency
-			if price, ok := c.prices[symbol]; ok {
-				marketValue = marketValue.Add(b.Total().Mul(price))
-			}
-		}
-	}
-
-	return marketValue, nil
 }
 
-func (c *AccountValueCalculator) NetValue(ctx context.Context) (fixedpoint.Value, error) {
+func (c *AccountValueCalculator) NetValue() fixedpoint.Value {
-	if len(c.prices) == 0 {
+	balances := c.session.Account.Balances()
-		if err := c.UpdatePrices(ctx); err != nil {
+	return totalValueInQuote(balances, c.priceSolver, c.quoteCurrency, func(
-			return fixedpoint.Zero, err
+		prev fixedpoint.Value, b types.Balance, price fixedpoint.Value,
+	) fixedpoint.Value {
+		return prev.Add(b.Net().Mul(price))
+	})
+}
+
+func totalValueInQuote(
+	balances types.BalanceMap,
+	priceSolver *pricesolver.SimplePriceSolver,
+	quoteCurrency string,
+	algo func(prev fixedpoint.Value, b types.Balance, price fixedpoint.Value) fixedpoint.Value,
+) (totalValue fixedpoint.Value) {
+	totalValue = fixedpoint.Zero
+
+	for _, b := range balances {
+		if b.Currency == quoteCurrency {
+			totalValue = algo(totalValue, b, fixedpoint.One)
+			continue
+		} else if price, ok := priceSolver.ResolvePrice(b.Currency, quoteCurrency); ok {
+			totalValue = algo(totalValue, b, price)
 		}
 	}
 
-	balances := c.session.Account.Balances()
+	return totalValue
-	accountValue := calculateNetValueInQuote(balances, c.priceSolver, c.quoteCurrency)
-	return accountValue, nil
 }
 
 func calculateNetValueInQuote(
-	balances types.BalanceMap, priceSolver *pricesolver.SimplePriceSolver, quoteCurrency string,
+	balances types.BalanceMap,
-) (accountValue fixedpoint.Value) {
+	priceSolver *pricesolver.SimplePriceSolver,
-	accountValue = fixedpoint.Zero
+	quoteCurrency string,
-	for _, b := range balances {
+) fixedpoint.Value {
-		if b.Currency == quoteCurrency {
+	return totalValueInQuote(balances, priceSolver, quoteCurrency, func(
-			accountValue = accountValue.Add(b.Net())
+		prev fixedpoint.Value, b types.Balance, price fixedpoint.Value,
-			continue
+	) fixedpoint.Value {
-		}
+		return prev.Add(b.Net().Mul(price))
-
+	})
-		if price, ok := priceSolver.ResolvePrice(b.Currency, quoteCurrency); ok {
-			accountValue = accountValue.Add(b.Net().Mul(price))
-		}
-	}
-
-	return accountValue
 }
 
-func (c *AccountValueCalculator) AvailableQuote(ctx context.Context) (fixedpoint.Value, error) {
+func (c *AccountValueCalculator) AvailableQuote() (fixedpoint.Value, error) {
 	accountValue := fixedpoint.Zero
 
-	if len(c.prices) == 0 {
-		if err := c.UpdatePrices(ctx); err != nil {
-			return accountValue, err
-		}
-	}
-
 	balances := c.session.Account.Balances()
 	for _, b := range balances {
 		if b.Currency == c.quoteCurrency {
@@ -162,13 +131,9 @@ func (c *AccountValueCalculator) AvailableQuote(ctx context.Context) (fixedpoint
 			continue
 		}
 
-		symbol := b.Currency + c.quoteCurrency
+		if price, ok := c.priceSolver.ResolvePrice(b.Currency, c.quoteCurrency); ok {
-		price, ok := c.prices[symbol]
+			accountValue = accountValue.Add(b.Net().Mul(price))
-		if !ok {
-			continue
 		}
-
-		accountValue = accountValue.Add(b.Net().Mul(price))
 	}
 
 	return accountValue, nil
@@ -176,20 +141,15 @@ func (c *AccountValueCalculator) AvailableQuote(ctx context.Context) (fixedpoint
 
 // MarginLevel calculates the margin level from the asset market value and the debt value
 // See https://www.binance.com/en/support/faq/360030493931
-func (c *AccountValueCalculator) MarginLevel(ctx context.Context) (fixedpoint.Value, error) {
+func (c *AccountValueCalculator) MarginLevel() (fixedpoint.Value, error) {
-	marginLevel := fixedpoint.Zero
+	marketValue := c.MarketValue()
-	marketValue, err := c.MarketValue(ctx)
+	debtValue := c.DebtValue()
-	if err != nil {
+
-		return marginLevel, err
+	if marketValue.IsZero() || debtValue.IsZero() {
+		return fixedpoint.NewFromFloat(999.0), nil
 	}
 
-	debtValue, err := c.DebtValue(ctx)
+	return marketValue.Div(debtValue), nil
-	if err != nil {
-		return marginLevel, err
-	}
-
-	marginLevel = marketValue.Div(debtValue)
-	return marginLevel, nil
 }
 
 func aggregateUsdNetValue(balances types.BalanceMap) fixedpoint.Value {
@@ -255,11 +215,7 @@ func CalculateBaseQuantity(
 		totalUsdValue = aggregateUsdNetValue(balances)
 	} else if len(restBalances) > 1 {
 		accountValue := NewAccountValueCalculator(session, nil, "USDT")
-		netValue, err := accountValue.NetValue(context.Background())
+		netValue := accountValue.NetValue()
-		if err != nil {
-			return quantity, err
-		}
-
 		totalUsdValue = netValue
 	} else {
 		// TODO: translate quote currency like BTC of ETH/BTC to usd value
@@ -362,7 +318,7 @@ func CalculateQuoteQuantity(
 
 	// using leverage -- starts from here
 	accountValue := NewAccountValueCalculator(session, nil, quoteCurrency)
-	availableQuote, err := accountValue.AvailableQuote(ctx)
+	availableQuote, err := accountValue.AvailableQuote()
 	if err != nil {
 		log.WithError(err).Errorf("can not update available quote")
 		return fixedpoint.Zero, err

pkg/bbgo/account_value_calc_test.go

@@ -3,7 +3,6 @@ package bbgo
 import (
 	"context"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/mock/gomock"
@@ -15,19 +14,6 @@ import (
 	"github.com/c9s/bbgo/pkg/types/mocks"
 )
 
-func newTestTicker() types.Ticker {
-	return types.Ticker{
-		Time:   time.Now(),
-		Volume: fixedpoint.Zero,
-		Last:   fixedpoint.NewFromFloat(19000.0),
-		Open:   fixedpoint.NewFromFloat(19500.0),
-		High:   fixedpoint.NewFromFloat(19900.0),
-		Low:    fixedpoint.NewFromFloat(18800.0),
-		Buy:    fixedpoint.NewFromFloat(19500.0),
-		Sell:   fixedpoint.NewFromFloat(18900.0),
-	}
-}
-
 func TestAccountValueCalculator_NetValue(t *testing.T) {
 	symbol := "BTCUSDT"
 	markets := AllMarkets()
@@ -36,12 +22,11 @@ func TestAccountValueCalculator_NetValue(t *testing.T) {
 		mockCtrl := gomock.NewController(t)
 		defer mockCtrl.Finish()
 
+		ticker := Ticker(symbol)
 		mockEx := mocks.NewMockExchange(mockCtrl)
 		// for market data stream and user data stream
 		mockEx.EXPECT().NewStream().Return(&types.StandardStream{}).Times(2)
-		mockEx.EXPECT().QueryTickers(gomock.Any(), []string{symbol}).Return(map[string]types.Ticker{
+		mockEx.EXPECT().QueryTicker(gomock.Any(), symbol).Return(&ticker, nil).AnyTimes()
-			"BTCUSDT": Ticker(symbol),
-		}, nil)
 
 		session := NewExchangeSession("test", mockEx)
 		session.Account.UpdateBalances(types.BalanceMap{
@@ -64,14 +49,12 @@ func TestAccountValueCalculator_NetValue(t *testing.T) {
 		})
 		assert.NotNil(t, session)
 
-		ctx := context.Background()
 		priceSolver := pricesolver.NewSimplePriceResolver(markets)
+		priceSolver.Update(symbol, ticker.GetValidPrice())
 
 		cal := NewAccountValueCalculator(session, priceSolver, "USDT")
-		assert.NotNil(t, cal)
 
-		netValue, err := cal.NetValue(ctx)
+		netValue := cal.NetValue()
-		assert.NoError(t, err)
 		assert.Equal(t, "20000", netValue.String())
 	})
 
@@ -79,12 +62,12 @@ func TestAccountValueCalculator_NetValue(t *testing.T) {
 		mockCtrl := gomock.NewController(t)
 		defer mockCtrl.Finish()
 
+		ticker := Ticker(symbol)
+
 		mockEx := mocks.NewMockExchange(mockCtrl)
 		// for market data stream and user data stream
 		mockEx.EXPECT().NewStream().Return(&types.StandardStream{}).Times(2)
-		mockEx.EXPECT().QueryTickers(gomock.Any(), []string{symbol}).Return(map[string]types.Ticker{
+		mockEx.EXPECT().QueryTicker(gomock.Any(), symbol).Return(&ticker, nil).AnyTimes()
-			symbol: Ticker(symbol),
-		}, nil)
 
 		session := NewExchangeSession("test", mockEx)
 		session.Account.UpdateBalances(types.BalanceMap{
@@ -105,16 +88,12 @@ func TestAccountValueCalculator_NetValue(t *testing.T) {
 				NetAsset:  fixedpoint.Zero,
 			},
 		})
-		assert.NotNil(t, session)
 
-		ctx := context.Background()
 		priceSolver := pricesolver.NewSimplePriceResolver(markets)
+		priceSolver.Update(symbol, ticker.GetValidPrice())
 
 		cal := NewAccountValueCalculator(session, priceSolver, "USDT")
-		assert.NotNil(t, cal)
+		netValue := cal.NetValue()
-
-		netValue, err := cal.NetValue(ctx)
-		assert.NoError(t, err)
 		assert.Equal(t, "2000", netValue.String()) // 21000-19000
 	})
 }
@@ -123,14 +102,13 @@ func TestNewAccountValueCalculator_MarginLevel(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()
 
-	ticker := newTestTicker()
+	symbol := "BTCUSDT"
+	ticker := Ticker(symbol)
 
 	mockEx := mocks.NewMockExchange(mockCtrl)
 	// for market data stream and user data stream
 	mockEx.EXPECT().NewStream().Return(&types.StandardStream{}).Times(2)
-	mockEx.EXPECT().QueryTickers(gomock.Any(), []string{"BTCUSDT"}).Return(map[string]types.Ticker{
+	mockEx.EXPECT().QueryTicker(gomock.Any(), symbol).Return(&ticker, nil).AnyTimes()
-		"BTCUSDT": ticker,
-	}, nil)
 
 	session := NewExchangeSession("test", mockEx)
 	session.Account.UpdateBalances(types.BalanceMap{
@@ -142,27 +120,20 @@ func TestNewAccountValueCalculator_MarginLevel(t *testing.T) {
 			Interest:  fixedpoint.NewFromFloat(0.003),
 			NetAsset:  fixedpoint.Zero,
 		},
-		"USDT": {
+		"USDT": Balance("USDT", Number(21000.0)),
-			Currency:  "USDT",
-			Available: fixedpoint.NewFromFloat(21000.0),
-			Locked:    fixedpoint.Zero,
-			Borrowed:  fixedpoint.Zero,
-			Interest:  fixedpoint.Zero,
-			NetAsset:  fixedpoint.Zero,
-		},
 	})
 	assert.NotNil(t, session)
 
 	ctx := context.Background()
 	markets := AllMarkets()
 	priceSolver := pricesolver.NewSimplePriceResolver(markets)
-	err := priceSolver.UpdateFromTickers(ctx, mockEx, "BTCUSDT")
+	err := priceSolver.UpdateFromTickers(ctx, mockEx, symbol)
 	assert.NoError(t, err)
 
 	cal := NewAccountValueCalculator(session, priceSolver, "USDT")
 	assert.NotNil(t, cal)
 
-	marginLevel, err := cal.MarginLevel(ctx)
+	marginLevel, err := cal.MarginLevel()
 	assert.NoError(t, err)
 
 	// expected (21000 / 19000 * 1.003)

pkg/pricesolver/simple.go

@@ -99,7 +99,6 @@ func (m *SimplePriceSolver) UpdateFromTickers(ctx context.Context, ex types.Exch
 }
 
 func (m *SimplePriceSolver) 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 {
@@ -122,10 +121,8 @@ func (m *SimplePriceSolver) inferencePrice(asset string, assetPrice fixedpoint.V
 	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
 				}
 			}
@@ -142,6 +139,12 @@ func (m *SimplePriceSolver) inferencePrice(asset string, assetPrice fixedpoint.V
 }
 
 func (m *SimplePriceSolver) ResolvePrice(asset string, preferredFiats ...string) (fixedpoint.Value, bool) {
+	if len(preferredFiats) == 0 {
+		return fixedpoint.Zero, false
+	} else if asset == preferredFiats[0] {
+		return fixedpoint.One, true
+	}
+
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	return m.inferencePrice(asset, fixedpoint.One, preferredFiats...)

pkg/strategy/harmonic/strategy.go

@@ -6,13 +6,14 @@ import (
 	"os"
 	"sync"
 
+	"github.com/sirupsen/logrus"
+
 	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/data/tsv"
 	"github.com/c9s/bbgo/pkg/datatype/floats"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/c9s/bbgo/pkg/indicator"
 	"github.com/c9s/bbgo/pkg/types"
-	"github.com/sirupsen/logrus"
 )
 
 const ID = "harmonic"
@@ -29,7 +30,7 @@ type Strategy struct {
 	Market      types.Market
 
 	types.IntervalWindow
-	//bbgo.OpenPositionOptions
+	// bbgo.OpenPositionOptions
 
 	// persistence fields
 	Position    *types.Position    `persistence:"position"`
@@ -239,7 +240,7 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 		// Cancel active orders
 		_ = s.orderExecutor.GracefulCancel(ctx)
 		// Close 100% position
-		//_ = s.ClosePosition(ctx, fixedpoint.One)
+		// _ = s.ClosePosition(ctx, fixedpoint.One)
 	})
 
 	s.session = session
@@ -258,7 +259,7 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 	s.orderExecutor.BindTradeStats(s.TradeStats)
 
 	// AccountValueCalculator
-	s.AccountValueCalculator = bbgo.NewAccountValueCalculator(s.session, s.Market.QuoteCurrency)
+	s.AccountValueCalculator = bbgo.NewAccountValueCalculator(s.session, nil, s.Market.QuoteCurrency)
 
 	// Accumulated profit report
 	if bbgo.IsBackTesting {

pkg/strategy/irr/strategy.go

@@ -255,7 +255,7 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 	s.orderExecutor.BindTradeStats(s.TradeStats)
 
 	// AccountValueCalculator
-	s.AccountValueCalculator = bbgo.NewAccountValueCalculator(s.session, s.Market.QuoteCurrency)
+	s.AccountValueCalculator = bbgo.NewAccountValueCalculator(s.session, nil, s.Market.QuoteCurrency)
 
 	// Accumulated profit report
 	if bbgo.IsBackTesting {

pkg/strategy/supertrend/strategy.go

@@ -255,7 +255,9 @@ func (s *Strategy) getSide(stSignal types.Direction, demaSignal types.Direction,
 	return side
 }
 
-func (s *Strategy) generateOrderForm(side types.SideType, quantity fixedpoint.Value, marginOrderSideEffect types.MarginOrderSideEffectType) types.SubmitOrder {
+func (s *Strategy) generateOrderForm(
+	side types.SideType, quantity fixedpoint.Value, marginOrderSideEffect types.MarginOrderSideEffectType,
+) types.SubmitOrder {
 	orderForm := types.SubmitOrder{
 		Symbol:           s.Symbol,
 		Market:           s.Market,
@@ -382,7 +384,7 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 	}
 
 	// AccountValueCalculator
-	s.AccountValueCalculator = bbgo.NewAccountValueCalculator(s.session, s.Market.QuoteCurrency)
+	s.AccountValueCalculator = bbgo.NewAccountValueCalculator(s.session, nil, s.Market.QuoteCurrency)
 
 	// For drawing
 	profitSlice := floats.Slice{1., 1.}

pkg/strategy/xmaker/strategy.go

@@ -187,6 +187,8 @@ type Strategy struct {
 	logger logrus.FieldLogger
 
 	metricsLabels prometheus.Labels
+
+	connectivityGroup *types.ConnectivityGroup
 }
 
 func (s *Strategy) ID() string {
@@ -654,40 +656,37 @@ func (s *Strategy) updateQuote(ctx context.Context) error {
 				hedgeAccount.MarginLevel.String(),
 				s.MinMarginLevel.String())
 
-			netValueInUsd, calcErr := s.accountValueCalculator.NetValue(ctx)
+			netValueInUsd := s.accountValueCalculator.NetValue()
-			if calcErr != nil {
-				s.logger.WithError(calcErr).Errorf("unable to calculate the net value")
-			} else {
-				// calculate credit buffer
-				s.logger.Infof("hedge account net value in usd: %f", netValueInUsd.Float64())
 
-				maximumValueInUsd := netValueInUsd.Mul(s.MaxHedgeAccountLeverage)
+			// calculate credit buffer
+			s.logger.Infof("hedge account net value in usd: %f", netValueInUsd.Float64())
 
-				s.logger.Infof("hedge account maximum leveraged value in usd: %f (%f x)", maximumValueInUsd.Float64(), s.MaxHedgeAccountLeverage.Float64())
+			maximumValueInUsd := netValueInUsd.Mul(s.MaxHedgeAccountLeverage)
 
-				if quote, ok := hedgeAccount.Balance(s.sourceMarket.QuoteCurrency); ok {
+			s.logger.Infof("hedge account maximum leveraged value in usd: %f (%f x)", maximumValueInUsd.Float64(), s.MaxHedgeAccountLeverage.Float64())
-					debt := quote.Debt()
-					quota := maximumValueInUsd.Sub(debt)
 
-					s.logger.Infof("hedge account quote balance: %s, debt: %s, quota: %s",
+			if quote, ok := hedgeAccount.Balance(s.sourceMarket.QuoteCurrency); ok {
-						quote.String(),
+				debt := quote.Debt()
-						debt.String(),
+				quota := maximumValueInUsd.Sub(debt)
-						quota.String())
 
-					hedgeQuota.QuoteAsset.Add(quota)
+				s.logger.Infof("hedge account quote balance: %s, debt: %s, quota: %s",
-				}
+					quote.String(),
+					debt.String(),
+					quota.String())
 
-				if base, ok := hedgeAccount.Balance(s.sourceMarket.BaseCurrency); ok {
+				hedgeQuota.QuoteAsset.Add(quota)
-					debt := base.Debt()
+			}
-					quota := maximumValueInUsd.Div(bestAsk.Price).Sub(debt)
 
-					s.logger.Infof("hedge account base balance: %s, debt: %s, quota: %s",
+			if base, ok := hedgeAccount.Balance(s.sourceMarket.BaseCurrency); ok {
-						base.String(),
+				debt := base.Debt()
-						debt.String(),
+				quota := maximumValueInUsd.Div(bestAsk.Price).Sub(debt)
-						quota.String())
 
-					hedgeQuota.BaseAsset.Add(quota)
+				s.logger.Infof("hedge account base balance: %s, debt: %s, quota: %s",
-				}
+					base.String(),
+					debt.String(),
+					quota.String())
+
+				hedgeQuota.BaseAsset.Add(quota)
 			}
 		}
 	} else {
@@ -1322,12 +1321,7 @@ func (s *Strategy) accountUpdater(ctx context.Context) {
 				return
 			}
 
-			netValue, err := s.accountValueCalculator.NetValue(ctx)
+			netValue := s.accountValueCalculator.NetValue()
-			if err != nil {
-				log.WithError(err).Errorf("unable to update account")
-				return
-			}
-
 			s.logger.Infof("hedge session net value ~= %f USD", netValue.Float64())
 		}
 	}
@@ -1419,7 +1413,7 @@ func (s *Strategy) CrossRun(
 		return fmt.Errorf("maker session market %s is not defined", s.Symbol)
 	}
 
-	s.accountValueCalculator = bbgo.NewAccountValueCalculator(s.sourceSession, s.sourceMarket.QuoteCurrency)
+	s.accountValueCalculator = bbgo.NewAccountValueCalculator(s.sourceSession, nil, s.sourceMarket.QuoteCurrency)
 
 	indicators := s.sourceSession.Indicators(s.Symbol)
 
@@ -1622,13 +1616,27 @@ func (s *Strategy) CrossRun(
 
 	s.stopC = make(chan struct{})
 
+	sourceConnectivity := types.NewConnectivity()
+	sourceConnectivity.Bind(s.sourceSession.UserDataStream)
+
+	s.connectivityGroup = types.NewConnectivityGroup(sourceConnectivity)
+
 	if s.RecoverTrade {
 		go s.tradeRecover(ctx)
 	}
 
-	go s.accountUpdater(ctx)
+	go func() {
-	go s.hedgeWorker(ctx)
+		select {
-	go s.quoteWorker(ctx)
+		case <-ctx.Done():
+		case <-s.connectivityGroup.AllAuthedC(ctx, 15*time.Second):
+		}
+
+		s.logger.Infof("all user data streams are connected, starting workers...")
+
+		go s.accountUpdater(ctx)
+		go s.hedgeWorker(ctx)
+		go s.quoteWorker(ctx)
+	}()
 
 	bbgo.OnShutdown(ctx, func(ctx context.Context, wg *sync.WaitGroup) {
 		// the ctx here is the shutdown context (not the strategy context)

pkg/testing/testhelper/ticker.go

@@ -30,12 +30,23 @@ var _tickers = map[string]types.Ticker{
 		Buy:    fixedpoint.NewFromFloat(2519.0),
 		Sell:   fixedpoint.NewFromFloat(2521.0),
 	},
+
+	"USDTTWD": {
+		Time:   time.Now(),
+		Volume: fixedpoint.Zero,
+		Open:   fixedpoint.NewFromFloat(32.1),
+		High:   fixedpoint.NewFromFloat(32.31),
+		Low:    fixedpoint.NewFromFloat(32.01),
+		Last:   fixedpoint.NewFromFloat(32.0),
+		Buy:    fixedpoint.NewFromFloat(32.0),
+		Sell:   fixedpoint.NewFromFloat(32.01),
+	},
 }
 
 func Ticker(symbol string) types.Ticker {
 	ticker, ok := _tickers[symbol]
 	if !ok {
-		panic(fmt.Errorf("%s test ticker not found, valid tickers: %+v", symbol, []string{"BTCUSDT", "ETHUSDT"}))
+		panic(fmt.Errorf("%s test ticker not found, valid tickers: %+v", symbol, []string{"BTCUSDT", "ETHUSDT", "USDTTWD"}))
 	}
 
 	return ticker