From 13274b140df1009fc52cd7d0f0ea0f0f3d5f8fe2 Mon Sep 17 00:00:00 2001 From: strange Date: Thu, 9 Oct 2025 18:43:12 +0800 Subject: [PATCH] vmdb price scale --- cmd/exchange/main.go | 5 +- cmd/test/generic.go | 2 +- cmd/test/test.go | 15 ++++- config/config.toml | 23 -------- config/exchange.toml | 24 ++++++++ internal/exchange/exchange_service.go | 30 +++++----- pkg/config/config.go | 4 -- pkg/storage/tsdb/victoria_metrics/metric.go | 64 ++++++++++++++++++++- pkg/storage/tsdb/victoria_metrics/vm.go | 22 ++++++- 9 files changed, 140 insertions(+), 49 deletions(-) diff --git a/cmd/exchange/main.go b/cmd/exchange/main.go index d58207c..f877ad0 100644 --- a/cmd/exchange/main.go +++ b/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 diff --git a/cmd/test/generic.go b/cmd/test/generic.go index 92f0f14..72cbe0a 100644 --- a/cmd/test/generic.go +++ b/cmd/test/generic.go @@ -18,7 +18,7 @@ import ( refv1 "google.golang.org/grpc/reflection/grpc_reflection_v1" ) -func main() { +func mainGeneric() { testGeneric() } diff --git a/cmd/test/test.go b/cmd/test/test.go index f2ba235..fcf9f87 100644 --- a/cmd/test/test.go +++ b/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" diff --git a/config/config.toml b/config/config.toml index 574c7d5..0ab286b 100644 --- a/config/config.toml +++ b/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 diff --git a/config/exchange.toml b/config/exchange.toml index 6e19a88..d901904 100644 --- a/config/exchange.toml +++ b/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 diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index 092e79f..6e6ec93 100644 --- a/internal/exchange/exchange_service.go +++ b/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"} - err = exchange.Subscriber.SubscribeKline(exchangeInstIds...) + // 订阅实时k线数据 + 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() - // historyMarkTs[task.interval] = max(historyMarkTs[task.interval], lastKlineTs) - // historyMarkTsMu.Unlock() + // 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()) @@ -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 { diff --git a/pkg/config/config.go b/pkg/config/config.go index c5b042a..03b82bd 100644 --- a/pkg/config/config.go +++ b/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 diff --git a/pkg/storage/tsdb/victoria_metrics/metric.go b/pkg/storage/tsdb/victoria_metrics/metric.go index 51aab3d..bd2e802 100644 --- a/pkg/storage/tsdb/victoria_metrics/metric.go +++ b/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 +} diff --git a/pkg/storage/tsdb/victoria_metrics/vm.go b/pkg/storage/tsdb/victoria_metrics/vm.go index e36e95d..b893799 100644 --- a/pkg/storage/tsdb/victoria_metrics/vm.go +++ b/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":