add strict start time, sync time checking for preventing back-test failure

related to #311

pkg/cmd/backtest.go

@@ -27,6 +27,8 @@ func init() {
 	BacktestCmd.Flags().Bool("sync", false, "sync backtest data")
 	BacktestCmd.Flags().Bool("sync-only", false, "sync backtest data only, do not run backtest")
 	BacktestCmd.Flags().String("sync-from", "", "sync backtest data from the given time, which will override the time range in the backtest config")
+	BacktestCmd.Flags().Bool("verify", false, "verify the kline back-test data")
+
 	BacktestCmd.Flags().Bool("base-asset-baseline", false, "use base asset performance as the competitive baseline performance")
 	BacktestCmd.Flags().CountP("verbose", "v", "verbose level")
 	BacktestCmd.Flags().String("config", "config/bbgo.yaml", "strategy config file")
@@ -73,6 +75,7 @@ var BacktestCmd = &cobra.Command{
 		if err != nil {
 			return err
 		}
+
 		jsonOutputEnabled := len(outputDirectory) > 0
 
 		syncOnly, err := cmd.Flags().GetBool("sync-only")
@@ -85,6 +88,11 @@ var BacktestCmd = &cobra.Command{
 			return err
 		}
 
+		shouldVerify, err := cmd.Flags().GetBool("verify")
+		if err != nil {
+			return err
+		}
+
 		exchangeNameStr, err := cmd.Flags().GetString("exchange")
 		if err != nil {
 			return err
@@ -145,7 +153,7 @@ var BacktestCmd = &cobra.Command{
 		environ.BacktestService = backtestService
 
 		if wantSync {
-			var syncFromTime = startTime
+			var syncFromTime time.Time
 
 			// override the sync from time if the option is given
 			if len(syncFromDateStr) > 0 {
@@ -159,61 +167,78 @@ var BacktestCmd = &cobra.Command{
 				}
 			} else {
 				// we need at least 1 month backward data for EMA and last prices
-				syncFromTime = syncFromTime.AddDate(0, -1, 0)
+				syncFromTime = startTime.AddDate(0, -1, 0)
-				log.Infof("adjusted sync start time to %s for backward market data", syncFromTime)
+				log.Infof("adjusted sync start time %s to %s for backward market data", startTime, syncFromTime)
 			}
 
 			log.Info("starting synchronization...")
 			for _, symbol := range userConfig.Backtest.Symbols {
+				firstKLine, err := backtestService.QueryFirstKLine(sourceExchange.Name(), symbol, types.Interval1m)
+				if err != nil {
+					return errors.Wrapf(err, "failed to query backtest kline")
+				}
+
+				// if we don't have klines before the start time endpoint, the back-test will fail.
+				// because the last price will be missing.
+				if syncFromTime.Before(firstKLine.EndTime) {
+					return fmt.Errorf("the sync-from-time you gave %s is earlier than the previous sync entry %s. "+
+						"re-syncing data from the earlier date before your first sync is not support,"+
+						"please clean up the kline table and restart a new sync",
+						syncFromTime,
+						firstKLine.EndTime)
+				}
+
 				if err := backtestService.Sync(ctx, sourceExchange, symbol, syncFromTime); err != nil {
 					return err
 				}
 			}
 			log.Info("synchronization done")
 
-			var corruptCnt = 0
+			if shouldVerify {
-			for _, symbol := range userConfig.Backtest.Symbols {
+				var corruptCnt = 0
-				log.Infof("verifying backtesting data...")
+				for _, symbol := range userConfig.Backtest.Symbols {
+					log.Infof("verifying backtesting data...")
 
-				for interval := range types.SupportedIntervals {
+					for interval := range types.SupportedIntervals {
-					log.Infof("verifying %s %s kline data...", symbol, interval)
+						log.Infof("verifying %s %s kline data...", symbol, interval)
 
-					klineC, errC := backtestService.QueryKLinesCh(startTime, time.Now(), sourceExchange, []string{symbol}, []types.Interval{interval})
+						klineC, errC := backtestService.QueryKLinesCh(startTime, time.Now(), sourceExchange, []string{symbol}, []types.Interval{interval})
-					var emptyKLine types.KLine
+						var emptyKLine types.KLine
-					var prevKLine types.KLine
+						var prevKLine types.KLine
-					for k := range klineC {
+						for k := range klineC {
-						if verboseCnt > 1 {
+							if verboseCnt > 1 {
-							fmt.Fprint(os.Stderr, ".")
+								fmt.Fprint(os.Stderr, ".")
-						}
-
-						if prevKLine != emptyKLine {
-							if prevKLine.StartTime.Add(interval.Duration()) != k.StartTime {
-								corruptCnt++
-								log.Errorf("found kline data corrupted at time: %s kline: %+v", k.StartTime, k)
-								log.Errorf("between %d and %d",
-									prevKLine.StartTime.Unix(),
-									k.StartTime.Unix())
 							}
+
+							if prevKLine != emptyKLine {
+								if prevKLine.StartTime.Add(interval.Duration()) != k.StartTime {
+									corruptCnt++
+									log.Errorf("found kline data corrupted at time: %s kline: %+v", k.StartTime, k)
+									log.Errorf("between %d and %d",
+										prevKLine.StartTime.Unix(),
+										k.StartTime.Unix())
+								}
+							}
+
+							prevKLine = k
 						}
 
-						prevKLine = k
+						if verboseCnt > 1 {
-					}
+							fmt.Fprintln(os.Stderr)
+						}
 
-					if verboseCnt > 1 {
+						if err := <-errC; err != nil {
-						fmt.Fprintln(os.Stderr)
+							return err
-					}
+						}
-
-					if err := <-errC; err != nil {
-						return err
 					}
 				}
-			}
 
-			log.Infof("backtest verification completed")
+				log.Infof("backtest verification completed")
-			if corruptCnt > 0 {
+				if corruptCnt > 0 {
-				log.Errorf("found %d corruptions", corruptCnt)
+					log.Errorf("found %d corruptions", corruptCnt)
-			} else {
+				} else {
-				log.Infof("found %d corruptions", corruptCnt)
+					log.Infof("found %d corruptions", corruptCnt)
+				}
 			}
 
 			if syncOnly {

pkg/service/backtest.go

@@ -2,6 +2,7 @@ package service
 
 import (
 	"context"
+	"strconv"
 	"strings"
 	"time"
 
@@ -20,13 +21,13 @@ 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())
 
-	lastKLine, err := s.QueryLast(exchange.Name(), symbol, interval)
+	lastKLine, err := s.QueryKLine(exchange.Name(), symbol, interval, "DESC", 1)
 	if err != nil {
 		return err
 	}
 
 	if lastKLine != nil {
-		log.Infof("found last checkpoint %s", lastKLine.EndTime)
+		log.Infof("found the last %s kline data checkpoint %s", symbol, lastKLine.EndTime)
 		startTime = lastKLine.StartTime.Add(time.Minute)
 	}
 
@@ -59,12 +60,21 @@ func (s *BacktestService) Sync(ctx context.Context, exchange types.Exchange, sym
 	return nil
 }
 
-// QueryLast queries the last order from the database
+func (s *BacktestService) QueryFirstKLine(ex types.ExchangeName, symbol string, interval types.Interval) (*types.KLine, error) {
-func (s *BacktestService) QueryLast(ex types.ExchangeName, symbol string, interval types.Interval) (*types.KLine, error) {
+	return s.QueryKLine(ex, symbol, interval, "ASC", 1)
+}
+
+// QueryLastKLine queries the last kline from the database
+func (s *BacktestService) QueryLastKLine(ex types.ExchangeName, symbol string, interval types.Interval) (*types.KLine, error) {
+	return s.QueryKLine(ex, symbol, interval, "DESC", 1)
+}
+
+// QueryKLine queries the klines from the database
+func (s *BacktestService) QueryKLine(ex types.ExchangeName, symbol string, interval types.Interval, orderBy string, limit int) (*types.KLine, error) {
 	log.Infof("querying last kline exchange = %s AND symbol = %s AND interval = %s", ex, symbol, interval)
 
 	// make the SQL syntax IDE friendly, so that it can analyze it.
-	sql := "SELECT * FROM binance_klines WHERE  `symbol` = :symbol AND `interval` = :interval ORDER BY end_time DESC LIMIT 1"
+	sql := "SELECT * FROM binance_klines WHERE  `symbol` = :symbol AND `interval` = :interval ORDER BY end_time " + orderBy + " LIMIT " + strconv.Itoa(limit)
 	sql = strings.ReplaceAll(sql, "binance_klines", ex.String()+"_klines")
 
 	rows, err := s.DB.NamedQuery(sql, map[string]interface{}{
@@ -74,7 +84,7 @@ func (s *BacktestService) QueryLast(ex types.ExchangeName, symbol string, interv
 	})
 
 	if err != nil {
-		return nil, errors.Wrap(err, "query last order error")
+		return nil, errors.Wrap(err, "query kline error")
 	}
 
 	if rows.Err() != nil {