grid2: integrate prometheus metrics

pkg/bbgo/activeorderbook.go

@@ -20,6 +20,8 @@ const CancelOrderWaitTime = 20 * time.Millisecond
 type ActiveOrderBook struct {
 	Symbol string
 	orders *types.SyncOrderMap
+
+	newCallbacks      []func(o types.Order)
 	filledCallbacks   []func(o types.Order)
 	canceledCallbacks []func(o types.Order)
 

pkg/bbgo/activeorderbook_callbacks.go

@@ -6,6 +6,16 @@ import (
 	"github.com/c9s/bbgo/pkg/types"
 )
 
+func (b *ActiveOrderBook) OnNew(cb func(o types.Order)) {
+	b.newCallbacks = append(b.newCallbacks, cb)
+}
+
+func (b *ActiveOrderBook) EmitNew(o types.Order) {
+	for _, cb := range b.newCallbacks {
+		cb(o)
+	}
+}
+
 func (b *ActiveOrderBook) OnFilled(cb func(o types.Order)) {
 	b.filledCallbacks = append(b.filledCallbacks, cb)
 }

pkg/strategy/grid2/metrics.go

@@ -0,0 +1,77 @@
+package grid2
+
+import "github.com/prometheus/client_golang/prometheus"
+
+var (
+	metricsGridNumOfOrders *prometheus.GaugeVec
+	metricsGridOrderPrices *prometheus.GaugeVec
+	metricsGridProfit      *prometheus.GaugeVec
+)
+
+func labelKeys(labels prometheus.Labels) []string {
+	var keys []string
+	for k := range labels {
+		keys = append(keys, k)
+	}
+
+	return keys
+}
+
+func mergeLabels(a, b prometheus.Labels) prometheus.Labels {
+	labels := prometheus.Labels{}
+	for k, v := range a {
+		labels[k] = v
+	}
+
+	for k, v := range b {
+		labels[k] = v
+	}
+	return labels
+}
+
+func initMetrics(extendedLabels []string) {
+	metricsGridNumOfOrders = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "bbgo_grid2_num_of_orders",
+			Help: "number of orders",
+		},
+		append([]string{
+			"strategy_instance",
+			"exchange", // exchange name
+			"symbol",   // symbol of the market
+		}, extendedLabels...),
+	)
+
+	metricsGridOrderPrices = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "bbgo_grid2_order_prices",
+			Help: "order prices",
+		},
+		append([]string{
+			"strategy_instance",
+			"exchange", // exchange name
+			"symbol",   // symbol of the market
+			"ith",
+		}, extendedLabels...),
+	)
+
+	metricsGridProfit = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "bbgo_grid2_grid_profit",
+			Help: "realized grid profit",
+		},
+		append([]string{
+			"strategy_instance",
+			"exchange", // exchange name
+			"symbol",   // symbol of the market
+		}, extendedLabels...),
+	)
+}
+
+func registerMetrics() {
+	prometheus.MustRegister(
+		metricsGridNumOfOrders,
+		metricsGridProfit,
+		metricsGridOrderPrices,
+	)
+}

pkg/strategy/grid2/strategy.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
 
 	"github.com/c9s/bbgo/pkg/bbgo"
