lychiyu/bbgo_origin
1
0
Fork 0
mirror of https://github.com/c9s/bbgo.git synced 2024-11-10 09:11:55 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #737 from c9s/refactor/reward-sync

Browse Source 
refactor: refactor reward service sync
This commit is contained in:
Yo-An Lin 2022-06-18 02:38:05 +08:00 committed by GitHub
commit 2eeb907452
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 62 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
pkg/bbgo/environment.go
View File

@ -617,7 +617,7 @@ func (environ *Environment) syncWithUserConfig(ctx context.Context, userConfig *
}
if userConfig.Sync.RewardHistory {
if err := environ.SyncService.SyncRewardHistory(ctx, session.Exchange); err != nil {
if err := environ.SyncService.SyncRewardHistory(ctx, session.Exchange, since); err != nil {
return err
}
}
@ -656,35 +656,11 @@ func (environ *Environment) Sync(ctx context.Context, userConfig ...*Config) err
return environ.syncWithUserConfig(ctx, userConfig[0])
}
since := time.Now().AddDate(0, -6, 0)
// the default sync logics
for _, session := range environ.sessions {
if err := environ.syncSession(ctx, session); err != nil {
return err
}
if len(userConfig) == 0 || userConfig[0].Sync == nil {
continue
}
if userConfig[0].Sync.DepositHistory {
if err := environ.SyncService.SyncDepositHistory(ctx, session.Exchange, since); err != nil {
return err
}
}
if userConfig[0].Sync.WithdrawHistory {
if err := environ.SyncService.SyncWithdrawHistory(ctx, session.Exchange, since); err != nil {
return err
}
}
if userConfig[0].Sync.RewardHistory {
if err := environ.SyncService.SyncRewardHistory(ctx, session.Exchange); err != nil {
return err
}
}
}
return nil

1
pkg/exchange/max/exchange.go
View File

