diff --git a/config/config.toml b/config/config.toml index 6b98c37..ec2c449 100644 --- a/config/config.toml +++ b/config/config.toml @@ -63,7 +63,7 @@ marketSubscribeLimit = 16 consumeBatch = 1024 consumeLater = 2000 # 时间到达later或者数据累计到batch触发consume # httpProxy = "http://192.168.1.6:7890" -httpProxy = "" +httpProxy = "http://10.255.183.209:7890" # 模拟盘API交易地址如下： # REST：https://www.okx.com diff --git a/internal/exchange/exchange.go b/internal/exchange/exchange.go index bc700cd..184f793 100644 --- a/internal/exchange/exchange.go +++ b/internal/exchange/exchange.go @@ -5,6 +5,7 @@ import ( "fmt" "sig-pub/pkg/types" "sig-pub/pkg/utils/collect" + "sync" ) // 交易所行情数据订阅 @@ -42,9 +43,25 @@ type Exchange struct { // 交易所交易产品 type ExchangeTradeInstance struct { Inst *types.TradeInstance - Status int32 // 交易产品状态, 0.初始化中 1.正常 - LiveMarkTs int64 // websocket订阅k线标记时间戳 - HistoryMarkTs int64 // 拉取历史k线标记时间戳 + Status int32 // 交易产品状态, 0.初始化中 1.正常 + LiveMarkTs int64 // websocket订阅k线标记时间戳 + LiveKStartTs map[types.Interval]int64 // ws开始订阅k线标记时间戳 + LiveKMarkTs map[types.Interval]int64 // ws实时订阅k线(confirmed)标记时间戳 + HistoryMarkTs map[types.Interval]int64 // 拉取历史k线标记时间戳 + Lock sync.RWMutex +} + +func (exInst *ExchangeTradeInstance) GetLiveMarkTs(interval types.Interval) (ts int64) { + exInst.Lock.RLock() + ts = exInst.LiveKMarkTs[interval] + exInst.Lock.RUnlock() + return +} + +func (exInst *ExchangeTradeInstance) SetLiveMarkTs(interval types.Interval, ts int64) { + exInst.Lock.Lock() + exInst.LiveKMarkTs[interval] = ts + exInst.Lock.Unlock() } func NewExchange(fetcher ExchangeFetcher, subscriber ExchangeSubscriber) *Exchange { diff --git a/internal/exchange/exchange_grpc_server.go b/internal/exchange/exchange_grpc_server.go index 012e2ce..3035ca4 100644 --- a/internal/exchange/exchange_grpc_server.go +++ b/internal/exchange/exchange_grpc_server.go @@ -10,6 +10,7 @@ import ( "sig-pub/pkg/data" "sig-pub/pkg/storage/kvrocks" "sig-pub/pkg/types" + "sig-pub/pkg/utils/collect" "sig-pub/pkg/utils/times" "sig-pub/pkg/zlog" "sync" @@ -85,7 +86,8 @@ func (svc *ExchangeGrpcServer) subscribeExchanges() { Inst: tradeInst, Status: 0, LiveMarkTs: 0, - HistoryMarkTs: 0, + LiveKMarkTs: make(map[types.Interval]int64), + HistoryMarkTs: make(map[types.Interval]int64), }) // 待初始化币种数据 if inst.Status == data.StatusProcessing { @@ -128,9 +130,14 @@ func (svc *ExchangeGrpcServer) consumerKline(exchange *Exchange, c <-chan *types if inst, ok := exchange.Insts.Load(channelK.ExgInstId); ok && inst != nil { exInst = inst.Inst // 标记交易产品开始订阅k线时间 - if inst.LiveMarkTs == 0 && len(channelK.Klines) > 0 { - inst.LiveMarkTs = channelK.Klines[0].Ts + if len(channelK.Klines) > 0 { + kline := collect.MustMax(channelK.Klines, func(k *types.Kline) int64 { return k.Ts }) + if inst.GetLiveMarkTs(kline.Interval) == 0 { + inst.SetLiveMarkTs(kline.Interval, kline.Ts) + } } + // confirmKlines := collect.Filter(channelK.Klines, func(_ int, k *types.Kline) bool { return k.Confirm }) + } else { zlog.Errorf("unknown exchange instId: %v, %s", channelK.Exchange, channelK.ExgInstId) continue @@ -312,13 +319,13 @@ func (svc *ExchangeGrpcServer) initialKlines(exchange *Exchange, insts []types.T var success, failed []types.TradeInstance for _, inst := range insts { - err := svc.initTradeInstanceKlines(exchange, inst) + err := svc.initialTradeInstanceKlines(exchange, inst) if err != nil { + zlog.Errorf("initial fetch trade instance error: %s(%s), err=%v", inst.InstId, inst.Exchange, err) failed = append(failed, inst) } else { success = append(success, inst) } - } zlog.Infof("%d insts initial finished, success %d, failed %d", len(insts), len(success), len(failed)) @@ -329,7 +336,7 @@ var ( ) // initTradeInstanceKlines 初始化交易产品历史k线数据 -func (svc *ExchangeGrpcServer) initTradeInstanceKlines(exchange *Exchange, inst types.TradeInstance) (err error) { +func (svc *ExchangeGrpcServer) initialTradeInstanceKlines(exchange *Exchange, inst types.TradeInstance) (err error) { // 并发数 concurrent := max(8, runtime.NumCPU()*2) // 任务 channel @@ -338,18 +345,35 @@ func (svc *ExchangeGrpcServer) initTradeInstanceKlines(exchange *Exchange, inst // 发布任务数, 成功任务数, 失败任务次数 var pubTasks, subTasks, failTasks atomic.Int32 - + var pubTaskDone atomic.Bool // 所有任务已发布 + var historyMarkTs = make(map[types.Interval]int64) // 已初始化最后k线时间戳 + var historyMarkTsMu sync.Mutex ctx, cancel := context.WithCancel(context.Background()) + defer func() { + if err != nil { + return + } + for interval, ts := range historyMarkTs { + tsKey := fmt.Sprintf(HistoryKlineTsKey, inst.Exchange, inst.InstId, interval) + if ex := svc.kvdb.SetI64(context.Background(), tsKey, ts); ex != nil { + zlog.Errorf("history mark inititaled ts error: key=%s, ts=%d", tsKey, ts) + } + } + }() + go func() { defer func() { - close(taskCh) - taskCh = nil - zlog.Infof("trade instance initial kline %s(%s), pub %d fetch tasks", pubTasks.Load(), inst.InstId, inst.Exchange) + pubTaskDone.Store(true) + // 无任务处理 + if subTasks.Load() == 0 { + cancel() + } + zlog.Infof("trade instance initial kline %s(%s), pub %d fetch tasks", inst.InstId, inst.Exchange, pubTasks.Load()) }() // for interval, intervalAdder := range types.SupportedIntervals { - interval := types.Interval1h + interval := types.Interval1d intervalAdder := types.SupportedIntervals[interval] // history 未补全前, history写 kvdb ts mark, 补全后 ws live 写 ts mark tsKey := fmt.Sprintf(HistoryKlineTsKey, inst.Exchange, inst.InstId, interval) @@ -364,6 +388,27 @@ func (svc *ExchangeGrpcServer) initTradeInstanceKlines(exchange *Exchange, inst beforeTs = intervalAdder(KlineBefore0, -1) } for { + // 对比 ws 获取的实时k线 + exchangeInst, ok := exchange.Insts.Load(inst.ExchangeInstId) + if !ok { + err = fmt.Errorf("not load exchange trade instance: %s", inst.ExchangeInstId) + cancel() + return + } + // 订阅完成 + if exchangeInst.LiveMarkTs != 0 && beforeTs >= exchangeInst.LiveMarkTs { + // history status -> ok + break + } + if beforeTs > time.Now().UnixMilli() { + if exchangeInst.LiveMarkTs != 0 { + // history status -> ok + } else { + // live status -> not ok + } + break + } + afterTs := intervalAdder(beforeTs, 101) task := fetchKlineTask{ inst: inst, @@ -382,25 +427,6 @@ func (svc *ExchangeGrpcServer) initTradeInstanceKlines(exchange *Exchange, inst } beforeTs = intervalAdder(afterTs, -1) - - // 对比 ws 获取的实时k线 - inst, ok := exchange.Insts.Load(inst.ExchangeInstId) - if !ok { - break - } - // 订阅完成 - if inst.LiveMarkTs != 0 && beforeTs > inst.LiveMarkTs { - // history status -> ok - break - } - if beforeTs > time.Now().UnixMilli() { - if inst.LiveMarkTs != 0 { - // history status -> ok - } else { - // live status -> not ok - } - break - } } }() @@ -412,23 +438,19 @@ func (svc *ExchangeGrpcServer) initTradeInstanceKlines(exchange *Exchange, inst defer wg.Done() var task fetchKlineTask - var ok bool for { select { case <-ctx.Done(): return - case task, ok = <-taskCh: - case task, ok = <-retryTaskCh: - } - if !ok { - continue + case task = <-taskCh: + case task = <-retryTaskCh: } if task.times > 0 { zlog.Infof("retry fetch history kline task %d times: task -> %s", task.times, task.logKey()) } - if ex := svc.fetchTaskKlines(exchange, task); ex != nil { + if lastKlineTs, ex := svc.fetchTaskKlines(exchange, task); ex != nil { failTasks.Add(1) if task.times >= SingleTaskMaxFailTimes { err = fmt.Errorf("task failed to many times %d, key: %s, err: %v", task.times, task.logKey(), ex) @@ -443,8 +465,23 @@ func (svc *ExchangeGrpcServer) initTradeInstanceKlines(exchange *Exchange, inst return } } else { + // 周期任务最后kline时间 + if lastKlineTs != 0 { + historyMarkTsMu.Lock() + if ts, ok := historyMarkTs[task.interval]; ok { + if lastKlineTs > ts { + historyMarkTs[task.interval] = lastKlineTs + } + } else { + historyMarkTs[task.interval] = lastKlineTs + } + historyMarkTsMu.Unlock() + } + + zlog.Debugf("trade instance initial kline tasks processing: %s(%s), pub %d, sub %d, fail %d", inst.InstId, inst.Exchange, pubTasks.Load(), subTasks.Load(), failTasks.Load()) // 任务都已执行成功结束 - if subTasks.Add(1) >= pubTasks.Load() { + subs := subTasks.Add(1) + if pubTaskDone.Load() && subs >= pubTasks.Load() { zlog.Infof("trade instance initial kline tasks success finished, %s(%s), pub %d, sub %d, fail %d", inst.InstId, inst.Exchange, pubTasks.Load(), subTasks.Load(), failTasks.Load()) cancel() return @@ -457,7 +494,7 @@ func (svc *ExchangeGrpcServer) initTradeInstanceKlines(exchange *Exchange, inst return } -func (svc *ExchangeGrpcServer) fetchTaskKlines(exchange *Exchange, task fetchKlineTask) (err error) { +func (svc *ExchangeGrpcServer) fetchTaskKlines(exchange *Exchange, task fetchKlineTask) (lastKlineTs int64, err error) { interval, afterTs, beforeTs := task.interval, task.afterTs, task.beforeTs klines, err := exchange.Fetcher.FetchHistoryKlines(context.Background(), task.inst.ExchangeInstId, interval, afterTs, beforeTs) if err != nil { @@ -471,9 +508,10 @@ func (svc *ExchangeGrpcServer) fetchTaskKlines(exchange *Exchange, task fetchKli loc, _ := time.LoadLocation("Asia/Shanghai") sts, ets := klines[0].Ts, klines[len(klines)-1].Ts + lastKlineTs = max(sts, ets) ss := time.UnixMilli(sts).In(loc).Format(times.FORMAT_DATE) ee := time.UnixMilli(ets).In(loc).Format(times.FORMAT_DATE) - zlog.Infof("fetch interval %s %d~%d klines: ret=%d~%d, %d klines, %s~%s", interval, afterTs, beforeTs, sts, ets, len(klines), ss, ee) + zlog.Infof("fetch interval %s %d~%d klines: ret=%d~%d, %d klines, %s~%s", interval, beforeTs, afterTs, ets, sts, len(klines), ee, ss) // store to tsdb err = svc.exchangeDataService.SaveKlines(task.inst, klines) diff --git a/pkg/types/interval.go b/pkg/types/interval.go index 6695719..9be8276 100644 --- a/pkg/types/interval.go +++ b/pkg/types/interval.go @@ -1,6 +1,8 @@ package types -import "time" +import ( + "time" +) var LossEmoji = "🔥" var ProfitEmoji = "💰" @@ -16,6 +18,15 @@ func (i Interval) AddMul(ts, mul int64) (int64, bool) { return c(ts, mul), true } +// Iota 返回interval唯一索引 +func (i Interval) Iota(ts, mul int64) (int, bool) { + index, ok := IntervalIotas[i] + if !ok { + return 0, false + } + return index, true +} + // func (i Interval) Minutes() (int64, bool) { // c, ok := SupportedIntervals[i] // if !ok || c <= 0 { @@ -97,3 +108,25 @@ var SupportedIntervals = IntervalMap{ Interval1mo: func(ts, mul int64) (ret int64) { return time.UnixMilli(ts).AddDate(0, int(mul), 0).UnixMilli() }, Interval3mo: func(ts, mul int64) (ret int64) { return time.UnixMilli(ts).AddDate(0, int(3*mul), 0).UnixMilli() }, } + +type IntervalIotaMap map[Interval]int + +var IntervalIotas = IntervalIotaMap{ + Interval1m: 2, + Interval3m: 3, + Interval5m: 4, + Interval15m: 5, + Interval30m: 6, + Interval1h: 7, + Interval2h: 8, + Interval4h: 9, + Interval6h: 10, + Interval12h: 11, + Interval1d: 12, + Interval2d: 13, + Interval3d: 14, + Interval5d: 15, + Interval1w: 16, + Interval1mo: 17, + Interval3mo: 18, +} diff --git a/pkg/utils/collect/collect.go b/pkg/utils/collect/collect.go index c154bdf..5901d1e 100644 --- a/pkg/utils/collect/collect.go +++ b/pkg/utils/collect/collect.go @@ -1,6 +1,9 @@ package collect -import "sort" +import ( + "cmp" + "sort" +) func In[T comparable](value T, values ...T) bool { if len(values) == 0 { @@ -69,17 +72,23 @@ func Sum[T int32 | int64 | int](nums []T) T { return sum } -func Max[T int32 | int64 | int](nums []T, dv T) T { - if len(nums) == 0 { - return dv +func MustMax[T any, C cmp.Ordered](slice []T, compare func(T) C) (max T) { + if len(slice) == 0 { + panic("MustMax slice length 0") + } + max = slice[0] + if len(slice) == 1 { + return } - var max T = nums[0] - for i := 1; i < len(nums); i++ { - if nums[i] > max { - max = nums[i] + + maxC := compare(max) + for i := 1; i < len(slice); i++ { + if c := compare(slice[i]); cmp.Compare(c, maxC) > 0 { + max = slice[i] + maxC = c } } - return max + return } func Slice2Map[T any, K comparable](slice []T, k func(int, T) K) map[K]T { diff --git a/pkg/zlog/exported.go b/pkg/zlog/exported.go index 6498074..f69bd21 100644 --- a/pkg/zlog/exported.go +++ b/pkg/zlog/exported.go @@ -1,5 +1,11 @@ package zlog +func Debug(args ...any) { + zSugar.Debug(args...) +} +func Debugf(format string, args ...any) { + zSugar.Debugf(format, args...) +} func Info(args ...any) { zSugar.Info(args...) }