exchange history kline stream

api/exchange.proto

@@ -17,6 +17,9 @@ service ExchangeService {
 
   // 查询历史k线
   rpc HistoryKline(ReqHistoryKline) returns (RspHistoryKline);
+
+  // 查询历史k线(流式返回)
+  rpc HistoryKlineStream(ReqHistoryKline) returns (stream RspHistoryKlineStream);
 }
 
 message ReqStreamSubscribeKline {
@@ -69,3 +72,12 @@ message RspHistoryKline {
   bool next = 5; // 是否有更多历史数据: 为true时, 可以用最后一根k线的ts作为before继续请求
   repeated Kline klines = 9;
 }
+
+message RspHistoryKlineStream {
+  ExchangeType exchange = 1; // 交易所
+  string instId = 2;
+  string interval = 3;
+  bool live = 4; // 第一根是否实时k线
+  bool next = 5; // 是否有更多历史数据: 为true时, 可以用最后一根k线的ts作为before继续请求
+  repeated Kline klines = 9;
+}

internal/exchange/exchange_grpc_server.go

@@ -136,3 +136,9 @@ func (svr *ExchangeGrpcServer) HistoryKline(ctx context.Context, req *pb.ReqHist
 	}
 	return
 }
+
+// HistoryKlineStream 获取交易产品历史k线(流式返回)
+func (svr *ExchangeGrpcServer) HistoryKlineStream(req *pb.ReqHistoryKline, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) {
+	err = svr.exchangeService.HistoryKlineStream(req, stream)
+	return
+}

internal/exchange/exchange_service.go

