trading kline store

api/exchange.proto

@@ -19,7 +19,7 @@ service ExchangeService {
   rpc HistoryKline(ReqHistoryKline) returns (RspHistoryKline);
 
   // 查询历史k线(流式返回)
-  rpc HistoryKlineStream(ReqHistoryKline) returns (stream RspHistoryKlineStream);
+  rpc HistoryKlineStream(ReqHistoryKlineStream) returns (stream RspHistoryKlineStream);
 }
 
 message ReqStreamSubscribeKline {
@@ -45,9 +45,12 @@ message RspExchanges {
 }
 
 message ReqExchangeInstanceState {
-  repeated string insts = 1;
-  bool allExchange = 2;
-  repeated ExchangeType exchanges = 3; // 交易所
+  bool allExchange = 1; // 所有交易所
+  bool allInsts = 2; // 所有交易产品
+  bool allStatus = 3; // 所有状态
+  repeated ExchangeType exchanges = 4; // 指定交易所
+  repeated string insts = 5; // 指定交易产品
+  repeated int32 status = 6; // 指定交易产品状态
 }
 message RspExchangeInstanceState {
   repeated TradeInstanceState instsState = 1; // 交易产品列表
@@ -69,15 +72,18 @@ message RspHistoryKline {
   string instId = 2;
   string interval = 3;
   bool live = 4; // 第一根是否实时k线
-  bool next = 5; // 是否有更多历史数据: 为true时, 可以用最后一根k线的ts作为before继续请求
+  // bool next = 5; // 是否有更多历史数据: 为true时, 可以用最后一根k线的ts作为before继续请求
   repeated Kline klines = 9;
 }
 
-message RspHistoryKlineStream {
+message ReqHistoryKlineStream {
   ExchangeType exchange = 1; // 交易所
   string instId = 2;
   string interval = 3;
-  bool live = 4; // 第一根是否实时k线
-  bool next = 5; // 是否有更多历史数据: 为true时, 可以用最后一根k线的ts作为before继续请求
-  repeated Kline klines = 9;
+  int64 before = 4;
+  int64 after = 5;
+  uint32 count = 6; // k线条数,before或after其中一个为0时有效
+}
+message RspHistoryKlineStream {
+  repeated Kline klines = 2;
 }

internal/exchange/exchange_grpc_server.go

@@ -109,7 +109,7 @@ func (svr *ExchangeGrpcServer) Exchanges(ctx context.Context, req *pb.ReqExchang
 
 // ExchangeInstanceState 获取交易产品系统状态
 func (svr *ExchangeGrpcServer) ExchangeInstanceState(ctx context.Context, req *pb.ReqExchangeInstanceState) (rsp *pb.RspExchangeInstanceState, err error) {
-	states, err := svr.exchangeService.ExchangeInstanceState(req.AllExchange, req.Exchanges, req.Insts)
+	states, err := svr.exchangeService.ExchangeInstanceState(req)
 	if err != nil {
 		return
 	}
@@ -138,7 +138,7 @@ func (svr *ExchangeGrpcServer) HistoryKline(ctx context.Context, req *pb.ReqHist
 }
 
 // HistoryKlineStream 获取交易产品历史k线(流式返回)
-func (svr *ExchangeGrpcServer) HistoryKlineStream(req *pb.ReqHistoryKline, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) {
+func (svr *ExchangeGrpcServer) HistoryKlineStream(req *pb.ReqHistoryKlineStream, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) {
 	err = svr.exchangeService.HistoryKlineStream(req, stream)
 	return
 }

internal/exchange/exchange_service.go

@@ -12,6 +12,7 @@ import (
 	"sig-pub/pkg/types"
 	"sig-pub/pkg/utils/collect"
 	"sig-pub/pkg/utils/conver"
+	"sig-pub/pkg/utils/retry"
 	"sig-pub/pkg/utils/times"
 	"sig-pub/pkg/zlog"
 	"sync"
@@ -142,10 +143,7 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch
 		if len(channelK.Klines) == 0 {
 			continue
 		}
-		// 排序后取出头尾k线
-		collect.SortAsc(channelK.Klines, func(k *types.Kline) int64 { return k.Ts })
-		firstKline, lastKline := channelK.Klines[0], channelK.Klines[len(channelK.Klines)-1]
-
+		receivedTs := time.Now().UnixMilli()
 		// 交易所 instid 转 sig-instid
 		var tradeInst *types.TradeInstance
 		exchangeInst, ok := exchange.ExchangeInsts.Load(channelK.ExgInstId)
@@ -155,15 +153,47 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch
 		}
 		tradeInst = exchangeInst.Inst
 
+		// 升序排序
+		collect.SortAsc(channelK.Klines, func(k *types.Kline) int64 { return k.Ts })
+
+		// 检查已确认k线是否连续并补齐
+		for _, kline := range channelK.Klines {
+			if kline.Confirm {
+				if kms, ok := kline.Interval.AddMul(kline.Ts, 1); ok {
+					delay := time.Now().UnixMilli() - kms
+					zlog.Debugf("recv confirm kline: delay=%dms, inst=%s(%s), interval=%s", delay, channelK.ExgInstId, channelK.Exchange, kline.Interval)
+
+					if lastConfirmK := exchangeInst.LastKline.Get(kline.Interval); lastConfirmK.Ts != 0 {
+						if expectTs, ok := lastConfirmK.Interval.AddMul(lastConfirmK.Ts, 1); ok && expectTs != kline.Ts {
+							zlog.Warningf("fetching padding klines: inst=%s(%s), interval=%s, ts=%d~%d", tradeInst.InstId, tradeInst.Exchange, kline.Interval, kline.Ts, lastConfirmK.Ts)
+							if err := svc.initialTradeInstanceKlines(exchange, *tradeInst); err != nil {
+								zlog.Errorf("fetch padding kline error: inst=%s(%s), interval=%s, ts=%d~%d, error=%v", tradeInst.InstId, tradeInst.Exchange, kline.Interval, kline.Ts, lastConfirmK.Ts, err)
+							}
+							// zlog.Warningf("fetching padding klines: inst=%s(%s), interval=%s, ts=%d~%d", tradeInst.InstId, tradeInst.Exchange, kline.Interval, kline.Ts, lastConfirmK.Ts)
+							// paddingKlines, err := exchange.Fetcher.FetchHistoryKlines(context.Background(), tradeInst.ExchangeInstId, kline.Interval, kline.Ts, lastConfirmK.Ts)
+							// if err != nil {
+							// 	zlog.Errorf("fetch padding kline error: inst=%s(%s), interval=%s, ts=%d~%d, error=%v", tradeInst.InstId, tradeInst.Exchange, kline.Interval, kline.Ts, lastConfirmK.Ts, err)
+							// } else {
+							// 	zlog.Debugf("fetched padding klines: inst=%s(%s), interval=%s, ts=%d~%d, %#v", tradeInst.InstId, tradeInst.Exchange, kline.Interval, kline.Ts, lastConfirmK.Ts, paddingKlines)
+							// 	channelK.Klines = append(paddingKlines, channelK.Klines...)
+							// 	collect.SortAsc(channelK.Klines, func(k *types.Kline) int64 { return k.Ts })
+							// }
+						}
+					}
+				}
+				break
+			}
+		}
+
+		// 取出头尾k线
+		lastKline := channelK.Klines[len(channelK.Klines)-1]
+
 		// 标记交易产品开始订阅k线时间
-		exchangeInst.LiveKStartTs.SetIf(firstKline.Interval, firstKline.Ts, func(old int64) bool { return old == 0 })
+		// exchangeInst.LiveKStartTs.SetIf(lastKline.Interval, lastKline.Ts, func(old int64) bool { return old == 0 })
 
 		// 记录实时k线
 		exchangeInst.LiveKline.Set(lastKline.Interval, *lastKline)
-		// 记录最后确认k线
-		if lastKline.Confirm {
-			exchangeInst.LastKline.Set(lastKline.Interval, *lastKline)
-		}
+
 		// 记录实时价格
 		exchangeInst.Last = lastKline.Close
 
@@ -172,6 +202,8 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch
 			// zlog.Infof("recv kline: %#v", kline)
 			confirm := 0
 			if kline.Confirm {
+				// 记录最后确认k线
+				exchangeInst.LastKline.Set(kline.Interval, *kline)
 				confirm = 1
 				confirmKlines = append(confirmKlines, kline)
 			}
@@ -188,7 +220,7 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch
 		}
 
 		// 存储到 tsdb
-		if _, ok := types.SupportedIntervals[firstKline.Interval]; ok && len(confirmKlines) > 0 {
+		if _, ok := types.SupportedIntervals[lastKline.Interval]; ok && len(confirmKlines) > 0 {
 			// tsdb storage todo 异步处理
 			err := svc.exchangeDataPersist.SaveKline(*tradeInst, confirmKlines)
 			// zlog.Infof("save confirm klines: instId=%s(%s), interval=%s, ts=%d", tradeInst.InstId, tradeInst.Exchange, firstKline.Interval, firstKline.Ts)
@@ -221,6 +253,10 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch
 				}
 			}
 		}
+
+		if useMs := time.Now().UnixMilli() - receivedTs; useMs > 10 {
+			zlog.Debugf("handle consume kline use: %dms", useMs)
+		}
 	}
 }
 
@@ -304,7 +340,15 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade
 			}
 		})
 		exchangeInst.Status.Store(int32(data.StatusOk))
