fix: legacy fixedpoint inf handling, refactor backtest kline consuming

pkg/backtest/exchange.go

@@ -70,6 +70,8 @@ type Exchange struct {
 	matchingBooksMutex sync.Mutex
 
 	markets types.MarketMap
+
+	Src *ExchangeDataSource
 }
 
 func NewExchange(sourceName types.ExchangeName, sourceExchange types.Exchange, srv *service.BacktestService, config *bbgo.Backtest) (*Exchange, error) {
@@ -362,36 +364,35 @@ func (e *Exchange) SubscribeMarketData(startTime, endTime time.Time, extraInterv
 	return klineC, nil
 }
 
-func (e *Exchange) ConsumeKLine(k types.KLine, handlers []func(types.KLine, *ExchangeDataSource), src *ExchangeDataSource) {
+func (e *Exchange) ConsumeKLine(k types.KLine) {
 	matching, ok := e.matchingBook(k.Symbol)
 	if !ok {
 		log.Errorf("matching book of %s is not initialized", k.Symbol)
 		return
 	}
-	if matching.ParamCache == nil {
-		matching.ParamCache = make(map[types.Interval]Param)
+	if matching.klineCache == nil {
+		matching.klineCache = make(map[types.Interval]types.KLine)
 	}
-	_, ok = matching.ParamCache[k.Interval]
+
+	kline1m, ok := matching.klineCache[k.Interval]
 	if ok { // pop out all the old
-		for _, param := range matching.ParamCache {
-			if param.kline.Interval == types.Interval1m {
+		if kline1m.Interval != types.Interval1m {
+			panic("expect 1m kline, get " + kline1m.Interval.String())
+		}
 		// here we generate trades and order updates
-				matching.processKLine(param.kline)
+		matching.processKLine(kline1m)
 		matching.NextKLine = &k
-			}
+		for _, kline := range matching.klineCache {
 			// log.Errorf("kline %v, next %v", param.kline, matching.NextKLine)
-			e.MarketDataStream.EmitKLineClosed(param.kline)
-			for _, h := range param.callbacks {
-				h(param.kline, param.src)
+			e.MarketDataStream.EmitKLineClosed(kline)
+			for _, h := range e.Src.Callbacks {
+				h(kline, e.Src)
 			}
 		}
-		matching.ParamCache = make(map[types.Interval]Param)
-	}
-	matching.ParamCache[k.Interval] = Param{
-		callbacks: handlers,
-		src:       src,
-		kline:     k,
+		// reset the paramcache
+		matching.klineCache = make(map[types.Interval]types.KLine)
 	}
+	matching.klineCache[k.Interval] = k
 }
 
 func (e *Exchange) CloseMarketData() error {

pkg/backtest/exchange_klinec.go

@@ -9,4 +9,5 @@ type ExchangeDataSource struct {
 	C         chan types.KLine
 	Exchange  *Exchange
 	Session   *bbgo.ExchangeSession
+	Callbacks []func(types.KLine, *ExchangeDataSource)
 }

pkg/backtest/matching.go

@@ -47,12 +47,6 @@ func init() {
 	}
 }
 
-type Param struct {
-	callbacks []func(types.KLine, *ExchangeDataSource)
-	kline     types.KLine
-	src       *ExchangeDataSource
-}
-
 // SimplePriceMatching implements a simple kline data driven matching engine for backtest
 //go:generate callbackgen -type SimplePriceMatching
 type SimplePriceMatching struct {
@@ -64,7 +58,7 @@ type SimplePriceMatching struct {
 	askOrders    []types.Order
 	closedOrders map[uint64]types.Order
 
-	ParamCache  map[types.Interval]Param
+	klineCache  map[types.Interval]types.KLine
 	LastPrice   fixedpoint.Value
 	LastKLine   types.KLine
 	NextKLine   *types.KLine

pkg/cmd/backtest.go

@@ -438,6 +438,9 @@ var BacktestCmd = &cobra.Command{
 		}
 
 		runCtx, cancelRun := context.WithCancel(ctx)
+		for _, exK := range exchangeSources {
+			exK.Callbacks = kLineHandlers
+		}
 		go func() {
 			defer cancelRun()
 
@@ -446,7 +449,7 @@ var BacktestCmd = &cobra.Command{
 			if numOfExchangeSources == 1 {
 				exSource := exchangeSources[0]
 				for k := range exSource.C {
-					exSource.Exchange.ConsumeKLine(k, kLineHandlers, &exSource)
+					exSource.Exchange.ConsumeKLine(k)
 				}
 
 				if err := exSource.Exchange.CloseMarketData(); err != nil {
@@ -467,7 +470,7 @@ var BacktestCmd = &cobra.Command{
 						break RunMultiExchangeData
 					}
 
-					exK.Exchange.ConsumeKLine(k, kLineHandlers, &exK)
+					exK.Exchange.ConsumeKLine(k)
 				}
 			}
 		}()
@@ -694,11 +697,13 @@ func toExchangeSources(sessions map[string]*bbgo.ExchangeSession, startTime, end
 		}
 
 		sessionCopy := session
-		exchangeSources = append(exchangeSources, backtest.ExchangeDataSource{
+		src := backtest.ExchangeDataSource{
 			C:        c,
 			Exchange: backtestEx,
 			Session:  sessionCopy,
-		})
+		}
+		backtestEx.Src = &src
+		exchangeSources = append(exchangeSources, src)
 	}
 	return exchangeSources, nil
 }

pkg/fixedpoint/convert.go

@@ -23,6 +23,8 @@ type Value int64
 const Zero = Value(0)
 const One = Value(1e8)
 const NegOne = Value(-1e8)
+const PosInf = Value(math.MaxInt64)
+const NegInf = Value(math.MinInt64)
 
 type RoundingMode int
 
@@ -81,6 +83,11 @@ func (v *Value) Scan(src interface{}) error {
 }
 
 func (v Value) Float64() float64 {
+	if v == PosInf {
+		return math.Inf(1)
+	} else if v == NegInf {
+		return math.Inf(-1)
+	}
 	return float64(v) / DefaultPow
 }
 
@@ -92,10 +99,20 @@ func (v Value) Abs() Value {
 }
 
 func (v Value) String() string {
+	if v == PosInf {
+		return "inf"
+	} else if v == NegInf {
+		return "-inf"
+	}
 	return strconv.FormatFloat(float64(v)/DefaultPow, 'f', -1, 64)
 }
 
 func (v Value) FormatString(prec int) string {
+	if v == PosInf {
+		return "inf"
+	} else if v == NegInf {
+		return "-inf"
+	}
 	pow := math.Pow10(prec)
 	return strconv.FormatFloat(
 		math.Trunc(float64(v)/DefaultPow*pow)/pow, 'f', prec, 64)
@@ -105,6 +122,11 @@ func (v Value) Percentage() string {
 	if v == 0 {
 		return "0"
 	}
+	if v == PosInf {
+		return "inf%"
+	} else if v == NegInf {
+		return "-inf%"
+	}
 	return strconv.FormatFloat(float64(v)/DefaultPow*100., 'f', -1, 64) + "%"
 }
 
@@ -112,6 +134,11 @@ func (v Value) FormatPercentage(prec int) string {
 	if v == 0 {
 		return "0"
 	}
+	if v == PosInf {
+		return "inf%"
+	} else if v == NegInf {
+		return "-inf%"
+	}
 	pow := math.Pow10(prec)
 	result := strconv.FormatFloat(
 		math.Trunc(float64(v)/DefaultPow*pow*100.)/pow, 'f', prec, 64)
@@ -407,6 +434,11 @@ func Must(v Value, err error) Value {
 }
 
 func NewFromFloat(val float64) Value {
+	if math.IsInf(val, 1) {
+		return PosInf
+	} else if math.IsInf(val, -1) {
+		return NegInf
+	}
 	return Value(int64(math.Trunc(val * DefaultPow)))
 }