From 419d3dbea34dc4d9288b91d8fe3eb58fde2dd625 Mon Sep 17 00:00:00 2001 From: strange Date: Tue, 29 Jul 2025 09:07:11 +0800 Subject: [PATCH] initial history klines --- cmd/exchange/main.go | 24 +- internal/exchange/exchange.go | 34 ++- internal/exchange/exchange_grpc_server.go | 250 ++++++++++++++---- internal/exchange/okx/channel_kline.go | 8 +- internal/exchange/okx/okx_fetch.go | 55 +++- internal/exchange/okx/okx_fetch_test.go | 6 +- internal/exchange/okx/okx_limiter.go | 12 + .../okx/{okx.go => okx_subscriber.go} | 16 +- internal/exchange/okx/types.go | 7 + pkg/aside/trade_instance_client.go | 10 +- pkg/storage/kvrocks/kvrocks.go | 53 ++++ pkg/types/exchange.go | 13 +- pkg/types/instance.go | 10 +- 13 files changed, 410 insertions(+), 88 deletions(-) create mode 100644 internal/exchange/okx/okx_limiter.go rename internal/exchange/okx/{okx.go => okx_subscriber.go} (74%) create mode 100644 pkg/storage/kvrocks/kvrocks.go diff --git a/cmd/exchange/main.go b/cmd/exchange/main.go index bfa2712..b0fa805 100644 --- a/cmd/exchange/main.go +++ b/cmd/exchange/main.go @@ -11,6 +11,7 @@ import ( "sig-pub/pkg/config" "sig-pub/pkg/grpc/discovery" "sig-pub/pkg/grpc/interceptor" + "sig-pub/pkg/storage/kvrocks" vmts "sig-pub/pkg/storage/tsdb/victoria_metrics" "sig-pub/pkg/utils/exit" "sig-pub/pkg/zlog" @@ -38,11 +39,6 @@ func main() { panic(err) } - okxExchange := okx.NewOkxExchange(conf.Exchange.Okx) - if err := okxExchange.Init(); err != nil { - panic(err) - } - etcdClient, err := clientv3.New(conf.Etcd) if err != nil { panic(err) @@ -65,7 +61,23 @@ func main() { } marketClient := pb.NewMarketClient(conn) tradeInstanceAside := aside.NewTradeInstanceAside(marketClient) - exchangeService := exchange.NewExchangeGrpcServer(tradeInstanceAside, tsdbService, okxExchange) + + // kvrocks db + kvdb := kvrocks.NewKVRocksDB(conf.Database.Kvrocks) + if err := kvdb.Ping(); err != nil { + panic(err) + } + + // okx exchange + okxSubscriber := okx.NewOkxSubscriber(conf.Exchange.Okx) + if err := okxSubscriber.Init(); err != nil { + panic(err) + } + okxFetcher := okx.NewOkxFetcher(conf.Exchange.Okx.HttpProxy) + okxExchange := exchange.NewExchange(okxFetcher, okxSubscriber) + + // exhcange main service + exchangeService := exchange.NewExchangeGrpcServer(tradeInstanceAside, tsdbService, kvdb, okxExchange) if err := exchangeService.Init(); err != nil { panic(err) } diff --git a/internal/exchange/exchange.go b/internal/exchange/exchange.go index 5b23d40..c1d524a 100644 --- a/internal/exchange/exchange.go +++ b/internal/exchange/exchange.go @@ -1,6 +1,11 @@ package exchange -import "sig-pub/pkg/types" +import ( + "context" + "fmt" + "sig-pub/pkg/types" + "sync" +) // 交易所行情数据订阅 type ExchangeSubscriber interface { @@ -19,4 +24,31 @@ type ExchangeSubscriber interface { // 交易所行情数据请求 type ExchangeFetcher interface { + // 交易所类型 + ExhcangeType() types.Exchange + // 获取区间内历史k线数据 + FetchHistoryKlines(ctx context.Context, instId string, interval types.Interval, after, before int64) (klines []*types.Kline, err error) +} + +// 交易所交互接口 +type Exchange struct { + ExType types.Exchange + Fetcher ExchangeFetcher + Subscriber ExchangeSubscriber + Insts map[string]*types.TradeInstance // + sync.RWMutex +} + +func NewExchange(fetcher ExchangeFetcher, subscriber ExchangeSubscriber) *Exchange { + exType := fetcher.ExhcangeType() + if exType != subscriber.ExhcangeType() { + panic(fmt.Errorf("exchange type not match: %#v, %#v", exType, subscriber.ExhcangeType())) + } + + return &Exchange{ + ExType: exType, + Fetcher: fetcher, + Subscriber: subscriber, + Insts: make(map[string]*types.TradeInstance), + } } diff --git a/internal/exchange/exchange_grpc_server.go b/internal/exchange/exchange_grpc_server.go index 9f3e990..6bdbbc9 100644 --- a/internal/exchange/exchange_grpc_server.go +++ b/internal/exchange/exchange_grpc_server.go @@ -6,7 +6,8 @@ import ( "io" "sig-pub/api/pb" "sig-pub/pkg/aside" - "sig-pub/pkg/data/entity" + "sig-pub/pkg/data" + "sig-pub/pkg/storage/kvrocks" "sig-pub/pkg/types" "sig-pub/pkg/zlog" "sync" @@ -15,43 +16,34 @@ import ( "google.golang.org/grpc" ) -type Exchange struct { - Type pb.Exchange - Subscriber ExchangeSubscriber - Insts map[string]*entity.TradeInstanceExchange - sync.RWMutex -} - type ExchangeGrpcServer struct { pb.UnimplementedExchangeServiceServer - exchangeMap map[pb.Exchange]*Exchange + exchangeMap map[types.Exchange]*Exchange tradeInstanceAside *aside.TradeInstanceAside exchangeDataService *ExchangeDataService + kvdb *kvrocks.KVRocksDB klineStreamId int64 klinePublisher *Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]] } // exchanges: 支持的数据源交易所 -func NewExchangeGrpcServer(tradeInstanceAside *aside.TradeInstanceAside, exchangeDataService *ExchangeDataService, exchanges ...ExchangeSubscriber) *ExchangeGrpcServer { - exchangeMap := make(map[pb.Exchange]*Exchange) +func NewExchangeGrpcServer( + tradeInstanceAside *aside.TradeInstanceAside, + exchangeDataService *ExchangeDataService, + kvdb *kvrocks.KVRocksDB, + exchanges ...*Exchange, +) *ExchangeGrpcServer { + exchangeMap := make(map[types.Exchange]*Exchange) for _, exchange := range exchanges { - exchangeType := exchange.ExhcangeType() - pbExchangeType, ok := exchangeType.Exchange2PB() - if !ok { - panic(fmt.Errorf("unknown exchange: %#v", exchangeType)) - } - exchangeMap[pbExchangeType] = &Exchange{ - Type: pbExchangeType, - Subscriber: exchange, - Insts: make(map[string]*entity.TradeInstanceExchange), - } + exchangeMap[exchange.ExType] = exchange } return &ExchangeGrpcServer{ exchangeMap: exchangeMap, tradeInstanceAside: tradeInstanceAside, exchangeDataService: exchangeDataService, + kvdb: kvdb, klinePublisher: NewPublisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]](16), } } @@ -68,16 +60,30 @@ func (svc *ExchangeGrpcServer) subscribeExchanges() { for _, exchange := range svc.exchangeMap { go func(exchange *Exchange) { // get exchange trade instances - insts, err := svc.tradeInstanceAside.ListExchangeTradeInstance(context.Background(), exchange.Type) + insts, err := svc.tradeInstanceAside.ListExchangeTradeInstance(context.Background(), exchange.ExType) if err != nil { zlog.Error(err) return } + var exchangeInstIds []string + var processingInsts []types.TradeInstance exchange.Lock() for _, inst := range insts { exchangeInstIds = append(exchangeInstIds, inst.ExchangeInstId) - exchange.Insts[inst.ExchangeInstId] = inst + tradeInst := &types.TradeInstance{ + InstId: inst.InstId, + Status: inst.Status, + PriceSz: 0, + QuantitySz: 0, + ExchangeInstId: inst.ExchangeInstId, + Exchange: exchange.ExType, + } + exchange.Insts[inst.ExchangeInstId] = tradeInst + // 待初始化币种数据 + if inst.Status == data.StatusProcessing { + processingInsts = append(processingInsts, *tradeInst) + } } exchange.Unlock() @@ -87,20 +93,23 @@ func (svc *ExchangeGrpcServer) subscribeExchanges() { zlog.Error(err) return } - c := exchange.Subscriber.ConsumerKline() - svc.consumerKline(exchange, c) - // todo subscribe books 订单簿 - zlog.Infof("unsubscribe exchange: %s", exchange.Type.String()) + + go func() { + c := exchange.Subscriber.ConsumerKline() + svc.consumerKline(exchange, c) + // todo subscribe books 订单簿 + zlog.Infof("unsubscribe exchange: %s", exchange.ExType) + }() + + // 初始化k线数据 + go svc.initialKlines(exchange, processingInsts) }(exchange) } } // consumerKline 消费交易所k线数据 func (svc *ExchangeGrpcServer) consumerKline(exchange *Exchange, c <-chan *types.ChannelKline) { - exchangeType, ok := types.ExchangePBParse(exchange.Type) - if !ok { - panic(fmt.Errorf("unknown exchange type: %v", exchange.Type)) - } + exchangeType := exchange.ExType for { channelK, ok := <-c @@ -109,7 +118,7 @@ func (svc *ExchangeGrpcServer) consumerKline(exchange *Exchange, c <-chan *types } // 交易所 instid 转 sig-instid - var exInst *entity.TradeInstanceExchange + var exInst *types.TradeInstance exchange.RLock() if inst, ok := exchange.Insts[channelK.InstId]; ok && inst != nil { exchange.RUnlock() @@ -124,12 +133,11 @@ func (svc *ExchangeGrpcServer) consumerKline(exchange *Exchange, c <-chan *types pubMsgMap := make(map[string]*pb.StreamKline) // instId := channelK.InstId - exchange, ok := channelK.Exchange.Exchange2PB() - if !ok { - zlog.Errorf("unknown exchange kline: %v", channelK.Exchange) + pbExType, err := channelK.Exchange.Exchange2PB() + if err != nil { + zlog.Error(err) continue } - exchangeName := exchange.String() var confirmKlines []*types.Kline for _, kline := range channelK.Klines { @@ -139,13 +147,13 @@ func (svc *ExchangeGrpcServer) consumerKline(exchange *Exchange, c <-chan *types confirm = 1 confirmKlines = append(confirmKlines, kline) } - pubKey := fmt.Sprintf("/kline/%s/%s/%s/%d", exchangeName, exInst.InstId, kline.Interval, confirm) + pubKey := fmt.Sprintf("/kline/%s/%s/%s/%d", exchangeType, exInst.InstId, kline.Interval, confirm) // todo 优化没有订阅者就跳过 msg, ok := pubMsgMap[pubKey] if !ok { msg = new(pb.StreamKline) msg.InstId = exInst.InstId - msg.Exchange = exchange + msg.Exchange = pbExType pubMsgMap[pubKey] = msg } @@ -155,14 +163,8 @@ func (svc *ExchangeGrpcServer) consumerKline(exchange *Exchange, c <-chan *types // tsdb storage if len(confirmKlines) > 0 { - typeInst := types.TradeInstance{ - InstId: exInst.InstId, // channelK.InstId - PriceSz: 0, - QuantitySz: 0, - Exchange: exchangeType, - } // todo 异步处理 - err := svc.exchangeDataService.SaveKlines(typeInst, confirmKlines) + err := svc.exchangeDataService.SaveKlines(*exInst, confirmKlines) if err != nil { zlog.Errorf("kline save to tsdb error: ", err) } @@ -279,3 +281,161 @@ func (svc *ExchangeGrpcServer) SubscribeKline(stream grpc.BidiStreamingServer[pb // } // } } + +const ( + KlineBefore0 int64 = 1672502400000 // k线开始数据 2023-01-01 00:00:00 GMT+8 + HistoryKlineTsKey string = "history-kline-ts:%s/%s/%s" // exchange:sig-instid:interval +) + +// initialKline 初始化交易产品历史k线数据 +func (svc *ExchangeGrpcServer) initialKlines(exchange *Exchange, insts []types.TradeInstance) { + var err error + defer func() { + if err != nil { + zlog.Error(err) + } + }() + + type task struct { + inst types.TradeInstance + interval types.Interval + afterTs int64 + beforeTs int64 + } + + // 任务 channel + ch := make(chan task) + + // 任务生成器 + go func() { + for _, inst := range insts { + // for interval, intervalAdder := range types.SupportedIntervals { + interval := types.Interval1h + intervalAdder := types.SupportedIntervals[interval] + tsKey := fmt.Sprintf(HistoryKlineTsKey, inst.Exchange, inst.InstId, interval) + beforeTs, e := svc.kvdb.GetI64(context.Background(), tsKey) + if e != nil { + err = e + return + } + if beforeTs == 0 { + beforeTs = intervalAdder(KlineBefore0, -1) + } + for { + afterTs := intervalAdder(beforeTs, 101) + ch <- task{ + inst: inst, + interval: interval, + afterTs: afterTs, + beforeTs: beforeTs, + } + beforeTs = afterTs + // todo compare ws live ts + } + // } + } + close(ch) + }() + + // 任务消费器 8协程并行 + wg := new(sync.WaitGroup) + for range 11 { + wg.Add(1) + go func() { + defer wg.Done() + for { + task, ok := <-ch + if !ok { + break + } + + interval, afterTs, beforeTs := task.interval, task.afterTs, task.beforeTs + klines, e := exchange.Fetcher.FetchHistoryKlines(context.Background(), task.inst.ExchangeInstId, interval, afterTs, beforeTs) + if e != nil { + // todo retry + err = e + return + } + if len(klines) == 0 { + continue // todo .... + } + + zlog.Infof("fetch interval %s %d~%d klines: ret=%d~%d, %d klines", interval, afterTs, beforeTs, klines[0].Ts, klines[len(klines)-1].Ts, len(klines)) + + // store to tsdb + err = svc.exchangeDataService.SaveKlines(task.inst, klines) + if err != nil { + return + } + } + }() + } + wg.Wait() + + zlog.Infof("%d insts initial finished", len(insts)) + + // var intervals []types.Interval + // for interval := range types.SupportedIntervals { + // intervals = append(intervals, interval) + // } + // var intervalsTs = make([]int, len(intervals)) + // var index int + + // var lock sync.Mutex + // var getTask = func() (interval types.Interval, afterTs, beforeTs int64) { + // lock.Lock() + // ts := intervalsTs[index] + // if ts == -1 { + // index++ + // } + + // lock.Unlock() + // return + // } + // var finishTask = func(interval types.Interval) { + + // } + + // ctx := context.Background() + + // inst := insts[0] + // // for interval, intervalAdder := range types.SupportedIntervals { + // interval := types.Interval1h + // intervalAdder := types.SupportedIntervals[interval] + // // kvrocks get exchange+inst+interval last/ts + // tsKey := fmt.Sprintf(HistoryKlineTsKey, inst.Exchange, inst.InstId, interval) + // beforeTs, e := svc.kvdb.GetI64(ctx, tsKey) + // if e != nil { + // err = e + // return + // } + // if beforeTs == 0 { + // beforeTs = intervalAdder(KlineBefore0, -1) + // } + + // for { + // afterTs := intervalAdder(beforeTs, 101) + // klines, e := exchange.Fetcher.FetchHistoryKlines(ctx, inst.ExchangeInstId, interval, afterTs, beforeTs) + // if e != nil { + // err = e + // return + // } + // if len(klines) == 0 { + // break + // } + + // zlog.Infof("fetch interval %s %d~%d klines: ret=%d~%d, %d klines", interval, afterTs, beforeTs, klines[0].Ts, klines[len(klines)-1].Ts, len(klines)) + + // // store to tsdb + // err = svc.exchangeDataService.SaveKlines(inst, klines) + // if err != nil { + // return + // } + + // // set kvdb inst ts mark + // beforeTs = klines[0].Ts + // } + // // } + + // zlog.Infof("%s %s initial finished", inst.Exchange, inst.InstId) +} diff --git a/internal/exchange/okx/channel_kline.go b/internal/exchange/okx/channel_kline.go index 0387168..8ce59c8 100644 --- a/internal/exchange/okx/channel_kline.go +++ b/internal/exchange/okx/channel_kline.go @@ -139,10 +139,10 @@ func candleData2Klines(channelData *ChannelData[*CandleData]) (r *types.ChannelK kline.Interval = types.Interval5d case "candle1W": kline.Interval = types.Interval1w - // case "candle1M": - // kline.Interval = types.Interval1mo - // case "candle3M": - // kline.Interval = types.Interval3mo + case "candle1M": + kline.Interval = types.Interval1mo + case "candle3M": + kline.Interval = types.Interval3mo } var kinds = []*decimal.Decimal{&kline.Open, &kline.High, &kline.Low, &kline.Close, &kline.Vol, nil, &kline.VolQuote} for i := 1; i <= 7; i++ { diff --git a/internal/exchange/okx/okx_fetch.go b/internal/exchange/okx/okx_fetch.go index 9bb5d42..4030578 100644 --- a/internal/exchange/okx/okx_fetch.go +++ b/internal/exchange/okx/okx_fetch.go @@ -7,22 +7,22 @@ import ( "net/http" "net/url" "sig-pub/pkg/types" + "strconv" "strings" "time" + "github.com/bytedance/sonic" "github.com/go-resty/resty/v2" - "golang.org/x/time/rate" + "github.com/govalues/decimal" ) const ( - HttpBaseUrl = "https://www.okx.com" - KlineBefore0 int64 = 1672502400000 // k线开始数据 2023-01-01 00:00:00 GMT+8 + HttpBaseUrl = "https://www.okx.com" ) type OkxFetcher struct { - client *resty.Client - httpProxy string - historyKlineLimiter *rate.Limiter + client *resty.Client + httpProxy string } func NewOkxFetcher(httpProxy string) (f *OkxFetcher) { @@ -39,13 +39,16 @@ func NewOkxFetcher(httpProxy string) (f *OkxFetcher) { } f = &OkxFetcher{ - client: client, - httpProxy: httpProxy, - historyKlineLimiter: rate.NewLimiter(rate.Every(100*time.Millisecond), 20), // rate: 20次/2s + client: client, + httpProxy: httpProxy, } return } +func (okx *OkxFetcher) ExhcangeType() types.Exchange { + return types.ExchangeOKX +} + // FetchHistoryKlines 获取交易产品历史K线数据 // https://my.okx.com/docs-v5/zh/#order-book-trading-market-data-get-candlesticks-history // 周期区间 after > before, (after, before) @@ -58,7 +61,7 @@ func (f *OkxFetcher) FetchHistoryKlines(ctx context.Context, okxInstId string, i err = errors.New("time range zero") return } - if err = f.historyKlineLimiter.Wait(ctx); err != nil { + if err = fetchHistoryKlineLimiter.Wait(ctx); err != nil { return } @@ -87,6 +90,36 @@ func (f *OkxFetcher) FetchHistoryKlines(ctx context.Context, okxInstId string, i return } - fmt.Println(string(resp.Body())) + r := new(RespHistoryKline) + if err = sonic.Unmarshal(resp.Body(), r); err != nil { + return + } + if r.Code != "0" { + err = fmt.Errorf("code: %s, msg: %s", r.Code, r.Msg) + return + } + for _, data := range r.Data { + ts, e := strconv.ParseInt(data[0], 10, 64) + if e != nil { + err = e + return + } + kline := &types.Kline{ + Interval: interval, + Ts: ts, + Confirm: data[8] == "1", + } + var kinds = []*decimal.Decimal{&kline.Open, &kline.High, &kline.Low, &kline.Close, &kline.Vol, nil, &kline.VolQuote} + for i := 1; i <= 7; i++ { + if kinds[i-1] == nil { + continue + } + *kinds[i-1], err = decimal.Parse(data[i]) + if err != nil { + return + } + } + klines = append(klines, kline) + } return } diff --git a/internal/exchange/okx/okx_fetch_test.go b/internal/exchange/okx/okx_fetch_test.go index 7aff070..ae049e3 100644 --- a/internal/exchange/okx/okx_fetch_test.go +++ b/internal/exchange/okx/okx_fetch_test.go @@ -8,6 +8,8 @@ import ( "time" ) +var klineBefore0 int64 = 1672502400000 // k线开始数据 2023-01-01 00:00:00 GMT+8 + func TestFetchHistoryKlines(t *testing.T) { okxFetcher := NewOkxFetcher("http://192.168.1.5:7890") @@ -18,7 +20,7 @@ func TestFetchHistoryKlines(t *testing.T) { // } interval := types.Interval5m intervalAdder := types.SupportedIntervals[interval] - before := intervalAdder(KlineBefore0, -1) + before := intervalAdder(klineBefore0, -1) for range 10 { after := intervalAdder(before, 10) klines, err := okxFetcher.FetchHistoryKlines(context.Background(), "BTC-USDT", interval, after, before) @@ -36,7 +38,7 @@ func TestFetchHistoryKlines(t *testing.T) { } func TestA(t *testing.T) { - begin := time.UnixMilli(KlineBefore0) + begin := time.UnixMilli(klineBefore0) before := begin.AddDate(0, -1, 0) after := begin.AddDate(0, 3, 0) fmt.Println("before:", before.UnixMilli()) diff --git a/internal/exchange/okx/okx_limiter.go b/internal/exchange/okx/okx_limiter.go new file mode 100644 index 0000000..11c556d --- /dev/null +++ b/internal/exchange/okx/okx_limiter.go @@ -0,0 +1,12 @@ +package okx + +import ( + "time" + + "golang.org/x/time/rate" +) + +var ( + // https://my.okx.com/docs-v5/zh/#order-book-trading-market-data-get-candlesticks-history + fetchHistoryKlineLimiter = rate.NewLimiter(rate.Every(100*time.Millisecond), 2) +) diff --git a/internal/exchange/okx/okx.go b/internal/exchange/okx/okx_subscriber.go similarity index 74% rename from internal/exchange/okx/okx.go rename to internal/exchange/okx/okx_subscriber.go index d4f8261..b67b53f 100644 --- a/internal/exchange/okx/okx.go +++ b/internal/exchange/okx/okx_subscriber.go @@ -31,18 +31,18 @@ var subscribeCandles = map[types.Interval]string{ } // kline -type OkxExchange struct { +type OkxSubscriber struct { conf config.OkxExchange channelCandle *ChannelCandle // K线频道 } -func NewOkxExchange(conf config.OkxExchange) *OkxExchange { - return &OkxExchange{ +func NewOkxSubscriber(conf config.OkxExchange) *OkxSubscriber { + return &OkxSubscriber{ conf: conf, } } -func (okx *OkxExchange) Init() (err error) { +func (okx *OkxSubscriber) Init() (err error) { // TODO 多个 ChannelCandle 实例 OkxAggregate okx.channelCandle = NewChannelCandle("candle-0", okx.conf.HttpProxy) if err = okx.channelCandle.Init(); err != nil { @@ -51,20 +51,20 @@ func (okx *OkxExchange) Init() (err error) { return } -func (okx *OkxExchange) ExhcangeType() types.Exchange { +func (okx *OkxSubscriber) ExhcangeType() types.Exchange { return types.ExchangeOKX } -func (okx *OkxExchange) ConsumerKline() <-chan *types.ChannelKline { +func (okx *OkxSubscriber) ConsumerKline() <-chan *types.ChannelKline { return okx.channelCandle.Consumer() } // 订阅产品k线行情 -func (okx *OkxExchange) SubscribeKline(instIds ...string) (err error) { +func (okx *OkxSubscriber) SubscribeKline(instIds ...string) (err error) { return okx.channelCandle.Subscribe(instIds...) } // 取消订阅产品k线行情 -func (okx *OkxExchange) UnsubscribeKline(instIds ...string) (err error) { +func (okx *OkxSubscriber) UnsubscribeKline(instIds ...string) (err error) { return okx.channelCandle.Unsubscribe(instIds...) } diff --git a/internal/exchange/okx/types.go b/internal/exchange/okx/types.go index 5e2b5f5..d607f83 100644 --- a/internal/exchange/okx/types.go +++ b/internal/exchange/okx/types.go @@ -65,3 +65,10 @@ type MarketData struct { Volume float64 Timestamp time.Time } + +// RespHistoryKline 历史k线响应 +type RespHistoryKline struct { + Code string `json:"code"` + Msg string `json:"msg"` + Data [][]string `json:"data"` +} diff --git a/pkg/aside/trade_instance_client.go b/pkg/aside/trade_instance_client.go index 70ace86..68d1627 100644 --- a/pkg/aside/trade_instance_client.go +++ b/pkg/aside/trade_instance_client.go @@ -5,6 +5,7 @@ import ( "sig-pub/api/pb" "sig-pub/pkg/data/entity" "sig-pub/pkg/mapping" + "sig-pub/pkg/types" "sig-pub/pkg/utils/kvcache" "time" @@ -61,9 +62,14 @@ func (c *TradeInstanceAside) getTradeInstance0(ctx context.Context, instId strin } // ListExchangeTradeInstance 获取交易所支持的交易实例 -func (c *TradeInstanceAside) ListExchangeTradeInstance(ctx context.Context, exchange pb.Exchange) (exInsts []*entity.TradeInstanceExchange, err error) { +func (c *TradeInstanceAside) ListExchangeTradeInstance(ctx context.Context, exchange types.Exchange) (exInsts []*entity.TradeInstanceExchange, err error) { + pbExType, err := exchange.Exchange2PB() + if err != nil { + return + } + rsp, err := c.client.ListExchangeTradeInstance(ctx, &pb.ReqListExchangeTradeInstance{ - Exchanges: []pb.Exchange{exchange}, + Exchanges: []pb.Exchange{pbExType}, }) if err != nil { return diff --git a/pkg/storage/kvrocks/kvrocks.go b/pkg/storage/kvrocks/kvrocks.go new file mode 100644 index 0000000..6f2c1ff --- /dev/null +++ b/pkg/storage/kvrocks/kvrocks.go @@ -0,0 +1,53 @@ +package kvrocks + +import ( + "context" + "fmt" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +type KVRocksDB struct { + client *redis.Client +} + +func NewKVRocksDB(conf redis.Options) *KVRocksDB { + client := redis.NewClient(&conf) + return &KVRocksDB{ + client: client, + } +} + +func (db *KVRocksDB) Ping() (err error) { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _, err = db.client.Ping(ctx).Result() + if err != nil { + return fmt.Errorf("can't connect kvrocks, %v", err) + } + return +} + +func (db *KVRocksDB) DB() *redis.Client { + return db.client +} + +func (db *KVRocksDB) GetI64(ctx context.Context, key string) (v int64, err error) { + v, err = db.DB().Get(ctx, key).Int64() + if err != nil { + if err == redis.Nil { + return 0, nil + } + return 0, err + } + return v, nil +} + +// todo set async +func (db *KVRocksDB) SetI64(ctx context.Context, key string, v int64) (err error) { + vs := strconv.FormatInt(v, 10) + err = db.DB().Set(ctx, key, vs, 0).Err() + return +} diff --git a/pkg/types/exchange.go b/pkg/types/exchange.go index cdc9a84..78e9c3c 100644 --- a/pkg/types/exchange.go +++ b/pkg/types/exchange.go @@ -1,6 +1,9 @@ package types -import "sig-pub/api/pb" +import ( + "fmt" + "sig-pub/api/pb" +) type Exchange string @@ -10,14 +13,14 @@ var ( ExchangeBINANCE = Exchange(pb.Exchange_BINANCE.String()) // 币安 ) -func (ex Exchange) Exchange2PB() (pb.Exchange, bool) { +func (ex Exchange) Exchange2PB() (pb.Exchange, error) { switch ex { case ExchangeOKX: - return pb.Exchange_OKX, true + return pb.Exchange_OKX, nil case ExchangeBINANCE: - return pb.Exchange_BINANCE, true + return pb.Exchange_BINANCE, nil default: - return pb.Exchange_SIG, false + return pb.Exchange_SIG, fmt.Errorf("unknown exchange: %v", ex) } } diff --git a/pkg/types/instance.go b/pkg/types/instance.go index 8ff07cc..1279ba2 100644 --- a/pkg/types/instance.go +++ b/pkg/types/instance.go @@ -2,8 +2,10 @@ package types // TradeInstance 交易产品 type TradeInstance struct { - InstId string // 交易产品系统id - Exchange Exchange // 当前处理交易产品交易所 - PriceSz int32 // 价格精度 - QuantitySz int32 // 交易量精度 + InstId string // 交易产品系统id + PriceSz int32 // 价格精度 + QuantitySz int32 // 交易量精度 + Status int32 // 交易所交易产品状态 + ExchangeInstId string // 交易所交易产品id + Exchange Exchange // 当前处理交易产品交易所 }