@@ -111,6 +112,13 @@ type Strategy struct {
 
 	ResetPositionWhenStart bool `json:"resetPositionWhenStart"`
 
+	// StrategyInstance
+	// an optional field for prometheus metrics
+	StrategyInstance string `json:"strategyInstance"`
+
+	// PrometheusLabels will be used as the base prometheus labels
+	PrometheusLabels prometheus.Labels `json:"prometheusLabels"`
+
 	// FeeRate is used for calculating the minimal profit spread.
 	// it makes sure that your grid configuration is profitable.
 	FeeRate fixedpoint.Value `json:"feeRate"`
@@ -135,6 +143,7 @@ type Strategy struct {
 	gridReadyCallbacks  []func()
 	gridProfitCallbacks []func(stats *GridProfitStats, profit *GridProfit)
 	gridClosedCallbacks []func()
+	gridErrorCallbacks  []func(err error)
 }
 
 func (s *Strategy) ID() string {
@@ -173,6 +182,13 @@ func (s *Strategy) Validate() error {
 	return nil
 }
 
+func (s *Strategy) Defaults() error {
+	if s.StrategyInstance == "" {
+		s.StrategyInstance = "main"
+	}
+	return nil
+}
+
 func (s *Strategy) Subscribe(session *bbgo.ExchangeSession) {
 	session.Subscribe(types.KLineChannel, s.Symbol, types.SubscribeOptions{Interval: types.Interval1m})
 
@@ -355,10 +371,7 @@ func (s *Strategy) processFilledOrder(o types.Order) {
 		profit := s.calculateProfit(o, newPrice, newQuantity)
 		s.logger.Infof("GENERATED GRID PROFIT: %+v", profit)
 		s.GridProfitStats.AddProfit(profit)
-
 		s.EmitGridProfit(s.GridProfitStats, profit)
-		bbgo.Notify(profit)
-		bbgo.Notify(s.GridProfitStats)
 
 	case types.SideTypeBuy:
 		newSide = types.SideTypeSell
@@ -668,6 +681,8 @@ func (s *Strategy) newOrderUpdateHandler(ctx context.Context, session *bbgo.Exch
 	return func(o types.Order) {
 		s.handleOrderFilled(o)
 		bbgo.Sync(ctx, s)
+
+		s.updateOpenOrderPricesMetrics(s.orderExecutor.ActiveMakerOrders().Orders())
 	}
 }
 
@@ -835,6 +850,10 @@ func (s *Strategy) openGrid(ctx context.Context, session *bbgo.ExchangeSession)
 		return err
 	}
 
+	// update the number of orders to metrics
+	baseLabels := s.newPrometheusLabels()
+	metricsGridNumOfOrders.With(baseLabels).Set(float64(len(createdOrders)))
+
 	var orderIds []uint64
 
 	for _, order := range createdOrders {
@@ -854,9 +873,30 @@ func (s *Strategy) openGrid(ctx context.Context, session *bbgo.ExchangeSession)
 
 	s.logger.Infof("ALL GRID ORDERS SUBMITTED")
 	s.EmitGridReady()
+
+	s.updateOpenOrderPricesMetrics(createdOrders)
 	return nil
 }
 
+func (s *Strategy) updateOpenOrderPricesMetrics(orders []types.Order) {
+	orders = sortOrdersByPriceAscending(orders)
+	num := len(orders)
+	for idx, order := range orders {
+		labels := s.newPrometheusLabels()
+		labels["ith"] = strconv.Itoa(num - idx)
+		metricsGridOrderPrices.With(labels).Set(order.Price.Float64())
+	}
+}
+
+func sortOrdersByPriceAscending(orders []types.Order) []types.Order {
+	sort.Slice(orders, func(i, j int) bool {
+		a := orders[i]
+		b := orders[j]
+		return a.Price.Compare(b.Price) < 0
+	})
+	return orders
+}
+
 func (s *Strategy) debugGridOrders(submitOrders []types.SubmitOrder, lastPrice fixedpoint.Value) {
 	s.logger.Infof("GRID ORDERS: [")
 	for i, order := range submitOrders {
@@ -1253,6 +1293,20 @@ func scanMissingPinPrices(orderBook *bbgo.ActiveOrderBook, pins []Pin) PriceMap
 	return missingPrices
 }
 
+func (s *Strategy) newPrometheusLabels() prometheus.Labels {
+	labels := prometheus.Labels{
+		"strategy_instance": s.StrategyInstance,
+		"exchange":          s.session.Name,
+		"symbol":            s.Symbol,
+	}
+
+	if s.PrometheusLabels == nil {
+		return labels
+	}
+
+	return mergeLabels(s.PrometheusLabels, labels)
+}
+
 func (s *Strategy) Run(ctx context.Context, _ bbgo.OrderExecutor, session *bbgo.ExchangeSession) error {
 	instanceID := s.InstanceID()
 
@@ -1290,6 +1344,14 @@ func (s *Strategy) Run(ctx context.Context, _ bbgo.OrderExecutor, session *bbgo.
 		s.Position = types.NewPositionFromMarket(s.Market)
 	}
 
+	// initialize and register prometheus metrics
+	if s.PrometheusLabels != nil {
+		initMetrics(labelKeys(s.PrometheusLabels))
+	} else {
+		initMetrics(nil)
+	}
+	registerMetrics()
+
 	if s.ResetPositionWhenStart {
 		s.Position.Reset()
 	}
@@ -1318,6 +1380,16 @@ func (s *Strategy) Run(ctx context.Context, _ bbgo.OrderExecutor, session *bbgo.
 
 	s.orderExecutor = orderExecutor
 
+	s.OnGridProfit(func(stats *GridProfitStats, profit *GridProfit) {
+		bbgo.Notify(profit)
+		bbgo.Notify(stats)
+	})
+
+	s.OnGridProfit(func(stats *GridProfitStats, profit *GridProfit) {
+		labels := s.newPrometheusLabels()
+		metricsGridProfit.With(labels).Set(stats.TotalQuoteProfit.Float64())
+	})
+
 	// TODO: detect if there are previous grid orders on the order book
 	if s.ClearOpenOrdersWhenStart {
 		if err := s.clearOpenOrders(ctx, session); err != nil {

pkg/strategy/grid2/strategy_callbacks.go

@@ -33,3 +33,13 @@ func (s *Strategy) EmitGridClosed() {
 		cb()
 	}
 }
+
+func (s *Strategy) OnGridError(cb func(err error)) {
+	s.gridErrorCallbacks = append(s.gridErrorCallbacks, cb)
+}
+
+func (s *Strategy) EmitGridError(err error) {
+	for _, cb := range s.gridErrorCallbacks {
+		cb(err)
+	}
+}