vmdb price scale

cmd/exchange/main.go

@@ -25,6 +25,7 @@ import (
 type ExchangeConf struct {
 	Register       discovery.Server
 	GrpcReflection bool
+	Okx            config.OkxExchange
 }
 
 func main() {
@@ -61,11 +62,11 @@ func main() {
 	}
 
 	// okx exchange
-	okxSubscriber := okx.NewOkxSubscriber(conf.Exchange.Okx)
+	okxSubscriber := okx.NewOkxSubscriber(exchangeConf.Okx)
 	if err := okxSubscriber.Init(); err != nil {
 		panic(err)
 	}
-	okxFetcher := okx.NewOkxFetcher(conf.Exchange.Okx.HttpProxy)
+	okxFetcher := okx.NewOkxFetcher(exchangeConf.Okx.HttpProxy)
 	okxExchange := exchange.NewExchange(okxFetcher, okxSubscriber)
 
 	// tsdb VictoriaMetrics

cmd/test/generic.go

@@ -18,7 +18,7 @@ import (
 	refv1 "google.golang.org/grpc/reflection/grpc_reflection_v1"
 )
 
-func main() {
+func mainGeneric() {
 	testGeneric()
 }
 

cmd/test/test.go

@@ -7,19 +7,30 @@ import (
 	"github.com/govalues/decimal"
 )
 
-func main_test() {
+func main() {
 	open := metrics.NewCounter("open")
 	open.Set(1234)
 
 	// curl -H 'Content-Type: application/json' --data-binary "@vmdata.json" -X POST http://localhost:8428/api/v1/import
 
-	testDecimal()
+	testDecimalScale()
 }
 
 type BTC struct {
 	Price decimal.Decimal
 }
 
+func testDecimalScale() {
+	price := decimal.MustParse("44268500")
+	scale, _ := decimal.Ten.PowInt(4)
+	// price, _ = price.Mul(scale)
+	// zlog.Info(scale)
+	// zlog.Info(price)
+
+	price, _ = price.Quo(scale)
+	zlog.Info(price)
+}
+
 func testDecimal() {
 	// zlog.Init()
 	// var interval = "1m"

config/config.toml

@@ -50,26 +50,3 @@ active = "victoriametrics"
 
 [tsdb.victoriametrics]
 addr = "http://127.0.0.1:8428"
-
-[exchange.okx]
-# apiKey = "7273282c-90f5-498f-9fe0-140ae07f5a73"
-# secretKey = "47F9F373CF43827497CA2EF99E889D48"
-# passphrase = "Sopod.2347."
-apiKey = "48be46ec-30ad-4f6f-a5e2-c9ea712df9cd"
-secretKey = "12323BA1B411A2235650BBD784850953"
-passphrase = "Tm.123456789"
-receiveBuffer = 4096
-marketSubscribeLimit = 16
-consumeBatch = 1024
-consumeLater = 2000 # 时间到达later或者数据累计到batch触发consume
-# httpProxy = "http://192.168.1.6:7890"
-httpProxy = "http://10.255.183.209:7890"
-
-# 模拟盘API交易地址如下：
-# REST：https://www.okx.com
-# WebSocket公共频道：wss://wspap.okx.com:8443/ws/v5/public
-# WebSocket私有频道：wss://wspap.okx.com:8443/ws/v5/private
-# WebSocket业务频道：wss://wspap.okx.com:8443/ws/v5/business
-
-# binance key RYgJrvqP4iGqdRth14r0ChgWo8eg0wPEcFqDttsKzvUJDyhOKvPiz42tXxjYIMiG
-# secret GLTzNNYzC0AbcINPAYfuKDjkWnAMQUhsyd1ed7ubcdzIRrFZBGUrOAkubqyjekVp

config/exchange.toml

@@ -5,3 +5,27 @@ grpcReflection = true     # 注册grpc反射服务
 nodeId = 1               # grpc服务节点id, 多实例唯一
 addr = ":8011"           # grpc 服务端口
 attrs = { weight = 10 }  # grpc 服务权重
+
+# okx exchange
+[okx]
+# apiKey = "7273282c-90f5-498f-9fe0-140ae07f5a73"
+# secretKey = "47F9F373CF43827497CA2EF99E889D48"
+# passphrase = "Sopod.2347."
+apiKey = "48be46ec-30ad-4f6f-a5e2-c9ea712df9cd"
+secretKey = "12323BA1B411A2235650BBD784850953"
+passphrase = "Tm.123456789"
+receiveBuffer = 4096
+marketSubscribeLimit = 16
+consumeBatch = 1024
+consumeLater = 2000 # 时间到达later或者数据累计到batch触发consume
+# httpProxy = "http://192.168.1.6:7890"
+httpProxy = "http://10.255.183.209:7890"
+
+# 模拟盘API交易地址如下：
+# REST：https://www.okx.com
+# WebSocket公共频道：wss://wspap.okx.com:8443/ws/v5/public
+# WebSocket私有频道：wss://wspap.okx.com:8443/ws/v5/private
+# WebSocket业务频道：wss://wspap.okx.com:8443/ws/v5/business
+
+# binance key RYgJrvqP4iGqdRth14r0ChgWo8eg0wPEcFqDttsKzvUJDyhOKvPiz42tXxjYIMiG
+# secret GLTzNNYzC0AbcINPAYfuKDjkWnAMQUhsyd1ed7ubcdzIRrFZBGUrOAkubqyjekVp

internal/exchange/exchange_service.go

@@ -63,7 +63,7 @@ func (svc *ExchangeService) subscribeExchanges() {
 	// 交易所订阅交易产品
 	for _, exchange := range svc.exchangeMap {
 		go func(exchange *Exchange) {
-			// get exchange trade instances
+			// get exchange all trade instances
 			insts, err := svc.tradeInstanceAside.ListExchangeTradeInstance(context.Background(), exchange.ExchangeType)
 			if err != nil {
 				zlog.Error(err)
@@ -83,26 +83,28 @@ func (svc *ExchangeService) subscribeExchanges() {
 					Leverages:      inst.Leverages,
 					Exchange:       exchange.ExchangeType,
 				}
+				// 待初始化币种数据
+				processingInsts = append(processingInsts, *tradeInst)
+
 				exchange.TradeInstIds.Store(inst.InstId, inst.ExchangeInstId)
-				exchange.ExchangeInsts.Store(inst.ExchangeInstId, &ExchangeTradeInstance{
+
+				exchangeInst := &ExchangeTradeInstance{
 					Inst:          tradeInst,
 					LiveKline:     types.NewIntervalState[types.Kline](),
 					LiveKStartTs:  types.NewIntervalState[int64](),
 					HistoryMarkTs: types.NewIntervalState[int64](),
-				})
-				// 待初始化币种数据
-				if inst.Status == int32(data.StatusProcessing) {
-					processingInsts = append(processingInsts, *tradeInst)
 				}
+				exchangeInst.Status.Store(int32(data.StatusProcessing))
+				exchange.ExchangeInsts.Store(inst.ExchangeInstId, exchangeInst)
 			}
 
-			// instIds := []string{"BTC-USDT", "DOGE-USDT-SWAP"}
+			// 订阅实时k线数据
-			err = exchange.Subscriber.SubscribeKline(exchangeInstIds...)
+			err = exchange.Subscriber.SubscribeKline(exchangeInstIds...) // []string{"BTC-USDT", "DOGE-USDT-SWAP"}
 			if err != nil {
 				zlog.Error(err)
 				return
 			}
-
+			// 消费实时k线数据
 			go func() {
 				c := exchange.Subscriber.ConsumerKline()
 				svc.consumerKline(exchange, c)
@@ -110,7 +112,7 @@ func (svc *ExchangeService) subscribeExchanges() {
 				zlog.Infof("unsubscribe exchange: %s", exchange.ExchangeType)
 			}()
 
-			// 初始化k线数据
+			// 初始化历史k线数据
 			go svc.initialKlines(exchange, processingInsts)
 		}(exchange)
 	}
@@ -355,7 +357,7 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade
 					zlog.Infof("retry fetch history kline task %d times: task -> %s", task.times, task.logKey())
 				}
 
-				if lastKlineTs, ex := svc.fetchTaskKlines(exchange, task); ex != nil {
+				if lastKlineTs, ex := svc.fetchTaskKlinesToTSDB(exchange, task); ex != nil {
 					failTasks.Add(1)
 					if task.times >= SingleKlineFetchTaskMaxFailTimes {
 						err = fmt.Errorf("task failed to many times %d, key: %s, err: %v", task.times, task.logKey(), ex)
@@ -375,9 +377,7 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade
 						exchangeInst.HistoryMarkTs.SetIf(task.interval, lastKlineTs, func(old int64) bool {
 							return lastKlineTs > old
 						})
-						// historyMarkTsMu.Lock()
+						// todo set trade instance status ok
-						// historyMarkTs[task.interval] = max(historyMarkTs[task.interval], lastKlineTs)
-						// historyMarkTsMu.Unlock()
 					}
 
 					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())
@@ -396,7 +396,7 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade
 	return
 }
 
-func (svc *ExchangeService) fetchTaskKlines(exchange *Exchange, task fetchKlineTask) (lastKlineTs int64, err error) {
+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)
 	if err != nil {

pkg/config/config.go

@@ -13,7 +13,6 @@ import (
 
 type Configuration struct {
 	// App      map[string]any
-	Exchange ExchangeConf
 
 	Grpc GrpcConfig
 	Etcd clientv3.Config
@@ -135,9 +134,6 @@ type VictoriaMetricsConfig struct {
 }
 
 // ============== exchange ==============
-type ExchangeConf struct {
-	Okx OkxExchange
-}
 type OkxExchange struct {
 	ApiKey               string
 	SecretKey            string

pkg/storage/tsdb/victoria_metrics/metric.go

@@ -5,6 +5,7 @@ import (
 	"sig-pub/pkg/types"
 
 	"github.com/bytedance/sonic"
+	"github.com/govalues/decimal"
 )
 
 var (
@@ -43,7 +44,7 @@ func (m *Metric) ToRowJson() ([]byte, error) {
 	return sonic.Marshal(m)
 }
 
-func Kline2Metrics(inst types.TradeInstance, klines []*types.Kline) (metrics []*Metric) {
+func Kline2Metrics0(inst types.TradeInstance, klines []*types.Kline) (metrics []*Metric) {
 	instMetrics := make(map[string][6]*Metric)
 	// id := inst.InstId
 	// inst.InstId = "doge_udst"
@@ -84,3 +85,64 @@ func Kline2Metrics(inst types.TradeInstance, klines []*types.Kline) (metrics []*
 	}
 	return
 }
+
+func Kline2Metrics(inst types.TradeInstance, klines []*types.Kline) (metrics []*Metric, err error) {
+	instMetrics := make(map[string][6]*Metric)
+
+	var priceSz, quantitySz decimal.Decimal
+	if priceSz, err = decimal.Ten.PowInt(int(inst.PriceSz)); err != nil {
+		return
+	}
+	if quantitySz, err = decimal.Ten.PowInt(int(inst.QuantitySz)); err != nil {
+		return
+	}
+	for _, kline := range klines {
+		ms, ok := instMetrics[inst.InstId]
+		if !ok || (rawValueLimit > 0 && len(ms[0].Values) >= rawValueLimit) {
+			ms = [6]*Metric{
+				NewMetric(inst.InstId, "interval", string(kline.Interval), "exchange", inst.Exchange.String(), "kind", "open"),
+				NewMetric(inst.InstId, "interval", string(kline.Interval), "exchange", inst.Exchange.String(), "kind", "high"),
+				NewMetric(inst.InstId, "interval", string(kline.Interval), "exchange", inst.Exchange.String(), "kind", "low"),
+				NewMetric(inst.InstId, "interval", string(kline.Interval), "exchange", inst.Exchange.String(), "kind", "close"),
+				NewMetric(inst.InstId, "interval", string(kline.Interval), "exchange", inst.Exchange.String(), "kind", "vol"),
+				NewMetric(inst.InstId, "interval", string(kline.Interval), "exchange", inst.Exchange.String(), "kind", "volQuote"),
+			}
+			instMetrics[inst.InstId] = ms
+			for i := range len(ms) {
+				metrics = append(metrics, ms[i])
+			}
+		}
+
+		// 存储精度处理
+		var open, high, low, close, vol, volQuote decimal.Decimal
+		if open, err = kline.Open.Mul(priceSz); err != nil {
+			return
+		}
+		if high, err = kline.High.Mul(priceSz); err != nil {
+			return
+		}
+		if low, err = kline.Low.Mul(priceSz); err != nil {
+			return
+		}
+		if close, err = kline.Close.Mul(priceSz); err != nil {
+			return
+		}
+		if vol, err = kline.Vol.Mul(quantitySz); err != nil {
+			return
+		}
+		if volQuote, err = kline.VolQuote.Mul(quantitySz); err != nil {
+			return
+		}
+
+		var klineValues = []decimal.Decimal{open, high, low, close, vol, volQuote}
+		for i, value := range klineValues {
+			value_f64, ok := value.Float64()
+			if !ok {
+				err = fmt.Errorf("kline value scale to float64 error: ", value)
+				return
+			}
+			ms[i].AddTsValue(kline.Ts, value_f64)
+		}
+	}
+	return
+}

pkg/storage/tsdb/victoria_metrics/vm.go

@@ -31,7 +31,10 @@ func NewVictoriaMetricsTSDB(conf config.VictoriaMetricsConfig) *VictoriaMetricsT
 }
 
 func (vm *VictoriaMetricsTSDB) SaveKlines(inst types.TradeInstance, klines []*types.Kline) (err error) {
-	metrics := Kline2Metrics(inst, klines)
+	metrics, err := Kline2Metrics(inst, klines)
+	if err != nil {
+		return
+	}
 	err = vm.batchWriteMetrics(metrics)
 	return
 }
@@ -113,6 +116,14 @@ func (vm *VictoriaMetricsTSDB) ListRangeKline(inst types.TradeInstance, interval
 		}
 	}()
 
+	var priceSz, quantitySz decimal.Decimal
+	if priceSz, err = decimal.Ten.PowInt(int(inst.PriceSz)); err != nil {
+		return
+	}
+	if quantitySz, err = decimal.Ten.PowInt(int(inst.QuantitySz)); err != nil {
+		return
+	}
+
 	// read response json line
 	reader := bufio.NewReader(resp.Body)
 	var vmLineBytes []byte
@@ -147,6 +158,15 @@ func (vm *VictoriaMetricsTSDB) ListRangeKline(inst types.TradeInstance, interval
 			}
 			value := vmMetric.Values[i]
 			switch vmMetric.Metric.Kind {
+			case "vol", "volQuote":
+				value, err = value.Quo(quantitySz)
+			default:
+				value, err = value.Quo(priceSz)
+			}
+			if err != nil {
+				return
+			}
+			switch vmMetric.Metric.Kind {
 			case "open":
 				kline.Open = value
 			case "close":