Merge pull request #1766 from c9s/c9s/refactor/account-value-calc

REFACTOR: refactor account value calculator with price solver

doc/topics/price-solver.md

@@ -0,0 +1,22 @@
+# Using Price Solver
+
+Price solver is a tool to calculate the price of a market based on the prices of other markets. It is useful when you
+want to calculate the price of a market that is not directly available on the exchange.
+
+## Simple Price Solver
+
+Simple price solver is a price solver that calculates the price of a market based on the prices of other markets.
+
+You may add a field to the struct to store the price solver:
+
+	priceSolver    *pricesolver.SimplePriceSolver
+
+To use the simple price solver, you need to create an instance of `SimplePriceSolver` and bind the market data stream of
+the source markets to it.
+
+	s.priceSolver = pricesolver.NewSimplePriceResolver(sourceMarkets)
+	s.priceSolver.BindStream(s.sourceSession.MarketDataStream)
+
+To update the price of the target market, you may call the `UpdatePrice` method of the price solver.
+    
+    s.priceSolver.Update(symbol, price)

pkg/bbgo/account_value_calc.go

@@ -3,12 +3,12 @@ package bbgo
 import (
 	"context"
 	"fmt"
-	"time"
 
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 
 	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/pricesolver"
 	"github.com/c9s/bbgo/pkg/risk"
 	"github.com/c9s/bbgo/pkg/types"
 )
@@ -20,25 +20,29 @@ var maxIsolatedMarginLeverage = fixedpoint.NewFromInt(10)
 var maxCrossMarginLeverage = fixedpoint.NewFromInt(3)
 
 type AccountValueCalculator struct {
+	priceSolver   *pricesolver.SimplePriceSolver
 	session       *ExchangeSession
 	quoteCurrency string
-	prices        map[string]fixedpoint.Value
-	tickers       map[string]types.Ticker
-	updateTime    time.Time
 }
 
-func NewAccountValueCalculator(session *ExchangeSession, quoteCurrency string) *AccountValueCalculator {
+func NewAccountValueCalculator(
+	session *ExchangeSession,
+	priceSolver *pricesolver.SimplePriceSolver,
+	quoteCurrency string,
+) *AccountValueCalculator {
 	return &AccountValueCalculator{
+		priceSolver:   priceSolver,
 		session:       session,
 		quoteCurrency: quoteCurrency,
-		prices:        make(map[string]fixedpoint.Value),
-		tickers:       make(map[string]types.Ticker),
 	}
 }
 
+// UpdatePrices updates the price index from the existing balances
 func (c *AccountValueCalculator) UpdatePrices(ctx context.Context) error {
 	balances := c.session.Account.Balances()
 	currencies := balances.Currencies()
+	markets := c.session.Markets()
+
 	var symbols []string
 	for _, currency := range currencies {
 		if currency == c.quoteCurrency {
@@ -46,116 +50,79 @@ func (c *AccountValueCalculator) UpdatePrices(ctx context.Context) error {
 		}
 
 		symbol := currency + c.quoteCurrency
-		symbols = append(symbols, symbol)
+		reversedSymbol := c.quoteCurrency + currency
-	}
+		if _, ok := markets[symbol]; ok {
-
+			symbols = append(symbols, symbol)
-	tickers, err := c.session.Exchange.QueryTickers(ctx, symbols...)
+		} else if _, ok2 := markets[reversedSymbol]; ok2 {
-	if err != nil {
+			symbols = append(symbols, reversedSymbol)
-		return err
-	}
-
-	c.tickers = tickers
-	for symbol, ticker := range tickers {
-		c.prices[symbol] = ticker.Last
-		if ticker.Time.After(c.updateTime) {
-			c.updateTime = ticker.Time
 		}
 	}
-	return nil
+
+	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))
-		}
+	})
 
-		symbol := b.Currency + c.quoteCurrency
-		price, ok := c.prices[symbol]
-		if !ok {
-			continue
-		}
-
-		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 {
-		if err := c.UpdatePrices(ctx); err != nil {
-			return fixedpoint.Zero, err
-		}
-	}
-
 	balances := c.session.Account.Balances()
