Merge pull request #1028 from c9s/feature/grid2

config/grid2-max.yaml

@@ -5,8 +5,8 @@ notifications:
     errorChannel: "bbgo-error"
   switches:
     trade: false
-    orderUpdate: false
-    submitOrder: false
+    orderUpdate: true
+    submitOrder: true
 
 sessions:
   max:
@@ -52,8 +52,12 @@ exchangeStrategies:
     ## for example, when the last price hit 17_000.0 then open a grid with the price range 13_000 to 20_000
     # triggerPrice: 16_900.0
 
+    ## stopLossPrice is used for closing the grid and sell all the inventory to stop loss.
+    ## (optional)
     stopLossPrice: 16_000.0
 
+    ## takeProfitPrice is used for closing the grid and sell all the inventory at higher price to take profit
+    ## (optional)
     takeProfitPrice: 20_000.0
 
     ## profitSpread is the profit spread of the arbitrage order (sell order)
@@ -83,3 +87,6 @@ exchangeStrategies:
     resetPositionWhenStart: true
     clearOpenOrdersWhenStart: false
     keepOrdersWhenShutdown: false
+
+    ## skipSpreadCheck skips the minimal spread check for the grid profit
+    skipSpreadCheck: true

config/grid2.yaml

@@ -13,6 +13,26 @@ sessions:
     exchange: binance
     envVarPrefix: binance
 
+sync:
+  # userDataStream is used to sync the trading data in real-time
+  # it uses the websocket connection to insert the trades
+  userDataStream:
+    trades: false
+    filledOrders: false
+
+  # since is the start date of your trading data
+  since: 2019-01-01
+
+  # sessions is the list of session names you want to sync
+  # by default, BBGO sync all your available sessions.
+  sessions:
+  - binance
+
+  # symbols is the list of symbols you want to sync
+  # by default, BBGO try to guess your symbols by your existing account balances.
+  symbols:
+  - BTCUSDT
+
 # example command:
 #   go run ./cmd/bbgo backtest --config config/grid2.yaml --base-asset-baseline
 backtest:

pkg/bbgo/order_executor_general.go

@@ -218,6 +218,10 @@ func (e *GeneralOrderExecutor) SubmitOrders(ctx context.Context, submitOrders ..
 	}
 
 	createdOrders, errIdx, err := BatchPlaceOrder(ctx, e.session.Exchange, formattedOrders...)
