bbgo_origin/pkg/exchange/batch/batch.go

166 lines
3.6 KiB
Go
Raw Normal View History

package batch
import (
"context"
2021-12-13 23:15:18 +00:00
"sort"
"time"
2021-12-25 19:04:21 +00:00
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"golang.org/x/time/rate"
"github.com/c9s/bbgo/pkg/types"
)
type KLineBatchQuery struct {
types.Exchange
}
2021-12-13 23:15:18 +00:00
func (e KLineBatchQuery) Query(ctx context.Context, symbol string, interval types.Interval, startTime, endTime time.Time) (c chan []types.KLine, errC chan error) {
c = make(chan []types.KLine, 1000)
errC = make(chan error, 1)
go func() {
defer close(c)
defer close(errC)
2021-12-13 23:15:18 +00:00
tryQueryKlineTimes := 0
2021-12-14 16:59:28 +00:00
var currentTime = startTime
for currentTime.Before(endTime) {
kLines, err := e.QueryKLines(ctx, symbol, interval, types.KLineQueryOptions{
2021-12-14 16:59:28 +00:00
StartTime: &currentTime,
EndTime: &endTime,
})
if err != nil {
errC <- err
return
}
2021-12-25 19:04:21 +00:00
sort.Slice(kLines, func(i, j int) bool { return kLines[i].StartTime.Unix() < kLines[j].StartTime.Unix() })
if len(kLines) == 0 {
return
2021-12-25 19:04:21 +00:00
} else if len(kLines) == 1 && kLines[0].StartTime.Unix() == currentTime.Unix() {
return
}
2021-12-25 19:04:21 +00:00
tryQueryKlineTimes++
2021-12-13 23:15:18 +00:00
const BatchSize = 200
2021-12-13 23:15:18 +00:00
var batchKLines = make([]types.KLine, 0, BatchSize)
for _, kline := range kLines {
2021-12-25 19:04:21 +00:00
// ignore any kline before the given start time of the batch query
if currentTime.Unix() != startTime.Unix() && kline.StartTime.Before(currentTime) {
continue
}
2021-12-25 19:04:21 +00:00
// if there is a kline after the endTime of the batch query, it means the data is out of scope, we should exit
if kline.StartTime.After(endTime) || kline.EndTime.After(endTime) {
if len(batchKLines) != 0 {
c <- batchKLines
batchKLines = nil
}
return
}
2021-12-13 23:15:18 +00:00
batchKLines = append(batchKLines, kline)
if len(batchKLines) == BatchSize {
c <- batchKLines
batchKLines = nil
2021-12-13 23:15:18 +00:00
}
2021-12-25 19:04:21 +00:00
// The issue is in FTX, prev endtime = next start time , so if add 1 ms , it would query forever.
currentTime = kline.StartTime.Time()
2021-12-13 23:15:18 +00:00
tryQueryKlineTimes = 0
}
2021-12-25 19:04:21 +00:00
// push the rest klines in the buffer
if len(batchKLines) > 0 {
c <- batchKLines
batchKLines = nil
}
2021-12-13 23:15:18 +00:00
if tryQueryKlineTimes > 10 { // it means loop 10 times
2021-12-25 19:04:21 +00:00
errC <- errors.Errorf("there's a dead loop in batch.go#Query , symbol: %s , interval: %s, startTime:%s ", symbol, interval, startTime.String())
2021-12-13 23:15:18 +00:00
return
}
}
}()
return c, errC
}
type RewardBatchQuery struct {
Service types.ExchangeRewardService
}
func (q *RewardBatchQuery) Query(ctx context.Context, startTime, endTime time.Time) (c chan types.Reward, errC chan error) {
c = make(chan types.Reward, 500)
errC = make(chan error, 1)
go func() {
limiter := rate.NewLimiter(rate.Every(5*time.Second), 2) // from binance (original 1200, use 1000 for safety)
defer close(c)
defer close(errC)
lastID := ""
rewardKeys := make(map[string]struct{}, 500)
for startTime.Before(endTime) {
if err := limiter.Wait(ctx); err != nil {
logrus.WithError(err).Error("rate limit error")
}
logrus.Infof("batch querying rewards %s <=> %s", startTime, endTime)
rewards, err := q.Service.QueryRewards(ctx, startTime)
if err != nil {
errC <- err
return
}
// empty data
if len(rewards) == 0 {
return
}
// there is no new data
if len(rewards) == 1 && rewards[0].UUID == lastID {
return
}
2021-03-17 17:15:29 +00:00
newCnt := 0
for _, o := range rewards {
if _, ok := rewardKeys[o.UUID]; ok {
continue
}
if o.CreatedAt.Time().After(endTime) {
// stop batch query
return
}
2021-03-17 17:15:29 +00:00
newCnt++
c <- o
rewardKeys[o.UUID] = struct{}{}
}
2021-03-17 16:46:10 +00:00
2021-03-17 17:15:29 +00:00
if newCnt == 0 {
return
}
end := len(rewards) - 1
2021-03-17 16:46:10 +00:00
startTime = rewards[end].CreatedAt.Time()
lastID = rewards[end].UUID
}
}()
return c, errC
}