IMPROVE: [xmaker] smooth latest aggregated signal

pkg/strategy/xmaker/signal/book.go

@@ -39,9 +39,9 @@
 	MinQuoteVolume fixedpoint.Value `json:"minQuoteVolume"`
 	MinDelta       fixedpoint.Value `json:"minDelta"`
 
-	Window          int    `json:"window"`
-	SmoothingWindow int    `json:"smoothingWindow"`
-	SmoothingType   string `json:"smoothingType"`
+	Window          int                       `json:"window"`
+	SmoothingWindow int                       `json:"smoothingWindow"`
+	SmoothingType   indicatorv2.SmoothingType `json:"smoothingType"`
 
 	symbol string
 	book   *types.StreamOrderBook
@@ -71,30 +71,15 @@
 
 	if s.Window > 0 {
 		s.bidVolumeSeries = types.NewFloat64Series()
-		s.bidVolumeIndicator = s.createVolumeIndicator(s.bidVolumeSeries)
+		s.bidVolumeIndicator = indicatorv2.NewSmoothedIndicator(s.bidVolumeSeries, s.Window, s.SmoothingWindow, s.SmoothingType)
 
 		s.askVolumeSeries = types.NewFloat64Series()
-		s.askVolumeIndicator = s.createVolumeIndicator(s.askVolumeSeries)
+		s.askVolumeIndicator = indicatorv2.NewSmoothedIndicator(s.askVolumeSeries, s.Window, s.SmoothingWindow, s.SmoothingType)
 	}
 
 	return nil
 }
 
