Merge pull request #655 from c9s/refactor/sync

fix: improve and fix kline sync

pkg/exchange/batch/closedorders.go

@@ -5,8 +5,6 @@ import (
 	"strconv"
 	"time"
 
-	"golang.org/x/time/rate"
-
 	"github.com/c9s/bbgo/pkg/types"
 )
 
@@ -17,7 +15,6 @@ type ClosedOrderBatchQuery struct {
 func (q *ClosedOrderBatchQuery) Query(ctx context.Context, symbol string, startTime, endTime time.Time, lastOrderID uint64) (c chan types.Order, errC chan error) {
 	query := &AsyncTimeRangedBatchQuery{
 		Type: types.Order{},
-		Limiter: rate.NewLimiter(rate.Every(5*time.Second), 2),
 		Q: func(startTime, endTime time.Time) (interface{}, error) {
 			orders, err := q.ExchangeTradeHistoryService.QueryClosedOrders(ctx, symbol, startTime, endTime, lastOrderID)
 			return orders, err

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/exchange/batch/time_range_query.go

@@ -8,6 +8,8 @@ import (
 
 	"github.com/sirupsen/logrus"
 	"golang.org/x/time/rate"
+
+	"github.com/c9s/bbgo/pkg/util"
 )
 
 var log = logrus.WithField("component", "batch")
@@ -52,6 +54,8 @@ func (q *AsyncTimeRangedBatchQuery) Query(ctx context.Context, ch interface{}, s
 
 			log.Debugf("batch querying %T: %v <=> %v", q.Type, startTime, endTime)
 
+			queryProfiler := util.StartTimeProfile("remoteQuery")
+
 			sliceInf, err := q.Q(startTime, endTime)
 			if err != nil {
 				errC <- err
@@ -60,9 +64,10 @@ func (q *AsyncTimeRangedBatchQuery) Query(ctx context.Context, ch interface{}, s
 
 			listRef := reflect.ValueOf(sliceInf)
 			listLen := listRef.Len()
-
 			log.Debugf("batch querying %T: %d remote records", q.Type, listLen)
 
+			queryProfiler.StopAndLog(log.Debugf)
+
 			if listLen == 0 {
 				if q.JumpIfEmpty > 0 {
 					startTime = startTime.Add(q.JumpIfEmpty)

pkg/exchange/batch/trade.go

@@ -4,8 +4,6 @@ import (
 	"context"
 	"time"
 
-	"golang.org/x/time/rate"
-
 	"github.com/c9s/bbgo/pkg/types"
 )
 
@@ -24,7 +22,6 @@ func (e TradeBatchQuery) Query(ctx context.Context, symbol string, options *type
 	endTime := *options.EndTime
 	query := &AsyncTimeRangedBatchQuery{
 		Type:    types.Trade{},
-		Limiter: rate.NewLimiter(rate.Every(5*time.Second), 2),
 		Q: func(startTime, endTime time.Time) (interface{}, error) {
 			return e.ExchangeTradeHistoryService.QueryTrades(ctx, symbol, options)
 		},

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().UTC()
+			},
+			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 {

pkg/util/profile.go

@@ -1,14 +1,21 @@
 package util
 
-import "time"
+import (
+	"time"
+)
 
 type TimeProfile struct {
+	Name               string
 	StartTime, EndTime time.Time
 	Duration           time.Duration
 }
 
-func StartTimeProfile() TimeProfile {
-	return TimeProfile{StartTime: time.Now()}
+func StartTimeProfile(args ...string) TimeProfile {
+	name := ""
+	if len(args) > 0 {
+		name = args[0]
+	}
+	return TimeProfile{StartTime: time.Now(), Name: name}
 }
 
 func (p *TimeProfile) TilNow() time.Duration {
@@ -20,3 +27,16 @@ func (p *TimeProfile) Stop() time.Duration {
 	p.Duration = p.EndTime.Sub(p.StartTime)
 	return p.Duration
 }
+
+type logFunction func(format string, args ...interface{})
+
+func (p *TimeProfile) StopAndLog(f logFunction) {
+	duration := p.Stop()
+	s := "[profile] "
+	if len(p.Name) > 0 {
+		s += p.Name
+	}
+
+	s += " " + duration.String()
+	f(s)
+}