+	if err != nil {
+		log.WithError(err).Errorf("place order error, will retry orders: %v", errIdx)
+	}
+
 	if len(errIdx) > 0 {
 		createdOrders2, err2 := BatchRetryPlaceOrder(ctx, e.session.Exchange, errIdx, formattedOrders...)
 		if err2 != nil {

pkg/exchange/binance/parse.go

@@ -75,7 +75,9 @@ type ExecutionReportEvent struct {
 	OrderPrice fixedpoint.Value `json:"p"`
 	StopPrice  fixedpoint.Value `json:"P"`
 
-	IsOnBook bool `json:"w"`
+	IsOnBook     bool                       `json:"w"`
+	WorkingTime  types.MillisecondTimestamp `json:"W"`
+	TrailingTime types.MillisecondTimestamp `json:"D"`
 
 	IsMaker bool `json:"m"`
 	Ignore  bool `json:"M"`

pkg/exchange/max/exchange.go

@@ -25,6 +25,7 @@ var closedOrderQueryLimiter = rate.NewLimiter(rate.Every(1*time.Second), 1)
 var tradeQueryLimiter = rate.NewLimiter(rate.Every(3*time.Second), 1)
 var accountQueryLimiter = rate.NewLimiter(rate.Every(3*time.Second), 1)
 var marketDataLimiter = rate.NewLimiter(rate.Every(2*time.Second), 10)
+var submitOrderLimiter = rate.NewLimiter(rate.Every(300*time.Millisecond), 10)
 
 var log = logrus.WithField("exchange", "max")
 
@@ -486,6 +487,10 @@ func (e *Exchange) Withdraw(ctx context.Context, asset string, amount fixedpoint
 }
 
 func (e *Exchange) SubmitOrder(ctx context.Context, order types.SubmitOrder) (createdOrder *types.Order, err error) {
+	if err := submitOrderLimiter.Wait(ctx); err != nil {
+		return nil, err
+	}
+
 	walletType := maxapi.WalletTypeSpot
 	if e.MarginSettings.IsMargin {
 		walletType = maxapi.WalletTypeMargin

pkg/strategy/grid2/profit.go

@@ -2,9 +2,13 @@ package grid2
 
 import (
 	"fmt"
+	"strconv"
 	"time"
 
+	"github.com/slack-go/slack"
+
 	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/style"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
@@ -22,3 +26,23 @@ func (p *GridProfit) String() string {
 func (p *GridProfit) PlainText() string {
 	return fmt.Sprintf("Grid profit: %f %s @ %s orderID %d", p.Profit.Float64(), p.Currency, p.Time.String(), p.Order.OrderID)
 }
+
+func (p *GridProfit) SlackAttachment() slack.Attachment {
+	title := fmt.Sprintf("Grid Profit %s %s", style.PnLSignString(p.Profit), p.Currency)
+	return slack.Attachment{
+		Title: title,
+		Color: "warning",
+		Fields: []slack.AttachmentField{
+			{
+				Title: "OrderID",
+				Value: strconv.FormatUint(p.Order.OrderID, 10),
+				Short: true,
+			},
+			{
+				Title: "Time",
+				Value: p.Time.String(),
+				Short: true,
+			},
+		},
+	}
+}

pkg/strategy/grid2/profit_stats.go

@@ -1,7 +1,14 @@
 package grid2
 
 import (
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/slack-go/slack"
+
 	"github.com/c9s/bbgo/pkg/fixedpoint"
+	"github.com/c9s/bbgo/pkg/style"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
@@ -16,6 +23,7 @@ type GridProfitStats struct {
 	Volume           fixedpoint.Value            `json:"volume,omitempty"`
 	Market           types.Market                `json:"market,omitempty"`
 	ProfitEntries    []*GridProfit               `json:"profitEntries,omitempty"`
+	Since            *time.Time                  `json:"since,omitempty"`
 }
 
 func newGridProfitStats(market types.Market) *GridProfitStats {
@@ -43,6 +51,11 @@ func (s *GridProfitStats) AddTrade(trade types.Trade) {
 	} else {
 		s.TotalFee[trade.FeeCurrency] = trade.Fee
 	}
+
+	if s.Since == nil {
+		t := trade.Time.Time()
+		s.Since = &t
+	}
 }
 
 func (s *GridProfitStats) AddProfit(profit *GridProfit) {
@@ -58,3 +71,68 @@ func (s *GridProfitStats) AddProfit(profit *GridProfit) {
 
 	s.ProfitEntries = append(s.ProfitEntries, profit)
 }
+
+func (s *GridProfitStats) SlackAttachment() slack.Attachment {
+	var fields = []slack.AttachmentField{
+		{
+			Title: "Arbitrage Count",
+			Value: strconv.Itoa(s.ArbitrageCount),
+			Short: true,
+		},
+	}
+
+	if !s.FloatProfit.IsZero() {
+		fields = append(fields, slack.AttachmentField{
+			Title: "Float Profit",
+			Value: style.PnLSignString(s.FloatProfit),
+			Short: true,
+		})
+	}
+
+	if !s.GridProfit.IsZero() {
+		fields = append(fields, slack.AttachmentField{
+			Title: "Total Grid Profit",
+			Value: style.PnLSignString(s.GridProfit),
+			Short: true,
+		})
+	}
+
+	if !s.TotalQuoteProfit.IsZero() {
+		fields = append(fields, slack.AttachmentField{
+			Title: "Total Quote Profit",
+			Value: style.PnLSignString(s.TotalQuoteProfit),
+			Short: true,
+		})
+	}
+
+	if !s.TotalBaseProfit.IsZero() {
+		fields = append(fields, slack.AttachmentField{
+			Title: "Total Base Profit",
+			Value: style.PnLSignString(s.TotalBaseProfit),
+			Short: true,
+		})
+	}
+
+	if len(s.TotalFee) > 0 {
+		for feeCurrency, fee := range s.TotalFee {
+			fields = append(fields, slack.AttachmentField{
+				Title: fmt.Sprintf("Fee (%s)", feeCurrency),
+				Value: fee.String() + " " + feeCurrency,
+				Short: true,
+			})
+		}
+	}
+
+	footer := "Total grid profit stats"
+	if s.Since != nil {
+		footer += fmt.Sprintf(" since %s", s.Since.String())
+	}
+
+	title := fmt.Sprintf("%s Grid Profit Stats", s.Symbol)
+	return slack.Attachment{
+		Title:  title,
+		Color:  "warning",
+		Fields: fields,
+		Footer: footer,
+	}
+}

pkg/strategy/grid2/strategy.go

@@ -17,6 +17,8 @@ import (
 
 const ID = "grid2"
 
+const orderTag = "grid2"
+
 var log = logrus.WithField("strategy", ID)
 
 var maxNumberOfOrderTradesQueryTries = 10
@@ -101,6 +103,8 @@ type Strategy struct {
 	// it makes sure that your grid configuration is profitable.
 	FeeRate fixedpoint.Value `json:"feeRate"`
 
+	SkipSpreadCheck bool `json:"skipSpreadCheck"`
+
 	GridProfitStats *GridProfitStats   `persistence:"grid_profit_stats"`
 	ProfitStats     *types.ProfitStats `persistence:"profit_stats"`
 	Position        *types.Position    `persistence:"position"`
@@ -139,8 +143,10 @@ func (s *Strategy) Validate() error {
 		return fmt.Errorf("gridNum can not be zero")
 	}
 
-	if err := s.checkSpread(); err != nil {
-		return errors.Wrapf(err, "spread is too small, please try to reduce your gridNum or increase the price range (upperPrice and lowerPrice)")
+	if !s.SkipSpreadCheck {
+		if err := s.checkSpread(); err != nil {
+			return errors.Wrapf(err, "spread is too small, please try to reduce your gridNum or increase the price range (upperPrice and lowerPrice)")
+		}
 	}
 
 	if !s.QuantityOrAmount.IsSet() && s.QuoteInvestment.IsZero() {
@@ -315,7 +321,7 @@ func (s *Strategy) processFilledOrder(o types.Order) {
 	if o.Side == types.SideTypeBuy {
 		baseSellQuantityReduction = s.aggregateOrderBaseFee(o)
 
-		s.logger.Infof("buy order base fee: %f %s", baseSellQuantityReduction.Float64(), s.Market.BaseCurrency)
+		s.logger.Infof("GRID BUY ORDER BASE FEE: %s %s", baseSellQuantityReduction.String(), s.Market.BaseCurrency)
 
 		newQuantity = newQuantity.Sub(baseSellQuantityReduction)
 	}
@@ -337,12 +343,13 @@ func (s *Strategy) processFilledOrder(o types.Order) {
 			newQuantity = fixedpoint.Max(orderQuoteQuantity.Div(newPrice), s.Market.MinQuantity)
 		}
 
-		// calculate profit
-		// TODO: send profit notification
 		profit := s.calculateProfit(o, newPrice, newQuantity)
 		s.logger.Infof("GENERATED GRID PROFIT: %+v", profit)
 		s.GridProfitStats.AddProfit(profit)
 
+		bbgo.Notify(profit)
+		bbgo.Notify(s.GridProfitStats)
+
 	case types.SideTypeBuy:
 		newSide = types.SideTypeSell
 		if !s.ProfitSpread.IsZero() {
@@ -366,7 +373,7 @@ func (s *Strategy) processFilledOrder(o types.Order) {
 		Side:        newSide,
 		TimeInForce: types.TimeInForceGTC,
 		Quantity:    newQuantity,
-		Tag:         "grid",
+		Tag:         orderTag,
 	}
 
 	s.logger.Infof("SUBMIT GRID REVERSE ORDER: %s", orderForm.String())
@@ -374,13 +381,14 @@ func (s *Strategy) processFilledOrder(o types.Order) {
 	if createdOrders, err := s.orderExecutor.SubmitOrders(context.Background(), orderForm); err != nil {
 		s.logger.WithError(err).Errorf("can not submit arbitrage order")
 	} else {
-		s.logger.Infof("order created: %+v", createdOrders)
+		s.logger.Infof("GRID REVERSE ORDER IS CREATED: %+v", createdOrders)
 	}
 }
 
 // handleOrderFilled is called when an order status is FILLED
 func (s *Strategy) handleOrderFilled(o types.Order) {
 	if s.grid == nil {
+		s.logger.Warn("grid is not opened yet, skip order update event")
 		return
 	}
 
@@ -646,6 +654,13 @@ func (s *Strategy) newTriggerPriceHandler(ctx context.Context, session *bbgo.Exc
 	})
 }
 
+func (s *Strategy) newOrderUpdateHandler(ctx context.Context, session *bbgo.ExchangeSession) func(o types.Order) {
+	return func(o types.Order) {
+		s.handleOrderFilled(o)
+		bbgo.Sync(ctx, s)
+	}
+}
+
 func (s *Strategy) newStopLossPriceHandler(ctx context.Context, session *bbgo.ExchangeSession) types.KLineCallback {
 	return types.KLineWith(s.Symbol, types.Interval1m, func(k types.KLine) {
 		if s.StopLossPrice.Compare(k.Low) < 0 {
@@ -803,17 +818,18 @@ func (s *Strategy) openGrid(ctx context.Context, session *bbgo.ExchangeSession)
 		return err
 	}
 
+	s.debugGridOrders(submitOrders, lastPrice)
+
 	createdOrders, err2 := s.orderExecutor.SubmitOrders(ctx, submitOrders...)
 	if err2 != nil {
 		return err
 	}
 
-	s.debugGridOrders(submitOrders, lastPrice)
-
 	for _, order := range createdOrders {
 		s.logger.Info(order.String())
 	}
 
+	s.logger.Infof("ALL GRID ORDERS SUBMITTED")
 	return nil
 }
 
@@ -864,7 +880,7 @@ func (s *Strategy) generateGridOrders(totalQuote, totalBase, lastPrice fixedpoin
 					Quantity:    quantity,
 					Market:      s.Market,
 					TimeInForce: types.TimeInForceGTC,
-					Tag:         "grid2",
+					Tag:         orderTag,
 				})
 				usedBase = usedBase.Add(quantity)
 			} else if i > 0 {
@@ -879,7 +895,7 @@ func (s *Strategy) generateGridOrders(totalQuote, totalBase, lastPrice fixedpoin
 					Quantity:    quantity,
 					Market:      s.Market,
 					TimeInForce: types.TimeInForceGTC,
-					Tag:         "grid2",
+					Tag:         orderTag,
 				})
 				quoteQuantity := quantity.Mul(price)
 				usedQuote = usedQuote.Add(quoteQuantity)
@@ -899,7 +915,7 @@ func (s *Strategy) generateGridOrders(totalQuote, totalBase, lastPrice fixedpoin
 				Quantity:    quantity,
 				Market:      s.Market,
 				TimeInForce: types.TimeInForceGTC,
-				Tag:         "grid",
+				Tag:         orderTag,
 			})
 			quoteQuantity := quantity.Mul(price)
 			usedQuote = usedQuote.Add(quoteQuantity)
@@ -1027,7 +1043,8 @@ func (s *Strategy) Run(ctx context.Context, _ bbgo.OrderExecutor, session *bbgo.
 	orderExecutor.TradeCollector().OnPositionUpdate(func(position *types.Position) {
 		bbgo.Sync(ctx, s)
 	})
-	orderExecutor.ActiveMakerOrders().OnFilled(s.handleOrderFilled)
+	orderExecutor.ActiveMakerOrders().OnFilled(s.newOrderUpdateHandler(ctx, session))
+
 	s.orderExecutor = orderExecutor
 
 	// TODO: detect if there are previous grid orders on the order book

pkg/strategy/grid2/strategy_test.go

@@ -444,7 +444,7 @@ func TestStrategy_handleOrderFilled(t *testing.T) {
 			Side:        types.SideTypeSell,
 			TimeInForce: types.TimeInForceGTC,
 			Market:      s.Market,
-			Tag:         "grid",
+			Tag:         orderTag,
 		}
 
 		orderExecutor := gridmocks.NewMockOrderExecutor(mockCtrl)
@@ -509,7 +509,7 @@ func TestStrategy_handleOrderFilled(t *testing.T) {
 			Side:        types.SideTypeSell,
 			TimeInForce: types.TimeInForceGTC,
 			Market:      s.Market,
-			Tag:         "grid",
+			Tag:         orderTag,
 		}
 
 		orderExecutor := gridmocks.NewMockOrderExecutor(mockCtrl)
@@ -577,7 +577,7 @@ func TestStrategy_handleOrderFilled(t *testing.T) {
 			Quantity:    number(0.09166666),
 			TimeInForce: types.TimeInForceGTC,
 			Market:      s.Market,
-			Tag:         "grid",
+			Tag:         orderTag,
 		}
 		orderExecutor.EXPECT().SubmitOrders(ctx, expectedSubmitOrder).Return([]types.Order{
 			{SubmitOrder: expectedSubmitOrder},
@@ -591,7 +591,7 @@ func TestStrategy_handleOrderFilled(t *testing.T) {
 			Quantity:    number(0.09999999),
 			TimeInForce: types.TimeInForceGTC,
 			Market:      s.Market,
-			Tag:         "grid",
+			Tag:         orderTag,
 		}
 		orderExecutor.EXPECT().SubmitOrders(ctx, expectedSubmitOrder2).Return([]types.Order{
 			{SubmitOrder: expectedSubmitOrder2},
@@ -664,7 +664,7 @@ func TestStrategy_handleOrderFilled(t *testing.T) {
 			Side:        types.SideTypeSell,
 			TimeInForce: types.TimeInForceGTC,
 			Market:      s.Market,
-			Tag:         "grid",
+			Tag:         orderTag,
 		}
 
 		orderExecutor := gridmocks.NewMockOrderExecutor(mockCtrl)
@@ -680,7 +680,7 @@ func TestStrategy_handleOrderFilled(t *testing.T) {
 			Side:        types.SideTypeBuy,
 			TimeInForce: types.TimeInForceGTC,
 			Market:      s.Market,
-			Tag:         "grid",
+			Tag:         orderTag,
 		}
 
 		orderExecutor.EXPECT().SubmitOrders(ctx, expectedSubmitOrder2).Return([]types.Order{

pkg/types/stream.go

@@ -242,7 +242,7 @@ func (s *StandardStream) Read(ctx context.Context, conn *websocket.Conn, cancel
 			if s.parser != nil {
 				e, err = s.parser(message)
 				if err != nil {
-					log.WithError(err).Errorf("websocket event parse error")
+					log.WithError(err).Errorf("websocket event parse error, message: %s", message)
 					continue
 				}
 			}