improve and fix kline sync

pkg/exchange/batch/kline.go

@@ -4,8 +4,6 @@ import (
 	"context"
 	"time"
 
-	"golang.org/x/time/rate"
-
 	"github.com/c9s/bbgo/pkg/types"
 )
 
@@ -16,7 +14,7 @@ type KLineBatchQuery struct {
 func (e *KLineBatchQuery) Query(ctx context.Context, symbol string, interval types.Interval, startTime, endTime time.Time) (c chan types.KLine, errC chan error) {
 	query := &AsyncTimeRangedBatchQuery{
 		Type:    types.KLine{},
-		Limiter: rate.NewLimiter(rate.Every(5*time.Second), 2),
+		Limiter: nil, // the rate limiter is handled in the exchange query method
 		Q: func(startTime, endTime time.Time) (interface{}, error) {
 			return e.Exchange.QueryKLines(ctx, symbol, interval, types.KLineQueryOptions{
 				StartTime: &startTime,
@@ -32,7 +30,7 @@ func (e *KLineBatchQuery) Query(ctx context.Context, symbol string, interval typ
 		},
 	}
 
-	c = make(chan types.KLine, 100)
+	c = make(chan types.KLine, 3000)
 	errC = query.Query(ctx, c, startTime, endTime)
 	return c, errC
 }

pkg/service/backtest.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 	"time"
 
+	sq "github.com/Masterminds/squirrel"
 	"github.com/jmoiron/sqlx"
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
@@ -23,16 +24,43 @@ type BacktestService struct {
 func (s *BacktestService) SyncKLineByInterval(ctx context.Context, exchange types.Exchange, symbol string, interval types.Interval, startTime, endTime time.Time) error {
 	log.Infof("synchronizing lastKLine for interval %s from exchange %s", interval, exchange.Name())
 
-	q := &batch.KLineBatchQuery{Exchange: exchange}
+	// TODO: use isFutures here
+	_, _, isIsolated, isolatedSymbol := getExchangeAttributes(exchange)
+	// override symbol if isolatedSymbol is not empty
+	if isIsolated && len(isolatedSymbol) > 0 {
+		symbol = isolatedSymbol
+	}
 
-	klineC, errC := q.Query(ctx, symbol, interval, startTime, endTime)
-	for kline := range klineC {
-		if err := s.Insert(kline); err != nil {
+	tasks := []SyncTask{
+		{
+			Type: types.KLine{},
+			Time: func(obj interface{}) time.Time {
+				return obj.(types.KLine).StartTime.Time()
+			},
+			ID: func(obj interface{}) string {
+				kline := obj.(types.KLine)
+				return kline.Symbol + kline.Interval.String() + strconv.FormatInt(kline.StartTime.UnixMilli(), 10)
+			},
+			Select: SelectLastKLines(exchange.Name(), symbol, interval, 100),
+			BatchQuery: func(ctx context.Context, startTime, endTime time.Time) (interface{}, chan error) {
+				q := &batch.KLineBatchQuery{Exchange: exchange}
+				return q.Query(ctx, symbol, interval, startTime, endTime)
+			},
+			Insert: func(obj interface{}) error {
+				kline := obj.(types.KLine)
+				return s.Insert(kline)
+			},
+		},
+	}
+
+	for _, sel := range tasks {
+		if err := sel.execute(ctx, s.DB, startTime); err != nil {
 			return err
 		}
 	}
 
-	return <-errC
+	return nil
+
 }
 
 func (s *BacktestService) Verify(symbols []string, startTime time.Time, endTime time.Time, sourceExchange types.Exchange, verboseCnt int) error {
@@ -334,3 +362,16 @@ func (s *BacktestService) SyncExist(ctx context.Context, exchange types.Exchange
 	}
 	return nil
 }
+
+// TODO: add is_futures column since the klines data is different
+func SelectLastKLines(ex types.ExchangeName, symbol string, interval types.Interval, limit uint64) sq.SelectBuilder {
+	return sq.Select("*").
+		From(strings.ToLower(ex.String()) + "_klines").
+		Where(sq.And{
+			sq.Eq{"symbol": symbol},
+			sq.Eq{"exchange": ex},
+			sq.Eq{"`interval`": interval.String()},
+		}).
+		OrderBy("start_time DESC").
+		Limit(limit)
+}

pkg/service/sync_task.go

@@ -40,7 +40,7 @@ type SyncTask struct {
 	BatchQuery func(ctx context.Context, startTime, endTime time.Time) (interface{}, chan error)
 }
 
-func (sel SyncTask) execute(ctx context.Context, db *sqlx.DB, startTime time.Time) error {
+func (sel SyncTask) execute(ctx context.Context, db *sqlx.DB, startTime time.Time, args ...time.Time) error {
 	// query from db
 	recordSlice, err := selectAndScanType(ctx, db, sel.Select, sel.Type)
 	if err != nil {
@@ -65,10 +65,15 @@ func (sel SyncTask) execute(ctx context.Context, db *sqlx.DB, startTime time.Tim
 	}
 
 	// default since time point
-	since := lastRecordTime(sel, recordSliceRef, startTime)
+	startTime = lastRecordTime(sel, recordSliceRef, startTime)
+
+	endTime := time.Now()
+	if len(args) > 0 {
+		endTime = args[0]
+	}
 
 	// asset "" means all assets
-	dataC, errC := sel.BatchQuery(ctx, since, time.Now())
+	dataC, errC := sel.BatchQuery(ctx, startTime, endTime)
 	dataCRef := reflect.ValueOf(dataC)
 
 	for {
@@ -148,3 +153,5 @@ func buildIdMap(sel SyncTask, recordSliceRef reflect.Value) map[string]struct{}
 
 	return ids
 }
+
+