-func (s *OrderBookBestPriceVolumeSignal) createVolumeIndicator(source types.Float64Source) types.Float64Calculator {
-	maxStream := indicatorv2.MAX(source, s.Window)
-	if s.SmoothingWindow > 0 {
-		switch s.SmoothingType {
-		case "rma":
-			return indicatorv2.RMA2(maxStream, s.SmoothingWindow, true)
-		case "ewma", "ema":
-			return indicatorv2.EWMA2(maxStream, s.SmoothingWindow)
-		default:
-			return indicatorv2.EWMA2(maxStream, s.SmoothingWindow)
-		}
-	}
-	return maxStream
-}
-
 func (s *OrderBookBestPriceVolumeSignal) CalculateSignal(ctx context.Context) (float64, error) {
 	bid, ask, ok := s.book.BestBidAndAsk()
 	if !ok {

pkg/strategy/xmaker/strategy.go

@@ -23,6 +23,7 @@ import (
 	"github.com/c9s/bbgo/pkg/dynamic"
 	"github.com/c9s/bbgo/pkg/exchange/sandbox"
 	"github.com/c9s/bbgo/pkg/fixedpoint"
+	indicatorv2 "github.com/c9s/bbgo/pkg/indicator/v2"
 	"github.com/c9s/bbgo/pkg/pricesolver"
 	"github.com/c9s/bbgo/pkg/profile/timeprofile"
 	"github.com/c9s/bbgo/pkg/risk/circuitbreaker"
@@ -100,6 +101,11 @@ type SignalMargin struct {
 	Threshold float64         `json:"threshold,omitempty"`
 }
 
+type AggregatedSignalConfig struct {
+	SmoothingWindow int                       `json:"smoothingWindow"`
+	SmoothingType   indicatorv2.SmoothingType `json:"smoothingType"`
+}
+
 type Strategy struct {
 	common.StrategyProfitFixer
 
@@ -142,6 +148,8 @@ type Strategy struct {
 	SignalReverseSideMargin       *SignalMargin `json:"signalReverseSideMargin,omitempty"`
 	SignalTrendSideMarginDiscount *SignalMargin `json:"signalTrendSideMarginDiscount,omitempty"`
 
+	AggregatedSignal *AggregatedSignalConfig `json:"aggregatedSignal,omitempty"`
+
 	// Margin is the default margin for the quote
 	Margin    fixedpoint.Value `json:"margin"`
 	BidMargin fixedpoint.Value `json:"bidMargin"`
@@ -288,6 +296,12 @@ type Strategy struct {
 	// TODO: use float64 series instead, so that we can store history signal values
 	lastAggregatedSignal MutexFloat64
 
+	// lastSmoothedAggregatedSignal stores the last smoothed aggregated signal with mutex
+	lastSmoothedAggregatedSignal MutexFloat64
+
+	aggregatedSignalSeries    *types.Float64Series
+	aggregatedSignalIndicator types.Float64Calculator
+
 	// lastDirectionDivergence stores the latest D2 value
 	lastDirectionDivergence MutexFloat64
 	lastDirectionMean       MutexFloat64
@@ -498,6 +512,11 @@ func (s *Strategy) Initialize() error {
 		}
 	}
 
+	if s.AggregatedSignal != nil && s.AggregatedSignal.SmoothingWindow > 0 {
+		s.aggregatedSignalSeries = types.NewFloat64Series()
+		s.aggregatedSignalIndicator = indicatorv2.NewSmoothedIndicator(s.aggregatedSignalSeries, 0, s.AggregatedSignal.SmoothingWindow, s.AggregatedSignal.SmoothingType)
+	}
+
 	s.positionExposure = bbgo.NewPositionExposure(s.Symbol)
 	s.positionExposure.SetLogger(s.logger)
 	s.positionExposure.SetMetricsLabels(ID, s.InstanceID(), s.MakerExchange, s.Symbol)
@@ -561,14 +580,7 @@ func (s *Strategy) getPositionHoldingPeriod(now time.Time) (time.Duration, bool)
 }
 
 func (s *Strategy) applySignalMargin(ctx context.Context, quote *Quote) error {
-	sig, err := s.AggregateSignal(ctx)
-	if err != nil {
-		return err
-	}
-
-	s.lastAggregatedSignal.Set(sig)
-	s.logger.Infof("aggregated signal: %f", sig)
-
+	sig := s.lastSmoothedAggregatedSignal.Get()
 	if sig == 0.0 {
 		return nil
 	}
@@ -986,8 +998,23 @@ func (s *Strategy) updateQuote(ctx context.Context) error {
 	}
 
 	s.logger.Infof("aggregated signal: %f", sig)
+	s.lastAggregatedSignal.Set(sig)
 	s.aggregatedSignalMetrics.Set(sig)
 
+	smoothedSig := sig
+	if s.aggregatedSignalIndicator != nil {
+		s.aggregatedSignalSeries.PushAndEmit(sig)
+		smoothedSig = s.aggregatedSignalIndicator.(types.Series).Index(0)
+	}
+	s.lastSmoothedAggregatedSignal.Set(smoothedSig)
+
+	// apply signal margin
+	if s.EnableSignalMargin {
+		if err := s.applySignalMargin(ctx, nil); err != nil {
+			s.logger.WithError(err).Errorf("unable to apply signal margin")
+		}
+	}
+
 	now := time.Now()
 	if s.CircuitBreaker != nil {
 		if reason, halted := s.CircuitBreaker.IsHalted(now); halted {
@@ -2011,7 +2038,7 @@ func (s *Strategy) hedge(ctx context.Context) {
 	)
 
 	lastPrice := s.lastPrice.Get()
-	sig := s.lastAggregatedSignal.Get()
+	sig := s.lastSmoothedAggregatedSignal.Get()
 
 	if lastPrice.IsZero() {
 		s.logger.Warnf("last price is zero, skip hedging")
@@ -2268,6 +2295,13 @@ func (s *Strategy) Defaults() error {
 		s.HedgeInterval = types.Duration(10 * time.Second)
 	}
 
+	if s.AggregatedSignal == nil {
+		s.AggregatedSignal = &AggregatedSignalConfig{
+			SmoothingWindow: 0,
+			SmoothingType:   indicatorv2.SmoothingTypeEMA,
+		}
+	}
+
 	if s.NumLayers == 0 {
 		s.NumLayers = 1
 	}