kline live ts mark

docker-compose.yml

@@ -88,7 +88,7 @@ services:
 
   sig-vm:
     container_name: sig-vm
-    image: victoriametrics/victoria-metrics:v1.116.0
+    image: victoriametrics/victoria-metrics:v1.127.0
     network_mode: host
     volumes:
       - "./fs/victoria-metrics-data:/victoria-metrics-data"

internal/exchange/exchange_service.go

@@ -9,6 +9,7 @@ import (
 	"sig-pub/pkg/aside"
 	"sig-pub/pkg/data"
 	"sig-pub/pkg/types"
+	"sig-pub/pkg/utils/collect"
 	"sig-pub/pkg/utils/times"
 	"sig-pub/pkg/zlog"
 	"sort"
@@ -23,7 +24,7 @@ import (
 type ExchangeService struct {
 	exchangeMap         map[pb.ExchangeType]*Exchange
 	tradeInstanceAside  *aside.TradeInstanceAside
-	exchangeDataService *ExchangeDataPersist
+	exchangeDataPersist *ExchangeDataPersist
 
 	klinePublisher *Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]]
 }
@@ -31,7 +32,7 @@ type ExchangeService struct {
 // exchanges: 支持的数据源交易所
 func NewExchangeService(
 	tradeInstanceAside *aside.TradeInstanceAside,
-	exchangeDataService *ExchangeDataPersist,
+	exchangeDataPersist *ExchangeDataPersist,
 	exchanges ...*Exchange,
 ) *ExchangeService {
 	exchangeMap := make(map[pb.ExchangeType]*Exchange)
@@ -42,7 +43,7 @@ func NewExchangeService(
 	return &ExchangeService{
 		exchangeMap:         exchangeMap,
 		tradeInstanceAside:  tradeInstanceAside,
-		exchangeDataService: exchangeDataService,
+		exchangeDataPersist: exchangeDataPersist,
 		klinePublisher:      NewPublisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]](16),
 	}
 }
@@ -181,16 +182,20 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch
 
 		if len(confirmKlines) > 0 {
 			// tsdb storage todo 异步处理
-			err := svc.exchangeDataService.SaveKline(*tradeInst, confirmKlines)
+			err := svc.exchangeDataPersist.SaveKline(*tradeInst, confirmKlines)
 			if err != nil {
-				zlog.Errorf("kline save to tsdb error: ", err)
-			}
-
-			// 初始化状态完成, 检查k线时间戳标记
-			if exchangeInst.Status.Load() == int32(data.StatusOk) {
-				lastConfirmKline := confirmKlines[len(confirmKlines)-1]
-				historyMark := exchangeInst.HistoryMarkTs.Get(lastConfirmKline.Interval)
-				_ = historyMark
+				zlog.Errorf("kline save to tsdb error: %v, %#v", err, confirmKlines)
+			} else {
+				// 初始化状态完成, 更新k线时间戳标记
+				if exchangeInst.Status.Load() == int32(data.StatusOk) {
+					latestK := collect.MustMax(confirmKlines, func(k *types.Kline) int64 { return k.Ts })
+					// 标记确认k线
+					tsKey, ex := svc.exchangeDataPersist.SaveHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, latestK.Interval, latestK.Ts)
+					if ex != nil {
+						zlog.Errorf("history mark inititaled ts error: key=%s, ts=%d, %v", tsKey, latestK.Ts, ex)
+					}
+					// todo k线完整性检查
+				}
 			}
 		}
 
@@ -271,7 +276,7 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade
 			return
 		}
 		exchangeInst.HistoryMarkTs.Range(func(_ int, interval types.Interval, ts int64) {
-			tsKey, ex := svc.exchangeDataService.SaveHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, interval, ts)
+			tsKey, ex := svc.exchangeDataPersist.SaveHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, interval, ts)
 			if ex != nil {
 				zlog.Errorf("history mark inititaled ts error: key=%s, ts=%d, %v", tsKey, ts, ex)
 			}
@@ -293,7 +298,7 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade
 			// interval := types.Interval1d
 			// intervalAdder := types.SupportedIntervals[interval]
 			// history 未补全前, history写 kvdb ts mark, 补全后 ws live 写 ts mark
-			beforeTs, ex := svc.exchangeDataService.GetHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, interval)
+			beforeTs, ex := svc.exchangeDataPersist.GetHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, interval)
 			if ex != nil {
 				err = ex
 				zlog.Error(err)
@@ -377,12 +382,11 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade
 						exchangeInst.HistoryMarkTs.SetIf(task.interval, lastKlineTs, func(old int64) bool {
 							return lastKlineTs > old
 						})
-						// todo set trade instance status ok
 					}
 
-					zlog.Debugf("trade instance initial kline tasks processing: %s(%s), pub %d, sub %d, fail %d", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load(), subTasks.Load(), failTasks.Load())
 					// 任务都已执行成功结束
 					subs := subTasks.Add(1)
