Add metrics for trade tracking.

api/exchange.go

@@ -237,6 +237,19 @@ type ExchangeShim interface {
 	FillTrackable
 }
 
+// TradeMetricsHandler is invoked by the MetricsTracker to process new trades
+type TradeMetricsHandler interface {
+	Read(trades []model.Trade)
+	Reset()
+	TotalBaseVolume() float64
+	NumTrades() int
+}
+
+// MetricsTracker knows how to track metrics, including trades
+type MetricsTracker interface {
+	RegisterHandler(handler TradeMetricsHandler)
+}
+
 // ConvertOperation2TM is a temporary adapter to support transitioning from the old Go SDK to the new SDK without having to bump the major version
 func ConvertOperation2TM(ops []txnbuild.Operation) []build.TransactionMutator {
 	muts := []build.TransactionMutator{}

cmd/trade.go

@@ -609,6 +609,9 @@ func runTradeCmd(options inputs) {
 		logger.Fatal(l, fmt.Errorf("could not generate metrics tracker: %s", e))
 	}
 
+	tradeMetricsHandler := plugins.MakeTradeMetricsHandler()
+	metricsTracker.RegisterHandler(tradeMetricsHandler)
+
 	e = metricsTracker.SendStartupEvent()
 	if e != nil {
 		logger.Fatal(l, fmt.Errorf("could not send startup event metric: %s", e))

plugins/tradeMetricsHandler.go

@@ -0,0 +1,42 @@
+package plugins
+
+import (
+	"github.com/stellar/kelp/model"
+)
+
+// TradeMetricsHandler tracks the number of trades
+type TradeMetricsHandler struct {
+	trades []model.Trade
+}
+
+// MakeTradeMetricsHandler is a factory method for the TradeMetricsHandler
+func MakeTradeMetricsHandler() *TradeMetricsHandler {
+	return &TradeMetricsHandler{
+		trades: []model.Trade{},
+	}
+}
+
+// Reset sets the handler's trade counter to zero.
+func (h *TradeMetricsHandler) Reset() {
+	h.trades = []model.Trade{}
+}
+
+// Read stores new trades internally.
+func (h *TradeMetricsHandler) Read(newTrades []model.Trade) {
+	for _, nt := range newTrades {
+		h.trades = append(h.trades, nt)
+	}
+}
+
+// NumTrades returns the number of trades.
+func (h *TradeMetricsHandler) NumTrades() int {
+	return len(h.trades)
+}
+
+// TotalBaseVolume returns the total base volume.
+func (h *TradeMetricsHandler) TotalBaseVolume() (total float64) {
+	for _, t := range h.trades {
+		total += t.Volume.AsFloat()
+	}
+	return
+}

support/metrics/metricsTracker.go

@@ -8,6 +8,8 @@ import (
 	"runtime/debug"
 	"time"
 
+	"github.com/stellar/kelp/api"
+	"github.com/stellar/kelp/model"
 	"github.com/stellar/kelp/support/networking"
 )
 
@@ -31,6 +33,9 @@ type MetricsTracker struct {
 	botStartTime        time.Time
 	isDisabled          bool
 	updateEventSentTime *time.Time
+
+	// uninitialized
+	handlers []api.TradeMetricsHandler
 }
 
 // TODO DS Investigate other fields to add to this top-level event.
@@ -81,6 +86,8 @@ type commonProps struct {
 	OperationalBufferNonNativePct    float64 `json:"operational_buffer_non_native_pct"`
 	SimMode                          bool    `json:"sim_mode"`
 	FixedIterations                  uint64  `json:"fixed_iterations"`
+	NumTradesSinceLastUpdate         int     `json:"num_trades_since_last_update"`
+	BaseVolumeTradesSinceLastUpdate  float64 `json:"base_volume_trades_since_last_update"`
 }
 
 // updateProps holds the properties for the update Amplitude event.
@@ -304,3 +311,41 @@ func (mt *MetricsTracker) sendEvent(eventType string, eventProps interface{}) er
 	}
 	return nil
 }
+
+// RegisterHandler adds an internal handler.
+func (mt *MetricsTracker) RegisterHandler(handler api.TradeMetricsHandler) {
+	mt.handlers = append(mt.handlers, handler)
+}
+
+// GetHandlers returns the handlers
+func (mt *MetricsTracker) GetHandlers() []api.TradeMetricsHandler {
+	return mt.handlers
+}
+
+// ResetHandlers resets all handlers
+func (mt *MetricsTracker) ResetHandlers() {
+	for _, h := range mt.handlers {
+		h.Reset()
+	}
+}
+
+// ReadIntoHandlers reads to all handlers
+func (mt *MetricsTracker) ReadIntoHandlers(trades []model.Trade) {
+	for _, h := range mt.handlers {
+		h.Read(trades)
+	}
+}
+
+func (mt *MetricsTracker) getTotalVolume() (total float64) {
+	for _, h := range mt.handlers {
+		total += h.TotalBaseVolume()
+	}
+	return
+}
+
+func (mt *MetricsTracker) getTotalNumTrades() (total int) {
+	for _, h := range mt.handlers {
+		total += h.NumTrades()
+	}
+	return
+}

trader/trader.go

@@ -311,6 +311,8 @@ func (t *Trader) synchronizeFetchBalancesOffersTrades() error {
 			buyingAOffers2,
 		) {
 			// this is the only success case
+			t.metricsTracker.ResetHandlers()
+			t.metricsTracker.ReadIntoHandlers(trades)
 			t.setBalances(baseBalance1, quoteBalance1)
 			t.setExistingOffers(sellingAOffers1, buyingAOffers1)
 			return nil