Merge pull request #1020 from c9s/feature/grid2

config/grid2.yaml

@@ -21,10 +21,13 @@ backtest:
   symbols:
   - BTCUSDT
   sessions: [binance]
+  feeMode: token
   accounts:
     binance:
+      makerFeeRate: 0.075%
+      takerFeeRate: 0.075%
       balances:
-        BTC: 0.0
+        BTC: 1.0
         USDT: 21_000.0
 
 exchangeStrategies:
@@ -32,9 +35,13 @@ exchangeStrategies:
 - on: binance
   grid2:
     symbol: BTCUSDT
-    upperPrice: 60_000.0
     lowerPrice: 28_000.0
-    gridNumber: 1000
+    upperPrice: 50_000.0
+
+    ## gridNumber is the total orders between the upper price and the lower price
+    ## gridSpread = (upperPrice - lowerPrice) / gridNumber
+    ## Make sure your gridNumber satisfy this: MIN(gridSpread/lowerPrice, gridSpread/upperPrice) > (makerFeeRate * 2)
+    gridNumber: 150
 
     ## compound is used for buying more inventory when the profit is made by the filled SELL order.
     ## when compound is disabled, fixed quantity is used for each grid order.
@@ -76,7 +83,7 @@ exchangeStrategies:
     quoteInvestment: 20_000
 
     ## baseInvestment (optional) can be useful when you have existing inventory, maybe bought at much lower price
-    baseInvestment: 0.0
+    baseInvestment: 1.0
 
     ## closeWhenCancelOrder (optional)
     ## default to false

pkg/backtest/exchange.go

@@ -358,8 +358,7 @@ func (e *Exchange) SubscribeMarketData(startTime, endTime time.Time, requiredInt
 		intervals = append(intervals, interval)
 	}
 
