interval: add 1s support

interval: add 1s support

interval: add 1s support

interval: fix 1s for backtesting

pkg/backtest/exchange.go

@@ -321,13 +321,13 @@ func (e *Exchange) BindUserData(userDataStream types.StandardStreamEmitter) {
 	e.matchingBooksMutex.Unlock()
 }
 
-func (e *Exchange) SubscribeMarketData(startTime, endTime time.Time, extraIntervals ...types.Interval) (chan types.KLine, error) {
+func (e *Exchange) SubscribeMarketData(startTime, endTime time.Time, requiredInterval types.Interval, extraIntervals ...types.Interval) (chan types.KLine, error) {
 	log.Infof("collecting backtest configurations...")
 
 	loadedSymbols := map[string]struct{}{}
 	loadedIntervals := map[types.Interval]struct{}{
 		// 1m interval is required for the backtest matching engine
-		types.Interval1m: {},
+		requiredInterval: {},
 	}
 
 	for _, it := range extraIntervals {
@@ -369,7 +369,7 @@ func (e *Exchange) SubscribeMarketData(startTime, endTime time.Time, extraInterv
 	return klineC, nil
 }
 
-func (e *Exchange) ConsumeKLine(k types.KLine) {
+func (e *Exchange) ConsumeKLine(k types.KLine, requiredInterval types.Interval) {
 	matching, ok := e.matchingBook(k.Symbol)
 	if !ok {
 		log.Errorf("matching book of %s is not initialized", k.Symbol)
@@ -379,14 +379,14 @@ func (e *Exchange) ConsumeKLine(k types.KLine) {
 		matching.klineCache = make(map[types.Interval]types.KLine)
 	}
 
-	kline1m, ok := matching.klineCache[k.Interval]
+	requiredKline, ok := matching.klineCache[k.Interval]
 	if ok { // pop out all the old
-		if kline1m.Interval != types.Interval1m {
+		if requiredKline.Interval.Seconds() < requiredInterval.Seconds() {
-			panic("expect 1m kline, got " + kline1m.Interval.String())
+			panic(fmt.Sprintf("expect required kline interval %s, got interval %s", requiredInterval.String(), requiredKline.Interval.String()))
 		}
-		e.currentTime = kline1m.EndTime.Time()
+		e.currentTime = requiredKline.EndTime.Time()
 		// here we generate trades and order updates
-		matching.processKLine(kline1m)
+		matching.processKLine(requiredKline)
 		matching.nextKLine = &k
 		for _, kline := range matching.klineCache {
 			e.MarketDataStream.EmitKLineClosed(kline)

pkg/cmd/backtest.go

@@ -4,6 +4,11 @@ import (
 	"bufio"
 	"context"
 	"fmt"
+	"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"
 	"os"
 	"path/filepath"
 	"sort"
@@ -11,8 +16,6 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/fatih/color"
-	"github.com/google/uuid"
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -21,13 +24,10 @@ import (
 	"github.com/c9s/bbgo/pkg/accounting/pnl"
 	"github.com/c9s/bbgo/pkg/backtest"
 	"github.com/c9s/bbgo/pkg/bbgo"
-	"github.com/c9s/bbgo/pkg/cmd/cmdutil"
-	"github.com/c9s/bbgo/pkg/data/tsv"
 	"github.com/c9s/bbgo/pkg/exchange"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
 	"github.com/c9s/bbgo/pkg/service"
 	"github.com/c9s/bbgo/pkg/types"
-	"github.com/c9s/bbgo/pkg/util"
 )
 
 func init() {
@@ -294,8 +294,8 @@ var BacktestCmd = &cobra.Command{
 			return err
 		}
 
-		backTestIntervals := []types.Interval{types.Interval1h, types.Interval1d}
+		allKLineIntervals, requiredInterval, backTestIntervals := collectSubscriptionIntervals(environ)
-		exchangeSources, err := toExchangeSources(environ.Sessions(), startTime, endTime, backTestIntervals...)
+		exchangeSources, err := toExchangeSources(environ.Sessions(), startTime, endTime, requiredInterval, backTestIntervals...)
 		if err != nil {
 			return err
 		}
@@ -455,7 +455,7 @@ var BacktestCmd = &cobra.Command{
 			if numOfExchangeSources == 1 {
 				exSource := exchangeSources[0]
 				for k := range exSource.C {
-					exSource.Exchange.ConsumeKLine(k)
+					exSource.Exchange.ConsumeKLine(k, requiredInterval)
 				}
 
 				if err := exSource.Exchange.CloseMarketData(); err != nil {
@@ -476,7 +476,7 @@ var BacktestCmd = &cobra.Command{
 						break RunMultiExchangeData
 					}
 
-					exK.Exchange.ConsumeKLine(k)
+					exK.Exchange.ConsumeKLine(k, requiredInterval)
 				}
 			}
 		}()
@@ -517,18 +517,6 @@ var BacktestCmd = &cobra.Command{
 			Symbols:              nil,
 		}
 
-		allKLineIntervals := map[types.Interval]struct{}{}
-		for _, interval := range backTestIntervals {
-			allKLineIntervals[interval] = struct{}{}
-		}
-
-		for _, session := range environ.Sessions() {
-			for _, sub := range session.Subscriptions {
-				if sub.Channel == types.KLineChannel {
-					allKLineIntervals[sub.Options.Interval] = struct{}{}
-				}
-			}
-		}
 		for interval := range allKLineIntervals {
 			summaryReport.Intervals = append(summaryReport.Intervals, interval)
 		}
@@ -601,6 +589,32 @@ 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 == types.Interval1s {
+					// if any subscription is 1s, then we will use 1s for back-testing
+					requiredInterval = sub.Options.Interval
+					log.Warnf("found 1s kline subscription, 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) (
 	*backtest.SessionSymbolReport,
 	error,
@@ -701,11 +715,11 @@ func confirmation(s string) bool {
 	}
 }
 
-func toExchangeSources(sessions map[string]*bbgo.ExchangeSession, startTime, endTime time.Time, extraIntervals ...types.Interval) (exchangeSources []*backtest.ExchangeDataSource, err error) {
+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, extraIntervals...)
+		c, err := backtestEx.SubscribeMarketData(startTime, endTime, requiredInterval, extraIntervals...)
 		if err != nil {
 			return exchangeSources, err
 		}

pkg/exchange/binance/exchange.go

@@ -1668,19 +1668,21 @@ func (e *Exchange) QueryPositionRisk(ctx context.Context, symbol string) (*types
 	return convertPositionRisk(risks[0])
 }
 
+// in seconds
 var SupportedIntervals = map[types.Interval]int{
-	types.Interval1m:  1,
+	types.Interval1s:  1,
-	types.Interval5m:  5,
+	types.Interval1m:  1 * 60,
-	types.Interval15m: 15,
+	types.Interval5m:  5 * 60,
-	types.Interval30m: 30,
+	types.Interval15m: 15 * 60,
-	types.Interval1h:  60,
+	types.Interval30m: 30 * 60,
-	types.Interval2h:  60 * 2,
+	types.Interval1h:  60 * 60,
-	types.Interval4h:  60 * 4,
+	types.Interval2h:  60 * 60 * 2,
-	types.Interval6h:  60 * 6,
+	types.Interval4h:  60 * 60 * 4,
-	types.Interval12h: 60 * 12,
+	types.Interval6h:  60 * 60 * 6,
-	types.Interval1d:  60 * 24,
+	types.Interval12h: 60 * 60 * 12,
-	types.Interval3d:  60 * 24 * 3,
+	types.Interval1d:  60 * 60 * 24,
-	types.Interval1w:  60 * 24 * 7,
+	types.Interval3d:  60 * 60 * 24 * 3,
+	types.Interval1w:  60 * 60 * 24 * 7,
 }
 
 func (e *Exchange) SupportedInterval() map[types.Interval]int {

pkg/types/interval.go

@@ -10,6 +10,14 @@ import (
 type Interval string
 
 func (i Interval) Minutes() int {
+	m, ok := SupportedIntervals[i]
+	if !ok {
+		return int(ParseInterval(i) / 60.)
+	}
+	return m
+}
+
+func (i Interval) Seconds() int {
 	m, ok := SupportedIntervals[i]
 	if !ok {
 		return ParseInterval(i)
@@ -18,7 +26,7 @@ func (i Interval) Minutes() int {
 }
 
 func (i Interval) Duration() time.Duration {
-	return time.Duration(i.Minutes()) * time.Minute
+	return time.Duration(i.Seconds()) * time.Second
 }
 
 func (i *Interval) UnmarshalJSON(b []byte) (err error) {
@@ -45,6 +53,7 @@ func (s IntervalSlice) StringSlice() (slice []string) {
 	return slice
 }
 
+var Interval1s = Interval("1s")
 var Interval1m = Interval("1m")
 var Interval3m = Interval("3m")
 var Interval5m = Interval("5m")
@@ -73,8 +82,10 @@ func ParseInterval(input Interval) int {
 		}
 	}
 	switch strings.ToLower(string(input[index:])) {
-	case "m":
+	case "s":
 		return t
+	case "m":
+		t *= 60
 	case "h":
 		t *= 60
 	case "d":
@@ -84,27 +95,28 @@ func ParseInterval(input Interval) int {
 	case "mo":
 		t *= 60 * 24 * 30
 	default:
-		panic("unknown input: " + input)
+		panic("unknown interval input: " + input)
 	}
 	return t
 }
 
 var SupportedIntervals = map[Interval]int{
-	Interval1m:  1,
+	Interval1s:  1,
-	Interval3m:  3,
+	Interval1m:  1 * 60,
-	Interval5m:  5,
+	Interval3m:  3 * 60,
-	Interval15m: 15,
+	Interval5m:  5 * 60,
-	Interval30m: 30,
+	Interval15m: 15 * 60,
-	Interval1h:  60,
+	Interval30m: 30 * 60,
-	Interval2h:  60 * 2,
+	Interval1h:  60 * 60,
-	Interval4h:  60 * 4,
+	Interval2h:  60 * 60 * 2,
-	Interval6h:  60 * 6,
+	Interval4h:  60 * 60 * 4,
-	Interval12h: 60 * 12,
+	Interval6h:  60 * 60 * 6,
-	Interval1d:  60 * 24,
+	Interval12h: 60 * 60 * 12,
-	Interval3d:  60 * 24 * 3,
+	Interval1d:  60 * 60 * 24,
-	Interval1w:  60 * 24 * 7,
+	Interval3d:  60 * 60 * 24 * 3,
-	Interval2w:  60 * 24 * 14,
+	Interval1w:  60 * 60 * 24 * 7,
-	Interval1mo: 60 * 24 * 30,
+	Interval2w:  60 * 60 * 24 * 14,
+	Interval1mo: 60 * 60 * 24 * 30,
 }
 
 // IntervalWindow is used by the indicators

pkg/types/interval_test.go

@@ -7,6 +7,7 @@ import (
 )
 
 func TestParseInterval(t *testing.T) {
+	assert.Equal(t, ParseIntervalSeconds("1s"), 1)
 	assert.Equal(t, ParseInterval("3m"), 3)
 	assert.Equal(t, ParseInterval("15h"), 15*60)
 	assert.Equal(t, ParseInterval("72d"), 72*24*60)