diff --git a/docker-compose.yml b/docker-compose.yml index a9b0fba..a9e2313 100644 --- a/docker-compose.yml +++ b/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" diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index 6e6ec93..4ee8329 100644 --- a/internal/exchange/exchange_service.go +++ b/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 } diff --git a/pkg/data/entity/trade_instance.go b/pkg/data/entity/trade_instance.go index 4e8e330..3e8e153 100644 --- a/pkg/data/entity/trade_instance.go +++ b/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) // 资金费率 资金周期 最小下单量 最大下单量 最小下单金额 最大下单金额 // 开仓手续费 平仓手续费 diff --git a/pkg/storage/tsdb/victoria_metrics/metric.go b/pkg/storage/tsdb/victoria_metrics/metric.go index bd2e802..164155e 100644 --- a/pkg/storage/tsdb/victoria_metrics/metric.go +++ b/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)