From d011868ccac53b845199df2d90b99381069dc275 Mon Sep 17 00:00:00 2001 From: strange Date: Thu, 30 Oct 2025 01:41:12 +0800 Subject: [PATCH] kline branch stream --- internal/exchange/exchange_grpc_server.go | 19 ++- internal/exchange/exchange_service.go | 175 ++++++++-------------- pkg/strategy/strategy.go | 6 +- pkg/types/series/klines.go | 14 +- pkg/utils/collect/collect.go | 10 +- 5 files changed, 96 insertions(+), 128 deletions(-) diff --git a/internal/exchange/exchange_grpc_server.go b/internal/exchange/exchange_grpc_server.go index 724605b..c3cbeab 100644 --- a/internal/exchange/exchange_grpc_server.go +++ b/internal/exchange/exchange_grpc_server.go @@ -133,20 +133,21 @@ func (svr *ExchangeGrpcServer) HistoryKline(ctx context.Context, req *pb.ReqHist return } + rsp = new(pb.RspHistoryKline) + rsp.Klines = make([]*pb.Kline, 0, total+1) rsp = &pb.RspHistoryKline{ Exchange: req.Series.Exchange, InstId: req.Series.InstId, Interval: req.Series.Interval, } - live, klines, err := svr.exchangeService.HistoryKline(ctx, req.Series) + live, err := svr.exchangeService.HistoryKline(req.Series, DefaultHistoryKlines, func(klines []*pb.Kline) error { + rsp.Klines = append(rsp.Klines, klines...) + return nil + }) if err != nil { return } rsp.Live = live - rsp.Klines = make([]*pb.Kline, 0, len(klines)) - for _, k := range klines { - rsp.Klines = append(rsp.Klines, k.ToPBKline()) - } return } @@ -156,6 +157,12 @@ func (svr *ExchangeGrpcServer) HistoryKlineStream(req *pb.ReqHistoryKlineStream, err = fmt.Errorf("series arg is required") return } - err = svr.exchangeService.HistoryKlineStream(req.Series, stream) + _, err = svr.exchangeService.HistoryKline(req.Series, 200, func(klines []*pb.Kline) error { + rsp := &pb.RspHistoryKlineStream{Klines: klines} + return stream.Send(rsp) + }) + if err != nil { + return + } return } diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index 947f365..b59f43b 100644 --- a/internal/exchange/exchange_service.go +++ b/internal/exchange/exchange_service.go @@ -733,7 +733,7 @@ func (svc *ExchangeService) CalcSeriesRange(arg *pb.SeriesRange) (after, before, } // HistoryKline 获取交易产品历史k线 (before < klines... < after) -func (svc *ExchangeService) HistoryKline(ctx context.Context, arg *pb.SeriesRange) (live bool, klines []*types.Kline, err error) { +func (svc *ExchangeService) HistoryKline(arg *pb.SeriesRange, recvBranch int, recvKline func(klines []*pb.Kline) error) (live bool, err error) { // 交易产品参数检查 exchange := svc.exchanges.Get(arg.Exchange) exchangeInstId, ok := exchange.TradeInstIds.Load(arg.InstId) @@ -748,46 +748,47 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, arg *pb.SeriesRang return } - // todo 交易产品初始化完成检查 exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId) if !ok { err = fmt.Errorf("trade instance not support for exchange: %s for %s", arg.InstId, arg.Exchange) return } + // 交易产品初始化完成检查 + if status := exchangeInst.Status.Load(); status != int32(data.StatusOk) { + err = fmt.Errorf("trade instance not ready: %s(%s) for %d", arg.InstId, arg.Exchange, status) + return + } // 限制最大时间范围 afterTs, beforeTs, total, err := svc.CalcSeriesRange(arg) if err != nil { return } - if total > MaxHistoryKlines { - err = fmt.Errorf("time range too large max %d", MaxHistoryKlines) - return - } + // if total > MaxHistoryKlines { + // err = fmt.Errorf("time range too large max %d", MaxHistoryKlines) + // return + // } if arg.Limit > 0 && total > int64(arg.Limit) { err = fmt.Errorf("time range %d out of limit %d", total, arg.Limit) return } - // 时间分段查询: 升降序, live, open - branch := int64(10) + // 分批查询 + branch := int64(2000) before, after := beforeTs, afterTs - if arg.Desc { - before = intervalAdder(after, -branch) - if before > afterTs { // int64 越界 - before = beforeTs + recvBuffer := make([]*pb.Kline, 0, recvBranch) + for range 10000 { + if arg.Desc { + before = max(intervalAdder(after, -branch+1), beforeTs) + } else { + after = min(intervalAdder(before, branch-1), afterTs) } - before = max(before, beforeTs) - } else { - after = intervalAdder(before, branch) - if after < beforeTs { // int64 越界 - after = afterTs + if after < KlineBefore0 || before > after { + break } - after = min(after, afterTs) - } - for { - klines, err = svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, before, after) - if err != nil { + klines, errK := svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, before, after) + if errK != nil { + err = errK zlog.Error("list vmtsdb kline error: ", err) return } @@ -795,107 +796,59 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, arg *pb.SeriesRang break } - if arg.Desc { - after = intervalAdder(klines[0].Ts, -1) - before = max(intervalAdder(after, -branch), beforeTs) - if after < KlineBefore0 || before > after { - break + // 检查k线是否连续进行补齐 + if err = svc.paddingKlinesIfNotSeries(exchange, arg.InstId, interval, klines); err != nil { + return + } + lastK := klines[len(klines)-1] + if after == afterTs && lastK.Ts != afterTs { + // vmtsdb 数据落盘30s延迟, 使用内存数据替代最新的一根k线 + lastConfirmK := exchangeInst.LastKline.Get(interval) + if lastConfirmK.Ts == lastK.Ts { + lastK = &lastConfirmK + klines[len(klines)-1] = lastK } - } else { - before = intervalAdder(klines[0].Ts, 1) - after = min(intervalAdder(before, 1000), afterTs) - if before > time.Now().UnixMilli() || before > after { - break + if lastConfirmK.Ts == afterTs && intervalAdder(lastK.Ts, 1) == afterTs { + lastK = &lastConfirmK + klines = append(klines, &lastConfirmK) } } - if arg.Desc { - collect.Reverse(klines) - } - tss := collect.Mapping(klines, func(_ int, k *types.Kline) int64 { return k.Ts }) - zlog.Info(tss) - } - - klines, err = svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, beforeTs, afterTs) - if err != nil { - zlog.Error("list vmtsdb kline error: ", err) - return - } - if len(klines) == 0 { - return - } - // 检查k线是否连续进行补齐 - if err = svc.paddingKlinesIfNotSeries(exchange, arg.InstId, interval, klines); err != nil { - return - } - - lastK := klines[len(klines)-1] - // vmtsdb 数据刷盘30s延迟, 使用内存数据替代第一根k线 - lastConfirmK := exchangeInst.LastKline.Get(interval) - if lastConfirmK.Ts == lastK.Ts { - lastK = &lastConfirmK - klines[len(klines)-1] = lastK - } - if lastConfirmK.Ts == afterTs && intervalAdder(lastK.Ts, 1) == afterTs { - lastK = &lastConfirmK - klines = append(klines, &lastConfirmK) - } - - // 降序排序 - if arg.Desc { - collect.Reverse(klines) - } - - // 实时k线 - if arg.Live && len(klines) > 0 { - liveK := exchangeInst.LiveKline.Get(interval) - if latest := intervalAdder(lastK.Ts, 1) == liveK.Ts; latest { - if arg.Desc { - klines = append([]*types.Kline{&liveK}, klines...) - } else { + // 实时k线 + if !live && arg.Live && lastK.Ts == afterTs { + liveK := exchangeInst.LiveKline.Get(interval) + if latest := intervalAdder(lastK.Ts, 1) == liveK.Ts; latest { klines = append(klines, &liveK) + live = true } - live = true } - } - return -} -// 查询历史k线(按时间升序流式返回) -func (svc *ExchangeService) HistoryKlineStream(arg *pb.SeriesRange, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) { - ctx := context.Background() - _, klines, err := svc.HistoryKline(ctx, arg) - if err != nil { - return - } + // next loop + if arg.Desc { + after = intervalAdder(klines[0].Ts, -1) + } else { + before = intervalAdder(klines[len(klines)-1].Ts, 1) + } - // 交易产品参数检查 - if !svc.exchanges.IsSupport(arg.Exchange) { - err = fmt.Errorf("exchange not support: %s", arg.Exchange) - return - } - exchange := svc.exchanges.Get(arg.Exchange) - interval := types.Interval(arg.Interval) + // 降序排序 + if arg.Desc { + collect.Reverse(klines) + } - // 检查k线是否连续进行补齐 - if err = svc.paddingKlinesIfNotSeries(exchange, arg.InstId, interval, klines); err != nil { - return - } + zlog.Infof("krange: %s, total=%d, %d~%d", arg.InstId, len(klines), klines[0].Ts, klines[len(klines)-1].Ts) - branch := 100 - length := len(klines) - kBuffer := make([]*pb.Kline, 0, branch) - for i, kline := range klines { - kBuffer = append(kBuffer, kline.ToPBKline()) - if len(kBuffer) < branch && i < length-1 { - continue - } - rsp := &pb.RspHistoryKlineStream{Klines: kBuffer} - if sendErr := stream.Send(rsp); sendErr != nil { - err = sendErr - return + // 分成小批量recv + length := len(klines) + for i, kline := range klines { + recvBuffer = append(recvBuffer, kline.ToPBKline()) + if len(recvBuffer) < recvBranch && i < length-1 { + continue + } + if err = recvKline(recvBuffer); err != nil { + break + } + recvBuffer = recvBuffer[:0] } - kBuffer = kBuffer[:0] } return } diff --git a/pkg/strategy/strategy.go b/pkg/strategy/strategy.go index b90cc4b..e3afecf 100644 --- a/pkg/strategy/strategy.go +++ b/pkg/strategy/strategy.go @@ -23,7 +23,7 @@ const ( // completed 同一时刻的所有其他周期都完成 func DriverIntervalKey(instId string, exchange pb.ExchangeType, completed bool, intervals ...types.Interval) string { types.IntervalsSort(intervals) - strIntervals := collect.Mapping(intervals, func(_ int, interval types.Interval) string { return string(interval) }) + strIntervals := collect.Mapping0(intervals, func(_ int, interval types.Interval) string { return string(interval) }) pubKey := fmt.Sprintf("/interval/%s/%s/%s/%d", instId, exchange.String(), strings.Join(strIntervals, ","), lang.Ternary(completed, 1, 0)) return pubKey } @@ -33,8 +33,8 @@ func DriverIntervalKey(instId string, exchange pb.ExchangeType, completed bool, func MulExchangeDriverIntervalKey(instId string, exchanges []pb.ExchangeType, intervals ...types.Interval) string { types.IntervalsSort(intervals) types.ExchangesSort(exchanges) - strIntervals := collect.Mapping(intervals, func(_ int, interval types.Interval) string { return string(interval) }) - strExchanges := collect.Mapping(exchanges, func(_ int, exchange pb.ExchangeType) string { return exchange.String() }) + strIntervals := collect.Mapping0(intervals, func(_ int, interval types.Interval) string { return string(interval) }) + strExchanges := collect.Mapping0(exchanges, func(_ int, exchange pb.ExchangeType) string { return exchange.String() }) pubKey := fmt.Sprintf("/interval/%s/%s/%s", instId, strings.Join(strExchanges, ","), strings.Join(strIntervals, ",")) return pubKey } diff --git a/pkg/types/series/klines.go b/pkg/types/series/klines.go index 8f1b146..f951a5d 100644 --- a/pkg/types/series/klines.go +++ b/pkg/types/series/klines.go @@ -9,29 +9,29 @@ import ( type Klines []types.Kline func (s Klines) Times() []int64 { - return collect.Mapping(s, func(_ int, k types.Kline) int64 { return k.Ts }) + return collect.Mapping0(s, func(_ int, k types.Kline) int64 { return k.Ts }) } func (s Klines) Open() Floats { - return collect.Mapping(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Open) }) + return collect.Mapping0(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Open) }) } func (s Klines) Close() Floats { - return collect.Mapping(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Close) }) + return collect.Mapping0(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Close) }) } func (s Klines) High() Floats { - return collect.Mapping(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.High) }) + return collect.Mapping0(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.High) }) } func (s Klines) Low() Floats { - return collect.Mapping(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Low) }) + return collect.Mapping0(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Low) }) } func (s Klines) Vol() Floats { - return collect.Mapping(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Vol) }) + return collect.Mapping0(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Vol) }) } func (s Klines) VolQuote() Floats { - return collect.Mapping(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.VolQuote) }) + return collect.Mapping0(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.VolQuote) }) } diff --git a/pkg/utils/collect/collect.go b/pkg/utils/collect/collect.go index afcb314..f76895c 100644 --- a/pkg/utils/collect/collect.go +++ b/pkg/utils/collect/collect.go @@ -155,7 +155,15 @@ func Slice2MapKv[T any, K comparable, V any](slice []T, mapping func(int, T) (K, return m } -func Mapping[T any, V any](slice []T, mapping func(int, T) V) (vs []V) { +func Mapping[T any, V any](slice []T, mapping func(T) V) (vs []V) { + vs = make([]V, 0, len(slice)) + for _, item := range slice { + vs = append(vs, mapping(item)) + } + return +} + +func Mapping0[T any, V any](slice []T, mapping func(int, T) V) (vs []V) { vs = make([]V, 0, len(slice)) for i, item := range slice { vs = append(vs, mapping(i, item))