diff --git a/pkg/bbgo/environment.go b/pkg/bbgo/environment.go index 3e7dd2ddd..41f80475d 100644 --- a/pkg/bbgo/environment.go +++ b/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 diff --git a/pkg/bbgo/server.go b/pkg/bbgo/server.go index 8490924e9..e6829fd71 100644 --- a/pkg/bbgo/server.go +++ b/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 { diff --git a/pkg/bbgo/session.go b/pkg/bbgo/session.go index a32c6c3df..27b3c01f2 100644 --- a/pkg/bbgo/session.go +++ b/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 +} diff --git a/pkg/service/trade.go b/pkg/service/trade.go index ba72bcb55..1e59fef84 100644 --- a/pkg/service/trade.go +++ b/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)