exchange history kline

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 {

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;

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 {

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

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),