-	log.Infof("using symbols: %v and intervals: %v for back-testing", symbols, intervals)
+	log.Infof("querying klines from database with exchange: %v symbols: %v and intervals: %v for back-testing", e.Name(), symbols, intervals)
-	log.Infof("querying klines from database...")
 	klineC, errC := e.srv.QueryKLinesCh(startTime, endTime, e, symbols, intervals)
 	go func() {
 		if err := <-errC; err != nil {

pkg/backtest/utils.go

@@ -0,0 +1,57 @@
+package backtest
+
+import (
+	"time"
+
+	"github.com/sirupsen/logrus"
+
+	"github.com/c9s/bbgo/pkg/bbgo"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+func CollectSubscriptionIntervals(environ *bbgo.Environment) (allKLineIntervals map[types.Interval]struct{}, requiredInterval types.Interval, backTestIntervals []types.Interval) {
+	// default extra back-test intervals
+	backTestIntervals = []types.Interval{types.Interval1h, types.Interval1d}
+	// all subscribed intervals
+	allKLineIntervals = make(map[types.Interval]struct{})
+
+	for _, interval := range backTestIntervals {
+		allKLineIntervals[interval] = struct{}{}
+	}
+	// default interval is 1m for all exchanges
+	requiredInterval = types.Interval1m
+	for _, session := range environ.Sessions() {
+		for _, sub := range session.Subscriptions {
+			if sub.Channel == types.KLineChannel {
+				if sub.Options.Interval.Seconds()%60 > 0 {
+					// if any subscription interval is less than 60s, then we will use 1s for back-testing
+					requiredInterval = types.Interval1s
+					logrus.Warnf("found kline subscription interval less than 60s, modify default backtest interval to 1s")
+				}
+				allKLineIntervals[sub.Options.Interval] = struct{}{}
+			}
+		}
+	}
+	return allKLineIntervals, requiredInterval, backTestIntervals
+}
+
+func InitializeExchangeSources(sessions map[string]*bbgo.ExchangeSession, startTime, endTime time.Time, requiredInterval types.Interval, extraIntervals ...types.Interval) (exchangeSources []*ExchangeDataSource, err error) {
+	for _, session := range sessions {
+		backtestEx := session.Exchange.(*Exchange)
+
+		c, err := backtestEx.SubscribeMarketData(startTime, endTime, requiredInterval, extraIntervals...)
+		if err != nil {
+			return exchangeSources, err
+		}
+
+		sessionCopy := session
+		src := &ExchangeDataSource{
+			C:        c,
+			Exchange: backtestEx,
+			Session:  sessionCopy,
+		}
+		backtestEx.Src = src
+		exchangeSources = append(exchangeSources, src)
+	}
+	return exchangeSources, nil
+}

pkg/cmd/backtest.go

@@ -11,11 +11,12 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/fatih/color"
+	"github.com/google/uuid"
+
 	"github.com/c9s/bbgo/pkg/cmd/cmdutil"
 	"github.com/c9s/bbgo/pkg/data/tsv"
 	"github.com/c9s/bbgo/pkg/util"
-	"github.com/fatih/color"
-	"github.com/google/uuid"
 
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
@@ -295,8 +296,8 @@ var BacktestCmd = &cobra.Command{
 			return err
 		}
 
-		allKLineIntervals, requiredInterval, backTestIntervals := collectSubscriptionIntervals(environ)
+		allKLineIntervals, requiredInterval, backTestIntervals := backtest.CollectSubscriptionIntervals(environ)
-		exchangeSources, err := toExchangeSources(environ.Sessions(), startTime, endTime, requiredInterval, backTestIntervals...)
+		exchangeSources, err := backtest.InitializeExchangeSources(environ.Sessions(), startTime, endTime, requiredInterval, backTestIntervals...)
 		if err != nil {
 			return err
 		}
@@ -593,32 +594,6 @@ var BacktestCmd = &cobra.Command{
 	},
 }
 
-func collectSubscriptionIntervals(environ *bbgo.Environment) (allKLineIntervals map[types.Interval]struct{}, requiredInterval types.Interval, backTestIntervals []types.Interval) {
-	// default extra back-test intervals
-	backTestIntervals = []types.Interval{types.Interval1h, types.Interval1d}
-	// all subscribed intervals
-	allKLineIntervals = make(map[types.Interval]struct{})
-
-	for _, interval := range backTestIntervals {
-		allKLineIntervals[interval] = struct{}{}
-	}
-	// default interval is 1m for all exchanges
-	requiredInterval = types.Interval1m
-	for _, session := range environ.Sessions() {
-		for _, sub := range session.Subscriptions {
-			if sub.Channel == types.KLineChannel {
-				if sub.Options.Interval.Seconds()%60 > 0 {
-					// if any subscription interval is less than 60s, then we will use 1s for back-testing
-					requiredInterval = types.Interval1s
-					log.Warnf("found kline subscription interval less than 60s, modify default backtest interval to 1s")
-				}
-				allKLineIntervals[sub.Options.Interval] = struct{}{}
-			}
-		}
-	}
-	return allKLineIntervals, requiredInterval, backTestIntervals
-}
-
 func createSymbolReport(userConfig *bbgo.Config, session *bbgo.ExchangeSession, symbol string, trades []types.Trade, intervalProfit *types.IntervalProfitCollector,
 	profitFactor, winningRatio fixedpoint.Value) (
 	*backtest.SessionSymbolReport,
@@ -722,27 +697,6 @@ func confirmation(s string) bool {
 	}
 }
 
-func toExchangeSources(sessions map[string]*bbgo.ExchangeSession, startTime, endTime time.Time, requiredInterval types.Interval, extraIntervals ...types.Interval) (exchangeSources []*backtest.ExchangeDataSource, err error) {
-	for _, session := range sessions {
-		backtestEx := session.Exchange.(*backtest.Exchange)
-
-		c, err := backtestEx.SubscribeMarketData(startTime, endTime, requiredInterval, extraIntervals...)
-		if err != nil {
-			return exchangeSources, err
-		}
-
-		sessionCopy := session
-		src := &backtest.ExchangeDataSource{
-			C:        c,
-			Exchange: backtestEx,
-			Session:  sessionCopy,
-		}
-		backtestEx.Src = src
-		exchangeSources = append(exchangeSources, src)
-	}
-	return exchangeSources, nil
-}
-
 func sync(ctx context.Context, userConfig *bbgo.Config, backtestService *service.BacktestService, sourceExchanges map[types.ExchangeName]types.Exchange, syncFrom, syncTo time.Time) error {
 	for _, symbol := range userConfig.Backtest.Symbols {
 		for _, sourceExchange := range sourceExchanges {

pkg/strategy/grid2/profit.go

@@ -1,6 +1,7 @@
 package grid2
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/c9s/bbgo/pkg/fixedpoint"
@@ -14,41 +15,6 @@ type GridProfit struct {
 	Order    types.Order      `json:"order"`
 }
 
-type GridProfitStats struct {
+func (p *GridProfit) String() string {
-	Symbol           string           `json:"symbol"`
+	return fmt.Sprintf("GRID PROFIT: %f %s @ %s orderID %d", p.Profit.Float64(), p.Currency, p.Time.String(), p.Order.OrderID)
-	TotalBaseProfit  fixedpoint.Value `json:"totalBaseProfit,omitempty"`
-	TotalQuoteProfit fixedpoint.Value `json:"totalQuoteProfit,omitempty"`
-	FloatProfit      fixedpoint.Value `json:"floatProfit,omitempty"`
-	GridProfit       fixedpoint.Value `json:"gridProfit,omitempty"`
-	ArbitrageCount   int              `json:"arbitrageCount,omitempty"`
-	TotalFee         fixedpoint.Value `json:"totalFee,omitempty"`
-	Volume           fixedpoint.Value `json:"volume,omitempty"`
-	Market           types.Market     `json:"market,omitempty"`
-	ProfitEntries    []*GridProfit    `json:"profitEntries,omitempty"`
-}
-
-func newGridProfitStats(market types.Market) *GridProfitStats {
-	return &GridProfitStats{
-		Symbol:           market.Symbol,
-		TotalBaseProfit:  fixedpoint.Zero,
-		TotalQuoteProfit: fixedpoint.Zero,
-		FloatProfit:      fixedpoint.Zero,
-		GridProfit:       fixedpoint.Zero,
-		ArbitrageCount:   0,
-		TotalFee:         fixedpoint.Zero,
-		Volume:           fixedpoint.Zero,
-		Market:           market,
-		ProfitEntries:    nil,
-	}
-}
-
-func (s *GridProfitStats) AddProfit(profit *GridProfit) {
-	switch profit.Currency {
-	case s.Market.QuoteCurrency:
-		s.TotalQuoteProfit = s.TotalQuoteProfit.Add(profit.Profit)
-	case s.Market.BaseCurrency:
-		s.TotalBaseProfit = s.TotalBaseProfit.Add(profit.Profit)
-	}
-
-	s.ProfitEntries = append(s.ProfitEntries, profit)
 }

pkg/strategy/grid2/profit_stats.go

@@ -0,0 +1,60 @@
+package grid2
+
+import (
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type GridProfitStats struct {
+	Symbol           string                      `json:"symbol"`
+	TotalBaseProfit  fixedpoint.Value            `json:"totalBaseProfit,omitempty"`
+	TotalQuoteProfit fixedpoint.Value            `json:"totalQuoteProfit,omitempty"`
+	FloatProfit      fixedpoint.Value            `json:"floatProfit,omitempty"`
+	GridProfit       fixedpoint.Value            `json:"gridProfit,omitempty"`
+	ArbitrageCount   int                         `json:"arbitrageCount,omitempty"`
+	TotalFee         map[string]fixedpoint.Value `json:"totalFee,omitempty"`
+	Volume           fixedpoint.Value            `json:"volume,omitempty"`
+	Market           types.Market                `json:"market,omitempty"`
+	ProfitEntries    []*GridProfit               `json:"profitEntries,omitempty"`
+}
+
+func newGridProfitStats(market types.Market) *GridProfitStats {
+	return &GridProfitStats{
+		Symbol:           market.Symbol,
+		TotalBaseProfit:  fixedpoint.Zero,
+		TotalQuoteProfit: fixedpoint.Zero,
+		FloatProfit:      fixedpoint.Zero,
+		GridProfit:       fixedpoint.Zero,
+		ArbitrageCount:   0,
+		TotalFee:         make(map[string]fixedpoint.Value),
+		Volume:           fixedpoint.Zero,
+		Market:           market,
+		ProfitEntries:    nil,
+	}
+}
+
+func (s *GridProfitStats) AddTrade(trade types.Trade) {
+	if s.TotalFee == nil {
+		s.TotalFee = make(map[string]fixedpoint.Value)
+	}
+
+	if fee, ok := s.TotalFee[trade.FeeCurrency]; ok {
+		s.TotalFee[trade.FeeCurrency] = fee.Add(trade.Fee)
+	} else {
+		s.TotalFee[trade.FeeCurrency] = trade.Fee
+	}
+}
+
+func (s *GridProfitStats) AddProfit(profit *GridProfit) {
+	// increase arbitrage count per profit round
+	s.ArbitrageCount++
+
+	switch profit.Currency {
+	case s.Market.QuoteCurrency:
+		s.TotalQuoteProfit = s.TotalQuoteProfit.Add(profit.Profit)
+	case s.Market.BaseCurrency:
+		s.TotalBaseProfit = s.TotalBaseProfit.Add(profit.Profit)
+	}
+
+	s.ProfitEntries = append(s.ProfitEntries, profit)
+}

pkg/strategy/grid2/strategy.go

@@ -19,6 +19,8 @@ const ID = "grid2"
 
 var log = logrus.WithField("strategy", ID)
 
+var maxNumberOfOrderTradesQueryTries = 10
+
 func init() {
 	// Register the pointer of the strategy struct,
 	// so that bbgo knows what struct to be used to unmarshal the configs (YAML or JSON)
@@ -131,12 +133,15 @@ func (s *Strategy) Validate() error {
 	}
 
 	if !s.ProfitSpread.IsZero() {
-		percent := s.ProfitSpread.Div(s.LowerPrice)
+		// the min fee rate from 2 maker/taker orders (with 0.1 rate for profit)
+		gridFeeRate := s.FeeRate.Mul(fixedpoint.NewFromFloat(2.01))
 
-		// the min fee rate from 2 maker/taker orders
+		if s.ProfitSpread.Div(s.LowerPrice).Compare(gridFeeRate) < 0 {
-		minProfitSpread := s.FeeRate.Mul(fixedpoint.NewFromInt(2))
+			return fmt.Errorf("profitSpread %f %s is too small for lower price, less than the fee rate: %s", s.ProfitSpread.Float64(), s.ProfitSpread.Div(s.LowerPrice).Percentage(), s.FeeRate.Percentage())
-		if percent.Compare(minProfitSpread) < 0 {
+		}
-			return fmt.Errorf("profitSpread %f %s is too small, less than the fee rate: %s", s.ProfitSpread.Float64(), percent.Percentage(), s.FeeRate.Percentage())
+
+		if s.ProfitSpread.Div(s.UpperPrice).Compare(gridFeeRate) < 0 {
+			return fmt.Errorf("profitSpread %f %s is too small for upper price, less than the fee rate: %s", s.ProfitSpread.Float64(), s.ProfitSpread.Div(s.UpperPrice).Percentage(), s.FeeRate.Percentage())
 		}
 	}
 
@@ -238,7 +243,50 @@ func (s *Strategy) verifyOrderTrades(o types.Order, trades []types.Trade) bool {
 	return true
 }
 
-// handleOrderFilled is called when a order status is FILLED
+// aggregateOrderBaseFee collects the base fee quantity from the given order
+// it falls back to query the trades via the RESTful API when the websocket trades are not all received.
+func (s *Strategy) aggregateOrderBaseFee(o types.Order) fixedpoint.Value {
+	// try to get the received trades (websocket trades)
+	orderTrades := s.historicalTrades.GetOrderTrades(o)
+	if len(orderTrades) > 0 {
+		s.logger.Infof("found filled order trades: %+v", orderTrades)
+	}
+
+	for maxTries := maxNumberOfOrderTradesQueryTries; maxTries > 0; maxTries-- {
+		// if one of the trades is missing, we need to query the trades from the RESTful API
+		if s.verifyOrderTrades(o, orderTrades) {
+			// if trades are verified
+			fees := collectTradeFee(orderTrades)
+			if fee, ok := fees[s.Market.BaseCurrency]; ok {
+				return fee
+			}
+			return fixedpoint.Zero
+		}
+
+		// if we don't support orderQueryService, then we should just skip
+		if s.orderQueryService == nil {
+			return fixedpoint.Zero
+		}
+
+		s.logger.Warnf("missing order trades or missing trade fee, pulling order trades from API")
+
+		// if orderQueryService is supported, use it to query the trades of the filled order
+		apiOrderTrades, err := s.orderQueryService.QueryOrderTrades(context.Background(), types.OrderQuery{
+			Symbol:  o.Symbol,
+			OrderID: strconv.FormatUint(o.OrderID, 10),
+		})
+		if err != nil {
+			s.logger.WithError(err).Errorf("query order trades error")
+		} else {
+			s.logger.Infof("fetched api trades: %+v", apiOrderTrades)
+			orderTrades = apiOrderTrades
+		}
+	}
+
+	return fixedpoint.Zero
+}
+
+// handleOrderFilled is called when an order status is FILLED
 func (s *Strategy) handleOrderFilled(o types.Order) {
 	if s.grid == nil {
 		return
@@ -259,38 +307,11 @@ func (s *Strategy) handleOrderFilled(o types.Order) {
 	// because when 1.0 BTC buy order is filled without FEE token, then we will actually get 1.0 * (1 - feeRate) BTC
 	// if we don't reduce the sell quantity, than we might fail to place the sell order
 	if o.Side == types.SideTypeBuy {
-		orderTrades := s.historicalTrades.GetOrderTrades(o)
+		baseSellQuantityReduction = s.aggregateOrderBaseFee(o)
-		if len(orderTrades) > 0 {
-			s.logger.Infof("FOUND FILLED ORDER TRADES: %+v", orderTrades)
-		}
 
-		if !s.verifyOrderTrades(o, orderTrades) {
+		s.logger.Infof("base fee: %f %s", baseSellQuantityReduction.Float64(), s.Market.BaseCurrency)
-			s.logger.Warnf("missing order trades or missing trade fee, pulling order trades from API")
 
-			// if orderQueryService is supported, use it to query the trades of the filled order
+		newQuantity = newQuantity.Sub(baseSellQuantityReduction)
-			if s.orderQueryService != nil {
-				apiOrderTrades, err := s.orderQueryService.QueryOrderTrades(context.Background(), types.OrderQuery{
-					Symbol:  o.Symbol,
-					OrderID: strconv.FormatUint(o.OrderID, 10),
-				})
-				if err != nil {
-					s.logger.WithError(err).Errorf("query order trades error")
-				} else {
-					orderTrades = apiOrderTrades
-				}
-			}
-		}
-
-		if s.verifyOrderTrades(o, orderTrades) {
-			// check if there is a BaseCurrency fee collected
-			fees := collectTradeFee(orderTrades)
-			if fee, ok := fees[s.Market.BaseCurrency]; ok {
-				baseSellQuantityReduction = fee
-				s.logger.Infof("baseSellQuantityReduction: %f %s", baseSellQuantityReduction.Float64(), s.Market.BaseCurrency)
-
-				newQuantity = newQuantity.Sub(baseSellQuantityReduction)
-			}
-		}
 	}
 
 	switch o.Side {
@@ -959,9 +980,14 @@ func (s *Strategy) Run(ctx context.Context, orderExecutor bbgo.OrderExecutor, se
 	s.orderExecutor.BindEnvironment(s.Environment)
 	s.orderExecutor.BindProfitStats(s.ProfitStats)
 	s.orderExecutor.Bind()
+
+	s.orderExecutor.TradeCollector().OnTrade(func(trade types.Trade, _, _ fixedpoint.Value) {
+		s.GridProfitStats.AddTrade(trade)
+	})
 	s.orderExecutor.TradeCollector().OnPositionUpdate(func(position *types.Position) {
 		bbgo.Sync(ctx, s)
 	})
+
 	s.orderExecutor.ActiveMakerOrders().OnFilled(s.handleOrderFilled)
 
 	// TODO: detect if there are previous grid orders on the order book

pkg/strategy/grid2/strategy_test.go

@@ -4,6 +4,7 @@ package grid2
 
 import (
 	"context"
+	"os"
 	"testing"
 
 	"github.com/sirupsen/logrus"
@@ -350,12 +351,20 @@ func TestBacktestStrategy(t *testing.T) {
 		GridNum:         100,
 		QuoteInvestment: number(9000.0),
 	}
+	RunBacktest(t, strategy)
+}
 
+func RunBacktest(t *testing.T, strategy bbgo.SingleExchangeStrategy) {
 	// TEMPLATE {{{ start backtest
-	startTime, err := types.ParseLooseFormatTime("2021-06-01")
+	const sqliteDbFile = "../../../data/bbgo_test.sqlite3"
+	const backtestExchangeName = "binance"
+	const backtestStartTime = "2022-06-01"
+	const backtestEndTime = "2022-06-30"
+
+	startTime, err := types.ParseLooseFormatTime(backtestStartTime)
 	assert.NoError(t, err)
 
-	endTime, err := types.ParseLooseFormatTime("2021-06-30")
+	endTime, err := types.ParseLooseFormatTime(backtestEndTime)
 	assert.NoError(t, err)
 
 	backtestConfig := &bbgo.Backtest{
@@ -364,7 +373,7 @@ func TestBacktestStrategy(t *testing.T) {
 		RecordTrades: false,
 		FeeMode:      bbgo.BacktestFeeModeToken,
 		Accounts: map[string]bbgo.BacktestAccount{
-			"binance": {
+			backtestExchangeName: {
 				MakerFeeRate: number(0.075 * 0.01),
 				TakerFeeRate: number(0.075 * 0.01),
 				Balances: bbgo.BacktestAccountBalanceMap{
@@ -374,7 +383,7 @@ func TestBacktestStrategy(t *testing.T) {
 			},
 		},
 		Symbols:       []string{"BTCUSDT"},
-		Sessions:      []string{"binance"},
+		Sessions:      []string{backtestExchangeName},
 		SyncSecKLines: false,
 	}
 
@@ -384,8 +393,14 @@ func TestBacktestStrategy(t *testing.T) {
 	environ := bbgo.NewEnvironment()
 	environ.SetStartTime(startTime.Time())
 
-	err = environ.ConfigureDatabaseDriver(ctx, "sqlite3", "../../../data/bbgo_test.sqlite3")
+	info, err := os.Stat(sqliteDbFile)
 	assert.NoError(t, err)
+	t.Logf("sqlite: %+v", info)
+
+	err = environ.ConfigureDatabaseDriver(ctx, "sqlite3", sqliteDbFile)
+	if !assert.NoError(t, err) {
+		return
+	}
 
 	backtestService := &service.BacktestService{DB: environ.DatabaseService.DB}
 	defer func() {
@@ -397,22 +412,24 @@ func TestBacktestStrategy(t *testing.T) {
 	bbgo.SetBackTesting(backtestService)
 	defer bbgo.SetBackTesting(nil)
 
-	exName, err := types.ValidExchangeName("binance")
+	exName, err := types.ValidExchangeName(backtestExchangeName)
 	if !assert.NoError(t, err) {
 		return
 	}
 
+	t.Logf("using exchange source: %s", exName)
+
 	publicExchange, err := exchange.NewPublic(exName)
 	if !assert.NoError(t, err) {
 		return
 	}
 
-	backtestExchange, err := backtest.NewExchange(publicExchange.Name(), publicExchange, backtestService, backtestConfig)
+	backtestExchange, err := backtest.NewExchange(exName, publicExchange, backtestService, backtestConfig)
 	if !assert.NoError(t, err) {
 		return
 	}
 
-	session := environ.AddExchange(exName.String(), backtestExchange)
+	session := environ.AddExchange(backtestExchangeName, backtestExchange)
 	assert.NotNil(t, session)
 
 	err = environ.Init(ctx)
@@ -430,11 +447,11 @@ func TestBacktestStrategy(t *testing.T) {
 		trader.DisableLogging()
 	}
 
-	// TODO: add grid2 to the user config and run backtest
 	userConfig := &bbgo.Config{
+		Backtest: backtestConfig,
 		ExchangeStrategies: []bbgo.ExchangeStrategyMount{
 			{
-				Mounts:   []string{"binance"},
+				Mounts:   []string{backtestExchangeName},
 				Strategy: strategy,
 			},
 		},
@@ -446,7 +463,32 @@ func TestBacktestStrategy(t *testing.T) {
 	err = trader.Run(ctx)
 	assert.NoError(t, err)
 
-	// TODO: feed data
+	allKLineIntervals, requiredInterval, backTestIntervals := backtest.CollectSubscriptionIntervals(environ)
+	t.Logf("requiredInterval: %s backTestIntervals: %v", requiredInterval, backTestIntervals)
 
+	_ = allKLineIntervals
+	exchangeSources, err := backtest.InitializeExchangeSources(environ.Sessions(), startTime.Time(), endTime.Time(), requiredInterval, backTestIntervals...)
+	if !assert.NoError(t, err) {
+		return
+	}
+
+	doneC := make(chan struct{})
+	go func() {
+		count := 0
+		exSource := exchangeSources[0]
+		for k := range exSource.C {
+			exSource.Exchange.ConsumeKLine(k, requiredInterval)
+			count++
+		}
+
+		err = exSource.Exchange.CloseMarketData()
+		assert.NoError(t, err)
+
+		assert.Greater(t, count, 0, "kLines count must be greater than 0, please check your backtest date range and symbol settings")
+
+		close(doneC)
+	}()
+
+	<-doneC
 	// }}}
 }