trading kline series

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;
 }

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://192.168.1.5:7890"
-# httpProxy = "http://10.255.183.209:7890"
+httpProxy = "http://10.255.183.209:7890"
 
 # 模拟盘API交易地址如下：
 # REST：https://www.okx.com

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
 }

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
 	}

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 {
+	if total > 100 {
-		err = fmt.Errorf("time range too large")
+		// err = fmt.Errorf("time range too large max 100")
-		return
+		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) {
-	// 开闭区间 confirmed
+		rsp.Next = false
-	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)
+		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
 }

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)

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 {
-		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:])
+		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

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)
+}

internal/trading/kline_store.go

@@ -1,14 +1,23 @@
 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
+	exchangeClient pb.ExchangeServiceClient
 	store          [3]*collect.ConcurrentMap[string, *KlineStoreInstance] // K线列表: []exchange<instId, interval, klines>
 }
 
@@ -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)
+	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
+// serial k线是否连续
-func (si *KlineStoreInstance) Update(kline *types.Kline) (ok bool) {
+func (si *KlineStoreInstance) Update(kline *types.Kline) (before int64, serial bool, err error) {
-	ks, ok := si.IntervalKlines.Load(kline.Interval)
+	if _, ok := types.SupportedIntervals[kline.Interval]; !ok {
-	if !ok {
+		err = fmt.Errorf("unsupport interval: %s", kline.Interval)
 		return
 	}
 
-	if lastTs, serial := ks.Update(kline); !serial {
+	si.Lock()
-		// k线不完整
+	defer si.Unlock()
-		// 拉取k线
-		start := lastTs
-		end := kline.Ts
-		_, _ = start, end
-	}
 
-	ok = true
+	ks := si.intervalKlines.Get(kline.Interval)
-	// emit kline event
+	before, serial = ks.Update(kline)
+	if !serial {
+		si.padding.Store(true)
+	}
 	return
 }