add trading volume query api

pkg/bbgo/environment.go

@@ -253,21 +253,6 @@ func (environ *Environment) Init(ctx context.Context) (err error) {
 		log.Infof("%s account", session.Name)
 		balances.Print()
 
-		for _, b := range balances {
-			priceSymbol := b.Currency + "USDT"
-			startTime := time.Now().Add(-10 * time.Minute)
-			klines, err := session.Exchange.QueryKLines(ctx, priceSymbol, types.Interval1m, types.KLineQueryOptions{
-				Limit:     100,
-				StartTime: &startTime,
-			})
-
-			if err != nil || len(klines) == 0 {
-				continue
-			}
-
-			session.lastPrices[priceSymbol] = klines[len(klines)-1].Close
-		}
-
 		session.Account.UpdateBalances(balances)
 		session.Account.BindStream(session.Stream)
 
@@ -543,6 +528,7 @@ func (environ *Environment) SyncTradesFrom(t time.Time) *Environment {
 	return environ
 }
 
+
 func (environ *Environment) Connect(ctx context.Context) error {
 	for n := range environ.sessions {
 		// avoid using the placeholder variable for the session because we use that in the callbacks

pkg/bbgo/server.go

@@ -8,7 +8,9 @@ import (
 
 	"github.com/gin-contrib/cors"
 	"github.com/gin-gonic/gin"
+	log "github.com/sirupsen/logrus"
 
+	"github.com/c9s/bbgo/pkg/service"
 	"github.com/c9s/bbgo/pkg/types"
 )
 
@@ -26,6 +28,31 @@ func RunServer(ctx context.Context, userConfig *Config, environ *Environment) er
 		c.JSON(http.StatusOK, gin.H{"message": "pong"})
 	})
 
+	r.GET("/api/trading-volume", func(c *gin.Context) {
+		period := c.DefaultQuery("period", "day")
+		startTimeString := c.DefaultQuery("start-time", time.Now().AddDate(0, 0, -7).Format(time.RFC3339))
+
+		startTime, err := time.Parse(time.RFC3339, startTimeString)
+		if err != nil {
+			c.Status(http.StatusBadRequest)
+			log.WithError(err).Error("start-time format incorrect")
+			return
+		}
+
+		rows, err := environ.TradeService.QueryTradingVolume(startTime, service.TradingVolumeQueryOptions{
+			GroupByExchange: false,
+			GroupByPeriod:   period,
+		})
+		if err != nil {
+			log.WithError(err).Error("trading volume query error")
+			c.Status(http.StatusInternalServerError)
+			return
+		}
+
+		c.JSON(http.StatusOK, rows)
+		return
+	})
+
 	r.GET("/api/sessions", func(c *gin.Context) {
 		c.JSON(http.StatusOK, gin.H{"sessions": userConfig.Sessions})
 	})
@@ -35,6 +62,13 @@ func RunServer(ctx context.Context, userConfig *Config, environ *Environment) er
 
 		for _, session := range environ.sessions {
 			balances := session.Account.Balances()
+
+			if err := session.UpdatePrices(ctx); err != nil {
+				log.WithError(err).Error("price update failed")
+				c.Status(http.StatusInternalServerError)
+				return
+			}
+
 			assets := balances.Assets(session.lastPrices)
 
 			for currency, asset := range assets {

pkg/bbgo/session.go

@@ -1,7 +1,9 @@
 package bbgo
 
 import (
+	"context"
 	"fmt"
+	"time"
 
 	"github.com/c9s/bbgo/pkg/indicator"
 	"github.com/c9s/bbgo/pkg/types"
@@ -241,3 +243,24 @@ func (session *ExchangeSession) FormatOrder(order types.SubmitOrder) (types.Subm
 	order.QuantityString = market.FormatQuantity(order.Quantity)
 	return order, nil
 }
+
+func (session *ExchangeSession) UpdatePrices(ctx context.Context) (err error) {
+	balances := session.Account.Balances()
+
+	for _, b := range balances {
+		priceSymbol := b.Currency + "USDT"
+		startTime := time.Now().Add(-10 * time.Minute)
+		klines, err := session.Exchange.QueryKLines(ctx, priceSymbol, types.Interval1m, types.KLineQueryOptions{
+			Limit:     100,
+			StartTime: &startTime,
+		})
+
+		if err != nil || len(klines) == 0 {
+			continue
+		}
+
+		session.lastPrices[priceSymbol] = klines[len(klines)-1].Close
+	}
+
+	return err
+}

pkg/service/trade.go

@@ -1,6 +1,9 @@
 package service
 
 import (
+	"strings"
+	"time"
+
 	"github.com/jmoiron/sqlx"
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
@@ -8,6 +11,21 @@ import (
 	"github.com/c9s/bbgo/pkg/types"
 )
 
+type TradingVolume struct {
+	Year        int       `db:"year" json:"year"`
+	Month       int       `db:"month" json:"month,omitempty"`
+	Day         int       `db:"day" json:"day,omitempty"`
+	Time        time.Time `json:"time,omitempty"`
+	Exchange    string    `db:"exchange" json:"exchange,omitempty"`
+	Symbol      string    `db:"symbol" json:"symbol,omitempty"`
+	QuoteVolume float64   `db:"quote_volume" json:"quoteVolume"`
+}
+
+type TradingVolumeQueryOptions struct {
+	GroupByExchange bool
+	GroupByPeriod   string
+}
+
 type TradeService struct {
 	DB *sqlx.DB
 }
@@ -16,6 +34,80 @@ func NewTradeService(db *sqlx.DB) *TradeService {
 	return &TradeService{db}
 }
 
+func (s *TradeService) QueryTradingVolume(startTime time.Time, options TradingVolumeQueryOptions) ([]TradingVolume, error) {
+	var sel []string
+	var groupBys []string
+	var orderBys []string
+	where := []string{"traded_at > :start_time"}
+	args := map[string]interface{}{
+		// "symbol":      symbol,
+		// "exchange":    ex,
+		// "is_margin":   isMargin,
+		// "is_isolated": isIsolated,
+		"start_time": startTime,
+	}
+
+	switch options.GroupByPeriod {
+
+	case "month":
+		sel = append(sel, "YEAR(traded_at) AS year", "MONTH(traded_at) AS month")
+		groupBys = append([]string{"MONTH(traded_at)", "YEAR(traded_at)"}, groupBys...)
+		orderBys = append(orderBys, "year ASC", "month ASC")
+
+	case "year":
+		sel = append(sel, "YEAR(traded_at) AS year")
+		groupBys = append([]string{"YEAR(traded_at)"}, groupBys...)
+		orderBys = append(orderBys, "year ASC")
+
+	case "day":
+		fallthrough
+
+	default:
+		sel = append(sel, "YEAR(traded_at) AS year", "MONTH(traded_at) AS month", "DAY(traded_at) AS day")
+		groupBys = append([]string{"DAY(traded_at)", "MONTH(traded_at)", "YEAR(traded_at)"}, groupBys...)
+		orderBys = append(orderBys, "year ASC", "month ASC", "day ASC")
+	}
+
+	if options.GroupByExchange {
+		sel = append(sel, "exchange")
+		groupBys = append([]string{"exchange"}, groupBys...)
+		orderBys = append(orderBys, "exchange")
+	}
+
+	sel = append(sel, "SUM(quantity * price) AS quote_volume")
+	sql := `SELECT ` + strings.Join(sel, ", ") + ` FROM trades` +
+		` WHERE ` + strings.Join(where, " AND ") +
+		` GROUP BY ` + strings.Join(groupBys, ", ") +
+		` ORDER BY ` + strings.Join(orderBys, ", ")
+
+	log.Info(sql)
+
+	rows, err := s.DB.NamedQuery(sql, args)
+	if err != nil {
+		return nil, errors.Wrap(err, "query last trade error")
+	}
+
+	if rows.Err() != nil {
+		return nil, rows.Err()
+	}
+
+	defer rows.Close()
+
+	var records []TradingVolume
+	for rows.Next() {
+		var record TradingVolume
+		err = rows.StructScan(&record)
+		if err != nil {
+			return records, err
+		}
+
+		record.Time = time.Date(record.Year, time.Month(record.Month), record.Day, 0, 0, 0, 0, time.UTC)
+		records = append(records, record)
+	}
+
+	return records, rows.Err()
+}
+
 // QueryLast queries the last trade from the database
 func (s *TradeService) QueryLast(ex types.ExchangeName, symbol string, isMargin bool, isIsolated bool) (*types.Trade, error) {
 	log.Infof("querying last trade exchange = %s AND symbol = %s AND is_margin = %v AND is_isolated = %v", ex, symbol, isMargin, isIsolated)