From 288d1e148cb6090160c8d94e76dceb23c78579c1 Mon Sep 17 00:00:00 2001 From: strange Date: Wed, 21 Jan 2026 18:36:26 +0800 Subject: [PATCH] exchange history kline --- api/exchange.proto | 2 +- api/pub.proto | 13 +++++++ internal/exchange/exchange_grpc_server.go | 43 +++++++++++++++++++---- internal/exchange/exchange_service.go | 19 +++++----- pkg/types/kline.go | 2 +- 5 files changed, 59 insertions(+), 20 deletions(-) diff --git a/api/exchange.proto b/api/exchange.proto index e5f607b..bc03a3e 100644 --- a/api/exchange.proto +++ b/api/exchange.proto @@ -77,7 +77,7 @@ message RspHistoryKline { string interval = 3; bool live = 4; // 第一根是否实时k线 // bool next = 5; // 是否有更多历史数据: 为true时, 可以用最后一根k线的ts作为before继续请求 - repeated Kline klines = 9; + KlineSeries klines = 9; } message ReqHistoryKlineStream { diff --git a/api/pub.proto b/api/pub.proto index dddf725..7a93640 100644 --- a/api/pub.proto +++ b/api/pub.proto @@ -120,6 +120,19 @@ message Kline { bool confirm = 10; // k线是否完结 } +message KlineSeries { + string interval = 2; // 周期 + repeated int64 time = 3; + repeated double open = 4; + repeated double high = 5; + repeated double low = 6; + repeated double close = 7; + repeated double vol = 8; // 交易量 + repeated double vol_quote = 9; // 交易额 + // bool confirm = 10; // k线是否完结 true + Kline live = 11; // 未完结实时k线 +} + // https://maicoin.github.io/max-websocket-docs/#/private_channels?id=snapshot message Order { ExchangeType exchange = 1; diff --git a/internal/exchange/exchange_grpc_server.go b/internal/exchange/exchange_grpc_server.go index 60d808a..705f4e8 100644 --- a/internal/exchange/exchange_grpc_server.go +++ b/internal/exchange/exchange_grpc_server.go @@ -4,8 +4,11 @@ import ( "context" "fmt" "io" + "math" "sig-pub/api/pb" "sig-pub/pkg/publish" + "sig-pub/pkg/types" + "sig-pub/pkg/types/decimals" "sig-pub/pkg/zlog" "sync/atomic" @@ -135,26 +138,48 @@ func (svr *ExchangeGrpcServer) HistoryKline(ctx context.Context, req *pb.ReqHist if err != nil { return } - if total > DefaultHistoryKlines { - err = fmt.Errorf("time range too large max %d", DefaultHistoryKlines) + if total > MaxHistoryKlines { + err = fmt.Errorf("time range too large max %d", MaxHistoryKlines) 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, err := svr.exchangeService.HistoryKline(req.Series, DefaultHistoryKlines, func(klines []*pb.Kline) error { - rsp.Klines = append(rsp.Klines, klines...) + + kSeries := &pb.KlineSeries{ + Interval: req.Series.Interval, + Time: make([]int64, 0, total), + Open: make([]float64, 0, total), + High: make([]float64, 0, total), + Low: make([]float64, 0, total), + Close: make([]float64, 0, total), + Vol: make([]float64, 0, total), + VolQuote: make([]float64, 0, total), + } + live, err := svr.exchangeService.HistoryKline(req.Series, math.MaxInt, func(liveK *types.Kline, klines []*types.Kline) error { + for _, kline := range klines { + kSeries.Time = append(kSeries.Time, kline.Ts) + kSeries.Open = append(kSeries.Open, decimals.MustToFloat64(kline.Open)) + kSeries.High = append(kSeries.High, decimals.MustToFloat64(kline.High)) + kSeries.Low = append(kSeries.Low, decimals.MustToFloat64(kline.Low)) + kSeries.Close = append(kSeries.Close, decimals.MustToFloat64(kline.Close)) + kSeries.Vol = append(kSeries.Vol, decimals.MustToFloat64(kline.Vol)) + kSeries.VolQuote = append(kSeries.VolQuote, decimals.MustToFloat64(kline.VolQuote)) + if liveK != nil { + kSeries.Live = liveK.ToPBKline() + } + } return nil }) if err != nil { return } rsp.Live = live + rsp.Klines = kSeries return } @@ -164,8 +189,12 @@ func (svr *ExchangeGrpcServer) HistoryKlineStream(req *pb.ReqHistoryKlineStream, err = fmt.Errorf("series arg is required") return } - _, err = svr.exchangeService.HistoryKline(req.Series, 200, func(klines []*pb.Kline) error { - rsp := &pb.RspHistoryKlineStream{Klines: klines} + _, err = svr.exchangeService.HistoryKline(req.Series, 200, func(_ *types.Kline, klines []*types.Kline) error { + rsp := &pb.RspHistoryKlineStream{} + rsp.Klines = make([]*pb.Kline, 0, len(klines)) + for _, kline := range klines { + rsp.Klines = append(rsp.Klines, kline.ToPBKline()) + } return stream.Send(rsp) }) if err != nil { diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index d45c98e..3d612d6 100644 --- a/internal/exchange/exchange_service.go +++ b/internal/exchange/exchange_service.go @@ -750,7 +750,7 @@ func (svc *ExchangeService) CalcSeriesRange(arg *pb.SeriesRange) (after, before, } // HistoryKline 获取交易产品历史k线 (before < klines... < after) -func (svc *ExchangeService) HistoryKline(arg *pb.SeriesRange, recvBranch int, recvKlineFn func(klines []*pb.Kline) error) (live bool, err error) { +func (svc *ExchangeService) HistoryKline(arg *pb.SeriesRange, recvBranch int, recvKlineFn func(liveK *types.Kline, klines []*types.Kline) error) (live bool, err error) { // 交易产品参数检查 exchange := svc.exchanges.Get(arg.Exchange) exchangeInstId, ok := exchange.TradeInstIds.Load(arg.InstId) @@ -792,7 +792,6 @@ func (svc *ExchangeService) HistoryKline(arg *pb.SeriesRange, recvBranch int, re // 分批查询 branch := int64(2000) before, after := beforeTs, afterTs - recvBuffer := make([]*pb.Kline, 0, min(recvBranch, int(branch))) var prevFirstK, prevLastK *types.Kline for range 10000 { if arg.Desc { @@ -837,10 +836,11 @@ func (svc *ExchangeService) HistoryKline(arg *pb.SeriesRange, recvBranch int, re } // 实时k线 + var liveKline *types.Kline 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) + liveKline = &liveK live = true } } @@ -860,16 +860,13 @@ func (svc *ExchangeService) HistoryKline(arg *pb.SeriesRange, recvBranch int, re // zlog.Debugf("krange: %s(%s), interval=%s, total=%d, %d~%d", arg.InstId, arg.Exchange, interval, len(klines), klines[0].Ts, klines[len(klines)-1].Ts) // 分成小批量recv - length := len(klines) - for i, kline := range klines { - recvBuffer = append(recvBuffer, kline.ToPBKline()) - if len(recvBuffer) < recvBranch && i < length-1 { - continue - } - if err = recvKlineFn(recvBuffer); err != nil { + for len(klines) > 0 { + length := min(100, len(klines)) + recvBuf := klines[:length] + if err = recvKlineFn(liveKline, recvBuf); err != nil { break } - recvBuffer = recvBuffer[:0] + klines = klines[length:] } } return diff --git a/pkg/types/kline.go b/pkg/types/kline.go index 3f7aece..a4729fc 100644 --- a/pkg/types/kline.go +++ b/pkg/types/kline.go @@ -45,7 +45,7 @@ func (k *Kline) ParsePBKline(exchange pb.ExchangeType, kline *pb.Kline) { k.VolQuote = decimals.MustFromFloat64(kline.VolQuote) } -func (k *Kline) ToPBKline() (kline *pb.Kline) { +func (k Kline) ToPBKline() (kline *pb.Kline) { kline = &pb.Kline{ Time: k.Ts, Interval: string(k.Interval),