-		zlog.Infof("initial history kline finish: instId=%s(%s), pub=%d, sub=%d, fail=%d, use %s", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load(), subTasks.Load(), failTimes.Load(), conver.TimeMilliFormat(time.Now().UnixMilli()-startTs, "/"))
+		zlog.Infof("initial history kline finish: instId=%s(%s), pub=%d, sub=%d, fail=%d, use %s", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load(), subTasks.Load(), failTimes.Load(), conver.TimeMilliFormat(time.Now().UnixMilli()-startTs, "."))
+
+		// flush vmtsdb to disk
+		retry.DoWithFixDelay(5, time.Second, func(retryTimes uint32) (_ struct{}, err error) {
+			if err = svc.exchangeDataPersist.vmtsdb.ForceFlush(); err != nil {
+				zlog.Errorf("flush vmts db error: ", err)
+			}
+			return
+		})
 	}()
 
 	// progress monitor
@@ -345,15 +389,21 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade
 			}
 			if beforeTs == 0 {
 				beforeTs = intervalAdder(KlineBefore0, -1)
+			} else {
+				// 不足100根，向前补齐100根一次拉取过来
+				total := (time.Now().UnixMilli() - beforeTs) / intervalAdder(0, 1)
+				if total < 100 {
+					beforeTs = max(intervalAdder(beforeTs, -100), KlineBefore0)
+				}
 			}
 
 			exchangeInst.HistoryMarkTs.Set(interval, beforeTs)
 			for {
 				// 判定订阅任务发布完成
-				liveStartTs := exchangeInst.LiveKStartTs.Get(interval)
-				if liveStartTs != 0 && beforeTs >= liveStartTs {
-					break
-				}
+				// liveStartTs := exchangeInst.LiveKStartTs.Get(interval)
+				// if liveStartTs != 0 && beforeTs >= liveStartTs {
+				// 	break
+				// }
 				if beforeTs > time.Now().UnixMilli() {
 					break
 				}
@@ -478,14 +528,14 @@ func (svc *ExchangeService) Exchanges() (exchanges []pb.ExchangeType, err error)
 }
 
 // ExchangeInstanceState 交易所交易产品状态