@ -230,6 +230,7 @@ func (e *Exchange) queryClosedOrdersByLastOrderID(ctx context.Context, symbol st
}
req.FromID(lastOrderID)
req.Limit(1000)
maxOrders, err := req.Do(ctx)
if err != nil {

95
pkg/service/reward.go
View File

@ -7,8 +7,8 @@ import (
"strings"
"time"
sq "github.com/Masterminds/squirrel"
"github.com/jmoiron/sqlx"
"github.com/sirupsen/logrus"
"github.com/c9s/bbgo/pkg/exchange/batch"
"github.com/c9s/bbgo/pkg/fixedpoint"
@ -23,74 +23,45 @@ type RewardService struct {
DB *sqlx.DB
}
func (s *RewardService) QueryLast(ex types.ExchangeName, limit int) ([]types.Reward, error) {
sql := "SELECT * FROM `rewards` WHERE `exchange` = :exchange ORDER BY `created_at` DESC LIMIT :limit"
rows, err := s.DB.NamedQuery(sql, map[string]interface{}{
"exchange": ex,
"limit": limit,
})
if err != nil {
return nil, err
}
defer rows.Close()
return s.scanRows(rows)
}
func (s *RewardService) Sync(ctx context.Context, exchange types.Exchange) error {
service, ok := exchange.(types.ExchangeRewardService)
func (s *RewardService) Sync(ctx context.Context, exchange types.Exchange, startTime time.Time) error {
api, ok := exchange.(types.ExchangeRewardService)
if !ok {
return ErrExchangeRewardServiceNotImplemented
}
var rewardKeys = map[string]struct{}{}
var startTime time.Time
records, err := s.QueryLast(exchange.Name(), 50)
if err != nil {
return err
isMargin, isFutures, _, _ := getExchangeAttributes(exchange)
if isMargin || isFutures {
return nil
}
if len(records) > 0 {
lastRecord := records[0]
startTime = lastRecord.CreatedAt.Time()
for _, record := range records {
rewardKeys[record.UUID] = struct{}{}
}
tasks := []SyncTask{
{
Type: types.Reward{},
Select: SelectLastRewards(exchange.Name(), 100),
BatchQuery: func(ctx context.Context, startTime, endTime time.Time) (interface{}, chan error) {
query := &batch.RewardBatchQuery{
Service: api,
}
return query.Query(ctx, startTime, endTime)
},
Time: func(obj interface{}) time.Time {
return obj.(types.Reward).CreatedAt.Time()
},
ID: func(obj interface{}) string {
reward := obj.(types.Reward)
return string(reward.Type) + "_" + reward.UUID
},
LogInsert: true,
},
}
batchQuery := &batch.RewardBatchQuery{Service: service}
rewardsC, errC := batchQuery.Query(ctx, startTime, time.Now())
for reward := range rewardsC {
select {
case <-ctx.Done():
return ctx.Err()
case err := <-errC:
if err != nil {
return err
}
default:
}
if _, ok := rewardKeys[reward.UUID]; ok {
continue
}
logrus.Infof("inserting reward: %s %s %s %f %s", reward.Exchange, reward.Type, reward.Currency, reward.Quantity.Float64(), reward.CreatedAt)
if err := s.Insert(reward); err != nil {
for _, sel := range tasks {
if err := sel.execute(ctx, s.DB, startTime); err != nil {
return err
}
}
return <-errC
return nil
}
type CurrencyPositionMap map[string]fixedpoint.Value
@ -215,3 +186,13 @@ func (s *RewardService) Insert(reward types.Reward) error {
reward)
return err
}
func SelectLastRewards(ex types.ExchangeName, limit uint64) sq.SelectBuilder {
return sq.Select("*").
From("rewards").
Where(sq.And{
sq.Eq{"exchange": ex},
}).
OrderBy("created_at DESC").
Limit(limit)
}

4
pkg/service/sync.go
View File

@ -75,13 +75,13 @@ func (s *SyncService) SyncMarginHistory(ctx context.Context, exchange types.Exch
return nil
}
func (s *SyncService) SyncRewardHistory(ctx context.Context, exchange types.Exchange) error {
func (s *SyncService) SyncRewardHistory(ctx context.Context, exchange types.Exchange, startTime time.Time) error {
if _, implemented := exchange.(types.ExchangeRewardService); !implemented {
return nil
}
log.Infof("syncing %s reward records...", exchange.Name())
if err := s.RewardService.Sync(ctx, exchange); err != nil {
if err := s.RewardService.Sync(ctx, exchange, startTime); err != nil {
return err
}

31
pkg/types/reward.go
View File

@ -1,6 +1,7 @@
package types
import (
"fmt"
"time"
"github.com/c9s/bbgo/pkg/fixedpoint"
@ -18,18 +19,26 @@ const (
)
type Reward struct {
GID int64 `json:"gid" db:"gid"`
UUID string `json:"uuid" db:"uuid"`
Exchange ExchangeName `json:"exchange" db:"exchange"`
Type RewardType `json:"reward_type" db:"reward_type"`
Currency string `json:"currency" db:"currency"`
Quantity fixedpoint.Value `json:"quantity" db:"quantity"`
State string `json:"state" db:"state"`
Note string `json:"note" db:"note"`
Spent bool `json:"spent" db:"spent"`
GID int64 `json:"gid" db:"gid"`
UUID string `json:"uuid" db:"uuid"`
Exchange ExchangeName `json:"exchange" db:"exchange"`
Type RewardType `json:"reward_type" db:"reward_type"`
Currency string `json:"currency" db:"currency"`
Quantity fixedpoint.Value `json:"quantity" db:"quantity"`
State string `json:"state" db:"state"`
Note string `json:"note" db:"note"`
Spent bool `json:"spent" db:"spent"`
CreatedAt Time `json:"created_at" db:"created_at"`
}
// Unix timestamp in seconds
CreatedAt Time `json:"created_at" db:"created_at"`
func (r Reward) String() (s string) {
s = fmt.Sprintf("reward %s %s %20s %20f %5s @ %s", r.Exchange, r.UUID, r.Type, r.Quantity.Float64(), r.Currency, r.CreatedAt.String())
if r.Note != "" {
s += ": " + r.Note
}
return s
}
type RewardSlice []Reward