Merge pull request c9s#889 from c9s/fix/backtest-last-kline-sync

service/backtest: check and filter kline by its endTime

pkg/service/backtest.go

@@ -37,13 +37,27 @@ func (s *BacktestService) SyncKLineByInterval(ctx context.Context, exchange type
 		_, _ = s.DB.Exec("PRAGMA synchronous = NORMAL")
 	}
 
+	now := time.Now()
 	tasks := []SyncTask{
 		{
 			Type:   types.KLine{},
 			Select: SelectLastKLines(exchange.Name(), symbol, interval, startTime, endTime, 100),
 			Time: func(obj interface{}) time.Time {
 				return obj.(types.KLine).StartTime.Time()
 			},
+			Filter: func(obj interface{}) bool {
+				k := obj.(types.KLine)
+				if k.EndTime.Before(k.StartTime.Time().Add(k.Interval.Duration() - time.Second)) {
+					return false
+				}
+
+				// Filter klines that has the endTime closed in the future
+				if k.EndTime.After(now) {
+					return false
+				}
+
+				return true
+			},
 			ID: func(obj interface{}) string {
 				kline := obj.(types.KLine)
 				return strconv.FormatInt(kline.StartTime.UnixMilli(), 10)

pkg/service/sync_task.go

@@ -35,6 +35,8 @@ type SyncTask struct {
 	OnLoad func(objs interface{})
 
 	// Filter is an optional field, which is used for filtering the remote records
+	// Return true to keep the record,
+	// Return false to filter the record.
 	Filter func(obj interface{}) bool
 
 	// BatchQuery is used for querying remote records.