Merge pull request #1730 from c9s/c9s/xmaker/market-trade-signal

FEATURE: [xmaker] add market trade signal

pkg/exchange/max/convert.go

@@ -247,29 +247,6 @@ func toGlobalTradeV3(t v3.Trade) ([]types.Trade, error) {
 	return trades, nil
 }
 
-func toGlobalTradeV2(t max.Trade) (*types.Trade, error) {
-	isMargin := t.WalletType == max.WalletTypeMargin
-	side := toGlobalSideType(t.Side)
-	return &types.Trade{
-		ID:            t.ID,
-		OrderID:       t.OrderID,
-		Price:         t.Price,
-		Symbol:        toGlobalSymbol(t.Market),
-		Exchange:      types.ExchangeMax,
-		Quantity:      t.Volume,
-		Side:          side,
-		IsBuyer:       t.IsBuyer(),
-		IsMaker:       t.IsMaker(),
-		Fee:           t.Fee,
-		FeeCurrency:   toGlobalCurrency(t.FeeCurrency),
-		QuoteQuantity: t.Funds,
-		Time:          types.Time(t.CreatedAt),
-		IsMargin:      isMargin,
-		IsIsolated:    false,
-		IsFutures:     false,
-	}, nil
-}
-
 func toGlobalDepositStatus(a max.DepositState) types.DepositStatus {
 	switch a {
 
@@ -285,6 +262,9 @@ func toGlobalDepositStatus(a max.DepositState) types.DepositStatus {
 	case max.DepositStateAccepted:
 		return types.DepositSuccess
 
+	case max.DepositStateFailed: // v3 state
+		return types.DepositRejected
+
 	case max.DepositStateProcessing: // v3 states
 		return types.DepositPending
 

pkg/exchange/max/maxapi/account.go

@@ -119,6 +119,7 @@ const (
 
 	// v3 states
 	DepositStateProcessing DepositState = "processing"
+	DepositStateFailed     DepositState = "failed"
 	DepositStateDone       DepositState = "done"
 )
 

pkg/strategy/xmaker/signal_trade.go

@@ -0,0 +1,111 @@
+package xmaker
+
+import (
+	"context"
+	"sync"
+	"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
+
+	mu sync.Mutex
+}
+
+func (s *TradeVolumeWindowSignal) handleTrade(trade types.Trade) {
+	s.mu.Lock()
+	s.trades = append(s.trades, trade)
+	s.mu.Unlock()
+}
+
+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
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	for idx, td := range s.trades {
+		// skip trades before the start time
+		if td.Time.Before(startTime) {
+			continue
+		}
+
+		startIdx = idx
+		break
+	}
+
+	trades := s.trades[startIdx:]
+	s.trades = trades
+	return trades
+}
+
+func (s *TradeVolumeWindowSignal) aggTradeVolume(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(_ context.Context) (float64, error) {
+	now := time.Now()
+	trades := s.filterTrades(now)
+	buyVolume, sellVolume := s.aggTradeVolume(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] %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,42 +371,33 @@ 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 {
-				continue
-			}
-
-			if signal.Weight > 0.0 {
-				sum += sig * signal.Weight
-				voters += signal.Weight
-			} else {
-				sum += sig
-				voters++
-			}
 		}
+
+		if err != nil {
+			return 0, err
+		} else if sig == 0.0 {
+			continue
+		}
+
+		if signal.Weight > 0.0 {
+			sum += sig * signal.Weight
+			voters += signal.Weight
+		} else {
+			sum += sig
+			voters++
+		}
+	}
+
+	if sum == 0.0 {
+		return 0.0, nil
 	}
 
 	return sum / voters, nil
@@ -1374,6 +1373,10 @@ func (s *Strategy) CrossRun(
 			if err := signalConfig.BollingerBandTrendSignal.Bind(ctx, s.sourceSession, s.Symbol); err != nil {
 				return err
 			}
+		} else if signalConfig.TradeVolumeWindowSignal != nil {
+			if err := signalConfig.TradeVolumeWindowSignal.Bind(ctx, s.sourceSession, s.Symbol); err != nil {
+				return err
+			}
 		}
 	}
 

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)
 }