-	accountValue := calculateNetValueInQuote(balances, c.prices, c.quoteCurrency)
+	return totalValueInQuote(balances, c.priceSolver, c.quoteCurrency, func(
-	return accountValue, nil
+		prev fixedpoint.Value, b types.Balance, price fixedpoint.Value,
+	) fixedpoint.Value {
+		return prev.Add(b.Net().Mul(price))
+	})
 }
 
-func calculateNetValueInQuote(balances types.BalanceMap, prices types.PriceMap, quoteCurrency string) (accountValue fixedpoint.Value) {
+func totalValueInQuote(
-	accountValue = fixedpoint.Zero
+	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 {
-			accountValue = accountValue.Add(b.Net())
+			totalValue = algo(totalValue, b, fixedpoint.One)
 			continue
-		}
+		} else if price, ok := priceSolver.ResolvePrice(b.Currency, quoteCurrency); ok {
-
+			totalValue = algo(totalValue, b, price)
-		symbol := b.Currency + quoteCurrency        // for BTC/USDT, ETH/USDT pairs
-		symbolReverse := quoteCurrency + b.Currency // for USDT/USDC or USDT/TWD pairs
-		if price, ok := prices[symbol]; ok {
-			accountValue = accountValue.Add(b.Net().Mul(price))
-		} else if priceReverse, ok2 := prices[symbolReverse]; ok2 {
-			accountValue = accountValue.Add(b.Net().Div(priceReverse))
 		}
 	}
 
-	return accountValue
+	return totalValue
 }
 