+					zlog.Debugf("trade instance initial kline tasks processing: %s(%s), pub %d, sub %d, fail %d", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load(), subTasks.Load(), failTasks.Load())
 					if pubTaskDone.Load() && subs >= pubTasks.Load() {
 						zlog.Infof("trade instance initial kline tasks success finished, %s(%s), pub %d, sub %d, fail %d", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load(), subTasks.Load(), failTasks.Load())
 						cancel()
@@ -413,10 +417,11 @@ func (svc *ExchangeService) fetchTaskKlinesToTSDB(exchange *Exchange, task fetch
 	lastKlineTs = max(sts, ets)
 	ss := time.UnixMilli(sts).In(loc).Format(times.FORMAT_DATE)
 	ee := time.UnixMilli(ets).In(loc).Format(times.FORMAT_DATE)
-	zlog.Infof("fetch interval %s %d~%d klines: ret=%d~%d, %d klines, %s~%s", interval, beforeTs, afterTs, ets, sts, len(klines), ee, ss)
+	// zlog.Infof("fetch interval %s %d~%d klines: ret=%d~%d, %d klines, %s~%s", interval, beforeTs, afterTs, ets, sts, len(klines), ee, ss)
+	zlog.Infof("fetch history interval klines: %s, %d klines, %s~%s", task.logKey(), len(klines), ee, ss)
 
 	// store to tsdb
-	err = svc.exchangeDataService.SaveKline(task.inst, klines)
+	err = svc.exchangeDataPersist.SaveKline(task.inst, klines)
 	if err != nil {
 		zlog.Errorf("save history klines to tsdb error: task -> %s, err -> %v", task.logKey(), err)
 		return
@@ -532,7 +537,7 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory
 		return
 	}
 
-	klines, err = svc.exchangeDataService.ListKline(*exchangeInst.Inst, interval, beforeTs, afterTs)
+	klines, err = svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, beforeTs, afterTs)
 	if err != nil || len(klines) == 0 {
 		return
 	}

pkg/data/entity/trade_instance.go

@@ -4,18 +4,20 @@ import "github.com/lib/pq"
 
 // 交易产品基础信息
 type TradeInstance struct {
-	InstId     string                   `gorm:"column:inst_id;primaryKey" json:"InstId"`      // 交易产品id BTC_USDT_SWAP
-	InstPair   string                   `gorm:"column:inst_pair" json:"instPair"`             // 交易对 BTCUSDT
-	InstType   int32                    `gorm:"column:inst_type" json:"instType"`             // 交易类型: 1现货2永续合约
-	InstCoin   string                   `gorm:"column:inst_coin" json:"instCoin"`             // 所属币种 BTC
-	Status     int32                    `gorm:"column:status" json:"status"`                  // 状态: 0禁用,1正常,4删除
-	PriceSz    int32                    `gorm:"column:price_sz" json:"priceSz"`               // 价格小数点位数
-	QuantitySz int32                    `gorm:"column:quantity_sz" json:"quantitySz"`         // 数量小数点位数
-	Icon       string                   `gorm:"column:icon" json:"icon"`                      // 币种图标
-	Leverages  pq.Int32Array            `gorm:"column:leverages;type:int[]" json:"leverages"` // 杠杆倍数[5,10,20,50,100]
-	UpdateBy   string                   `gorm:"column:update_by" json:"updateBy"`             // 更新人
-	UpdateTime int64                    `gorm:"column:update_time" json:"updateTime"`         // 更新时间戳毫秒
-	Exchanges  []*TradeInstanceExchange `gorm:"-" json:"exchanges"`                           // 交易产品支持交易所
+	InstId        string                   `gorm:"column:inst_id;primaryKey" json:"InstId"`      // 交易产品id BTC_USDT_SWAP
+	InstPair      string                   `gorm:"column:inst_pair" json:"instPair"`             // 交易对 BTCUSDT
+	InstType      int32                    `gorm:"column:inst_type" json:"instType"`             // 交易类型: 1现货2永续合约
+	InstCoin      string                   `gorm:"column:inst_coin" json:"instCoin"`             // 所属币种 BTC
+	Status        int32                    `gorm:"column:status" json:"status"`                  // 状态: 0禁用,1正常,4删除
+	PriceSz       int32                    `gorm:"column:price_sz" json:"priceSz"`               // 价格小数点左移位数(float64计算时)
+	QuantitySz    int32                    `gorm:"column:quantity_sz" json:"quantitySz"`         // 数量小数点左移位数(float64计算时)
+	PriceScale    int32                    `gorm:"column:price_scale" json:"priceScale"`         // 价格有效小数点位数
+	QuantityScale int32                    `gorm:"column:quantity_scale" json:"quantityScale"`   // 数量有效小数点位数
+	Icon          string                   `gorm:"column:icon" json:"icon"`                      // 币种图标
+	Leverages     pq.Int32Array            `gorm:"column:leverages;type:int[]" json:"leverages"` // 杠杆倍数[5,10,20,50,100]
+	UpdateBy      string                   `gorm:"column:update_by" json:"updateBy"`             // 更新人
+	UpdateTime    int64                    `gorm:"column:update_time" json:"updateTime"`         // 更新时间戳毫秒
+	Exchanges     []*TradeInstanceExchange `gorm:"-" json:"exchanges"`                           // 交易产品支持交易所
 	// 合约面值(0.0001/BTC) 杠杆倍数[5,10,20,50,100] 前端显示(BTC/USDT) 是否可交易 开空 开多 市价开空 市价开多 交易对(btcusdt)
 	// 资金费率 资金周期 最小下单量 最大下单量 最小下单金额 最大下单金额
 	// 开仓手续费 平仓手续费

pkg/storage/tsdb/victoria_metrics/metric.go

@@ -138,7 +138,7 @@ func Kline2Metrics(inst types.TradeInstance, klines []*types.Kline) (metrics []*
 		for i, value := range klineValues {
 			value_f64, ok := value.Float64()
 			if !ok {
-				err = fmt.Errorf("kline value scale to float64 error: ", value)
+				err = fmt.Errorf("kline value scale to float64 error: %s", value.String())
 				return
 			}
 			ms[i].AddTsValue(kline.Ts, value_f64)