@@ -23,7 +23,8 @@ import (
 
 // ExchangeService 交易所服务
 type ExchangeService struct {
-	exchangeMap         map[pb.ExchangeType]*Exchange
+	// exchanges         map[pb.ExchangeType]*Exchange
+	exchanges           *types.ExchangeState[*Exchange]
 	tradeInstanceAside  *client.TradeInstanceAside
 	exchangeDataPersist *ExchangeDataPersist
 
@@ -36,13 +37,16 @@ func NewExchangeService(
 	exchangeDataPersist *ExchangeDataPersist,
 	exchanges ...*Exchange,
 ) *ExchangeService {
-	exchangeMap := make(map[pb.ExchangeType]*Exchange)
+	exchangeState := types.NewExchangeState[*Exchange]()
 	for _, exchange := range exchanges {
-		exchangeMap[exchange.ExchangeType] = exchange
+		if !exchangeState.IsSupport(exchange.ExchangeType) {
+			panic(fmt.Errorf("unsupport exchange: %s", exchange.ExchangeType.String()))
+		}
+		exchangeState.Set(exchange.ExchangeType, exchange)
 	}
 
 	return &ExchangeService{
-		exchangeMap:         exchangeMap,
+		exchanges:           exchangeState,
 		tradeInstanceAside:  tradeInstanceAside,
 		exchangeDataPersist: exchangeDataPersist,
 		klinePublisher:      NewPublisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]](16),
@@ -63,7 +67,7 @@ func (svc *ExchangeService) GetKlineSubscriber() (subscriber *Publisher[int64, g
 // 订阅交易所推送行情
 func (svc *ExchangeService) subscribeExchanges() {
 	// 交易所订阅交易产品
-	for _, exchange := range svc.exchangeMap {
+	svc.exchanges.Range(func(_ pb.ExchangeType, exchange *Exchange) {
 		go func(exchange *Exchange) {
 			// get exchange all trade instances
 			insts, err := svc.tradeInstanceAside.ListExchangeTradeInstance(context.Background(), exchange.ExchangeType)
@@ -118,7 +122,7 @@ func (svc *ExchangeService) subscribeExchanges() {
 			// 初始化历史k线数据
 			go svc.initialKlines(exchange, processingInsts)
 		}(exchange)
-	}
+	})
 }
 
 // consumerKline 消费交易所k线数据
@@ -467,9 +471,9 @@ func (svc *ExchangeService) fetchTaskKlinesToTSDB(exchange *Exchange, task fetch
 
 // Exchanges 支持的交易所列表
 func (svc *ExchangeService) Exchanges() (exchanges []pb.ExchangeType, err error) {
-	for exchange := range svc.exchangeMap {
+	svc.exchanges.Range(func(exchange pb.ExchangeType, _ *Exchange) {
 		exchanges = append(exchanges, exchange)
-	}
+	})
 	return
 }
 
@@ -477,17 +481,16 @@ func (svc *ExchangeService) Exchanges() (exchanges []pb.ExchangeType, err error)
 func (svc *ExchangeService) ExchangeInstanceState(allExchange bool, exchangeTypes []pb.ExchangeType, instIds []string) (states []*pb.TradeInstanceState, err error) {
 	var exchanges []*Exchange
 	if allExchange {
-		for _, exg := range svc.exchangeMap {
+		svc.exchanges.Range(func(_ pb.ExchangeType, exchange *Exchange) {
-			exchanges = append(exchanges, exg)
+			exchanges = append(exchanges, exchange)
-		}
+		})
 	} else {
 		for _, exchangeType := range exchangeTypes {
-			exg, ok := svc.exchangeMap[exchangeType]
+			if !svc.exchanges.IsSupport(exchangeType) {
-			if !ok {
 				err = fmt.Errorf("not support exchange: %v", exchangeType)
 				return
 			}
-			exchanges = append(exchanges, exg)
+			exchanges = append(exchanges, svc.exchanges.Get(exchangeType))
 		}
 	}
 	if len(exchanges) == 0 {
@@ -522,11 +525,7 @@ func (svc *ExchangeService) ExchangeInstanceState(allExchange bool, exchangeType
 // HistoryKline 获取交易产品历史k线 (before < klines... < after)
 func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistoryKline, rsp *pb.RspHistoryKline) (klines []*types.Kline, err error) {
 	// 交易产品参数检查
-	exchange, ok := svc.exchangeMap[req.Exchange]
+	exchange := svc.exchanges.Get(req.Exchange)
-	if !ok {
-		err = fmt.Errorf("exchange not support: %s", req.Exchange)
-		return
-	}
 	exchangeInstId, ok := exchange.TradeInstIds.Load(req.InstId)
 	if !ok {
 		err = fmt.Errorf("trade instance not support: %s", req.InstId)
@@ -623,3 +622,86 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory
 	}
 	return
 }
+
+// 查询历史k线(流式返回)
+func (svc *ExchangeService) HistoryKlineStream(req *pb.ReqHistoryKline, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) {
+	// 交易产品参数检查
+	if !svc.exchanges.IsSupport(req.Exchange) {
+		err = fmt.Errorf("exchange not support: %s", req.Exchange)
+		return
+	}
+	exchange := svc.exchanges.Get(req.Exchange)
+	exchangeInstId, ok := exchange.TradeInstIds.Load(req.InstId)
+	if !ok {
+		err = fmt.Errorf("trade instance not support: %s", req.InstId)
+		return
+	}
+	interval := types.Interval(req.Interval)
+	intervalAdder, ok := types.SupportedIntervals[interval]
+	if !ok {
+		err = fmt.Errorf("interval not support: %s", req.Interval)
+		return
+	}
+	// todo 交易产品初始化完成检查
+	exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId)
+	if !ok {
+		err = fmt.Errorf("trade instance not support for exchange: %s for %s", req.InstId, req.Exchange)
+		return
+	}
+
+	// k线长度检查
+	afterTs, beforeTs, count := int64(req.After), int64(req.Before), int64(req.Count)
+	if count == 0 {
+		count = 100
+	}
+	liveK := exchangeInst.LiveKline.Get(interval)
+	lastTs := liveK.Ts
+	if !liveK.Confirm {
+		lastTs = intervalAdder(liveK.Ts, -1)
+	}
+	if afterTs == 0 && beforeTs == 0 {
+		beforeTs = max(intervalAdder(lastTs, -count+1), KlineBefore0)
+		afterTs = lastTs
+	}
+	if afterTs == 0 {
+		afterTs = min(intervalAdder(beforeTs, count-1), lastTs)
+	}
+	if beforeTs == 0 {
+		beforeTs = max(intervalAdder(afterTs, -count+1), KlineBefore0)
+	}
+	if beforeTs > afterTs {
+		err = fmt.Errorf("time range invalid: before must less then after")
+		return
+	}
+
+	branch := 100
+	curBeforeTs, curAfterTs := beforeTs, intervalAdder(beforeTs, int64(branch)-1)
+	for i := 0; curAfterTs <= afterTs; i++ {
+		if i > 0 {
+			curBeforeTs = intervalAdder(curAfterTs, 1)
+			curAfterTs = min(intervalAdder(curBeforeTs, int64(branch)-1), afterTs)
+		}
+
+		klines, kerr := svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, curBeforeTs, curAfterTs)
+		if kerr != nil {
+			err = kerr
+			return
+		}
+		if len(klines) == 0 {
+			break
+		}
+
+		resp := new(pb.RspHistoryKlineStream)
+		resp.Klines = collect.Mapping(klines, func(_ int, k *types.Kline) *pb.Kline { return k.ToPBKline() })
+		// 是否还有更多
+		resp.Next = len(klines) == int(count)
+		if sendErr := stream.Send(resp); sendErr != nil {
+			err = sendErr
+			return
+		}
+		if !resp.Next {
+			break
+		}
+	}
+	return
+}

pkg/types/exchange.go

@@ -49,7 +49,7 @@ func (s *ExchangeState[T]) Set(exchange pb.ExchangeType, v T) {
 	s.state[exchange] = v
 }
 
-func (s *ExchangeState[T]) Range(f func(exchange pb.ExchangeType, m T)) {
+func (s *ExchangeState[T]) Range(f func(exchange pb.ExchangeType, v T)) {
 	for _, exchange := range SupportedExchanges {
 		f(exchange, s.state[exchange])
 	}

pkg/utils/container/heap.go

@@ -0,0 +1,59 @@
+package container
+
+import (
+	"cmp"
+	"container/heap"
+)
+
+// 堆封装
+type heapQueue[T any, C cmp.Ordered] struct {
+	queue   []T
+	compare func(T) C
+}
+
+func (h *heapQueue[T, C]) Less(i, j int) bool {
+	return h.compare(h.queue[i]) < h.compare(h.queue[j])
+}
+
+func (h *heapQueue[T, C]) Swap(i, j int) {
+	h.queue[i], h.queue[j] = h.queue[j], h.queue[i]
+}
+
+func (h *heapQueue[T, C]) Len() int {
+	return len(h.queue)
+}
+
+func (h *heapQueue[T, C]) Pop() (v any) {
+	h.queue, v = h.queue[:h.Len()-1], h.queue[h.Len()-1]
+	return
+}
+
+func (h *heapQueue[T, C]) Push(v any) {
+	h.queue = append(h.queue, v.(T))
+}
+
+// Heap 封装堆操作
+type Heap[T any, C cmp.Ordered] struct {
+	queue *heapQueue[T, C]
+}
+
+func NewHeap[T any, C cmp.Ordered](compare func(T) C) *Heap[T, C] {
+	return &Heap[T, C]{
+		queue: &heapQueue[T, C]{compare: compare},
+	}
+}
+
+func (h *Heap[T, C]) Push(v T) {
+	heap.Push(h.queue, v)
+}
+
+func (h *Heap[T, C]) Pop() T {
+	return heap.Pop(h.queue).(T)
+}
+
+func (h *Heap[T, C]) Peek() (v T, ok bool) {
+	if h.queue.Len() == 0 {
+		return v, false
+	}
+	return h.queue.queue[0], true
+}

pkg/utils/timer/timer.go

@@ -0,0 +1,56 @@
+package timer
+
+import (
+	"cmp"
+	"sig-pub/pkg/utils/container"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+type timerTask struct {
+	ExecuteAt int64
+	Canceled  atomic.Bool
+	Handler   func()
+}
+
+// Timer todo 使用二叉树堆的定时任务执行器
+type Timer[T any] struct {
+	mu     sync.RWMutex
+	heap   *container.Heap[T, int64]
+	tick   time.Duration
+	ticker *time.Ticker
+}
+
+func NewTimer[T any, C cmp.Ordered](tick time.Duration, compare func(T) int64) *Timer[T] {
+	return &Timer[T]{
+		heap: container.NewHeap(compare),
+		tick: tick,
+	}
+}
+
+func (t *Timer[T]) Init() {
+	t.ticker = time.NewTicker(t.tick)
+	go func() {
+		for {
+			now, ok := <-t.ticker.C
+			if !ok {
+				return
+			}
+			t.mu.RLock()
+			v, ok := t.heap.Peek()
+			t.mu.RUnlock()
+			if !ok {
+				continue
+			}
+			_, _ = now, v
+			// todo handler
+		}
+	}()
+	// time.Unix()
+	// t.ticker.C
+}
+
+func (t *Timer[T]) Push(v T, duration time.Duration) {
+	t.heap.Push(v)
+}

pkg/utils/timer/timer_test.go

@@ -0,0 +1,41 @@
+package timer
+
+import (
+	"container/heap"
+	"fmt"
+	"testing"
+)
+
+type myHeap []int
+
+func (h *myHeap) Less(i, j int) bool {
+	return (*h)[i] < (*h)[j]
+}
+
+func (h *myHeap) Swap(i, j int) {
+	(*h)[i], (*h)[j] = (*h)[j], (*h)[i]
+}
+
+func (h *myHeap) Len() int {
+	return len(*h)
+}
+
+func (h *myHeap) Pop() (v any) {
+	*h, v = (*h)[:h.Len()-1], (*h)[h.Len()-1]
+	return
+}
+
+func (h *myHeap) Push(v any) {
+	*h = append(*h, v.(int))
+}
+
+func TestMyHelp(t *testing.T) {
+	h := new(myHeap)
+	for _, v := range []int{5, 3, 8, 1, 2, 7} {
+		heap.Push(h, v)
+	}
+	fmt.Println(h)
+	for range len(*h) {
+		fmt.Println(heap.Pop(h), h)
+	}
+}