-func (svc *ExchangeService) ExchangeInstanceState(allExchange bool, exchangeTypes []pb.ExchangeType, instIds []string) (states []*pb.TradeInstanceState, err error) {
+func (svc *ExchangeService) ExchangeInstanceState(req *pb.ReqExchangeInstanceState) (states []*pb.TradeInstanceState, err error) {
 	var exchanges []*Exchange
-	if allExchange {
+	if req.AllExchange {
 		svc.exchanges.Range(func(_ pb.ExchangeType, exchange *Exchange) {
 			exchanges = append(exchanges, exchange)
 		})
 	} else {
-		for _, exchangeType := range exchangeTypes {
+		for _, exchangeType := range req.Exchanges {
 			if !svc.exchanges.IsSupport(exchangeType) {
 				err = fmt.Errorf("not support exchange: %v", exchangeType)
 				return
@@ -499,20 +549,34 @@ func (svc *ExchangeService) ExchangeInstanceState(allExchange bool, exchangeType
 	}
 
 	for _, exchange := range exchanges {
-		for _, instId := range instIds {
-			// trade instId to exchangeInstId
-			exchangeInstId, ok := exchange.TradeInstIds.Load(instId)
-			if !ok {
-				continue
-			}
-			exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId)
-			if !ok {
-				continue
+		insts := make([]*ExchangeTradeInstance, 0, 8)
+		if req.AllInsts {
+			exchange.ExchangeInsts.Range(func(_ string, inst *ExchangeTradeInstance) bool {
+				if req.AllStatus || collect.In(inst.Status.Load(), req.Status...) {
+					insts = append(insts, inst)
+				}
+				return true
+			})
+		} else {
+			for _, instId := range req.Insts {
+				// trade instId 转 exchangeInstId
+				exchangeInstId, ok := exchange.TradeInstIds.Load(instId)
+				if !ok {
+					continue
+				}
+				inst, ok := exchange.ExchangeInsts.Load(exchangeInstId)
+				if !ok {
+					continue
+				}
+				if req.AllStatus || collect.In(inst.Status.Load(), req.Status...) {
+					insts = append(insts, inst)
+				}
 			}
-
+		}
+		for _, exchangeInst := range insts {
 			state := &pb.TradeInstanceState{
 				Exchange: exchange.ExchangeType,
-				InstId:   instId,
+				InstId:   exchangeInst.Inst.InstId,
 				Status:   exchangeInst.Status.Load(),
 				Last:     exchangeInst.Last.String(),
 			}
@@ -522,6 +586,10 @@ func (svc *ExchangeService) ExchangeInstanceState(allExchange bool, exchangeType
 	return
 }
 
+const (
+	MaxHistoryKlines = 100
+)
+
 // HistoryKline 获取交易产品历史k线 (before < klines... < after)
 func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistoryKline, rsp *pb.RspHistoryKline) (klines []*types.Kline, err error) {
 	// 交易产品参数检查
@@ -548,59 +616,56 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory
 	// k线长度检查
 	afterTs, beforeTs, count := int64(req.After), int64(req.Before), int64(req.Count)
 	if count == 0 {
-		count = 100
+		count = MaxHistoryKlines
 	}
-	nowTime := time.Now().UnixMilli()
-	if afterTs == 0 && beforeTs == 0 {
-		// 拉取最新的
+	// 拉取最新的
+	lastTs := int64(0)
+	if afterTs == 0 {
 		liveK := exchangeInst.LiveKline.Get(interval)
-		lastTs := liveK.Ts
+		lastTs = liveK.Ts
 		if !liveK.Confirm {
 			lastTs = intervalAdder(liveK.Ts, -1)
 		}
-		if !req.Asc {
-			afterTs = lastTs
-		} else {
-			// 从低到高拉取
-			beforeTs = max(intervalAdder(lastTs, -count+1), KlineBefore0)
-		}
+	}
+	if afterTs == 0 && beforeTs == 0 {
+		afterTs = lastTs
 	}
 	if afterTs == 0 {
-		afterTs = min(intervalAdder(beforeTs, count), nowTime)
+		afterTs = min(intervalAdder(beforeTs, count-1), lastTs)
 	}
 	if beforeTs == 0 {
-		beforeTs = max(intervalAdder(afterTs, -count), KlineBefore0)
+		beforeTs = max(intervalAdder(afterTs, -count+1), KlineBefore0)
 	}
 	if beforeTs > afterTs {
 		err = fmt.Errorf("time range invalid: before must less then after")
 		return
 	}
 	// 限制最大时间范围
-	total := (afterTs - beforeTs) / intervalAdder(0, 1)
-	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
+	total := (afterTs-beforeTs)/intervalAdder(0, 1) + 1
+	if total > MaxHistoryKlines {
+		err = fmt.Errorf("time range too large max %d", MaxHistoryKlines)
+		return
 	}
 
 	klines, err = svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, beforeTs, afterTs)
-	if err != nil || len(klines) == 0 {
-		rsp.Next = false
+	if err != nil {
+		zlog.Error("list vmtsdb kline error: ", err)
 		return
 	}
-	if len(klines) < int(count) {
-		rsp.Next = false
+	if len(klines) == 0 {
+		return
 	}
-	lastK := klines[len(klines)-1]
 
+	lastK := klines[len(klines)-1]
 	// vmtsdb 数据刷盘30s延迟, 使用内存数据替代第一根k线
-	if lastConfirmK := exchangeInst.LastKline.Get(interval); lastConfirmK.Ts == lastK.Ts {
-		klines[len(klines)-1] = &lastConfirmK
+	lastConfirmK := exchangeInst.LastKline.Get(interval)
+	if lastConfirmK.Ts == lastK.Ts {
+		lastK = &lastConfirmK
+		klines[len(klines)-1] = lastK
+	}
+	if lastConfirmK.Ts == afterTs && intervalAdder(lastK.Ts, 1) == afterTs {
 		lastK = &lastConfirmK
+		klines = append(klines, &lastConfirmK)
 	}
 
 	// 降序排序
@@ -624,7 +689,7 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory
 }
 
 // 查询历史k线(流式返回)
-func (svc *ExchangeService) HistoryKlineStream(req *pb.ReqHistoryKline, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) {
+func (svc *ExchangeService) HistoryKlineStream(req *pb.ReqHistoryKlineStream, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) {
 	// 交易产品参数检查
 	if !svc.exchanges.IsSupport(req.Exchange) {
 		err = fmt.Errorf("exchange not support: %s", req.Exchange)
@@ -674,34 +739,39 @@ func (svc *ExchangeService) HistoryKlineStream(req *pb.ReqHistoryKline, stream g
 		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, err := svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, beforeTs, afterTs)
+	if err != nil {
+		zlog.Error("fetch history kline stream error: ", err)
+		return
+	}
+	if len(klines) == 0 {
+		return
+	}
+	lastK := klines[len(klines)-1]
+	// vmtsdb 数据刷盘30s延迟, 使用内存数据替代第一根k线
+	lastConfirmK := exchangeInst.LastKline.Get(interval)
+	if lastConfirmK.Ts == lastK.Ts {
+		klines[len(klines)-1] = &lastConfirmK
+		lastK = klines[len(klines)-1]
+	}
+	if lastConfirmK.Ts == afterTs && intervalAdder(lastK.Ts, 1) == afterTs {
+		klines = append(klines, &lastConfirmK)
+	}
 
-		klines, kerr := svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, curBeforeTs, curAfterTs)
-		if kerr != nil {
-			err = kerr
-			return
-		}
-		if len(klines) == 0 {
-			break
+	branch := 100
+	length := len(klines)
+	kBuffer := make([]*pb.Kline, 0, branch)
+	for i, kline := range klines {
+		kBuffer = append(kBuffer, kline.ToPBKline())
+		if len(kBuffer) < branch && i < length-1 {
+			continue
 		}
-
-		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 {
+		rsp := &pb.RspHistoryKlineStream{Klines: kBuffer}
+		if sendErr := stream.Send(rsp); sendErr != nil {
 			err = sendErr
 			return
 		}
-		if !resp.Next {
-			break
-		}
+		kBuffer = kBuffer[:0]
 	}
 	return
 }

internal/exchange/okx/okx_fetch.go

@@ -53,7 +53,7 @@ func (okx *OkxFetcher) ExhcangeType() pb.ExchangeType {
 
 // FetchHistoryKlines 获取交易产品历史K线数据
 // https://my.okx.com/docs-v5/zh/#order-book-trading-market-data-get-candlesticks-history
-// 周期区间 after > before, (after, before)
+// 开区间降序响应 after > before, (after, before)
 func (f *OkxFetcher) FetchHistoryKlines(ctx context.Context, okxInstId string, interval types.Interval, after, before int64) (klines []*types.Kline, err error) {
 	if okxInstId == "" {
 		err = errors.New("instid is empty")

internal/trading/kline_series.go

@@ -3,10 +3,12 @@ package trading
 import (
 	"fmt"
 	"sig-pub/api/pb"
+	"sig-pub/pkg/data"
 	"sig-pub/pkg/types"
 	"sig-pub/pkg/types/series"
 	"sig-pub/pkg/zlog"
 	"sync"
+	"sync/atomic"
 )
 
 const (
@@ -14,6 +16,23 @@ const (
 	MaxCapSeriesKlines = 1280 // 最大k线数量, slice扩容12次后cap=1280
 )
 
+// TradeInstanceKlineSeries 单个交易产品所有周期k线
+type TradeInstanceKlineSeries struct {
+	IntervalKlines *types.IntervalState[*KlineSeries]
+	Status         atomic.Int32 // 交易产品状态
+}
+
+func NewTradeInstanceKlineSeries(exchange pb.ExchangeType, instId string) *TradeInstanceKlineSeries {
+	si := new(TradeInstanceKlineSeries)
+	si.Status.Store(int32(data.StatusNone))
+	si.IntervalKlines = types.NewIntervalState[*KlineSeries]()
+
+	for interval := range types.SupportedIntervals {
+		si.IntervalKlines.Set(interval, NewKlineSeries(exchange, instId, interval))
+	}
+	return si
+}
+
 type KlineSeries struct {
 	sync.RWMutex
 	Exchange      pb.ExchangeType
@@ -38,8 +57,7 @@ func NewKlineSeries(exchange pb.ExchangeType, instId string, interval types.Inte
 	}
 }
 
-// Get
-// [0]当前k线
+// Get [0]当前k线
 func (s *KlineSeries) Get(start int16) types.Kline {
 	index := len(s.klines) - 1 - int(start)
 	if index >= 0 && index < len(s.klines)-1 {
@@ -59,6 +77,8 @@ func (s *KlineSeries) Get(start int16) types.Kline {
 
 // Series [start...end]
 func (s *KlineSeries) Series(start, end int16) (klines series.Klines) {
+	s.RLock()
+	defer s.RUnlock()
 	endTs := s.Interval.MustAddMul(s.lastTs, int64(-start))
 	startTs := s.Interval.MustAddMul(s.lastTs, int64(-end))
 	_ = endTs
@@ -90,6 +110,12 @@ func (s *KlineSeries) Update(kline *types.Kline) (lastTs int64, serial bool) {
 
 	if len(s.klines) < MaxCapSeriesKlines {
 		s.klines = append(s.klines, kline)
+		// 容量超过0.65, 直接扩容到最大
+		if capacity := cap(s.klines); capacity < MaxCapSeriesKlines && capacity > MaxCapSeriesKlines*0.65 {
+			klines := make([]*types.Kline, len(s.klines), MaxCapSeriesKlines)
+			copy(klines, s.klines)
+			s.klines = klines
+		}
 	} else {
 		// 循环复用切片空间，避免扩容
 		length := len(s.klines)

internal/trading/kline_series_test.go

@@ -1,34 +0,0 @@
-package trading
-
-import (
-	"fmt"
-	"testing"
-)
-
-func TestSliceCap(t *testing.T) {
-	Max, MaxCap := 1000, 1280
-	_ = MaxCap
-	var arr []int
-	prevCap := cap(arr)
-	for i := range 1600 {
-		ele := i
-		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, len=%d\n", i, c, len(arr))
-		}
-	}
-	fmt.Println("----------------------------------------------------------")
-	fmt.Println(arr)
-}

internal/trading/kline_store.go

@@ -2,7 +2,6 @@ package trading
 
 import (
 	"context"
-	"fmt"
 	"io"
 	"math"
 	"sig-pub/api/pb"
@@ -12,8 +11,6 @@ import (
 	"sig-pub/pkg/utils/collect"
 	"sig-pub/pkg/utils/retry"
 	"sig-pub/pkg/zlog"
-	"sync"
-	"sync/atomic"
 	"time"
 
 	"google.golang.org/grpc"
@@ -22,7 +19,7 @@ import (
 type KlineStore struct {
 	exchangeClient pb.ExchangeServiceClient
 
-	store             *types.ExchangeState[*collect.ConcurrentMap[string, *KlineStoreInstance]]        // K线列表: []exchange<instId, interval, klines>
+	store             *types.ExchangeState[*collect.ConcurrentMap[string, *TradeInstanceKlineSeries]]  // K线列表: []exchange<instId, interval, klines>
 	subKlineIntervals []string                                                                         // 订阅的k线的周期列表
 	subKlineInsts     *types.ExchangeState[*collect.SyncMap[string, bool]]                             // 订阅k线中的交易产品列表
 	subKlineStream    grpc.BidiStreamingClient[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline] // 订阅k线的stream
@@ -36,12 +33,12 @@ func NewKlineSeriesStore(exchangeClient pb.ExchangeServiceClient) (kss *KlineSto
 		return string(interval)
 	})
 	// 产品列表
-	kss.subKlineInsts = types.NewExchangeState0(func() *collect.SyncMap[string, bool] {
+	kss.subKlineInsts = types.NewExchangeStateInit(func() *collect.SyncMap[string, bool] {
 		return collect.NewSyncMap[string, bool]()
 	})
 	// 各交易所 store 初始化
-	kss.store = types.NewExchangeState0(func() *collect.ConcurrentMap[string, *KlineStoreInstance] {
-		return collect.NewConcurrentMap[string, *KlineStoreInstance](64, func(s string) string {
+	kss.store = types.NewExchangeStateInit(func() *collect.ConcurrentMap[string, *TradeInstanceKlineSeries] {
+		return collect.NewConcurrentMap[string, *TradeInstanceKlineSeries](64, func(s string) string {
 			return s
 		})
 	})
@@ -57,11 +54,27 @@ func (s *KlineStore) Init() (err error) {
 		func(msg *mq.PublishExchangeTradeInstanceInited) (err error) {
 			// 初始化k线, 开始订阅k线
 			zlog.Infof("subscribed TopicExchangeTradeInstanceInited: %#v", msg)
-			go s.initKlineSeries(msg.Exchange, msg.InstId)
+			go s.inititalKlineSeries(msg.Exchange, msg.InstId)
 			return
 		})
 
-	// todo 拉取已初始化完成交易产品, 初始化k线, 开始订阅k线
+	go func() {
+		//  拉取已初始化完成交易产品, 初始化k线, 开始订阅k线
+		rsp, _ := retry.DoWithFixDelay(math.MaxInt32, time.Second, func(retryTimes uint32) (rsp *pb.RspExchangeInstanceState, err error) {
+			rsp, err = s.exchangeClient.ExchangeInstanceState(context.Background(), &pb.ReqExchangeInstanceState{
+				AllExchange: true,
+				AllInsts:    true,
+				Status:      []int32{int32(data.StatusOk)},
+			})
+			if err != nil {
+				zlog.Errorf("ExchangeInstanceState error: retry=%d, %v", retryTimes, err)
+			}
+			return
+		})
+		for _, inst := range rsp.InstsState {
+			s.inititalKlineSeries(inst.Exchange, inst.InstId)
+		}
+	}()
 	return
 }
 
@@ -115,7 +128,10 @@ func (s *KlineStore) connectSubscribeKline(reconnect bool) {
 		for _, k := range msg.Kline.Klines {
 			kline := new(types.Kline)
 			kline.ParsePBKline(msg.Kline.Exchange, k)
-			zlog.Debugf("recv: streamId=%d, %v, %s, %#v", msg.Kline.StreamId, msg.Kline.Exchange, msg.Kline.InstId, kline)
+			if kms, ok := kline.Interval.AddMul(kline.Ts, 1); ok {
+				delay := time.Now().UnixMilli() - kms
+				zlog.Debugf("recv kline: streamId=%d, delay=%dms, inst=%s(%v), interval=%s, close=%s(%v)", msg.Kline.StreamId, delay, msg.Kline.InstId, msg.Kline.Exchange, kline.Interval, kline.Close.String(), kline.Confirm)
+			}
 			// kline klineStore -> klineSeries -> strategy -> indicator -> klineSeries.Series
 			s.Update(msg.Kline.Exchange, msg.Kline.InstId, kline)
 		}
@@ -146,9 +162,9 @@ func (s *KlineStore) sendSubscribeKline(save bool, exchange pb.ExchangeType, ins
 		if s.subKlineStream == nil {
 			return
 		}
-		zlog.Debugf("send stream subscribe kline msg: retry=%d, %#v", retry, subMsg)
+		zlog.Debugf("sending stream subscribe kline: retry=%d, instId=%s%v", retry, exchange, instIds)
 		if err = s.subKlineStream.Send(subMsg); err != nil {
-			zlog.Errorf("send stream subscribe kline msg error: %v", subMsg, err)
+			zlog.Errorf("send stream subscribe kline msg error: %#v", subMsg, err)
 			return
 		}
 		return
@@ -160,160 +176,107 @@ func (s *KlineStore) sendSubscribeKline(save bool, exchange pb.ExchangeType, ins
 	go retry.DoWithFixDelay(math.MaxInt32, time.Second, doSend)
 }
 
-func (s *KlineStore) initKlineSeries(exchange pb.ExchangeType, instId string) {
+func (s *KlineStore) inititalKlineSeries(exchange pb.ExchangeType, instId string) {
 	if !s.store.IsSupport(exchange) {
+		zlog.Errorf("unsupport exchange %s", exchange)
 		return
 	}
-
-	storeInst := s.store.Get(exchange).ComputeIfAbsent(instId, func(k string) *KlineStoreInstance {
-		return NewKlineStoreInstance(exchange, k)
+	storeInst := s.store.Get(exchange).ComputeIfAbsent(instId, func(k string) *TradeInstanceKlineSeries {
+		return NewTradeInstanceKlineSeries(exchange, k)
 	})
+	// 交易产品已初始化过
+	if !storeInst.Status.CompareAndSwap(int32(data.StatusNone), int32(data.StatusProcessing)) {
+		return
+	}
 
+	zlog.Infof("initial kline series starting: %s(%s),", instId, exchange)
 	// 初始化最新的 klineSeries
 	for _, interval := range s.subKlineIntervals {
-		for {
-			before := 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:    0,
-					Before:   before,
-					Live:     false,
-					Asc:      true,
-				}, grpc.UseCompressor("snappy"))
-				if err != nil {
-					zlog.Errorf("fetch missing klines error: instId=%s(%s) before=%d, err=%v", instId, exchange, before, err)
-				}
-				return
-			})
-			if err != nil {
-				zlog.Errorf("trade instance initial failed: %s(%s), %v", instId, exchange, err)
-				return
-			}
-			klines := rsp.Klines
-			if len(klines) == 0 {
-				break
-			}
-			for _, kline := range 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)
-			}
-
-			before = klines[len(klines)-1].Ts
-			if !rsp.Next {
-				break
-			}
-		}
+		retry.DoWithFixDelay(math.MaxInt32, 2*time.Second, func(retryTimes uint32) (_ struct{}, err error) {
+			err = s.fetchHistoryKlineToSeries(exchange, instId, interval, 0, 0, MaxSeriesKlines)
+			return
+		})
 	}
 	// 初始化历史k线完成, 开始订阅k线
 	storeInst.Status.Store(int32(data.StatusOk))
 	s.sendSubscribeKline(true, exchange, instId)
+	zlog.Infof("initial kline series success: %s(%s),", instId, exchange)
 }
 
-// Update
-// kline klineStore -> klineSeries -> strategy -> indicator -> klineSeries.Series
-func (s *KlineStore) Update(exchange pb.ExchangeType, instId string, kline *types.Kline) {
-	storeInst := s.store.Get(exchange).ComputeIfAbsent(instId, func(k string) *KlineStoreInstance {
-		return NewKlineStoreInstance(exchange, k)
-	})
-
-	// 只处理已初始化完成的交易产品k线
-	if storeInst.Status.Load() != int32(data.StatusOk) {
-		return
+// fetchHistoryKlineToSeries 拉去历史k线数据更新series
+func (s *KlineStore) fetchHistoryKlineToSeries(exchange pb.ExchangeType, instId, interval string, before, after int64, count uint32) (err error) {
+	// 拉取最新的1000条k线
+	req := &pb.ReqHistoryKlineStream{
+		Exchange: exchange,
+		InstId:   instId,
+		Interval: interval,
+		Count:    count,
+		Before:   before,
+		After:    after,
 	}
-	before, serial, err := storeInst.Update(kline)
+	stream, err := s.exchangeClient.HistoryKlineStream(context.Background(), req, grpc.UseCompressor("snappy"))
 	if err != nil {
-		zlog.Error("update kline series error: instId=%s(%s)", instId, exchange, err)
+		zlog.Errorf("fetch history kline stream error: instId=%s(%s), interval=%s, %#v, err=%v", instId, exchange, interval, req, 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:    after,
-			Before:   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)
+	for {
+		msg, err0 := stream.Recv()
+		if err0 == io.EOF {
+			// zlog.Debugf("fetch kline stream connection server closed")
+			break
 		}
-		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)
+		if err0 != nil {
+			err = err0
+			zlog.Error("fetch kline stream recv error: ", err0)
+			return
+		}
+		// zlog.Debugf("recv: %s(%s), %s, branch=%d, ts=%d~%d", instId, exchange, interval, len(msg.Klines), msg.Klines[0].Ts, msg.Klines[len(msg.Klines)-1].Ts)
+		for _, k := range msg.Klines {
+			kline := new(types.Kline)
+			kline.ParsePBKline(exchange, k)
+			s.Update(exchange, instId, kline)
 		}
 	}
+	return
 }
 
-// KlineStoreInstance 单个交易产品所有周期k线
-type KlineStoreInstance struct {
-	sync.RWMutex
-	Exchange       pb.ExchangeType
-	InstId         string
-	intervalKlines *types.IntervalState[*KlineSeries]
-	Status         atomic.Int32 // 交易产品状态
-}
-
-func NewKlineStoreInstance(exchange pb.ExchangeType, instId string) *KlineStoreInstance {
-	si := &KlineStoreInstance{
-		Exchange:       exchange,
-		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))
-	}
-	return si
-}
-
-// Update 更新k线
-// serial k线是否连续
-func (si *KlineStoreInstance) Update(kline *types.Kline) (before int64, serial bool, err error) {
+// Update
+// kline klineStore -> klineSeries -> strategy -> indicator -> klineSeries.Series
+func (s *KlineStore) Update(exchange pb.ExchangeType, instId string, kline *types.Kline) {
 	if _, ok := types.SupportedIntervals[kline.Interval]; !ok {
-		err = fmt.Errorf("unsupport interval: %s", kline.Interval)
+		zlog.Warningf("unsupport interval: %s", kline.Interval)
+		return
+	}
+	if !s.store.IsSupport(exchange) {
+		zlog.Warningf("unsupport exchange: %v", exchange)
 		return
 	}
 
-	si.Lock()
-	defer si.Unlock()
+	instSeries := s.store.Get(exchange).ComputeIfAbsent(instId, func(k string) *TradeInstanceKlineSeries {
+		return NewTradeInstanceKlineSeries(exchange, k)
+	})
 
-	ks := si.intervalKlines.Get(kline.Interval)
-	before, serial = ks.Update(kline)
-	if !serial {
-		// si.status.Store(int32(data.StatusProcessing))
+	before, serial := instSeries.IntervalKlines.Get(kline.Interval).Update(kline)
+	if !serial && instSeries.Status.CompareAndSwap(int32(data.StatusOk), int32(data.StatusProcessing)) {
+		func() {
+			defer instSeries.Status.Store(int32(data.StatusOk))
+			// 拉取缺失的k线
+			after := kline.Ts
+			zlog.Debugf("fetching padding kline series: instId=%s(%s), interval=%s, ts=%d~%d", instId, exchange, kline.Interval, before, after)
+			err := s.fetchHistoryKlineToSeries(exchange, instId, string(kline.Interval), before, after, 0)
+			if err != nil {
+				zlog.Error("fetch padding kline series error: instId=%s(%s), interval=%s, ts=%d~%d, err=%v", instId, exchange, kline.Interval, before, after, err)
+				return
+			}
+		}()
+	}
+
+	if expTs, ok := kline.Interval.AddMul(kline.Ts, 2); ok {
+		// k线已过期则不执行策略
+		if expTs < time.Now().Unix() {
+			return
+		}
+		// todo emit kline update, calc indicator...
 	}
-	return
 }

pkg/backtrace/backtrace.go

@@ -0,0 +1,13 @@
+package backtrace
+
+import "sig-pub/pkg/strategy"
+
+// 回测引擎
+type BacktraceEngine struct {
+	strategy strategy.Strategy
+}
+
+// 多周期策略回测引擎
+type MultiIntervalBacktraceEngine struct {
+	strategy strategy.MultiIntervalStrategy
+}

pkg/client/trade_instance_client.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"math"
 	"sig-pub/api/pb"
 	"sig-pub/pkg/data/entity"
 	"sig-pub/pkg/mapping"
@@ -64,7 +65,7 @@ func (c *TradeInstanceAside) getTradeInstance0(ctx context.Context, instId strin
 
 // ListExchangeTradeInstance 获取交易所支持的交易实例
 func (c *TradeInstanceAside) ListExchangeTradeInstance(ctx context.Context, exchange pb.ExchangeType) (marketInsts []*pb.MarketTradeInstance, err error) {
-	return retry.DoWithStepDelay(10, time.Second, func(retryTimes uint32) ([]*pb.MarketTradeInstance, error) {
+	return retry.DoWithFixDelay(math.MaxInt32, time.Second, func(retryTimes uint32) ([]*pb.MarketTradeInstance, error) {
 		rsp, err := c.marketClient.ListMarketTradeInstance(ctx, &pb.ReqListMarketTradeInstance{Exchange: exchange})
 		if err != nil {
 			zlog.Errorf("list market trade instances error: retry %d times, %v", retryTimes, err)

pkg/data/common.go

@@ -6,7 +6,7 @@ import "errors"
 type Status int32
 
 const (
-	StatusDisabled   Status = 0
+	StatusNone       Status = 0
 	StatusOk         Status = 1
 	StatusProcessing Status = 2
 	StatusDeleted    Status = 4

pkg/indicator/base.go

@@ -12,12 +12,12 @@ type IIndicator interface {
 
 // IKlineSeries k线序列, strategy服务提供
 type IKlineSeries interface {
-	Get(start int16) (kline types.Kline)
-	Series(start, end int16) (klines series.Klines)
+	Get(offset int16) (kline types.Kline)
+	Series(offset, count int16) (klines series.Klines)
 }
 
 // IIndicatorSeries 指标序列, 供策略读取, strategy服务提供
 type IIndicatorSeries interface {
-	Get(start int16) (vector float64)
-	Series(start, end int16) (matrix series.Floats)
+	Get(offset int16) (vector float64)
+	Series(offset, count int16) (matrix series.Floats)
 }

pkg/strategy/gold_x.go

@@ -0,0 +1,31 @@
+package strategy
+
+// GoldX 金叉策略
+type GoldX struct {
+}
+
+func (s *GoldX) New() Strategy {
+	return &GoldX{}
+}
+
+func (s *GoldX) Meta() StrategyMeta {
+	return StrategyMeta{
+		Name: "GoldX",
+	}
+}
+
+func (s *GoldX) Update(ctx StrategyContext) {
+	sma14 := ctx.IndicatorW("sma", 14)
+	sma28 := ctx.IndicatorW("sma", 28)
+	// 包装方法
+	s14 := sma14.Series(0, 2)
+	s28 := sma28.Series(0, 2)
+	crossover := s14[0] > s28[0] && s14[1] < s28[1]  // 上穿
+	crossunder := s14[0] < s28[0] && s14[1] > s28[1] // 下穿
+	if crossover {
+		ctx.Buy()
+	}
+	if crossunder {
+		ctx.Sell()
+	}
+}

pkg/strategy/strategy.go

@@ -0,0 +1,35 @@
+package strategy
+
+import (
+	"sig-pub/pkg/indicator"
+	"sig-pub/pkg/types"
+	"sig-pub/pkg/types/series"
+)
+
+// todo Exit 止盈止损策略(trading service 管理)
+// todo Meta 策略调参, 回测引擎自动调参回测(最佳参数) argGenerator.next() (arg, ok)
+type Strategy interface {
+	New() Strategy
+	Meta() StrategyMeta
+	Update(ctx StrategyContext)
+}
+
+type StrategyMeta struct {
+	// Id     string `json:"id"` // 策略注册/执行器系统分配
+	Name string
+	Desc string
+}
+
+// StrategyContext 策略外部访问能力
+// klineSeries, Indicator
+type StrategyContext interface {
+	Buy()  // 发出多信号
+	Sell() // 发出空信号
+
+	// Get [0]当前k线
+	Get(offset int16) types.Kline
+	// Series [offset...end]
+	Series(offset, count int16) (klines series.Klines)
+	// 获取窗口类型指标
+	IndicatorW(name string, window int) indicator.IIndicatorSeries
+}

pkg/strategy/strategy_multi_interval.go

@@ -0,0 +1,18 @@
+package strategy
+
+import "sig-pub/pkg/types"
+
+// 多k线周期策略
+type MultiIntervalStrategy interface {
+	Strategy
+	DriverInterval() types.Interval       // 驱动k线周期, 当驱动周期k线更新时则判断调用Update方法
+	SubscribeIntervals() []types.Interval // 订阅k线周期, 当同一时间的订阅周期都更新时调用Update方法
+}
+
+type MultiExchangeStrategy interface {
+	Strategy
+}
+
+type MultiIntervalExchangeStrategy interface {
+	Strategy
+}

pkg/types/exchange.go

@@ -23,10 +23,10 @@ type ExchangeState[T any] struct {
 }
 
 func NewExchangeState[T any]() *ExchangeState[T] {
-	return NewExchangeState0[T](func() (v T) { return })
+	return NewExchangeStateInit(func() (v T) { return })
 }
 
-func NewExchangeState0[T any](newer func() T) *ExchangeState[T] {
+func NewExchangeStateInit[T any](newer func() T) *ExchangeState[T] {
 	maxExchange := collect.MustMax(SupportedExchanges, func(e pb.ExchangeType) int32 { return int32(e) })
 	es := &ExchangeState[T]{
 		state: make([]T, maxExchange+1),

pkg/types/interval.go

@@ -78,7 +78,7 @@ var SupportedIntervals = IntervalMap{
 	Interval1h:  func(ts, mul int64) (ret int64) { return ts + (60 * 60 * 1000 * mul) },
 	Interval2h:  func(ts, mul int64) (ret int64) { return ts + (2 * 60 * 60 * 1000 * mul) },
 	Interval4h:  func(ts, mul int64) (ret int64) { return ts + (4 * 60 * 60 * 1000 * mul) },
-	Interval6h:  func(ts, mul int64) (ret int64) { return ts + (4 * 60 * 60 * 1000 * mul) },
+	Interval6h:  func(ts, mul int64) (ret int64) { return ts + (6 * 60 * 60 * 1000 * mul) },
 	Interval12h: func(ts, mul int64) (ret int64) { return ts + (12 * 60 * 60 * 1000 * mul) },
 	Interval1d:  func(ts, mul int64) (ret int64) { return ts + (24 * 60 * 60 * 1000 * mul) },
 	Interval2d:  func(ts, mul int64) (ret int64) { return ts + (2 * 24 * 60 * 60 * 1000 * mul) },