xmaker: add TradeVolumeWindowSignal

pkg/strategy/xmaker/signal_trade.go

@@ -0,0 +1,102 @@
+package xmaker
+
+import (
+	"context"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/c9s/bbgo/pkg/bbgo"
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+var tradeVolumeWindowSignalMetrics = prometheus.NewGaugeVec(
+	prometheus.GaugeOpts{
+		Name: "xmaker_trade_volume_window_signal",
+		Help: "",
+	}, []string{"symbol"})
+
+func init() {
+	prometheus.MustRegister(tradeVolumeWindowSignalMetrics)
+}
+
+type TradeVolumeWindowSignal struct {
+	Threshold fixedpoint.Value `json:"threshold"`
+	Window    types.Duration   `json:"window"`
+
+	trades []types.Trade
+	symbol string
+}
+
+func (s *TradeVolumeWindowSignal) handleTrade(trade types.Trade) {
+	s.trades = append(s.trades, trade)
+}
+
+func (s *TradeVolumeWindowSignal) Bind(ctx context.Context, session *bbgo.ExchangeSession, symbol string) error {
+	s.symbol = symbol
+
+	if s.Window == 0 {
+		s.Window = types.Duration(time.Minute)
+	}
+
+	if s.Threshold.IsZero() {
+		s.Threshold = fixedpoint.NewFromFloat(0.7)
+	}
+
+	session.MarketDataStream.OnMarketTrade(s.handleTrade)
+	return nil
+}
+
+func (s *TradeVolumeWindowSignal) filterTrades(now time.Time) []types.Trade {
+	startTime := now.Add(-time.Duration(s.Window))
+	startIdx := 0
+
+	for idx, td := range s.trades {
+		// skip trades before the start time
+		if td.Time.Before(startTime) {
+			continue
+		}
+
+		startIdx = idx
+		break
+	}
+
+	s.trades = s.trades[startIdx:]
+	return s.trades
+}
+
+func (s *TradeVolumeWindowSignal) calculateTradeVolume(trades []types.Trade) (buyVolume, sellVolume float64) {
+	for _, td := range trades {
+		if td.IsBuyer {
+			buyVolume += td.Quantity.Float64()
+		} else {
+			sellVolume += td.Quantity.Float64()
+		}
+	}
+
+	return buyVolume, sellVolume
+}
+
+func (s *TradeVolumeWindowSignal) CalculateSignal(ctx context.Context) (float64, error) {
+	now := time.Now()
+	trades := s.filterTrades(now)
+	buyVolume, sellVolume := s.calculateTradeVolume(trades)
+	totalVolume := buyVolume + sellVolume
+
+	threshold := s.Threshold.Float64()
+	buyRatio := buyVolume / totalVolume
+	sellRatio := sellVolume / totalVolume
+
+	sig := 0.0
+	if buyRatio > threshold {
+		sig = (buyRatio - threshold) / 2.0
+	} else if sellRatio > threshold {
+		sig = -(sellRatio - threshold) / 2.0
+	}
+
+	log.Infof("[TradeVolumeWindowSignal] sig: %f buy/sell = %f/%f", sig, buyVolume, sellVolume)
+
+	tradeVolumeWindowSignalMetrics.WithLabelValues(s.symbol).Set(sig)
+	return sig, nil
+}

pkg/strategy/xmaker/signal_trade_test.go

@@ -0,0 +1,55 @@
+package xmaker
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/types"
+
+	. "github.com/c9s/bbgo/pkg/testing/testhelper"
+)
+
+var tradeId = 0
+
+func Trade(symbol string, side types.SideType, price, quantity fixedpoint.Value, t time.Time) types.Trade {
+	tradeId++
+	return types.Trade{
+		ID:       uint64(tradeId),
+		Symbol:   symbol,
+		Side:     side,
+		Price:    price,
+		IsBuyer:  side == types.SideTypeBuy,
+		Quantity: quantity,
+		Time:     types.Time(t),
+	}
+}
+
+func TestMarketTradeWindowSignal(t *testing.T) {
+	now := time.Now()
+	symbol := "BTCUSDT"
+	sig := &TradeVolumeWindowSignal{
+		symbol:    symbol,
+		Threshold: fixedpoint.NewFromFloat(0.65),
+		Window:    types.Duration(time.Minute),
+	}
+
+	sig.trades = []types.Trade{
+		Trade(symbol, types.SideTypeBuy, Number(18000.0), Number(1.0), now.Add(-2*time.Minute)),
+		Trade(symbol, types.SideTypeSell, Number(18000.0), Number(0.5), now.Add(-2*time.Second)),
+		Trade(symbol, types.SideTypeBuy, Number(18000.0), Number(1.0), now.Add(-1*time.Second)),
+	}
+
+	ctx := context.Background()
+	sigNum, err := sig.CalculateSignal(ctx)
+	if assert.NoError(t, err) {
+		// buy ratio: 1/1.5 = 0.6666666666666666
+		// sell ratio: 0.5/1.5 = 0.3333333333333333
+		assert.InDelta(t, 0.0083333, sigNum, 0.0001)
+	}
+
+	assert.Len(t, sig.trades, 2)
+}

