From c8ff5a969cd3a8b3b31efaa316d2d91bb099a1c1 Mon Sep 17 00:00:00 2001 From: strange Date: Thu, 16 Oct 2025 18:31:31 +0800 Subject: [PATCH] trading kline store --- api/exchange.proto | 4 +- internal/exchange/exchange_service.go | 20 +++--- internal/trading/kline_series.go | 5 +- internal/trading/kline_series_test.go | 11 ++-- internal/trading/kline_store.go | 95 ++++++++++++++++++++++++--- internal/trading/trading_service.go | 20 ++---- pkg/mq/nats_data.go | 6 +- pkg/types/interval.go | 5 +- 8 files changed, 119 insertions(+), 47 deletions(-) diff --git a/api/exchange.proto b/api/exchange.proto index 567a53f..11c94a8 100644 --- a/api/exchange.proto +++ b/api/exchange.proto @@ -54,8 +54,8 @@ message ReqHistoryKline { ExchangeType exchange = 1; // 交易所 string instId = 2; string interval = 3; - uint64 before = 4; - uint64 after = 5; + int64 before = 4; + int64 after = 5; uint32 count = 6; // k线条数,before或after其中一个为0时有效 // bool open = 7; // 是否开区间, 不包含 before/after bool live = 8; // 实时k线, before和after为0时是否追加实时k线 diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index c522588..48c90e6 100644 --- a/internal/exchange/exchange_service.go +++ b/internal/exchange/exchange_service.go @@ -245,20 +245,22 @@ func (svc *ExchangeService) initialKlines(exchange *Exchange, insts []types.Trad for _, inst := range insts { err := svc.initialTradeInstanceKlines(exchange, inst) + status := data.StatusFailed if err != nil { zlog.Errorf("initial fetch trade instance error: %s(%s), err=%v", inst.InstId, inst.Exchange, err) failed = append(failed, inst) } else { success = append(success, inst) - - // 发布交易产品初始化完成事件 - publish := &mq.PublishExchangeTradeInstanceInited{ - Exchange: inst.Exchange, - InstId: inst.InstId, - } - if err := mq.NatsPublish(mq.TopicExchangeTradeInstanceInited, publish); err != nil { - zlog.Errorf("publish trade instance inited error: %s(%s), err=%v", inst.InstId, inst.Exchange, err) - } + status = data.StatusOk + } + // 发布交易产品初始化完成事件 + publish := &mq.PublishExchangeTradeInstanceInited{ + Exchange: inst.Exchange, + InstId: inst.InstId, + Status: status, + } + if err := mq.NatsPublish(mq.TopicExchangeTradeInstanceInited, publish); err != nil { + zlog.Errorf("publish trade instance inited error: %s(%s), err=%v", inst.InstId, inst.Exchange, err) } } diff --git a/internal/trading/kline_series.go b/internal/trading/kline_series.go index 6a927b0..bf85748 100644 --- a/internal/trading/kline_series.go +++ b/internal/trading/kline_series.go @@ -10,7 +10,8 @@ import ( ) const ( - MaxSeriesKlines = 1280 // slice扩容12次后cap=1280 + MaxSeriesKlines = 1000 // 有效k线数量 + MaxCapSeriesKlines = 1280 // 最大k线数量, slice扩容12次后cap=1280 ) type KlineSeries struct { @@ -87,7 +88,7 @@ func (s *KlineSeries) Update(kline *types.Kline) (lastTs int64, serial bool) { } } - if len(s.klines) < MaxSeriesKlines { + if len(s.klines) < MaxCapSeriesKlines { s.klines = append(s.klines, kline) } else { // 循环复用切片空间，避免扩容 diff --git a/internal/trading/kline_series_test.go b/internal/trading/kline_series_test.go index 9a9e660..1559a98 100644 --- a/internal/trading/kline_series_test.go +++ b/internal/trading/kline_series_test.go @@ -6,26 +6,29 @@ import ( ) func TestSliceCap(t *testing.T) { - Max := 1280 + Max, MaxCap := 1000, 1280 + _ = MaxCap var arr []int prevCap := cap(arr) - for i := range 1500 { + for i := range 1600 { ele := i - if len(arr) < Max { + if len(arr) < MaxCap { arr = append(arr, ele) } else { length := len(arr) copy(arr, arr[length-Max+1:]) arr[Max-1] = ele arr = arr[:Max] + fmt.Printf("move=%d, %d, arr=(%d~%d)\n", i, len(arr), arr[0], arr[len(arr)-1]) } c := cap(arr) if c != prevCap { // 发生扩容 prevCap = c - fmt.Printf("i=%d, cap=%d\n", i, c) + fmt.Printf("i=%d, cap=%d, len=%d\n", i, c, len(arr)) } } + fmt.Println("----------------------------------------------------------") fmt.Println(arr) } diff --git a/internal/trading/kline_store.go b/internal/trading/kline_store.go index 1453621..b404723 100644 --- a/internal/trading/kline_store.go +++ b/internal/trading/kline_store.go @@ -5,6 +5,8 @@ import ( "fmt" "math" "sig-pub/api/pb" + "sig-pub/pkg/data" + "sig-pub/pkg/mq" "sig-pub/pkg/types" "sig-pub/pkg/utils/collect" "sig-pub/pkg/utils/retry" @@ -17,24 +19,96 @@ import ( ) type KlineStore struct { - exchangeClient pb.ExchangeServiceClient - store [3]*collect.ConcurrentMap[string, *KlineStoreInstance] // K线列表: []exchange + exchangeClient pb.ExchangeServiceClient + subscribeKlineIntervals []string + store [3]*collect.ConcurrentMap[string, *KlineStoreInstance] // K线列表: []exchange } -func NewKlineSeriesStore() (kss *KlineStore) { - kss = &KlineStore{} +func NewKlineSeriesStore(exchangeClient pb.ExchangeServiceClient) (kss *KlineStore) { + // 订阅实时k线周期列表 + subscribeKlineIntervals := collect.Map2Slice(types.SupportedIntervals, func(interval types.Interval, _ types.IntervalAdder) string { + return string(interval) + }) + kss = &KlineStore{ + subscribeKlineIntervals: subscribeKlineIntervals, + exchangeClient: exchangeClient, + } kss.store[pb.ExchangeType_OKX] = collect.NewConcurrentMap[string, *KlineStoreInstance](64, func(s string) string { return s }) // kss.klines[pb.ExchangeType_BINANCE] = return } +func (s *KlineStore) Init() (err error) { + // 拉取已初始化完成交易产品, 初始化k线, 开始订阅k线 + + // 订阅交易产品初始化完成事件 + mq.NatsCreateConsumer("trading", mq.StreamExchange, mq.TopicExchangeTradeInstanceInited, func() *mq.PublishExchangeTradeInstanceInited { return new(mq.PublishExchangeTradeInstanceInited) }, + func(msg *mq.PublishExchangeTradeInstanceInited) (err error) { + // 初始化k线, 开始订阅k线 + zlog.Infof("subscribed TopicExchangeTradeInstanceInited: %#v", msg) + go s.subscribeKlines(msg.Exchange, msg.InstId) + return + }) + return +} + +func (s *KlineStore) subscribeKlines(exchange pb.ExchangeType, instId string) { + storeInst := s.store[exchange].ComputeIfAbsent(instId, func(k string) *KlineStoreInstance { + return NewKlineStoreInstance(exchange, k) + }) + + // 初始化最新的 klineSeries + for _, interval := range s.subscribeKlineIntervals { + for { + after, before := int64(0), int64(0) + rsp, err := retry.DoWithFixDelay(math.MaxInt32, 2*time.Second, func(retryTimes uint32) (rsp *pb.RspHistoryKline, err error) { + rsp, err = s.exchangeClient.HistoryKline(context.Background(), &pb.ReqHistoryKline{ + Exchange: exchange, + InstId: instId, + Interval: string(interval), + Count: MaxSeriesKlines, + After: after, + Before: before, + Live: false, + Asc: true, + }, grpc.UseCompressor("snappy")) + if err != nil { + zlog.Errorf("fetch missing klines error: instId=%s(%s) after=%d before=%d, err=%v", instId, exchange, after, before, err) + } + return + }) + if err != nil { + zlog.Errorf("trade instance initial failed: %s(%s), %v", instId, exchange, err) + return + } + for _, kline := range rsp.Klines { + k := new(types.Kline) + k.ParsePBKline(exchange, kline) + _, _, err = storeInst.Update(k) + if err != nil { + zlog.Error("update initial kline series error: instId=%s(%s)", instId, exchange, err) + return + } + + s.Update(exchange, instId, k) + after = k.Ts + } + + } + } + + // 开始订阅k线 +} + // Update // kline klineStore -> klineSeries -> strategy -> indicator -> klineSeries.Series func (s *KlineStore) Update(exchange pb.ExchangeType, instId string, kline *types.Kline) { storeInst := s.store[exchange].ComputeIfAbsent(instId, func(k string) *KlineStoreInstance { return NewKlineStoreInstance(exchange, k) }) - if storeInst.padding.Load() { + + // 只处理已初始化完成的交易产品k线 + if storeInst.status.Load() != int32(data.StatusOk) { return } before, serial, err := storeInst.Update(kline) @@ -57,8 +131,8 @@ func (s *KlineStore) paddingMissKlines(storeInst *KlineStoreInstance, exchange p Exchange: exchange, InstId: instId, Interval: string(interval), - After: uint64(after), - Before: uint64(before), + After: after, + Before: before, Live: false, }, grpc.UseCompressor("snappy")) if err != nil { @@ -88,8 +162,7 @@ type KlineStoreInstance struct { Exchange pb.ExchangeType InstId string intervalKlines *types.IntervalState[*KlineSeries] - // tickK *types.Kline // 秒级k线 - padding atomic.Bool // 是否正在拉取历史k线 + status atomic.Int32 // 交易产品状态 } func NewKlineStoreInstance(exchange pb.ExchangeType, instId string) *KlineStoreInstance { @@ -98,6 +171,8 @@ func NewKlineStoreInstance(exchange pb.ExchangeType, instId string) *KlineStoreI InstId: instId, intervalKlines: types.NewIntervalState[*KlineSeries](), } + si.status.Store(int32(data.StatusProcessing)) + for interval := range types.SupportedIntervals { si.intervalKlines.Set(interval, NewKlineSeries(exchange, instId, interval)) } @@ -118,7 +193,7 @@ func (si *KlineStoreInstance) Update(kline *types.Kline) (before int64, serial b ks := si.intervalKlines.Get(kline.Interval) before, serial = ks.Update(kline) if !serial { - si.padding.Store(true) + // si.status.Store(int32(data.StatusProcessing)) } return } diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index 1536716..57ae2d4 100644 --- a/internal/trading/trading_service.go +++ b/internal/trading/trading_service.go @@ -5,7 +5,6 @@ import ( "io" "sig-pub/api/pb" "sig-pub/pkg/client" - "sig-pub/pkg/mq" "sig-pub/pkg/types" "sig-pub/pkg/utils/collect" "sig-pub/pkg/zlog" @@ -36,13 +35,16 @@ func NewTradingService( return &TradingService{ marketClientAside: marketClientAside, exchangeClient: exchangeClient, + klineStore: NewKlineSeriesStore(exchangeClient), subscribeKlineIntervals: subscribeKlineIntervals, } } // 初始化历史k线, 订阅实时k线 func (svr *TradingService) Init() (err error) { - svr.klineStore = NewKlineSeriesStore() + if err = svr.klineStore.Init(); err != nil { + return + } // get instance exchangeTradeInsts, err := svr.marketClientAside.ListExchangeTradeInstance(context.Background(), pb.ExchangeType_OKX) @@ -53,25 +55,11 @@ func (svr *TradingService) Init() (err error) { svr.subKlineInsts[exInst.Exchange] = append(svr.subKlineInsts[exInst.Exchange], exInst.InstId) } - // 拉取已初始化完成交易产品, 初始化k线, 开始订阅k线 - - // 订阅交易产品初始化完成事件 - mq.NatsCreateConsumer("trading", mq.StreamExchange, mq.TopicExchangeTradeInstanceInited, func() *mq.PublishExchangeTradeInstanceInited { return new(mq.PublishExchangeTradeInstanceInited) }, - func(msg *mq.PublishExchangeTradeInstanceInited) (err error) { - // 初始化k线, 开始订阅k线 - zlog.Infof("trade instance inited: %#v", msg) - return - }) - // 订阅k线 go svr.subscribeStreamKlines(false) return } -func (svr *TradingService) subscribeKlines() { - -} - func (svr *TradingService) subscribeStreamKlines(reconnect bool) { defer func() { if svr.subKlineStream != nil { diff --git a/pkg/mq/nats_data.go b/pkg/mq/nats_data.go index c1d91f8..aff7b16 100644 --- a/pkg/mq/nats_data.go +++ b/pkg/mq/nats_data.go @@ -1,9 +1,13 @@ package mq -import "sig-pub/api/pb" +import ( + "sig-pub/api/pb" + "sig-pub/pkg/data" +) // 交易产品初始化完成通知 type PublishExchangeTradeInstanceInited struct { Exchange pb.ExchangeType `json:"exchange"` // 交易所类型 InstId string `json:"instId"` // 交易产品id + Status data.Status `json:"status"` // 初始化状态结果 } diff --git a/pkg/types/interval.go b/pkg/types/interval.go index da06195..66e66ce 100644 --- a/pkg/types/interval.go +++ b/pkg/types/interval.go @@ -2,7 +2,6 @@ package types import ( "fmt" - "sig-pub/pkg/zlog" "sort" "time" ) @@ -113,8 +112,8 @@ func init() { intervalIotas[interval] = index intervalIotaMax = max(intervalIotaMax, index) } - zlog.Debugf("init intervals: %#v", iotasIntervals) - zlog.Debugf("init intervalsIotas: %#v", intervalIotas) + // zlog.Debugf("init intervals: %#v", iotasIntervals) + // zlog.Debugf("init intervalsIotas: %#v", intervalIotas) } type IntervalState[T any] struct {