Revert "use cursor to fix [b[[loopSaveCandle function"
This reverts commit b927864dd47f732fee6d48bfe111e1caac8aeffc.
This commit is contained in:
zhangkun9038@dingtalk.com
parent
b927864dd4
commit
a92106313f
35
core/core.go
35
core/core.go
@ -1,35 +0,0 @@
|
||||
package core
|
||||
|
||||
import (
|
||||
"context"
|
||||
)
|
||||
|
||||
type Core struct {
|
||||
// Add any necessary fields here
|
||||
}
|
||||
|
||||
func (c *Core) GetScoreList(count int) []string {
|
||||
// Implementation of GetScoreList
|
||||
}
|
||||
|
||||
func (c *Core) GetScoreListWithContext(ctx context.Context, count int) ([]string, error) {
|
||||
done := make(chan []string, 1)
|
||||
errChan := make(chan error, 1)
|
||||
|
||||
go func() {
|
||||
list := c.GetScoreList(count)
|
||||
select {
|
||||
case done <- list:
|
||||
case <-ctx.Done():
|
||||
}
|
||||
}()
|
||||
|
||||
select {
|
||||
case list := <-done:
|
||||
return list, nil
|
||||
case err := <-errChan:
|
||||
return nil, err
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
1
go.mod
1
go.mod
@ -7,7 +7,6 @@ require (
|
||||
github.com/phyer/core v0.1.90
|
||||
github.com/phyer/v5sdkgo v0.1.4
|
||||
github.com/sirupsen/logrus v1.9.3
|
||||
golang.org/x/sync v0.10.0
|
||||
)
|
||||
|
||||
require (
|
||||
|
||||
3
go.sum
3
go.sum
@ -81,10 +81,7 @@ golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
|
||||
golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
|
||||
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9 h1:SQFwaSi55rU7vdNs9Yr0Z324VNlrF+0wMqRXT4St8ck=
|
||||
golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
|
||||
golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
|
||||
golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
|
||||
163
main.go
163
main.go
@ -3,8 +3,6 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
// "fmt"
|
||||
"math/rand"
|
||||
"os"
|
||||
@ -14,7 +12,6 @@ import (
|
||||
|
||||
"github.com/phyer/v5sdkgo/rest"
|
||||
"github.com/sirupsen/logrus"
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
// "v5sdk_go/ws"
|
||||
// "v5sdk_go/ws/wImpl"
|
||||
@ -172,128 +169,66 @@ func ShowSysTime(cr *core.Core) {
|
||||
// range: 随机的范围，从0开始到range个周期，作为查询的after值，也就是随机n个周期，去取之前的记录,对于2D，5D等数据，可以用来补全数据, range值越大，随机散点的范围越大, 越失焦
|
||||
|
||||
func LoopAllCoinsList(period int64, delay int64, mdura int, barPeriod string, onceCount int, rge int) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
cr := core.Core{}
|
||||
cr.Init()
|
||||
|
||||
// Increase buffer size to accommodate multiple updates
|
||||
// Buffer size should be large enough to handle typical batch size
|
||||
// but not so large that it consumes too much memory
|
||||
const channelBufferSize = 100
|
||||
allScoreChan := make(chan []string, channelBufferSize)
|
||||
|
||||
// Add channel overflow protection
|
||||
sendWithTimeout := func(scores []string) {
|
||||
select {
|
||||
case allScoreChan <- scores:
|
||||
// Successfully sent
|
||||
case <-time.After(5 * time.Second):
|
||||
logrus.Warn("Failed to send scores to channel - buffer full")
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
mainTicker := time.NewTicker(time.Duration(period) * time.Second)
|
||||
defer mainTicker.Stop()
|
||||
|
||||
var eg errgroup.Group
|
||||
|
||||
// Producer goroutine
|
||||
eg.Go(func() error {
|
||||
allScoreChan := make(chan []string)
|
||||
// logrus.Info("start LoopAllCoinsList: period: ", period, " delay: ", delay, " mdura:", mdura, " barPeriod: ", barPeriod, " onceCount: ", onceCount, " rge:", rge)
|
||||
per1 := 1 * time.Minute
|
||||
ticker := time.NewTicker(per1)
|
||||
go func() {
|
||||
for {
|
||||
tsi := time.Now().Unix()
|
||||
//logrus.Info("tsi, period, delay, tsi%(period): ", tsi, period, delay, tsi%(period))
|
||||
if tsi%(period) != delay {
|
||||
time.Sleep(1 * time.Second)
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case <-mainTicker.C:
|
||||
if time.Now().Unix()%period != delay {
|
||||
continue
|
||||
}
|
||||
|
||||
_, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
list := cr.GetScoreList(-1)
|
||||
cancel()
|
||||
|
||||
// Use the new send function with timeout
|
||||
if len(list) > 0 {
|
||||
sendWithTimeout(list)
|
||||
}
|
||||
case <-ticker.C:
|
||||
go func() {
|
||||
// -1 是获取全部coin列表
|
||||
list := cr.GetScoreList(-1)
|
||||
// logrus.Info("allCoins3", list)
|
||||
allScoreChan <- list
|
||||
}()
|
||||
}
|
||||
}
|
||||
})
|
||||
}()
|
||||
for {
|
||||
allScore, _ := <-allScoreChan
|
||||
logrus.Debug("allCoins allScore", allScore)
|
||||
if len(allScore) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
// Consumer goroutine
|
||||
eg.Go(func() error {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case allScore := <-allScoreChan:
|
||||
if len(allScore) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
// Process in batches to avoid overwhelming the system
|
||||
if err := processScores(&cr, allScore, mdura, barPeriod, onceCount, rge); err != nil {
|
||||
logrus.WithError(err).Error("Failed to process scores")
|
||||
// Consider implementing exponential backoff retry here
|
||||
}
|
||||
utils.TickerWrapper(time.Duration(mdura)*time.Second, allScore, func(i int, ary []string) error {
|
||||
nw := time.Now()
|
||||
rand.Seed(nw.UnixNano())
|
||||
ct := rand.Intn(rge)
|
||||
minutes, _ := cr.PeriodToMinutes(barPeriod)
|
||||
tmi := nw.UnixMilli()
|
||||
tmi = tmi - tmi%60000
|
||||
tmi = tmi - (int64(ct) * minutes * 60000)
|
||||
lm := strconv.Itoa(onceCount)
|
||||
// logrus.Info("instId: ", ary[i], " limit: ", lm, " onceCount:", onceCount)
|
||||
if lm == "0" {
|
||||
lm = "100"
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
if err := eg.Wait(); err != nil {
|
||||
logrus.WithError(err).Error("LoopAllCoinsList failed")
|
||||
restQ := core.RestQueue{
|
||||
InstId: ary[i],
|
||||
Bar: barPeriod,
|
||||
WithWs: false,
|
||||
Limit: lm,
|
||||
After: tmi,
|
||||
}
|
||||
js, err := json.Marshal(restQ)
|
||||
logrus.Debug("allCoins lpush js:", string(js))
|
||||
cr.RedisLocalCli.LPush("restQueue", js)
|
||||
return err
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func processScores(cr *core.Core, allScore []string, mdura int, barPeriod string, onceCount int, rge int) error {
|
||||
utils.TickerWrapper(time.Duration(mdura)*time.Second, allScore, func(i int, ary []string) error {
|
||||
nw := time.Now()
|
||||
// 使用更安全的随机数生成
|
||||
randSource := rand.NewSource(nw.UnixNano())
|
||||
randGen := rand.New(randSource)
|
||||
ct := randGen.Intn(rge)
|
||||
|
||||
minutes, err := cr.PeriodToMinutes(barPeriod)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get period minutes: %w", err)
|
||||
}
|
||||
|
||||
tmi := nw.UnixMilli()
|
||||
tmi = tmi - tmi%60000
|
||||
tmi = tmi - (int64(ct) * minutes * 60000)
|
||||
|
||||
limit := onceCount
|
||||
if limit == 0 {
|
||||
limit = 100
|
||||
}
|
||||
|
||||
restQ := core.RestQueue{
|
||||
InstId: ary[i],
|
||||
Bar: barPeriod,
|
||||
WithWs: false,
|
||||
Limit: strconv.Itoa(limit),
|
||||
After: tmi,
|
||||
}
|
||||
|
||||
js, err := json.Marshal(restQ)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to marshal RestQueue: %w", err)
|
||||
}
|
||||
logrus.WithFields(logrus.Fields{
|
||||
"instId": ary[i],
|
||||
"bar": barPeriod,
|
||||
"limit": limit,
|
||||
}).Debug("Pushing to restQueue")
|
||||
|
||||
return cr.RedisLocalCli.LPush("restQueue", js).Err()
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func main() {
|
||||
cr := core.Core{}
|
||||
// level := os.Getenv("TEXUS_LOGLEVEL")
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user