pkg/strategy/xmaker/strategy.go

@@ -65,6 +65,7 @@ type SignalConfig struct {
 	BollingerBandTrendSignal *BollingerBandTrendSignal       `json:"bollingerBandTrend,omitempty"`
 	OrderBookBestPriceSignal *OrderBookBestPriceVolumeSignal `json:"orderBookBestPrice,omitempty"`
 	KLineShapeSignal         *KLineShapeSignal               `json:"klineShape,omitempty"`
+	TradeVolumeWindowSignal  *TradeVolumeWindowSignal        `json:"tradeVolumeWindow,omitempty"`
 }
 
 func init() {
@@ -205,7 +206,14 @@ func (s *Strategy) CrossSubscribe(sessions map[string]*bbgo.ExchangeSession) {
 	if !ok {
 		panic(fmt.Errorf("maker session %s is not defined", s.MakerExchange))
 	}
+
 	makerSession.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{Interval: "1m"})
+
+	for _, sig := range s.SignalConfigList {
+		if sig.TradeVolumeWindowSignal != nil {
+			sourceSession.Subscribe(types.MarketTradeChannel, s.Symbol, types.SubscribeOptions{})
+		}
+	}
 }
 
 func aggregatePrice(pvs types.PriceVolumeSlice, requiredQuantity fixedpoint.Value) (price fixedpoint.Value) {
@@ -363,31 +371,19 @@ func (s *Strategy) calculateSignal(ctx context.Context) (float64, error) {
 	sum := 0.0
 	voters := 0.0
 	for _, signal := range s.SignalConfigList {
+		var sig float64
+		var err error
 		if signal.OrderBookBestPriceSignal != nil {
-			sig, err := signal.OrderBookBestPriceSignal.CalculateSignal(ctx)
+			sig, err = signal.OrderBookBestPriceSignal.CalculateSignal(ctx)
-			if err != nil {
-				return 0, err
-			}
-
-			if sig == 0.0 {
-				continue
-			}
-
-			if signal.Weight > 0.0 {
-				sum += sig * signal.Weight
-				voters += signal.Weight
-			} else {
-				sum += sig
-				voters++
-			}
-
 		} else if signal.BollingerBandTrendSignal != nil {
-			sig, err := signal.BollingerBandTrendSignal.CalculateSignal(ctx)
+			sig, err = signal.BollingerBandTrendSignal.CalculateSignal(ctx)
-			if err != nil {
+		} else if signal.TradeVolumeWindowSignal != nil {
-				return 0, err
+			sig, err = signal.TradeVolumeWindowSignal.CalculateSignal(ctx)
 		}
 
-			if sig == 0.0 {
+		if err != nil {
+			return 0, err
+		} else if sig == 0.0 {
 			continue
 		}
 
@@ -399,6 +395,9 @@ func (s *Strategy) calculateSignal(ctx context.Context) (float64, error) {
 			voters++
 		}
 	}
+
+	if sum == 0.0 {
+		return 0.0, nil
 	}
 
 	return sum / voters, nil

pkg/types/price_volume_slice.go

@@ -13,6 +13,13 @@ type PriceVolume struct {
 	Price, Volume fixedpoint.Value
 }
 
+func NewPriceVolume(p, v fixedpoint.Value) PriceVolume {
+	return PriceVolume{
+		Price:  p,
+		Volume: v,
+	}
+}
+
 func (p PriceVolume) InQuote() fixedpoint.Value {
 	return p.Price.Mul(p.Volume)
 }