feat(quotes): switch Binance driver from streaming raw trades to aggregated trades (#477)

Author: maxpushka

pkg/quotes/binance.go

@@ -129,7 +129,7 @@ func (b *binance) Subscribe(market Market) error {
 	}
 
 	idle := time.NewTimer(b.idlePeriod)
-	doneCh, stopCh, err := gobinance.WsTradeServe(symbol, b.handleTrade(market, idle), b.handleErr(market))
+	doneCh, stopCh, err := gobinance.WsAggTradeServe(symbol, b.handleTrade(market, idle), b.handleErr(market))
 	if err != nil {
 		return fmt.Errorf("%s: %w: %w", market, ErrFailedSub, err)
 	}
@@ -216,8 +216,8 @@ func (b *binance) updateAssets() {
 func (b *binance) handleTrade(
 	market Market,
 	idle *time.Timer,
-) func(*gobinance.WsTradeEvent) {
-	return func(event *gobinance.WsTradeEvent) {
+) func(*gobinance.WsAggTradeEvent) {
+	return func(event *gobinance.WsAggTradeEvent) {
 		idle.Reset(b.idlePeriod)
 
 		tradeEvent, err := b.buildEvent(event, market)
@@ -256,7 +256,7 @@ func (b *binance) handleErr(market Market) func(error) {
 	}
 }
 
-func (b *binance) buildEvent(tr *gobinance.WsTradeEvent, market Market) (TradeEvent, error) {
+func (b *binance) buildEvent(tr *gobinance.WsAggTradeEvent, market Market) (TradeEvent, error) {
 	price, err := decimal.NewFromString(tr.Price)
 	if err != nil {
 		return TradeEvent{}, fmt.Errorf("failed to parse price: %+v", tr.Price)
@@ -328,7 +328,7 @@ func (b *binance) HistoricalData(ctx context.Context, market Market, window time
 	// Convert aggregated trades to a trade events
 	trades = make([]TradeEvent, 0, limit)
 	for _, aggTrade := range aggTrades {
-		trade, err := b.buildEvent(&gobinance.WsTradeEvent{
+		trade, err := b.buildEvent(&gobinance.WsAggTradeEvent{
 			Price:        aggTrade.Price,
 			Quantity:     aggTrade.Quantity,
 			TradeTime:    aggTrade.Timestamp,

pkg/quotes/testing/app.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	_ "net/http/pprof"
 	"os"
+	"time"
 
 	"github.com/ipfs/go-log/v2"
 
@@ -62,7 +63,9 @@ func main() {
 				"market", e.Market,
 				"side", e.TakerType.String(),
 				"price", e.Price.String(),
-				"amount", e.Amount.String())
+				"amount", e.Amount.String(),
+				"created_received_diff", time.Now().Sub(e.CreatedAt).String(),
+			)
 		}
 		outboxStop <- struct{}{}
 	}()