-func (c *AccountValueCalculator) AvailableQuote(ctx context.Context) (fixedpoint.Value, error) {
+func calculateNetValueInQuote(
-	accountValue := fixedpoint.Zero
+	balances types.BalanceMap,
+	priceSolver *pricesolver.SimplePriceSolver,
+	quoteCurrency string,
+) fixedpoint.Value {
+	return totalValueInQuote(balances, priceSolver, quoteCurrency, func(
+		prev fixedpoint.Value, b types.Balance, price fixedpoint.Value,
+	) fixedpoint.Value {
+		return prev.Add(b.Net().Mul(price))
+	})
+}
 
-	if len(c.prices) == 0 {
+func (c *AccountValueCalculator) AvailableQuote() (fixedpoint.Value, error) {
-		if err := c.UpdatePrices(ctx); err != nil {
+	accountValue := fixedpoint.Zero
-			return accountValue, err
-		}
-	}
 
 	balances := c.session.Account.Balances()
 	for _, b := range balances {
@@ -164,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
@@ -178,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 {
@@ -220,7 +178,9 @@ func usdFiatBalances(balances types.BalanceMap) (fiats types.BalanceMap, rest ty
 	return fiats, rest
 }
 
-func CalculateBaseQuantity(session *ExchangeSession, market types.Market, price, quantity, leverage fixedpoint.Value) (fixedpoint.Value, error) {
+func CalculateBaseQuantity(
+	session *ExchangeSession, market types.Market, price, quantity, leverage fixedpoint.Value,
+) (fixedpoint.Value, error) {
 	// default leverage guard
 	if leverage.IsZero() {
 		leverage = defaultLeverage
@@ -249,17 +209,18 @@ func CalculateBaseQuantity(session *ExchangeSession, market types.Market, price,
 
 	usdBalances, restBalances := usdFiatBalances(balances)
 
-	// for isolated margin we can calculate from these two pair
+	// for isolated margin, we can calculate from these two pair
 	totalUsdValue := fixedpoint.Zero
 	if len(restBalances) == 1 && types.IsUSDFiatCurrency(market.QuoteCurrency) {
 		totalUsdValue = aggregateUsdNetValue(balances)
 	} else if len(restBalances) > 1 {
-		accountValue := NewAccountValueCalculator(session, "USDT")
+		priceSolver := pricesolver.NewSimplePriceResolver(session.Markets())
-		netValue, err := accountValue.NetValue(context.Background())
+		accountValue := NewAccountValueCalculator(session, priceSolver, "USDT")
-		if err != nil {
+		if err := accountValue.UpdatePrices(context.Background()); err != nil {
-			return quantity, err
+			return fixedpoint.Zero, err
 		}
 
+		netValue := accountValue.NetValue()
 		totalUsdValue = netValue
 	} else {
 		// TODO: translate quote currency like BTC of ETH/BTC to usd value
@@ -329,7 +290,9 @@ func CalculateBaseQuantity(session *ExchangeSession, market types.Market, price,
 		errors.New("quantity is zero, can not submit sell order, please check your settings"))
 }
 
-func CalculateQuoteQuantity(ctx context.Context, session *ExchangeSession, quoteCurrency string, leverage fixedpoint.Value) (fixedpoint.Value, error) {
+func CalculateQuoteQuantity(
+	ctx context.Context, session *ExchangeSession, quoteCurrency string, leverage fixedpoint.Value,
+) (fixedpoint.Value, error) {
 	// default leverage guard
 	if leverage.IsZero() {
 		leverage = defaultLeverage
@@ -359,8 +322,14 @@ func CalculateQuoteQuantity(ctx context.Context, session *ExchangeSession, quote
 	}
 
 	// using leverage -- starts from here
-	accountValue := NewAccountValueCalculator(session, quoteCurrency)
+	priceSolver := pricesolver.NewSimplePriceResolver(session.Markets())
-	availableQuote, err := accountValue.AvailableQuote(ctx)
+
+	accountValue := NewAccountValueCalculator(session, priceSolver, quoteCurrency)
+	if err := accountValue.UpdatePrices(ctx); err != nil {
+		return fixedpoint.Zero, err
+	}
+
+	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,41 +3,30 @@ package bbgo
 import (
 	"context"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/mock/gomock"
 
 	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/pricesolver"
+	. "github.com/c9s/bbgo/pkg/testing/testhelper"
 	"github.com/c9s/bbgo/pkg/types"
 	"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()
 
 	t.Run("borrow and available", func(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{"BTCUSDT"}).Return(map[string]types.Ticker{
+		mockEx.EXPECT().QueryTicker(gomock.Any(), symbol).Return(&ticker, nil).AnyTimes()
-			"BTCUSDT": newTestTicker(),
-		}, nil)
 
 		session := NewExchangeSession("test", mockEx)
 		session.Account.UpdateBalances(types.BalanceMap{
@@ -60,12 +49,12 @@ func TestAccountValueCalculator_NetValue(t *testing.T) {
 		})
 		assert.NotNil(t, session)
 
-		cal := NewAccountValueCalculator(session, "USDT")
+		priceSolver := pricesolver.NewSimplePriceResolver(markets)
-		assert.NotNil(t, cal)
+		priceSolver.Update(symbol, ticker.GetValidPrice())
 
-		ctx := context.Background()
+		cal := NewAccountValueCalculator(session, priceSolver, "USDT")
-		netValue, err := cal.NetValue(ctx)
+
-		assert.NoError(t, err)
+		netValue := cal.NetValue()
 		assert.Equal(t, "20000", netValue.String())
 	})
 
@@ -73,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{"BTCUSDT"}).Return(map[string]types.Ticker{
+		mockEx.EXPECT().QueryTicker(gomock.Any(), symbol).Return(&ticker, nil).AnyTimes()
-			"BTCUSDT": newTestTicker(),
-		}, nil)
 
 		session := NewExchangeSession("test", mockEx)
 		session.Account.UpdateBalances(types.BalanceMap{
@@ -99,14 +88,12 @@ func TestAccountValueCalculator_NetValue(t *testing.T) {
 				NetAsset:  fixedpoint.Zero,
 			},
 		})
-		assert.NotNil(t, session)
 
-		cal := NewAccountValueCalculator(session, "USDT")
+		priceSolver := pricesolver.NewSimplePriceResolver(markets)
-		assert.NotNil(t, cal)
+		priceSolver.Update(symbol, ticker.GetValidPrice())
 
-		ctx := context.Background()
+		cal := NewAccountValueCalculator(session, priceSolver, "USDT")
-		netValue, err := cal.NetValue(ctx)
+		netValue := cal.NetValue()
-		assert.NoError(t, err)
 		assert.Equal(t, "2000", netValue.String()) // 21000-19000
 	})
 }
@@ -115,12 +102,13 @@ func TestNewAccountValueCalculator_MarginLevel(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	defer mockCtrl.Finish()
 
+	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": newTestTicker(),
-	}, nil)
 
 	session := NewExchangeSession("test", mockEx)
 	session.Account.UpdateBalances(types.BalanceMap{
@@ -132,22 +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)
 
-	cal := NewAccountValueCalculator(session, "USDT")
+	ctx := context.Background()
+	markets := AllMarkets()
+	priceSolver := pricesolver.NewSimplePriceResolver(markets)
+	err := priceSolver.UpdateFromTickers(ctx, mockEx, symbol)
+	assert.NoError(t, err)
+
+	cal := NewAccountValueCalculator(session, priceSolver, "USDT")
 	assert.NotNil(t, cal)
 
-	ctx := context.Background()
+	marginLevel, err := cal.MarginLevel()
-	marginLevel, err := cal.MarginLevel(ctx)
 	assert.NoError(t, err)
 
 	// expected (21000 / 19000 * 1.003)
@@ -172,14 +158,13 @@ func Test_aggregateUsdValue(t *testing.T) {
 		{
 			name: "mixed",
 			args: args{
-				balances: types.BalanceMap{
+				balances: BalancesFromText(`
-					"USDC": types.Balance{Currency: "USDC", Available: number(70.0)},
+					USDC, 150.0
-					"USDT": types.Balance{Currency: "USDT", Available: number(100.0)},
+					USDT, 100.0
-					"BUSD": types.Balance{Currency: "BUSD", Available: number(80.0)},
+					BTC, 0.01
-					"BTC":  types.Balance{Currency: "BTC", Available: number(0.01)},
+					`),
-				},
 			},
-			want: number(250.0),
+			want: Number(250.0),
 		},
 	}
 	for _, tt := range tests {
@@ -201,21 +186,19 @@ func Test_usdFiatBalances(t *testing.T) {
 	}{
 		{
 			args: args{
-				balances: types.BalanceMap{
+				balances: BalancesFromText(`
-					"USDC": types.Balance{Currency: "USDC", Available: number(70.0)},
+					USDC, 150.0
-					"USDT": types.Balance{Currency: "USDT", Available: number(100.0)},
+					USDT, 100.0
-					"BUSD": types.Balance{Currency: "BUSD", Available: number(80.0)},
+					BTC, 0.01
-					"BTC":  types.Balance{Currency: "BTC", Available: number(0.01)},
+				`),
-				},
-			},
-			wantFiats: types.BalanceMap{
-				"USDC": types.Balance{Currency: "USDC", Available: number(70.0)},
-				"USDT": types.Balance{Currency: "USDT", Available: number(100.0)},
-				"BUSD": types.Balance{Currency: "BUSD", Available: number(80.0)},
-			},
-			wantRest: types.BalanceMap{
-				"BTC": types.Balance{Currency: "BTC", Available: number(0.01)},
 			},
+			wantFiats: BalancesFromText(`
+				USDC, 150.0
+				USDT, 100.0
+				`),
+			wantRest: BalancesFromText(`
+				BTC, 0.01
+				`),
 		},
 	}
 	for _, tt := range tests {
@@ -242,52 +225,46 @@ func Test_calculateNetValueInQuote(t *testing.T) {
 			name: "positive asset",
 			args: args{
 				balances: types.BalanceMap{
-					"USDC": types.Balance{Currency: "USDC", Available: number(70.0)},
+					"USDC": types.Balance{Currency: "USDC", Available: number(70.0 + 80.0)},
 					"USDT": types.Balance{Currency: "USDT", Available: number(100.0)},
-					"BUSD": types.Balance{Currency: "BUSD", Available: number(80.0)},
 					"BTC":  types.Balance{Currency: "BTC", Available: number(0.01)},
 				},
 				prices: types.PriceMap{
-					"USDCUSDT": number(1.0),
+					"USDCUSDT": Number(1.0),
-					"BUSDUSDT": number(1.0),
+					"BTCUSDT":  Number(19000.0),
-					"BTCUSDT":  number(19000.0),
 				},
 				quoteCurrency: "USDT",
 			},
-			wantAccountValue: number(19000.0*0.01 + 100.0 + 80.0 + 70.0),
+			wantAccountValue: Number(19000.0*0.01 + 100.0 + 80.0 + 70.0),
 		},
 		{
 			name: "reversed usdt price",
 			args: args{
 				balances: types.BalanceMap{
-					"USDC": types.Balance{Currency: "USDC", Available: number(70.0)},
+					"USDC": types.Balance{Currency: "USDC", Available: Number(70.0 + 80.0)},
-					"TWD":  types.Balance{Currency: "TWD", Available: number(3000.0)},
+					"TWD":  types.Balance{Currency: "TWD", Available: Number(3000.0)},
-					"USDT": types.Balance{Currency: "USDT", Available: number(100.0)},
+					"USDT": types.Balance{Currency: "USDT", Available: Number(100.0)},
-					"BUSD": types.Balance{Currency: "BUSD", Available: number(80.0)},
+					"BTC":  types.Balance{Currency: "BTC", Available: Number(0.01)},
-					"BTC":  types.Balance{Currency: "BTC", Available: number(0.01)},
 				},
 				prices: types.PriceMap{
-					"USDTTWD":  number(30.0),
+					"USDTTWD":  Number(30.0),
-					"USDCUSDT": number(1.0),
+					"USDCUSDT": Number(1.0),
-					"BUSDUSDT": number(1.0),
+					"BTCUSDT":  Number(19000.0),
-					"BTCUSDT":  number(19000.0),
 				},
 				quoteCurrency: "USDT",
 			},
-			wantAccountValue: number(19000.0*0.01 + 100.0 + 80.0 + 70.0 + (3000.0 / 30.0)),
+			wantAccountValue: Number(19000.0*0.01 + 100.0 + 80.0 + 70.0 + (3000.0 / 30.0)),
 		},
 		{
 			name: "borrow base asset",
 			args: args{
 				balances: types.BalanceMap{
-					"USDT": types.Balance{Currency: "USDT", Available: number(20000.0 * 2)},
+					"USDT": types.Balance{Currency: "USDT", Available: Number(20000.0*2 + 80.0)},
-					"USDC": types.Balance{Currency: "USDC", Available: number(70.0)},
+					"USDC": types.Balance{Currency: "USDC", Available: Number(70.0)},
-					"BUSD": types.Balance{Currency: "BUSD", Available: number(80.0)},
+					"BTC":  types.Balance{Currency: "BTC", Available: Number(0), Borrowed: Number(2.0)},
-					"BTC":  types.Balance{Currency: "BTC", Available: number(0), Borrowed: number(2.0)},
 				},
 				prices: types.PriceMap{
 					"USDCUSDT": number(1.0),
-					"BUSDUSDT": number(1.0),
 					"BTCUSDT":  number(19000.0),
 				},
 				quoteCurrency: "USDT",
@@ -298,26 +275,35 @@ func Test_calculateNetValueInQuote(t *testing.T) {
 			name: "multi base asset",
 			args: args{
 				balances: types.BalanceMap{
-					"USDT": types.Balance{Currency: "USDT", Available: number(20000.0 * 2)},
+					"USDT": types.Balance{Currency: "USDT", Available: Number(20000.0*2 + 80.0)},
-					"USDC": types.Balance{Currency: "USDC", Available: number(70.0)},
+					"USDC": types.Balance{Currency: "USDC", Available: Number(70.0)},
-					"BUSD": types.Balance{Currency: "BUSD", Available: number(80.0)},
+					"ETH":  types.Balance{Currency: "ETH", Available: Number(10.0)},
-					"ETH":  types.Balance{Currency: "ETH", Available: number(10.0)},
+					"BTC":  types.Balance{Currency: "BTC", Available: Number(0), Borrowed: Number(2.0)},
-					"BTC":  types.Balance{Currency: "BTC", Available: number(0), Borrowed: number(2.0)},
 				},
 				prices: types.PriceMap{
-					"USDCUSDT": number(1.0),
+					"USDCUSDT": Number(1.0),
-					"BUSDUSDT": number(1.0),
+					"BTCUSDT":  Number(19000.0),
-					"ETHUSDT":  number(1700.0),
+					"ETHUSDT":  Number(1700.0),
-					"BTCUSDT":  number(19000.0),
 				},
 				quoteCurrency: "USDT",
 			},
-			wantAccountValue: number(19000.0*-2.0 + 1700.0*10.0 + 20000.0*2 + 80.0 + 70.0),
+			wantAccountValue: Number(19000.0*-2.0 + 1700.0*10.0 + 20000.0*2 + 80.0 + 70.0),
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			assert.Equalf(t, tt.wantAccountValue, calculateNetValueInQuote(tt.args.balances, tt.args.prices, tt.args.quoteCurrency), "calculateNetValueInQuote(%v, %v, %v)", tt.args.balances, tt.args.prices, tt.args.quoteCurrency)
+			markets := AllMarkets()
+			priceSolver := pricesolver.NewSimplePriceResolver(markets)
+
+			for symbol, price := range tt.args.prices {
+				priceSolver.Update(symbol, price)
+			}
+
+			assert.InDeltaf(t,
+				tt.wantAccountValue.Float64(),
+				calculateNetValueInQuote(tt.args.balances, priceSolver, tt.args.quoteCurrency).Float64(),
+				0.01,
+				"calculateNetValueInQuote(%v, %v, %v)", tt.args.balances, tt.args.prices, tt.args.quoteCurrency)
 		})
 	}
 }

pkg/pricesolver/simple.go

@@ -78,6 +78,12 @@ func (m *SimplePriceSolver) BindStream(stream types.Stream) {
 
 func (m *SimplePriceSolver) UpdateFromTickers(ctx context.Context, ex types.Exchange, symbols ...string) error {
 	for _, symbol := range symbols {
+		// only query the ticker for the symbol that is in the market map
+		_, ok := m.markets[symbol]
+		if !ok {
+			continue
+		}
+
 		ticker, err := ex.QueryTicker(ctx, symbol)
 		if err != nil {
 			return err
@@ -93,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 {
@@ -116,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
 				}
 			}
@@ -136,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,15 @@ 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/pricesolver"
 	"github.com/c9s/bbgo/pkg/types"
-	"github.com/sirupsen/logrus"
 )
 
 const ID = "harmonic"
@@ -29,7 +31,7 @@ type Strategy struct {
 	Market      types.Market
 
 	types.IntervalWindow
-	//bbgo.OpenPositionOptions
+	// bbgo.OpenPositionOptions
 
 	// persistence fields
 	Position    *types.Position    `persistence:"position"`
@@ -239,7 +241,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 +260,12 @@ 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)
+	priceSolver := pricesolver.NewSimplePriceResolver(session.Markets())
+	priceSolver.BindStream(s.session.MarketDataStream)
+	s.AccountValueCalculator = bbgo.NewAccountValueCalculator(s.session, priceSolver, s.Market.QuoteCurrency)
+	if err := s.AccountValueCalculator.UpdatePrices(ctx); err != nil {
+		return err
+	}
 
 	// Accumulated profit report
 	if bbgo.IsBackTesting {

pkg/strategy/irr/strategy.go

@@ -11,6 +11,7 @@ import (
 	"github.com/c9s/bbgo/pkg/datatype/floats"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/c9s/bbgo/pkg/indicator"
+	"github.com/c9s/bbgo/pkg/pricesolver"
 	"github.com/c9s/bbgo/pkg/types"
 
 	"github.com/sirupsen/logrus"
@@ -255,7 +256,13 @@ 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)
+	priceSolver := pricesolver.NewSimplePriceResolver(session.Markets())
+	priceSolver.BindStream(session.MarketDataStream)
+
+	s.AccountValueCalculator = bbgo.NewAccountValueCalculator(s.session, priceSolver, s.Market.QuoteCurrency)
+	if err := s.AccountValueCalculator.UpdatePrices(ctx); err != nil {
+		return err
+	}
 
 	// Accumulated profit report
 	if bbgo.IsBackTesting {

pkg/strategy/supertrend/strategy.go

@@ -11,6 +11,7 @@ import (
 	"github.com/sirupsen/logrus"
 
 	"github.com/c9s/bbgo/pkg/datatype/floats"
+	"github.com/c9s/bbgo/pkg/pricesolver"
 
 	"github.com/c9s/bbgo/pkg/bbgo"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
@@ -255,7 +256,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,
@@ -381,8 +384,14 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 		s.ProfitStatsTracker.Bind(s.session, s.orderExecutor.TradeCollector())
 	}
 
+	priceSolver := pricesolver.NewSimplePriceResolver(session.Markets())
+	priceSolver.BindStream(session.MarketDataStream)
+
 	// AccountValueCalculator
-	s.AccountValueCalculator = bbgo.NewAccountValueCalculator(s.session, s.Market.QuoteCurrency)
+	s.AccountValueCalculator = bbgo.NewAccountValueCalculator(s.session, priceSolver, s.Market.QuoteCurrency)
+	if err := s.AccountValueCalculator.UpdatePrices(ctx); err != nil {
+		return err
+	}
 
 	// 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 {
@@ -1336,12 +1335,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())
 		}
 	}
@@ -1433,8 +1427,6 @@ 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)
-
 	indicators := s.sourceSession.Indicators(s.Symbol)
 
 	s.boll = indicators.BOLL(types.IntervalWindow{
@@ -1494,6 +1486,11 @@ func (s *Strategy) CrossRun(
 	s.priceSolver = pricesolver.NewSimplePriceResolver(sourceMarkets)
 	s.priceSolver.BindStream(s.sourceSession.MarketDataStream)
 
+	s.accountValueCalculator = bbgo.NewAccountValueCalculator(s.sourceSession, s.priceSolver, s.sourceMarket.QuoteCurrency)
+	if err := s.accountValueCalculator.UpdatePrices(ctx); err != nil {
+		return err
+	}
+
 	s.sourceSession.MarketDataStream.OnKLineClosed(types.KLineWith(s.Symbol, types.Interval1m, func(k types.KLine) {
 		s.priceSolver.Update(k.Symbol, k.Close)
 		feeToken := s.sourceSession.Exchange.PlatformFeeCurrency()
@@ -1636,13 +1633,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/balance.go

@@ -0,0 +1,43 @@
+package testhelper
+
+import (
+	"strings"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+func BalancesFromText(str string) types.BalanceMap {
+	balances := make(types.BalanceMap)
+	lines := strings.Split(str, "\n")
+	for _, line := range lines {
+		line = strings.TrimSpace(line)
+		if len(line) == 0 {
+			continue
+		}
+
+		cols := strings.SplitN(line, ",", 2)
+		if len(cols) < 2 {
+			panic("column length should be 2")
+		}
+
+		currency := strings.TrimSpace(cols[0])
+		available := fixedpoint.MustNewFromString(strings.TrimSpace(cols[1]))
+		balances[currency] = Balance(currency, available)
+	}
+
+	return balances
+}
+
+// Balance returns a balance object with the given currency and available amount
+func Balance(currency string, available fixedpoint.Value) types.Balance {
+	return types.Balance{
+		Currency:          currency,
+		Available:         available,
+		Locked:            fixedpoint.Zero,
+		Borrowed:          fixedpoint.Zero,
+		Interest:          fixedpoint.Zero,
+		NetAsset:          fixedpoint.Zero,
+		MaxWithdrawAmount: fixedpoint.Zero,
+	}
+}

pkg/testing/testhelper/market.go

@@ -7,36 +7,64 @@ import (
 	"github.com/c9s/bbgo/pkg/types"
 )
 
-var markets = map[string]types.Market{
+var _markets = types.MarketMap{
 	"BTCUSDT": {
 		Symbol:          "BTCUSDT",
 		PricePrecision:  2,
 		VolumePrecision: 8,
 		QuoteCurrency:   "USDT",
 		BaseCurrency:    "BTC",
-		MinNotional:     fixedpoint.MustNewFromString("0.001"),
+		MinNotional:     fixedpoint.MustNewFromString("10.0"),
 		MinAmount:       fixedpoint.MustNewFromString("10.0"),
 		MinQuantity:     fixedpoint.MustNewFromString("0.001"),
 		TickSize:        fixedpoint.MustNewFromString("0.01"),
 	},
 
 	"ETHUSDT": {
-		Symbol:          "ETH",
+		Symbol:          "ETHUSDT",
 		PricePrecision:  2,
 		VolumePrecision: 8,
 		QuoteCurrency:   "USDT",
 		BaseCurrency:    "ETH",
-		MinNotional:     fixedpoint.MustNewFromString("0.005"),
+		MinNotional:     fixedpoint.MustNewFromString("10.0"),
 		MinAmount:       fixedpoint.MustNewFromString("10.0"),
 		MinQuantity:     fixedpoint.MustNewFromString("0.001"),
 		TickSize:        fixedpoint.MustNewFromString("0.01"),
 	},
+
+	"USDCUSDT": {
+		Symbol:          "USDCUSDT",
+		PricePrecision:  5,
+		VolumePrecision: 2,
+		QuoteCurrency:   "USDT",
+		BaseCurrency:    "USDC",
+		MinNotional:     fixedpoint.MustNewFromString("10.0"),
+		MinAmount:       fixedpoint.MustNewFromString("10.0"),
+		MinQuantity:     fixedpoint.MustNewFromString("10.0"),
+		TickSize:        fixedpoint.MustNewFromString("0.0001"),
+	},
+
+	"USDTTWD": {
+		Symbol:          "USDTTWD",
+		PricePrecision:  2,
+		VolumePrecision: 1,
+		QuoteCurrency:   "TWD",
+		BaseCurrency:    "USDT",
+		MinNotional:     fixedpoint.MustNewFromString("10.0"),
+		MinAmount:       fixedpoint.MustNewFromString("10.0"),
+		MinQuantity:     fixedpoint.MustNewFromString("10.0"),
+		TickSize:        fixedpoint.MustNewFromString("0.01"),
+	},
+}
+
+func AllMarkets() types.MarketMap {
+	return _markets
 }
 
 func Market(symbol string) types.Market {
-	market, ok := markets[symbol]
+	market, ok := _markets[symbol]
 	if !ok {
-		panic(fmt.Errorf("%s market not found, valid markets: %+v", symbol, markets))
+		panic(fmt.Errorf("%s test market not found, valid markets: %+v", symbol, _markets))
 	}
 
 	return market

pkg/testing/testhelper/ticker.go

@@ -0,0 +1,53 @@
+package testhelper
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+var _tickers = map[string]types.Ticker{
+	"BTCUSDT": {
+		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),
+	},
+
+	"ETHUSDT": {
+		Time:   time.Now(),
+		Volume: fixedpoint.Zero,
+		Open:   fixedpoint.NewFromFloat(2510.0),
+		High:   fixedpoint.NewFromFloat(2530.0),
+		Low:    fixedpoint.NewFromFloat(2505.0),
+		Last:   fixedpoint.NewFromFloat(2520.0),
+		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", "USDTTWD"}))
+	}
+
+	return ticker
+}