From d24ba35e1c7b33197dbc0b58b71d705edcacb22b Mon Sep 17 00:00:00 2001 From: strange Date: Fri, 17 Oct 2025 18:34:38 +0800 Subject: [PATCH] exchange history kline stream --- api/exchange.proto | 12 +++ internal/exchange/exchange_grpc_server.go | 6 ++ internal/exchange/exchange_service.go | 120 ++++++++++++++++++---- pkg/types/exchange.go | 2 +- pkg/utils/container/heap.go | 59 +++++++++++ pkg/utils/timer/timer.go | 56 ++++++++++ pkg/utils/timer/timer_test.go | 41 ++++++++ 7 files changed, 276 insertions(+), 20 deletions(-) create mode 100644 pkg/utils/container/heap.go create mode 100644 pkg/utils/timer/timer.go create mode 100644 pkg/utils/timer/timer_test.go diff --git a/api/exchange.proto b/api/exchange.proto index 11c94a8..0ccf7a8 100644 --- a/api/exchange.proto +++ b/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; +} diff --git a/internal/exchange/exchange_grpc_server.go b/internal/exchange/exchange_grpc_server.go index c6cfee7..7483b4a 100644 --- a/internal/exchange/exchange_grpc_server.go +++ b/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 +} diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index 75f735f..41676fd 100644 --- a/internal/exchange/exchange_service.go +++ b/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 { - exchanges = append(exchanges, exg) - } + svc.exchanges.Range(func(_ pb.ExchangeType, exchange *Exchange) { + exchanges = append(exchanges, exchange) + }) } else { for _, exchangeType := range exchangeTypes { - exg, ok := svc.exchangeMap[exchangeType] - if !ok { + if !svc.exchanges.IsSupport(exchangeType) { 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] - if !ok { - 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) @@ -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 +} diff --git a/pkg/types/exchange.go b/pkg/types/exchange.go index d5df50a..d7d8802 100644 --- a/pkg/types/exchange.go +++ b/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]) } diff --git a/pkg/utils/container/heap.go b/pkg/utils/container/heap.go new file mode 100644 index 0000000..a04a70e --- /dev/null +++ b/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 +} diff --git a/pkg/utils/timer/timer.go b/pkg/utils/timer/timer.go new file mode 100644 index 0000000..4017ce3 --- /dev/null +++ b/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) +} diff --git a/pkg/utils/timer/timer_test.go b/pkg/utils/timer/timer_test.go new file mode 100644 index 0000000..a973d89 --- /dev/null +++ b/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) + } +}