From 959de63913f9849fd3b28385c977d5cbab58dcb3 Mon Sep 17 00:00:00 2001 From: strange Date: Wed, 15 Oct 2025 18:41:54 +0800 Subject: [PATCH] trading kline series --- api/exchange.proto | 5 +- config/exchange.toml | 4 +- internal/exchange/exchange_data_persist.go | 2 - internal/exchange/exchange_grpc_server.go | 4 +- internal/exchange/exchange_service.go | 62 ++++++++----- internal/exchange/okx/okx_fetch.go | 1 + internal/trading/kline_series.go | 20 ++--- internal/trading/kline_series_test.go | 31 +++++++ internal/trading/kline_store.go | 100 ++++++++++++++++----- 9 files changed, 165 insertions(+), 64 deletions(-) create mode 100644 internal/trading/kline_series_test.go diff --git a/api/exchange.proto b/api/exchange.proto index 3329442..567a53f 100644 --- a/api/exchange.proto +++ b/api/exchange.proto @@ -57,12 +57,15 @@ message ReqHistoryKline { uint64 before = 4; uint64 after = 5; uint32 count = 6; // k线条数,before或after其中一个为0时有效 - bool open = 7; // 是否开区间, 不包含 before/after + // bool open = 7; // 是否开区间, 不包含 before/after bool live = 8; // 实时k线, before和after为0时是否追加实时k线 + bool asc = 9; // 是否升序, 默认降序 } message RspHistoryKline { 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/config/exchange.toml b/config/exchange.toml index c0a42b7..4d567a5 100644 --- a/config/exchange.toml +++ b/config/exchange.toml @@ -18,8 +18,8 @@ receiveBuffer = 4096 marketSubscribeLimit = 16 consumeBatch = 1024 consumeLater = 2000 # 时间到达later或者数据累计到batch触发consume -httpProxy = "http://192.168.1.5:7890" -# httpProxy = "http://10.255.183.209:7890" +# httpProxy = "http://192.168.1.5:7890" +httpProxy = "http://10.255.183.209:7890" # 模拟盘API交易地址如下： # REST：https://www.okx.com diff --git a/internal/exchange/exchange_data_persist.go b/internal/exchange/exchange_data_persist.go index 7edaafd..4d717cd 100644 --- a/internal/exchange/exchange_data_persist.go +++ b/internal/exchange/exchange_data_persist.go @@ -7,7 +7,6 @@ import ( "sig-pub/pkg/storage/kvrocks" vmts "sig-pub/pkg/storage/tsdb/victoria_metrics" "sig-pub/pkg/types" - "sig-pub/pkg/utils/collect" ) type ExchangeDataPersist struct { @@ -53,6 +52,5 @@ func (p *ExchangeDataPersist) SaveHistoryKlineMarkTs(exchange pb.ExchangeType, i // ListKline 查询k线列表 func (p *ExchangeDataPersist) ListKline(inst types.TradeInstance, interval types.Interval, start, end int64) (klines []*types.Kline, err error) { klines, err = p.vmtsdb.ListRangeKline(inst, interval, start, end) - collect.Reverse(klines) return } diff --git a/internal/exchange/exchange_grpc_server.go b/internal/exchange/exchange_grpc_server.go index 4635123..c6cfee7 100644 --- a/internal/exchange/exchange_grpc_server.go +++ b/internal/exchange/exchange_grpc_server.go @@ -123,7 +123,9 @@ func (svr *ExchangeGrpcServer) HistoryKline(ctx context.Context, req *pb.ReqHist InstId: req.InstId, Interval: req.Interval, } - klines, err := svr.exchangeService.HistoryKline(ctx, req) + // for before := req.Before; ; { + // } + klines, err := svr.exchangeService.HistoryKline(ctx, req, rsp) if err != nil { return } diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index 1ce981e..c2da483 100644 --- a/internal/exchange/exchange_service.go +++ b/internal/exchange/exchange_service.go @@ -419,7 +419,9 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade func (svc *ExchangeService) fetchTaskKlinesToTSDB(exchange *Exchange, task fetchKlineTask) (lastKlineTs int64, err error) { interval, afterTs, beforeTs := task.interval, task.afterTs, task.beforeTs - klines, err := exchange.Fetcher.FetchHistoryKlines(context.Background(), task.inst.ExchangeInstId, interval, afterTs, beforeTs) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + klines, err := exchange.Fetcher.FetchHistoryKlines(ctx, task.inst.ExchangeInstId, interval, afterTs, beforeTs) if err != nil { zlog.Errorf("fetch history kline task error: task -> %s, err -> %v", task.logKey(), err) return @@ -501,7 +503,7 @@ func (svc *ExchangeService) ExchangeInstanceState(allExchange bool, exchangeType } // HistoryKline 获取交易产品历史k线 (before < klines... < after) -func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistoryKline) (klines []*types.Kline, err error) { +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 { @@ -528,51 +530,63 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory // k线长度检查 afterTs, beforeTs, count := int64(req.After), int64(req.Before), int64(req.Count) - latest := false + nowTime := time.Now().UnixMilli() if afterTs == 0 && beforeTs == 0 { - afterTs = time.Now().UnixMilli() - latest = true + // 拉取最新的 + afterTs = nowTime } if count == 0 { - count = 120 + count = 100 } if afterTs == 0 { - afterTs = intervalAdder(beforeTs, count) + afterTs = min(intervalAdder(beforeTs, count), nowTime) } if beforeTs == 0 { - beforeTs = intervalAdder(afterTs, -count) + beforeTs = max(intervalAdder(afterTs, -count), KlineBefore0) } if beforeTs > afterTs { err = fmt.Errorf("time range invalid: before must less then after") return } - if beforeTs == afterTs { - return - } + // 限制最大时间范围 total := (afterTs - beforeTs) / intervalAdder(0, 1) - if total > 1000 { - err = fmt.Errorf("time range too large") - return + if total > 100 { + // err = fmt.Errorf("time range too large max 100") + if req.Asc { + afterTs = min(intervalAdder(beforeTs, 100), nowTime) + } else { + beforeTs = max(intervalAdder(afterTs, -100), KlineBefore0) + } + rsp.Next = true } klines, err = svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, beforeTs, afterTs) if err != nil || len(klines) == 0 { + rsp.Next = false return } + if len(klines) < int(count) { + rsp.Next = false + } - // 开闭区间 confirmed - if req.Open { - if klines[0].Ts == afterTs { - klines = klines[1:] - } - if len(klines) > 0 && klines[len(klines)-1].Ts == beforeTs { - klines = klines[:len(klines)-1] - } + lastK := klines[len(klines)-1] + + // 降序排序 + if !req.Asc { + collect.Reverse(klines) } - if latest && req.Live { + // 实时k线 + if req.Live && len(klines) > 0 { liveK := exchangeInst.LiveKline.Get(interval) - klines = append([]*types.Kline{&liveK}, klines...) + if latest := intervalAdder(lastK.Ts, 1) == liveK.Ts; latest { + if req.Asc { + klines = append(klines, &liveK) + } else { + klines = append([]*types.Kline{&liveK}, klines...) + } + rsp.Live = true + } } return } diff --git a/internal/exchange/okx/okx_fetch.go b/internal/exchange/okx/okx_fetch.go index 10f4980..73d3865 100644 --- a/internal/exchange/okx/okx_fetch.go +++ b/internal/exchange/okx/okx_fetch.go @@ -34,6 +34,7 @@ func NewOkxFetcher(httpProxy string) (f *OkxFetcher) { MaxConnsPerHost: 10, IdleConnTimeout: 90 * time.Second, TLSHandshakeTimeout: 10 * time.Second, + DisableKeepAlives: false, }) if httpProxy != "" { client.SetProxy(httpProxy) diff --git a/internal/trading/kline_series.go b/internal/trading/kline_series.go index 881595e..6a927b0 100644 --- a/internal/trading/kline_series.go +++ b/internal/trading/kline_series.go @@ -10,7 +10,7 @@ import ( ) const ( - MaxSeriesKlines = 1000 + MaxSeriesKlines = 1280 // slice扩容12次后cap=1280 ) type KlineSeries struct { @@ -75,30 +75,26 @@ func (s *KlineSeries) Update(kline *types.Kline) (lastTs int64, serial bool) { serial = true lastTs = s.lastTs if kline.Ts <= s.lastTs { - // 已存在的k线，直接忽略 return } // 检查k线是否连续 if len(s.klines) > 0 { expectTs := s.Interval.MustAddMul(s.lastTs, 1) if kline.Ts != expectTs { - serial = false - zlog.Warningf("k线不连续: last.Ts=%d, expected=%d, got=%d", s.lastTs, expectTs, kline.Ts) + zlog.Warningf("k线不连续: instId=%s(%s), interval=%s, lastTs=%d, expected=%d, got=%d", s.InstId, s.Exchange, s.Interval, s.lastTs, expectTs, kline.Ts) return } } - if cap(s.klines) < MaxSeriesKlines { + if len(s.klines) < MaxSeriesKlines { s.klines = append(s.klines, kline) } else { - if len(s.klines) < MaxSeriesKlines { - s.klines = append(s.klines, kline) - } else { - // 循环复用切片空间，避免扩容 - copy(s.klines, s.klines[1:]) - s.klines[MaxSeriesKlines-1] = kline - } + // 循环复用切片空间，避免扩容 + length := len(s.klines) + copy(s.klines, s.klines[length-MaxSeriesKlines+1:]) + s.klines[MaxSeriesKlines-1] = kline + s.klines = s.klines[:MaxSeriesKlines] } s.lastTs = kline.Ts return diff --git a/internal/trading/kline_series_test.go b/internal/trading/kline_series_test.go new file mode 100644 index 0000000..9a9e660 --- /dev/null +++ b/internal/trading/kline_series_test.go @@ -0,0 +1,31 @@ +package trading + +import ( + "fmt" + "testing" +) + +func TestSliceCap(t *testing.T) { + Max := 1280 + var arr []int + prevCap := cap(arr) + for i := range 1500 { + ele := i + if len(arr) < Max { + arr = append(arr, ele) + } else { + length := len(arr) + copy(arr, arr[length-Max+1:]) + arr[Max-1] = ele + arr = arr[:Max] + } + + c := cap(arr) + if c != prevCap { + // 发生扩容 + prevCap = c + fmt.Printf("i=%d, cap=%d\n", i, c) + } + } + fmt.Println(arr) +} diff --git a/internal/trading/kline_store.go b/internal/trading/kline_store.go index 4df07c5..1453621 100644 --- a/internal/trading/kline_store.go +++ b/internal/trading/kline_store.go @@ -1,15 +1,24 @@ package trading import ( + "context" + "fmt" + "math" "sig-pub/api/pb" - vmts "sig-pub/pkg/storage/tsdb/victoria_metrics" "sig-pub/pkg/types" "sig-pub/pkg/utils/collect" + "sig-pub/pkg/utils/retry" + "sig-pub/pkg/zlog" + "sync" + "sync/atomic" + "time" + + "google.golang.org/grpc" ) type KlineStore struct { - vmdb vmts.VictoriaMetricsTSDB - store [3]*collect.ConcurrentMap[string, *KlineStoreInstance] // K线列表: []exchange + exchangeClient pb.ExchangeServiceClient + store [3]*collect.ConcurrentMap[string, *KlineStoreInstance] // K线列表: []exchange } func NewKlineSeriesStore() (kss *KlineStore) { @@ -19,50 +28,97 @@ func NewKlineSeriesStore() (kss *KlineStore) { return } -func (s *KlineStore) Update(exchange pb.ExchangeType, instId string, kline *types.Kline) (k types.Kline) { +// 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) }) - storeInst.Update(kline) - return + if storeInst.padding.Load() { + return + } + before, serial, err := storeInst.Update(kline) + if err != nil { + zlog.Error("update kline series error: instId=%s(%s)", instId, exchange, err) + return + } + if !serial { + // 拉取缺失的k线 + s.paddingMissKlines(storeInst, exchange, instId, kline.Interval, kline.Ts, before) + // emit kline event + } +} + +// 拉取缺失的k线 +func (s *KlineStore) paddingMissKlines(storeInst *KlineStoreInstance, exchange pb.ExchangeType, instId string, interval types.Interval, after, before int64) { + // 拉取缺失的k线 + 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), + After: uint64(after), + Before: uint64(before), + Live: false, + }, 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 { + return + } + + collect.Reverse(rsp.Klines) + for _, kline := range rsp.Klines { + k := new(types.Kline) + k.ParsePBKline(exchange, kline) + zlog.Infof("padding missing kline: instId=%s(%s) %s %#v", instId, exchange, interval, k) + _, _, err := storeInst.Update(k) + if err != nil { + zlog.Error("update missing kline series error: instId=%s(%s)", instId, exchange, err) + } + } } // KlineStoreInstance 单个交易产品所有周期k线 type KlineStoreInstance struct { + sync.RWMutex Exchange pb.ExchangeType InstId string - IntervalKlines *collect.SyncMap[types.Interval, *KlineSeries] + intervalKlines *types.IntervalState[*KlineSeries] + // tickK *types.Kline // 秒级k线 + padding atomic.Bool // 是否正在拉取历史k线 } func NewKlineStoreInstance(exchange pb.ExchangeType, instId string) *KlineStoreInstance { si := &KlineStoreInstance{ Exchange: exchange, InstId: instId, - IntervalKlines: collect.NewSyncMap[types.Interval, *KlineSeries](), + intervalKlines: types.NewIntervalState[*KlineSeries](), } for interval := range types.SupportedIntervals { - si.IntervalKlines.Store(interval, NewKlineSeries(exchange, instId, interval)) + si.intervalKlines.Set(interval, NewKlineSeries(exchange, instId, interval)) } return si } // Update 更新k线 -// kline klineStore -> klineSeries -> strategy -> indicator -> klineSeries.Series -func (si *KlineStoreInstance) Update(kline *types.Kline) (ok bool) { - ks, ok := si.IntervalKlines.Load(kline.Interval) - if !ok { +// serial k线是否连续 +func (si *KlineStoreInstance) Update(kline *types.Kline) (before int64, serial bool, err error) { + if _, ok := types.SupportedIntervals[kline.Interval]; !ok { + err = fmt.Errorf("unsupport interval: %s", kline.Interval) return } - if lastTs, serial := ks.Update(kline); !serial { - // k线不完整 - // 拉取k线 - start := lastTs - end := kline.Ts - _, _ = start, end - } + si.Lock() + defer si.Unlock() - ok = true - // emit kline event + ks := si.intervalKlines.Get(kline.Interval) + before, serial = ks.Update(kline) + if !serial { + si.padding.Store(true) + } return }