diff --git a/README.md b/README.md index 610d146..dade7a3 100644 --- a/README.md +++ b/README.md @@ -48,3 +48,19 @@ k线推送 价格/交易量精度->tsdb读写存储 kline时间窗口 + +tradingview advanced-charts: + - https://cn.tradingview.com/advanced-charts/ +tradingview lightweight-charts: + - https://tradingview.github.io/lightweight-charts/ + - https://github.com/tradingview/lightweight-charts +d3js: + - https://d3js.org/ + +DXcharts: + - https://devexperts.com/dxcharts/ + - https://github.com/devexperts/dxcharts-lite + - https://financefeeds.com/zh-CN/dxcharts-%E5%9B%BE%E8%A1%A8%E7%9A%84%E5%B4%9B%E8%B5%B7%E6%88%90%E4%B8%BA%E4%BA%A4%E6%98%93%E5%B9%B3%E5%8F%B0%E9%87%91%E8%9E%8D%E7%A7%91%E6%8A%80%E7%BD%91%E7%AB%99%E7%9A%84%E7%AB%9E%E4%BA%89%E4%BC%98%E5%8A%BF/ + +golang-clash: + - https://hub.docker.com/r/dreamacro/clash diff --git a/api/exchange.proto b/api/exchange.proto index b0b85a7..07410b9 100644 --- a/api/exchange.proto +++ b/api/exchange.proto @@ -6,22 +6,42 @@ option go_package = "./pb"; // 订阅/查询/存储,交易所k线图/热度/深度等基础数据指标 service ExchangeService { - rpc SubscribeKline(stream ReqStreamSubscribeKline) - returns (stream RspStreamSubscribeKline); + // 订阅实时K线 + rpc SubscribeKline(stream ReqStreamSubscribeKline) returns (stream RspStreamSubscribeKline); + + // 获取支持的交易所列表 + rpc Exchanges(ReqExchanges) returns (RspExchanges); + + // 获取交易所交易产品状态 + rpc ExchangeInstanceState(ReqExchangeInstanceState) returns (RspExchangeInstanceState); } message ReqStreamSubscribeKline { SubscribeType subType = 1; - repeated Exchange exchanges = 2; // 交易所 + repeated ExchangeType exchanges = 2; // 交易所 repeated string instIds = 3; repeated string intervals = 4; bool onlyConfirm = 5; // 只要confirm的k线 } - message RspStreamSubscribeKline { StreamKline kline = 1; } message StreamKline { repeated Kline klines = 2; - Exchange exchange = 3; // 交易所 + ExchangeType exchange = 3; // 交易所 string instId = 4; // 交易产品id } + +message ReqExchanges { +} +message RspExchanges { + repeated ExchangeType exchanges = 1; +} + +message ReqExchangeInstanceState { + repeated string insts = 1; + bool allExchange = 2; + repeated ExchangeType exchanges = 3; // 交易所 +} +message RspExchangeInstanceState { + repeated TradeInstanceState instsState = 1; // 交易产品列表 +} diff --git a/api/indicator.proto b/api/indicator.proto index cfd27d2..7e4d512 100644 --- a/api/indicator.proto +++ b/api/indicator.proto @@ -16,7 +16,7 @@ message IndicatorSubReq { } message Indicator { - Exchange exhcange = 1; + ExchangeType exhcange = 1; string instId = 2; string indicator = 3; string sub = 4; // 指标子标题, MA5, MA10, MA20 diff --git a/api/market.proto b/api/market.proto index e4c5497..d093e3e 100644 --- a/api/market.proto +++ b/api/market.proto @@ -5,7 +5,7 @@ import "api/pub.proto"; option go_package = "./pb"; // 交易产品管理 -service Market { +service MarketService { rpc GetTradeInstance(ReqGetTradeInstance) returns (RspGetTradeInstance); rpc AddTradeInstance(ReqAddTradeInstance) returns (RspAddTradeInstance); // rpc UpdateTradeInstance(ReqUpdateTradeInstance) returns (RspUpdateTradeInstance); @@ -46,18 +46,18 @@ message RspStatusTradeInstance { message ReqListTradeInstance { int32 page = 1; int32 pageSize = 2; - Exchange exchange = 3; // 交易所 + ExchangeType exchange = 3; // 交易所 string topic = 4; string instId = 5; } message RspListTradeInstance { repeated Kline klines = 1; - Exchange exchange = 2; // 交易所 + ExchangeType exchange = 2; // 交易所 string instId = 3; // 交易产品id } message ReqListExchangeTradeInstance { - repeated Exchange exchanges = 1; + repeated ExchangeType exchanges = 1; } message RspListExchangeTradeInstance { repeated TradeInstanceExchange exchangeInsts = 1; diff --git a/api/pub.proto b/api/pub.proto index 1754fab..86a0de6 100644 --- a/api/pub.proto +++ b/api/pub.proto @@ -8,7 +8,7 @@ enum SubscribeType { UnsubscribeAll = 2; } -enum Exchange { +enum ExchangeType { SIG = 0; OKX = 1; BINANCE = 2; @@ -79,11 +79,19 @@ message TradeInstanceExchange { string exchangeInstId = 1; string instId = 2; int32 status = 3; - int32 exchange = 4; + ExchangeType exchange = 4; string updateBy = 5; int64 updateTime = 6; } +// 交易所交易产品状态 +message TradeInstanceState { + ExchangeType exchange = 1; + string instId = 2; + int32 status = 3; + string last = 4; // 最后成交价格 +} + message Kline { int64 Ts = 2; string interval = 3; // 周期 @@ -98,7 +106,7 @@ message Kline { // https://maicoin.github.io/max-websocket-docs/#/private_channels?id=snapshot message Order { - string exchange = 1; + ExchangeType exchange = 1; string symbol = 2; string id = 3; Side side = 4; diff --git a/cmd/exchange/main.go b/cmd/exchange/main.go index b1311f7..8f795c5 100644 --- a/cmd/exchange/main.go +++ b/cmd/exchange/main.go @@ -42,7 +42,7 @@ func main() { resolver := dis.Resolver() // new market grpc client - marketUrl := discovery.ConsulDialUrl(pb.Market_ServiceDesc.ServiceName) + marketUrl := discovery.ConsulDialUrl(pb.MarketService_ServiceDesc.ServiceName) marketConn, err := grpc.NewClient(marketUrl, grpc.WithResolvers(resolver), grpc.WithTransportCredentials(insecure.NewCredentials()), @@ -50,7 +50,7 @@ func main() { if err != nil { panic(err) } - marketClient := pb.NewMarketClient(marketConn) + marketClient := pb.NewMarketServiceClient(marketConn) tradeInstanceAside := aside.NewTradeInstanceAside(marketClient) // kvrocks db @@ -69,17 +69,20 @@ func main() { // tsdb VictoriaMetrics vmtsdb := vmts.NewVictoriaMetricsTSDB(conf.Tsdb.Victoriametrics) - tsdbService := exchange.NewExchangeDataService(vmtsdb, nil) + tsdbService := exchange.NewExchangeDataService(vmtsdb, kvdb) if err := tsdbService.Init(); err != nil { panic(err) } - // exhcange main service - exchangeService := exchange.NewExchangeGrpcServer(tradeInstanceAside, tsdbService, kvdb, okxExchange) + // exchange main service + exchangeService := exchange.NewExchangeService(tradeInstanceAside, tsdbService, okxExchange) if err := exchangeService.Init(); err != nil { panic(err) } - + exchangeGrpcServer := exchange.NewExchangeGrpcServer(exchangeService) + if err := exchangeGrpcServer.Init(); err != nil { + panic(err) + } grpcServer := grpc.NewServer(config.GetGrpcOptions( conf.Grpc, grpc.UnaryInterceptor(interceptor.RecoverInterceptor))..., @@ -88,15 +91,13 @@ func main() { // 注册反射服务 reflection.Register(grpcServer) } - pb.RegisterExchangeServiceServer(grpcServer, exchangeService) + pb.RegisterExchangeServiceServer(grpcServer, exchangeGrpcServer) exit.AddHook(grpcServer.GracefulStop, exit.WithOrderFront()) // consul 服务注册 register := exchangeConf.Register - if register.Name == "" { - register.Name = pb.ExchangeService_ServiceDesc.ServiceName - } + register.Name = pb.ExchangeService_ServiceDesc.ServiceName if err := dis.Registry(grpcServer, register); err != nil { panic(err) } diff --git a/cmd/indicator/main.go b/cmd/indicator/main.go index 75701a1..702f61e 100644 --- a/cmd/indicator/main.go +++ b/cmd/indicator/main.go @@ -81,7 +81,7 @@ func subscribeEvents(client pb.ExchangeServiceClient) { instIds := []string{"BTC-USDT", "DOGE-USDT-SWAP"} msg := &pb.ReqStreamSubscribeKline{ SubType: pb.SubscribeType_Subscribe, - Exchanges: []pb.Exchange{pb.Exchange_OKX}, + Exchanges: []pb.ExchangeType{pb.ExchangeType_OKX}, InstIds: instIds, Intervals: []string{ string(types.Interval1s), @@ -99,7 +99,7 @@ func subscribeEvents(client pb.ExchangeServiceClient) { <-time.After(10 * time.Second) msg := &pb.ReqStreamSubscribeKline{ SubType: pb.SubscribeType_Unsubscribe, - Exchanges: []pb.Exchange{pb.Exchange_OKX}, + Exchanges: []pb.ExchangeType{pb.ExchangeType_OKX}, InstIds: []string{"BTC-USDT"}, Intervals: []string{ string(types.Interval1s), diff --git a/cmd/market/main.go b/cmd/market/main.go index 029b40d..5065fa8 100644 --- a/cmd/market/main.go +++ b/cmd/market/main.go @@ -59,15 +59,13 @@ func main() { // 注册反射服务 reflection.Register(grpcServer) } - pb.RegisterMarketServer(grpcServer, marketGrpcServer) + pb.RegisterMarketServiceServer(grpcServer, marketGrpcServer) exit.AddHook(grpcServer.GracefulStop, exit.WithOrderFront()) // consul 服务注册 register := marketConf.Register - if register.Name == "" { - register.Name = pb.Market_ServiceDesc.ServiceName - } + register.Name = pb.MarketService_ServiceDesc.ServiceName dis := discovery.NewConsulDiscovery(client) if err := dis.Registry(grpcServer, register); err != nil { panic(err) diff --git a/config/exchange.toml b/config/exchange.toml index 44021d9..6e19a88 100644 --- a/config/exchange.toml +++ b/config/exchange.toml @@ -1,5 +1,5 @@ -grpcReflection = false # 注册grpc反射服务 +grpcReflection = true # 注册grpc反射服务 [register] nodeId = 1 # grpc服务节点id, 多实例唯一 diff --git a/go.mod b/go.mod index 16e1bb8..ae70fc4 100644 --- a/go.mod +++ b/go.mod @@ -87,7 +87,7 @@ require ( github.com/sagikazarmark/locafero v0.7.0 // indirect github.com/sourcegraph/conc v0.3.0 // indirect github.com/spf13/afero v1.12.0 // indirect - github.com/spf13/cast v1.7.1 // indirect + github.com/spf13/cast v1.10.0 // indirect github.com/spf13/pflag v1.0.6 // indirect github.com/subosito/gotenv v1.6.0 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect diff --git a/go.sum b/go.sum index e573182..54de9d6 100644 --- a/go.sum +++ b/go.sum @@ -268,6 +268,8 @@ github.com/spf13/afero v1.12.0 h1:UcOPyRBYczmFn6yvphxkn9ZEOY65cpwGKb5mL36mrqs= github.com/spf13/afero v1.12.0/go.mod h1:ZTlWwG4/ahT8W7T0WQ5uYmjI9duaLQGy3Q2OAl4sk/4= github.com/spf13/cast v1.7.1 h1:cuNEagBQEHWN1FnbGEjCXL2szYEXqfJPbP2HNUaca9Y= github.com/spf13/cast v1.7.1/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo= +github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY= +github.com/spf13/cast v1.10.0/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo= github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o= github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/spf13/viper v1.20.0 h1:zrxIyR3RQIOsarIrgL8+sAvALXul9jeEPa06Y0Ph6vY= diff --git a/internal/exchange/exchange.go b/internal/exchange/exchange.go index 184f793..1d2d407 100644 --- a/internal/exchange/exchange.go +++ b/internal/exchange/exchange.go @@ -3,15 +3,18 @@ package exchange import ( "context" "fmt" + "sig-pub/api/pb" "sig-pub/pkg/types" "sig-pub/pkg/utils/collect" - "sync" + "sync/atomic" + + "github.com/govalues/decimal" ) // 交易所行情数据订阅 type ExchangeSubscriber interface { // 交易所类型 - ExhcangeType() types.Exchange + ExhcangeType() pb.ExchangeType // 消费k线行情数据 ConsumerKline() <-chan *types.ChannelKline @@ -26,42 +29,28 @@ type ExchangeSubscriber interface { // 交易所行情数据请求 type ExchangeFetcher interface { // 交易所类型 - ExhcangeType() types.Exchange + ExhcangeType() pb.ExchangeType // 获取区间内历史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 *collect.SyncMap[string, *ExchangeTradeInstance] // map[string]*ExchangeTradeInstance // - // sync.RWMutex + ExchangeType pb.ExchangeType + Fetcher ExchangeFetcher + Subscriber ExchangeSubscriber + TradeInstIds *collect.SyncMap[string, string] // map[string]*ExchangeTradeInstance // + ExchangeInsts *collect.SyncMap[string, *ExchangeTradeInstance] // map[string]*ExchangeTradeInstance // } // 交易所交易产品 type ExchangeTradeInstance struct { Inst *types.TradeInstance - Status int32 // 交易产品状态, 0.初始化中 1.正常 - LiveMarkTs int64 // websocket订阅k线标记时间戳 - LiveKStartTs map[types.Interval]int64 // ws开始订阅k线标记时间戳 - LiveKMarkTs map[types.Interval]int64 // ws实时订阅k线(confirmed)标记时间戳 - HistoryMarkTs map[types.Interval]int64 // 拉取历史k线标记时间戳 - Lock sync.RWMutex -} - -func (exInst *ExchangeTradeInstance) GetLiveMarkTs(interval types.Interval) (ts int64) { - exInst.Lock.RLock() - ts = exInst.LiveKMarkTs[interval] - exInst.Lock.RUnlock() - return -} - -func (exInst *ExchangeTradeInstance) SetLiveMarkTs(interval types.Interval, ts int64) { - exInst.Lock.Lock() - exInst.LiveKMarkTs[interval] = ts - exInst.Lock.Unlock() + Status atomic.Int32 // 交易产品状态, 0.初始化中 1.正常 + LiveKline *types.IntervalState[types.Kline] // 实时k线数据 + LiveKStartTs *types.IntervalState[int64] // ws开始订阅k线标记时间戳 + HistoryMarkTs *types.IntervalState[int64] // 拉取历史k线标记时间戳 + Last decimal.Decimal // 交易产品实时价格tick更新 } func NewExchange(fetcher ExchangeFetcher, subscriber ExchangeSubscriber) *Exchange { @@ -71,9 +60,10 @@ func NewExchange(fetcher ExchangeFetcher, subscriber ExchangeSubscriber) *Exchan } return &Exchange{ - ExType: exType, - Fetcher: fetcher, - Subscriber: subscriber, - Insts: collect.NewSyncMap[string, *ExchangeTradeInstance](), + ExchangeType: exType, + Fetcher: fetcher, + Subscriber: subscriber, + TradeInstIds: collect.NewSyncMap[string, string](), + ExchangeInsts: collect.NewSyncMap[string, *ExchangeTradeInstance](), } } diff --git a/internal/exchange/exchange_data_persist.go b/internal/exchange/exchange_data_persist.go new file mode 100644 index 0000000..d65508e --- /dev/null +++ b/internal/exchange/exchange_data_persist.go @@ -0,0 +1,50 @@ +package exchange + +import ( + "context" + "fmt" + "sig-pub/api/pb" + "sig-pub/pkg/storage/kvrocks" + vmts "sig-pub/pkg/storage/tsdb/victoria_metrics" + "sig-pub/pkg/types" +) + +type ExchangeDataPersist struct { + vmtsdb *vmts.VictoriaMetricsTSDB + kvdb *kvrocks.KVRocksDB + // exchangeFetch ExchangeFetcher +} + +func NewExchangeDataService( + vmdb *vmts.VictoriaMetricsTSDB, + kvdb *kvrocks.KVRocksDB, +) *ExchangeDataPersist { + return &ExchangeDataPersist{ + vmtsdb: vmdb, + kvdb: kvdb, + } +} + +func (p *ExchangeDataPersist) Init() (err error) { + return +} + +func (p *ExchangeDataPersist) SaveKlines(inst types.TradeInstance, klines []*types.Kline) (err error) { + err = p.vmtsdb.SaveKlines(inst, klines) + if err != nil { + return + } + return +} + +func (p *ExchangeDataPersist) GetHistoryKlineMarkTs(exchange pb.ExchangeType, instId string, interval types.Interval) (ts int64, err error) { + tsKey := fmt.Sprintf(HistoryKlineTsKey, exchange, instId, interval) + ts, err = p.kvdb.GetI64(context.Background(), tsKey) + return +} + +func (p *ExchangeDataPersist) SaveHistoryKlineMarkTs(exchange pb.ExchangeType, instId string, interval types.Interval, ts int64) (tsKey string, err error) { + tsKey = fmt.Sprintf(HistoryKlineTsKey, exchange, instId, interval) + err = p.kvdb.SetI64(context.Background(), tsKey, ts) + return +} diff --git a/internal/exchange/exchange_data_service.go b/internal/exchange/exchange_data_service.go deleted file mode 100644 index 547021b..0000000 --- a/internal/exchange/exchange_data_service.go +++ /dev/null @@ -1,34 +0,0 @@ -package exchange - -import ( - vmts "sig-pub/pkg/storage/tsdb/victoria_metrics" - "sig-pub/pkg/types" -) - -type ExchangeDataService struct { - vmdb *vmts.VictoriaMetricsTSDB - exchangeFetch ExchangeFetcher -} - -func NewExchangeDataService( - vmdb *vmts.VictoriaMetricsTSDB, - exchangeFetch ExchangeFetcher, -) *ExchangeDataService { - return &ExchangeDataService{ - vmdb: vmdb, - exchangeFetch: exchangeFetch, - } -} - -func (svc *ExchangeDataService) Init() (err error) { - return -} - -func (svc *ExchangeDataService) SaveKlines(inst types.TradeInstance, klines []*types.Kline) (err error) { - err = svc.vmdb.SaveKlines(inst, klines) - if err != nil { - return - } - // todo klines 时间点回溯检查 - return -} diff --git a/internal/exchange/exchange_grpc_server.go b/internal/exchange/exchange_grpc_server.go index 3035ca4..eb836ca 100644 --- a/internal/exchange/exchange_grpc_server.go +++ b/internal/exchange/exchange_grpc_server.go @@ -4,203 +4,36 @@ import ( "context" "fmt" "io" - "runtime" "sig-pub/api/pb" - "sig-pub/pkg/aside" - "sig-pub/pkg/data" - "sig-pub/pkg/storage/kvrocks" - "sig-pub/pkg/types" - "sig-pub/pkg/utils/collect" - "sig-pub/pkg/utils/times" "sig-pub/pkg/zlog" - "sync" "sync/atomic" - "time" "google.golang.org/grpc" ) type ExchangeGrpcServer struct { pb.UnimplementedExchangeServiceServer - exchangeMap map[types.Exchange]*Exchange - tradeInstanceAside *aside.TradeInstanceAside - exchangeDataService *ExchangeDataService - kvdb *kvrocks.KVRocksDB + exchangeService *ExchangeService + klineStreamId int64 - klineStreamId int64 - klinePublisher *Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]] + klineSubscriber *Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]] } // exchanges: 支持的数据源交易所 -func NewExchangeGrpcServer( - tradeInstanceAside *aside.TradeInstanceAside, - exchangeDataService *ExchangeDataService, - kvdb *kvrocks.KVRocksDB, - exchanges ...*Exchange, -) *ExchangeGrpcServer { - exchangeMap := make(map[types.Exchange]*Exchange) - for _, exchange := range exchanges { - exchangeMap[exchange.ExType] = exchange - } - +func NewExchangeGrpcServer(exchangeService *ExchangeService) *ExchangeGrpcServer { return &ExchangeGrpcServer{ - exchangeMap: exchangeMap, - tradeInstanceAside: tradeInstanceAside, - exchangeDataService: exchangeDataService, - kvdb: kvdb, - klinePublisher: NewPublisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]](16), + exchangeService: exchangeService, } } -func (svc *ExchangeGrpcServer) Init() (err error) { - svc.subscribeExchanges() +func (svr *ExchangeGrpcServer) Init() (err error) { + svr.klineSubscriber = svr.exchangeService.GetKlineSubscriber() return } -// 订阅交易所推送行情 -func (svc *ExchangeGrpcServer) subscribeExchanges() { - // consumerKline - // 交易所订阅交易产品 - for _, exchange := range svc.exchangeMap { - go func(exchange *Exchange) { - // get exchange trade instances - insts, err := svc.tradeInstanceAside.ListExchangeTradeInstance(context.Background(), exchange.ExType) - if err != nil { - zlog.Error(err) - return - } - - var exchangeInstIds []string - var processingInsts []types.TradeInstance - for _, inst := range insts { - exchangeInstIds = append(exchangeInstIds, inst.ExchangeInstId) - tradeInst := &types.TradeInstance{ - InstId: inst.InstId, - Status: inst.Status, - PriceSz: 0, - QuantitySz: 0, - ExchangeInstId: inst.ExchangeInstId, - Exchange: exchange.ExType, - } - exchange.Insts.Store(inst.ExchangeInstId, &ExchangeTradeInstance{ - Inst: tradeInst, - Status: 0, - LiveMarkTs: 0, - LiveKMarkTs: make(map[types.Interval]int64), - HistoryMarkTs: make(map[types.Interval]int64), - }) - // 待初始化币种数据 - if inst.Status == data.StatusProcessing { - processingInsts = append(processingInsts, *tradeInst) - } - } - - // instIds := []string{"BTC-USDT", "DOGE-USDT-SWAP"} - err = exchange.Subscriber.SubscribeKline(exchangeInstIds...) - if err != nil { - zlog.Error(err) - return - } - - 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 := exchange.ExType - - for { - channelK, ok := <-c - if !ok { - return - } - - // 交易所 instid 转 sig-instid - var exInst *types.TradeInstance - if inst, ok := exchange.Insts.Load(channelK.ExgInstId); ok && inst != nil { - exInst = inst.Inst - // 标记交易产品开始订阅k线时间 - if len(channelK.Klines) > 0 { - kline := collect.MustMax(channelK.Klines, func(k *types.Kline) int64 { return k.Ts }) - if inst.GetLiveMarkTs(kline.Interval) == 0 { - inst.SetLiveMarkTs(kline.Interval, kline.Ts) - } - } - // confirmKlines := collect.Filter(channelK.Klines, func(_ int, k *types.Kline) bool { return k.Confirm }) - - } else { - zlog.Errorf("unknown exchange instId: %v, %s", channelK.Exchange, channelK.ExgInstId) - continue - } - - // publish to subscribers - pubMsgMap := make(map[string]*pb.StreamKline) - - // instId := channelK.InstId - pbExType, err := channelK.Exchange.Exchange2PB() - if err != nil { - zlog.Error(err) - continue - } - - var confirmKlines []*types.Kline - for _, kline := range channelK.Klines { - // zlog.Infof("recv kline: %#v", kline) - confirm := 0 - if kline.Confirm { - confirm = 1 - confirmKlines = append(confirmKlines, kline) - } - 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 = pbExType - pubMsgMap[pubKey] = msg - } - - pbk := kline.ToPBKline() - msg.Klines = append(msg.Klines, pbk) - } - - // tsdb storage - if len(confirmKlines) > 0 { - // todo 异步处理 - err := svc.exchangeDataService.SaveKlines(*exInst, confirmKlines) - if err != nil { - zlog.Errorf("kline save to tsdb error: ", err) - } - } - - for pubKey, msg := range pubMsgMap { - if len(msg.Klines) == 0 { - continue - } - subs := svc.klinePublisher.Publisher(pubKey) - for _, sub := range subs { - if err := sub.Send(&pb.RspStreamSubscribeKline{Kline: msg}); err != nil { - zlog.Error(err) - } - } - } - } -} - // SubscribeKline 订阅k线stream -func (svc *ExchangeGrpcServer) SubscribeKline(stream grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]) (err0 error) { - streamId := atomic.AddInt64(&svc.klineStreamId, 1) +func (svr *ExchangeGrpcServer) SubscribeKline(stream grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]) (err0 error) { + streamId := atomic.AddInt64(&svr.klineStreamId, 1) // subKey = /kline/exchange/instId/interval/confirm // 接收消息的goroutine @@ -228,17 +61,17 @@ func (svc *ExchangeGrpcServer) SubscribeKline(stream grpc.BidiStreamingServer[pb select { case <-stream.Context().Done(): // 客户端断开连接 - svc.klinePublisher.UnsubscribeAll(streamId) + svr.klineSubscriber.UnsubscribeAll(streamId) return stream.Context().Err() case msg, ok := <-recvChan: if !ok { // 接收通道关闭，结束流 - svc.klinePublisher.UnsubscribeAll(streamId) + svr.klineSubscriber.UnsubscribeAll(streamId) return } if msg.SubType == pb.SubscribeType_UnsubscribeAll { - svc.klinePublisher.UnsubscribeAll(streamId) + svr.klineSubscriber.UnsubscribeAll(streamId) continue } for _, exchange := range msg.Exchanges { @@ -253,9 +86,9 @@ func (svc *ExchangeGrpcServer) SubscribeKline(stream grpc.BidiStreamingServer[pb zlog.Infof("stream: %d sub: %s", streamId, subKey) switch msg.SubType { case pb.SubscribeType_Subscribe: - svc.klinePublisher.Subscribe(subKey, streamId, stream) + svr.klineSubscriber.Subscribe(subKey, streamId, stream) case pb.SubscribeType_Unsubscribe: - svc.klinePublisher.Unsubscribe(subKey, streamId) + svr.klineSubscriber.Unsubscribe(subKey, streamId) } } } @@ -263,261 +96,21 @@ func (svc *ExchangeGrpcServer) SubscribeKline(stream grpc.BidiStreamingServer[pb } } } - - // for { - // select { - // case <-ctx.Done(): - // // 客户端断开连接 - // // log.Printf("Client disconnected from topic: %s", topic) - // return - // default: - // // 模拟事件生成 - // event := &pb.SubscribeKlineStreamRsp{ - // Kline: &pb.StreamKline{ - // InstId: "DOGE/USDT", - // Exchange: pb.Exchange_OKX, - // Klines: []*pb.Kline{ - // { - // Ts: time.Now().Unix(), - // }, - // }, - // }, - // } - - // // 推送事件 - // if err := streamRsp.Send(event); err != nil { - // log.Printf("Failed to send event to client: %v", err) - // return err - // } - - // // 模拟事件间隔 - // time.Sleep(2 * time.Second) - // } - // } -} - -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 -) - -type fetchKlineTask struct { - inst types.TradeInstance - interval types.Interval - afterTs int64 - beforeTs int64 - times int32 // 重试次数 } -func (t fetchKlineTask) logKey() string { - return fmt.Sprintf("%s:%s:%s:%d:%d", t.inst.Exchange, t.inst.InstId, t.interval, t.beforeTs, t.afterTs) -} - -// initialKline 初始化交易产品历史k线数据 -func (svc *ExchangeGrpcServer) initialKlines(exchange *Exchange, insts []types.TradeInstance) { - // 记录成功和失败的交易产品 - var success, failed []types.TradeInstance - - for _, inst := range insts { - err := svc.initialTradeInstanceKlines(exchange, inst) - if err != nil { - zlog.Errorf("initial fetch trade instance error: %s(%s), err=%v", inst.InstId, inst.Exchange, err) - failed = append(failed, inst) - } else { - success = append(success, inst) - } - } - - zlog.Infof("%d insts initial finished, success %d, failed %d", len(insts), len(success), len(failed)) -} - -var ( - SingleTaskMaxFailTimes int32 = 10 -) - -// initTradeInstanceKlines 初始化交易产品历史k线数据 -func (svc *ExchangeGrpcServer) initialTradeInstanceKlines(exchange *Exchange, inst types.TradeInstance) (err error) { - // 并发数 - concurrent := max(8, runtime.NumCPU()*2) - // 任务 channel - taskCh := make(chan fetchKlineTask, concurrent) - retryTaskCh := make(chan fetchKlineTask, concurrent) - - // 发布任务数, 成功任务数, 失败任务次数 - var pubTasks, subTasks, failTasks atomic.Int32 - var pubTaskDone atomic.Bool // 所有任务已发布 - var historyMarkTs = make(map[types.Interval]int64) // 已初始化最后k线时间戳 - var historyMarkTsMu sync.Mutex - ctx, cancel := context.WithCancel(context.Background()) - - defer func() { - if err != nil { - return - } - for interval, ts := range historyMarkTs { - tsKey := fmt.Sprintf(HistoryKlineTsKey, inst.Exchange, inst.InstId, interval) - if ex := svc.kvdb.SetI64(context.Background(), tsKey, ts); ex != nil { - zlog.Errorf("history mark inititaled ts error: key=%s, ts=%d", tsKey, ts) - } - } - }() - - go func() { - defer func() { - pubTaskDone.Store(true) - // 无任务处理 - if subTasks.Load() == 0 { - cancel() - } - zlog.Infof("trade instance initial kline %s(%s), pub %d fetch tasks", inst.InstId, inst.Exchange, pubTasks.Load()) - }() - - // for interval, intervalAdder := range types.SupportedIntervals { - interval := types.Interval1d - intervalAdder := types.SupportedIntervals[interval] - // history 未补全前, history写 kvdb ts mark, 补全后 ws live 写 ts mark - tsKey := fmt.Sprintf(HistoryKlineTsKey, inst.Exchange, inst.InstId, interval) - beforeTs, ex := svc.kvdb.GetI64(context.Background(), tsKey) - if ex != nil { - err = ex - zlog.Error(err) - cancel() - return - } - if beforeTs == 0 { - beforeTs = intervalAdder(KlineBefore0, -1) - } - for { - // 对比 ws 获取的实时k线 - exchangeInst, ok := exchange.Insts.Load(inst.ExchangeInstId) - if !ok { - err = fmt.Errorf("not load exchange trade instance: %s", inst.ExchangeInstId) - cancel() - return - } - // 订阅完成 - if exchangeInst.LiveMarkTs != 0 && beforeTs >= exchangeInst.LiveMarkTs { - // history status -> ok - break - } - if beforeTs > time.Now().UnixMilli() { - if exchangeInst.LiveMarkTs != 0 { - // history status -> ok - } else { - // live status -> not ok - } - break - } - - afterTs := intervalAdder(beforeTs, 101) - task := fetchKlineTask{ - inst: inst, - interval: interval, - afterTs: afterTs, - beforeTs: beforeTs, - times: 0, - } - - // 发布任务 - select { - case taskCh <- task: - pubTasks.Add(1) - case <-ctx.Done(): - return - } - - beforeTs = intervalAdder(afterTs, -1) - } - }() - - // 任务消费器 8协程并行 - wg := new(sync.WaitGroup) - for range concurrent { - wg.Add(1) - go func() { - defer wg.Done() - - var task fetchKlineTask - for { - select { - case <-ctx.Done(): - return - case task = <-taskCh: - case task = <-retryTaskCh: - } - - if task.times > 0 { - zlog.Infof("retry fetch history kline task %d times: task -> %s", task.times, task.logKey()) - } - - if lastKlineTs, ex := svc.fetchTaskKlines(exchange, task); ex != nil { - failTasks.Add(1) - if task.times >= SingleTaskMaxFailTimes { - err = fmt.Errorf("task failed to many times %d, key: %s, err: %v", task.times, task.logKey(), ex) - cancel() - return - } - // retry task - task.times++ - select { - case retryTaskCh <- task: - case <-ctx.Done(): - return - } - } else { - // 周期任务最后kline时间 - if lastKlineTs != 0 { - historyMarkTsMu.Lock() - if ts, ok := historyMarkTs[task.interval]; ok { - if lastKlineTs > ts { - historyMarkTs[task.interval] = lastKlineTs - } - } else { - historyMarkTs[task.interval] = lastKlineTs - } - historyMarkTsMu.Unlock() - } - - zlog.Debugf("trade instance initial kline tasks processing: %s(%s), pub %d, sub %d, fail %d", inst.InstId, inst.Exchange, pubTasks.Load(), subTasks.Load(), failTasks.Load()) - // 任务都已执行成功结束 - subs := subTasks.Add(1) - if pubTaskDone.Load() && subs >= pubTasks.Load() { - zlog.Infof("trade instance initial kline tasks success finished, %s(%s), pub %d, sub %d, fail %d", inst.InstId, inst.Exchange, pubTasks.Load(), subTasks.Load(), failTasks.Load()) - cancel() - return - } - } - } - }() - } - wg.Wait() - return -} - -func (svc *ExchangeGrpcServer) fetchTaskKlines(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) +// 获取支持的交易所列表 +func (svr *ExchangeGrpcServer) Exchanges(ctx context.Context, req *pb.ReqExchanges) (rsp *pb.RspExchanges, err error) { + exchanges, err := svr.exchangeService.Exchanges() if err != nil { - zlog.Errorf("fetch history kline task error: task -> %s, err -> %v", task.logKey(), err) - return - } - if len(klines) == 0 { return } + return &pb.RspExchanges{Exchanges: exchanges}, nil +} - loc, _ := time.LoadLocation("Asia/Shanghai") - - sts, ets := klines[0].Ts, klines[len(klines)-1].Ts - 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) - - // store to tsdb - err = svc.exchangeDataService.SaveKlines(task.inst, klines) +func (svr *ExchangeGrpcServer) ExchangeInstanceState(ctx context.Context, req *pb.ReqExchangeInstanceState) (rsp *pb.RspExchangeInstanceState, err error) { + states, err := svr.exchangeService.ExchangeInstanceState(req.AllExchange, req.Exchanges, req.Insts) if err != nil { - zlog.Errorf("save history klines to tsdb error: task -> %s, err -> %v", task.logKey(), err) return } - return + return &pb.RspExchangeInstanceState{InstsState: states}, nil } diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go new file mode 100644 index 0000000..a23377e --- /dev/null +++ b/internal/exchange/exchange_service.go @@ -0,0 +1,478 @@ +package exchange + +import ( + "context" + "errors" + "fmt" + "runtime" + "sig-pub/api/pb" + "sig-pub/pkg/aside" + "sig-pub/pkg/data" + "sig-pub/pkg/types" + "sig-pub/pkg/utils/times" + "sig-pub/pkg/zlog" + "sort" + "sync" + "sync/atomic" + "time" + + "google.golang.org/grpc" +) + +// ExchangeService 交易所服务 +type ExchangeService struct { + exchangeMap map[pb.ExchangeType]*Exchange + tradeInstanceAside *aside.TradeInstanceAside + exchangeDataService *ExchangeDataPersist + + klinePublisher *Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]] +} + +// exchanges: 支持的数据源交易所 +func NewExchangeService( + tradeInstanceAside *aside.TradeInstanceAside, + exchangeDataService *ExchangeDataPersist, + exchanges ...*Exchange, +) *ExchangeService { + exchangeMap := make(map[pb.ExchangeType]*Exchange) + for _, exchange := range exchanges { + exchangeMap[exchange.ExchangeType] = exchange + } + + return &ExchangeService{ + exchangeMap: exchangeMap, + tradeInstanceAside: tradeInstanceAside, + exchangeDataService: exchangeDataService, + klinePublisher: NewPublisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]](16), + } +} + +func (svc *ExchangeService) Init() (err error) { + svc.subscribeExchanges() + return +} + +// GetKlineSubscriber 订阅k线订阅器 +func (svc *ExchangeService) GetKlineSubscriber() (subscriber *Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]]) { + subscriber = svc.klinePublisher + return +} + +// 订阅交易所推送行情 +func (svc *ExchangeService) subscribeExchanges() { + // consumerKline + // 交易所订阅交易产品 + for _, exchange := range svc.exchangeMap { + go func(exchange *Exchange) { + // get exchange trade instances + insts, err := svc.tradeInstanceAside.ListExchangeTradeInstance(context.Background(), exchange.ExchangeType) + if err != nil { + zlog.Error(err) + return + } + + var exchangeInstIds []string + var processingInsts []types.TradeInstance + for _, inst := range insts { + exchangeInstIds = append(exchangeInstIds, inst.ExchangeInstId) + tradeInst := &types.TradeInstance{ + InstId: inst.InstId, + Status: inst.Status, + PriceSz: 0, + QuantitySz: 0, + ExchangeInstId: inst.ExchangeInstId, + Exchange: exchange.ExchangeType, + } + exchange.TradeInstIds.Store(inst.InstId, inst.ExchangeInstId) + exchange.ExchangeInsts.Store(inst.ExchangeInstId, &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) + } + } + + // instIds := []string{"BTC-USDT", "DOGE-USDT-SWAP"} + err = exchange.Subscriber.SubscribeKline(exchangeInstIds...) + if err != nil { + zlog.Error(err) + return + } + + go func() { + c := exchange.Subscriber.ConsumerKline() + svc.consumerKline(exchange, c) + // todo subscribe books 订单簿 + zlog.Infof("unsubscribe exchange: %s", exchange.ExchangeType) + }() + + // 初始化k线数据 + go svc.initialKlines(exchange, processingInsts) + }(exchange) + } +} + +// consumerKline 消费交易所k线数据 +func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.ChannelKline) { + exchangeType := exchange.ExchangeType + + // publish to subscribers + pubStreamKlineMap := make(map[string]*pb.StreamKline) + + for { + clear(pubStreamKlineMap) + + channelK, ok := <-c + if !ok { + return + } + if len(channelK.Klines) == 0 { + continue + } + if len(channelK.Klines) > 1 { + sort.Slice(channelK.Klines, func(i, j int) bool { + return channelK.Klines[i].Ts < channelK.Klines[j].Ts + }) + } + firstKline, lastKline := channelK.Klines[0], channelK.Klines[len(channelK.Klines)-1] + + // 交易所 instid 转 sig-instid + var tradeInst *types.TradeInstance + exchangeInst, ok := exchange.ExchangeInsts.Load(channelK.ExgInstId) + if !ok || exchangeInst == nil || exchangeInst.Inst == nil { + zlog.Errorf("unknown exchange instId: %v, %s", channelK.Exchange, channelK.ExgInstId) + continue + } + tradeInst = exchangeInst.Inst + + // 标记交易产品开始订阅k线时间 + exchangeInst.LiveKStartTs.SetIf(firstKline.Interval, firstKline.Ts, func(old int64) bool { return old == 0 }) + + // 标记实时k线 + exchangeInst.LiveKline.Set(lastKline.Interval, *lastKline) + // 记录实时价格 + exchangeInst.Last = lastKline.Close + + var confirmKlines []*types.Kline + for _, kline := range channelK.Klines { + // zlog.Infof("recv kline: %#v", kline) + confirm := 0 + if kline.Confirm { + confirm = 1 + confirmKlines = append(confirmKlines, kline) + } + pubKey := fmt.Sprintf("/kline/%s/%s/%s/%d", exchangeType, tradeInst.InstId, kline.Interval, confirm) + msg, ok := pubStreamKlineMap[pubKey] + if !ok { + msg = new(pb.StreamKline) + msg.InstId = tradeInst.InstId + msg.Exchange = channelK.Exchange + pubStreamKlineMap[pubKey] = msg + } + pbk := kline.ToPBKline() + msg.Klines = append(msg.Klines, pbk) + } + + if len(confirmKlines) > 0 { + // tsdb storage todo 异步处理 + err := svc.exchangeDataService.SaveKlines(*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 + } + } + + // publish grpc stream klines + for pubKey, kline := range pubStreamKlineMap { + if len(kline.Klines) == 0 { + continue + } + subs := svc.klinePublisher.Publisher(pubKey) + for _, sub := range subs { + if err := sub.Send(&pb.RspStreamSubscribeKline{Kline: kline}); err != nil { + zlog.Error(err) + } + } + } + } +} + +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 + SingleKlineFetchTaskMaxFailTimes int32 = 100 // 单个k线拉取任务最大失败次数 +) + +type fetchKlineTask struct { + inst types.TradeInstance + interval types.Interval + afterTs int64 + beforeTs int64 + times int32 // 重试次数 +} + +func (t fetchKlineTask) logKey() string { + return fmt.Sprintf("%s:%s:%s:%d:%d", t.inst.Exchange, t.inst.InstId, t.interval, t.beforeTs, t.afterTs) +} + +// initialKline 初始化交易产品历史k线数据 +func (svc *ExchangeService) initialKlines(exchange *Exchange, insts []types.TradeInstance) { + // 记录成功和失败的交易产品 + var success, failed []types.TradeInstance + + for _, inst := range insts { + err := svc.initialTradeInstanceKlines(exchange, inst) + if err != nil { + zlog.Errorf("initial fetch trade instance error: %s(%s), err=%v", inst.InstId, inst.Exchange, err) + failed = append(failed, inst) + } else { + success = append(success, inst) + } + } + + zlog.Infof("%d insts initial finished, success %d, failed %d", len(insts), len(success), len(failed)) +} + +// initTradeInstanceKlines 初始化交易产品历史k线数据 +func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, tradeInst types.TradeInstance) (err error) { + exchangeInst, ok := exchange.ExchangeInsts.Load(tradeInst.ExchangeInstId) + if !ok { + err = fmt.Errorf("not load exchange trade instance: %s", tradeInst.ExchangeInstId) + return + } + + // 并发数 + concurrent := max(8, runtime.NumCPU()*2) + // 任务 channel + taskCh := make(chan fetchKlineTask, concurrent) + retryTaskCh := make(chan fetchKlineTask, concurrent) + + // 发布任务数, 成功任务数, 失败任务次数 + var pubTasks, subTasks, failTasks atomic.Int32 + var pubTaskDone atomic.Bool // 所有任务已发布 + ctx, cancel := context.WithCancel(context.Background()) + + defer func() { + if err != nil { + // 交易所k线初始化失败 + exchangeInst.Status.Store(int32(data.StatusFailed)) + return + } + exchangeInst.HistoryMarkTs.Range(func(_ int, interval types.Interval, ts int64) { + tsKey, ex := svc.exchangeDataService.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) + } + }) + exchangeInst.Status.Store(int32(data.StatusOk)) + }() + + go func() { + defer func() { + pubTaskDone.Store(true) + // 无任务处理 + if subTasks.Load() == 0 { + cancel() + } + zlog.Infof("trade instance initial kline %s(%s), pub %d fetch tasks", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load()) + }() + + for interval, intervalAdder := range types.SupportedIntervals { + // 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) + if ex != nil { + err = ex + zlog.Error(err) + cancel() + return + } + if beforeTs == 0 { + beforeTs = intervalAdder(KlineBefore0, -1) + } + + exchangeInst.HistoryMarkTs.Set(interval, beforeTs) + for { + // 判定订阅任务发布完成 + liveStartTs := exchangeInst.LiveKStartTs.Get(interval) + if liveStartTs != 0 && beforeTs >= liveStartTs { + break + } + if beforeTs > time.Now().UnixMilli() { + break + } + + afterTs := intervalAdder(beforeTs, 101) + task := fetchKlineTask{ + inst: tradeInst, + interval: interval, + afterTs: afterTs, + beforeTs: beforeTs, + times: 0, + } + + // 发布任务 + select { + case taskCh <- task: + pubTasks.Add(1) + case <-ctx.Done(): + return + } + + beforeTs = intervalAdder(afterTs, -1) + } + } + }() + + // 任务消费器 多协程并行 + wg := new(sync.WaitGroup) + for range concurrent { + wg.Add(1) + go func() { + defer wg.Done() + + var task fetchKlineTask + for { + select { + case <-ctx.Done(): + return + case task = <-taskCh: + case task = <-retryTaskCh: + } + + if task.times > 0 { + zlog.Infof("retry fetch history kline task %d times: task -> %s", task.times, task.logKey()) + } + + if lastKlineTs, ex := svc.fetchTaskKlines(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) + cancel() + return + } + // retry task + task.times++ + select { + case retryTaskCh <- task: + case <-ctx.Done(): + return + } + } else { + // 周期任务最后kline时间 + if lastKlineTs != 0 { + 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() + } + + 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) + 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() + return + } + } + } + }() + } + wg.Wait() + return +} + +func (svc *ExchangeService) fetchTaskKlines(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 { + zlog.Errorf("fetch history kline task error: task -> %s, err -> %v", task.logKey(), err) + return + } + if len(klines) == 0 { + return + } + + loc, _ := time.LoadLocation("Asia/Shanghai") + + sts, ets := klines[0].Ts, klines[len(klines)-1].Ts + 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) + + // store to tsdb + err = svc.exchangeDataService.SaveKlines(task.inst, klines) + if err != nil { + zlog.Errorf("save history klines to tsdb error: task -> %s, err -> %v", task.logKey(), err) + return + } + return +} + +// Exchanges 支持的交易所列表 +func (svc *ExchangeService) Exchanges() (exchanges []pb.ExchangeType, err error) { + for exchange := range svc.exchangeMap { + exchanges = append(exchanges, exchange) + } + return +} + +// ExchangeInstanceState 交易所交易产品状态 +func (svc *ExchangeService) ExchangeInstanceState(allExchange bool, exchangeTypes []pb.ExchangeType, instIds []string) (states []*pb.TradeInstanceState, err error) { + var exchanges []*Exchange + if allExchange { + for _, exg := range svc.exchangeMap { + exchanges = append(exchanges, exg) + } + } else { + for _, exchangeType := range exchangeTypes { + exg, ok := svc.exchangeMap[exchangeType] + if !ok { + err = fmt.Errorf("not support exchange: %v", exchangeType) + return + } + exchanges = append(exchanges, exg) + } + } + if len(exchanges) == 0 { + err = errors.New("no support exchanges") + return + } + + for _, exchange := range exchanges { + for _, instId := range instIds { + // trade instId to exchangeInstId + exchangeInstId, ok := exchange.TradeInstIds.Load(instId) + if !ok { + continue + } + inst, ok := exchange.ExchangeInsts.Load(exchangeInstId) + if !ok { + continue + } + + state := &pb.TradeInstanceState{ + Exchange: exchange.ExchangeType, + InstId: instId, + Last: inst.Last.String(), + } + states = append(states, state) + } + } + return +} diff --git a/internal/exchange/okx/channel_kline.go b/internal/exchange/okx/channel_kline.go index 677d33a..68187f9 100644 --- a/internal/exchange/okx/channel_kline.go +++ b/internal/exchange/okx/channel_kline.go @@ -2,6 +2,7 @@ package okx import ( "fmt" + "sig-pub/api/pb" "sig-pub/pkg/types" "strconv" @@ -87,7 +88,7 @@ func (c *ChannelCandle) GetSubscribes() (instIds []string) { func candleData2Klines(channelData *ChannelData[*CandleData]) (r *types.ChannelKline, err error) { r = &types.ChannelKline{ ExgInstId: channelData.InstId, - Exchange: types.ExchangeOKX, + Exchange: pb.ExchangeType_OKX, } for _, data := range *channelData.Data { ts, e := strconv.ParseInt(data[0], 10, 64) diff --git a/internal/exchange/okx/okx_fetch.go b/internal/exchange/okx/okx_fetch.go index 53f0ae8..10f4980 100644 --- a/internal/exchange/okx/okx_fetch.go +++ b/internal/exchange/okx/okx_fetch.go @@ -6,6 +6,7 @@ import ( "fmt" "net/http" "net/url" + "sig-pub/api/pb" "sig-pub/pkg/types" "strconv" "strings" @@ -45,8 +46,8 @@ func NewOkxFetcher(httpProxy string) (f *OkxFetcher) { return } -func (okx *OkxFetcher) ExhcangeType() types.Exchange { - return types.ExchangeOKX +func (okx *OkxFetcher) ExhcangeType() pb.ExchangeType { + return pb.ExchangeType_OKX } // FetchHistoryKlines 获取交易产品历史K线数据 diff --git a/internal/exchange/okx/okx_subscriber.go b/internal/exchange/okx/okx_subscriber.go index b67b53f..139108b 100644 --- a/internal/exchange/okx/okx_subscriber.go +++ b/internal/exchange/okx/okx_subscriber.go @@ -1,6 +1,7 @@ package okx import ( + "sig-pub/api/pb" "sig-pub/pkg/config" "sig-pub/pkg/types" ) @@ -51,8 +52,8 @@ func (okx *OkxSubscriber) Init() (err error) { return } -func (okx *OkxSubscriber) ExhcangeType() types.Exchange { - return types.ExchangeOKX +func (okx *OkxSubscriber) ExhcangeType() pb.ExchangeType { + return pb.ExchangeType_OKX } func (okx *OkxSubscriber) ConsumerKline() <-chan *types.ChannelKline { diff --git a/internal/indicator/indicator.go b/internal/indicator/indicator.go index a543720..d1463fe 100644 --- a/internal/indicator/indicator.go +++ b/internal/indicator/indicator.go @@ -1,3 +1,20 @@ package indicator // load indicator plugin + +// 热指标 自动加载/实时更新/内存缓存 +// 历史指标 实时计算 +// 自定义插件化指标 + +type IndicatorService struct { +} + +func NewIndicatorService() *IndicatorService { + return &IndicatorService{} +} + +// 加载热指标 +// 订阅k线数据 更新指标 +func (svc *IndicatorService) Init() { + +} diff --git a/internal/market/market_grpc_server.go b/internal/market/market_grpc_server.go index 07a19e3..45f68dc 100644 --- a/internal/market/market_grpc_server.go +++ b/internal/market/market_grpc_server.go @@ -9,7 +9,7 @@ import ( ) type MarketGrpcServer struct { - pb.UnimplementedMarketServer + pb.UnimplementedMarketServiceServer tradeInstanceService *TradeInstanceService } diff --git a/internal/market/trade_instance_service.go b/internal/market/trade_instance_service.go index 61c13d5..97537af 100644 --- a/internal/market/trade_instance_service.go +++ b/internal/market/trade_instance_service.go @@ -140,7 +140,7 @@ func (s *TradeInstanceService) UpdateInstanceStatus(inst *args.UpdateTradeInstan func (s *TradeInstanceService) ListExchangeTradeInstance(exchanges []int32) (exchangesInsts []*entity.TradeInstanceExchange, err error) { err = s.db.Select(&exchangesInsts, ` select * from t_trade_instance_exchange where exchange in ? and status in ? order by inst_id, exchange - `, exchanges, []int{data.StatusOk, data.StatusProcessing}) + `, exchanges, []data.Status{data.StatusOk, data.StatusProcessing}) if err != nil { return } diff --git a/internal/market/validation.go b/internal/market/validation.go index 6952246..bec8cbe 100644 --- a/internal/market/validation.go +++ b/internal/market/validation.go @@ -5,7 +5,6 @@ import ( "fmt" "sig-pub/api/pb" "sig-pub/pkg/data/entity" - "sig-pub/pkg/types" "sig-pub/pkg/utils/validator" ) @@ -17,12 +16,13 @@ func validateTradeInstance(inst *entity.TradeInstance) (err error) { if validator.IsAnyBlank(inst.InstId, inst.InstPair, inst.InstCoin) { return errors.New("InstId InstPair InstCoin can't be blank") } - // todo validate others + // 交易所 for _, ex := range inst.Exchanges { - _, ok := types.ExchangePBParse(pb.Exchange(ex.Exchange)) - if !ok { + _, ok := pb.ExchangeType_name[ex.Exchange] + if !ok || ex.Exchange == 0 { return fmt.Errorf("exchange %d not valid", ex.Exchange) } } + // todo validate others return } diff --git a/internal/sig/sig_server.go b/internal/sig/sig_server.go index bdc5ab7..70ac1de 100644 --- a/internal/sig/sig_server.go +++ b/internal/sig/sig_server.go @@ -11,6 +11,7 @@ import ( "sig-pub/pkg/resp" "sig-pub/pkg/utils/strs" "sig-pub/pkg/zlog" + "strings" "time" "github.com/gin-gonic/gin" @@ -61,14 +62,16 @@ func (s *SigServer) Run(addr string) (err error) { } func (s *SigServer) handleGrpcGenericCall(c *gin.Context) { - svc := strs.UpperInitialLetter(c.Param("svr")) + svr := strs.UpperInitialLetter(c.Param("svr")) method := strs.UpperInitialLetter(c.Param("method")) - if svc == "" || method == "" { + if svr == "" || method == "" { c.JSON(http.StatusBadRequest, resp.Error("service not found")) return } - + if !strings.HasSuffix(svr, "Service") { + svr += "Service" + } // todo service white list // get request body @@ -80,7 +83,7 @@ func (s *SigServer) handleGrpcGenericCall(c *gin.Context) { ctx1, cancel := context.WithTimeout(context.Background(), time.Second*5) defer cancel() - grpcGenericClient, err := s.grpcGenericClientFactory.GetClient(ctx1, svc) + grpcGenericClient, err := s.grpcGenericClientFactory.GetClient(ctx1, svr) if err != nil { zlog.Error(err) c.JSON(http.StatusForbidden, resp.Error(err.Error())) diff --git a/pkg/aside/trade_instance_client.go b/pkg/aside/trade_instance_client.go index 68d1627..da660dd 100644 --- a/pkg/aside/trade_instance_client.go +++ b/pkg/aside/trade_instance_client.go @@ -5,7 +5,6 @@ import ( "sig-pub/api/pb" "sig-pub/pkg/data/entity" "sig-pub/pkg/mapping" - "sig-pub/pkg/types" "sig-pub/pkg/utils/kvcache" "time" @@ -17,14 +16,14 @@ import ( // todo nats 更新监听 更新缓存 // grpc trade instance client type TradeInstanceAside struct { - client pb.MarketClient - cache *kvcache.KVCache[*entity.TradeInstance] - cacheSf singleflight.Group + marketClient pb.MarketServiceClient + cache *kvcache.KVCache[*entity.TradeInstance] + cacheSf singleflight.Group } -func NewTradeInstanceAside(client pb.MarketClient) *TradeInstanceAside { +func NewTradeInstanceAside(marketClient pb.MarketServiceClient) *TradeInstanceAside { return &TradeInstanceAside{ - client: client, + marketClient: marketClient, cache: kvcache.NewExpireStore[*entity.TradeInstance]( time.Minute, cache.WithShards(16), @@ -52,7 +51,7 @@ func (c *TradeInstanceAside) getTradeInstance0(ctx context.Context, instId strin return } // grpc 获取 - rsp, err := c.client.GetTradeInstance(ctx, &pb.ReqGetTradeInstance{InstId: instId}) + rsp, err := c.marketClient.GetTradeInstance(ctx, &pb.ReqGetTradeInstance{InstId: instId}) if err != nil { return } @@ -62,14 +61,9 @@ func (c *TradeInstanceAside) getTradeInstance0(ctx context.Context, instId strin } // ListExchangeTradeInstance 获取交易所支持的交易实例 -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{pbExType}, +func (c *TradeInstanceAside) ListExchangeTradeInstance(ctx context.Context, exchange pb.ExchangeType) (exInsts []*entity.TradeInstanceExchange, err error) { + rsp, err := c.marketClient.ListExchangeTradeInstance(ctx, &pb.ReqListExchangeTradeInstance{ + Exchanges: []pb.ExchangeType{exchange}, }) if err != nil { return diff --git a/pkg/data/common.go b/pkg/data/common.go index 4286c4c..f6dcdc6 100644 --- a/pkg/data/common.go +++ b/pkg/data/common.go @@ -6,10 +6,11 @@ import "errors" type Status int32 const ( - StatusDisabled = 0 - StatusOk = 1 - StatusProcessing = 2 - StatusDeleted = 4 + StatusDisabled Status = 0 + StatusOk Status = 1 + StatusProcessing Status = 2 + StatusDeleted Status = 4 + StatusFailed Status = 5 ) var ( diff --git a/pkg/indicator/rsi.go b/pkg/indicator/rsi.go new file mode 100644 index 0000000..6e6ba3e --- /dev/null +++ b/pkg/indicator/rsi.go @@ -0,0 +1,44 @@ +package indicator + +import ( + "fmt" + "sig-pub/pkg/types" + "sig-pub/pkg/types/series" + + "github.com/spf13/cast" +) + +// RSI: 相对强弱指数 (RSI) +// rsi define: https://www.investopedia.com/terms/r/rsi.asp +type RSI struct { + series.Series + values series.Floats + prices series.Floats + + baseline int32 +} + +func NewRSI(baseline int32) *RSI { + return &RSI{ + baseline: baseline, + } +} + +func (ind *RSI) Init(prams map[string]any) { + cast.ToIntE("1") +} + +func (ind *RSI) Update(klines []types.Kline) (err error) { + for _, kline := range klines { + c, ok := kline.Close.Float64() + if !ok { + err = fmt.Errorf("kline close to float64 error: %s", kline.Close.String()) + return + } + ind.prices.Push(c) + } + + diff := ind.prices.Diff() + _ = diff + return +} diff --git a/pkg/mapping/market.go b/pkg/mapping/market.go index 366be1c..e733f72 100644 --- a/pkg/mapping/market.go +++ b/pkg/mapping/market.go @@ -32,7 +32,7 @@ func ExchangeInstance2Proto(exInst *entity.TradeInstanceExchange) (pbExInst *pb. ExchangeInstId: exInst.ExchangeInstId, InstId: exInst.InstId, Status: exInst.Status, - Exchange: exInst.Exchange, + Exchange: pb.ExchangeType(exInst.Exchange), UpdateBy: exInst.UpdateBy, UpdateTime: exInst.UpdateTime, } @@ -66,7 +66,7 @@ func Proto2ExchangeTradeInstance(pbExInst *pb.TradeInstanceExchange) (exInst *en ExchangeInstId: pbExInst.ExchangeInstId, InstId: pbExInst.InstId, Status: pbExInst.Status, - Exchange: pbExInst.Exchange, + Exchange: int32(pbExInst.Exchange), UpdateBy: pbExInst.UpdateBy, UpdateTime: pbExInst.UpdateTime, } diff --git a/pkg/storage/tsdb/victoria_metrics/metric.go b/pkg/storage/tsdb/victoria_metrics/metric.go index b40e192..51aab3d 100644 --- a/pkg/storage/tsdb/victoria_metrics/metric.go +++ b/pkg/storage/tsdb/victoria_metrics/metric.go @@ -55,12 +55,12 @@ func Kline2Metrics(inst types.TradeInstance, klines []*types.Kline) (metrics []* 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", string(inst.Exchange), "kind", "open"), - NewMetric(inst.InstId, "interval", string(kline.Interval), "exchange", string(inst.Exchange), "kind", "high"), - NewMetric(inst.InstId, "interval", string(kline.Interval), "exchange", string(inst.Exchange), "kind", "low"), - NewMetric(inst.InstId, "interval", string(kline.Interval), "exchange", string(inst.Exchange), "kind", "close"), - NewMetric(inst.InstId, "interval", string(kline.Interval), "exchange", string(inst.Exchange), "kind", "vol"), - NewMetric(inst.InstId, "interval", string(kline.Interval), "exchange", string(inst.Exchange), "kind", "volQuote"), + 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) { diff --git a/pkg/storage/tsdb/victoria_metrics/vm.go b/pkg/storage/tsdb/victoria_metrics/vm.go index 7ec2eb2..909709d 100644 --- a/pkg/storage/tsdb/victoria_metrics/vm.go +++ b/pkg/storage/tsdb/victoria_metrics/vm.go @@ -100,7 +100,7 @@ func (vm *VictoriaMetricsTSDB) GetRangeKline(inst types.TradeInstance, interval return } - match := fmt.Sprintf("%s{interval=\"%s\"}", inst.InstId, interval) + match := fmt.Sprintf("%s{exchange=\"%s\", interval=\"%s\"}", inst.InstId, inst.Exchange.String(), interval) params := fmt.Sprintf("start=%d&end=%d&match[]=%s", start, end, url.QueryEscape(match)) resp, err := http.Get(fmt.Sprintf("%s/api/v1/export?%s", vm.addr, params)) if err != nil { diff --git a/pkg/storage/tsdb/victoria_metrics/vm_test.go b/pkg/storage/tsdb/victoria_metrics/vm_test.go index 6f5b8b2..dc05f45 100644 --- a/pkg/storage/tsdb/victoria_metrics/vm_test.go +++ b/pkg/storage/tsdb/victoria_metrics/vm_test.go @@ -1,6 +1,7 @@ package vmts import ( + "sig-pub/api/pb" "sig-pub/pkg/config" "sig-pub/pkg/types" "testing" @@ -13,7 +14,7 @@ func TestGetRangeKline(t *testing.T) { vmdb := NewVictoriaMetricsTSDB(config.VictoriaMetricsConfig{Addr: test_addr}) inst := types.TradeInstance{ InstId: "BTC_USDT", - Exchange: types.ExchangeOKX, + Exchange: pb.ExchangeType_OKX, } vmdb.GetRangeKline(inst, types.Interval1m, 1753368047691, time.Now().UnixMilli()) diff --git a/pkg/types/exchange.go b/pkg/types/exchange.go deleted file mode 100644 index 78e9c3c..0000000 --- a/pkg/types/exchange.go +++ /dev/null @@ -1,36 +0,0 @@ -package types - -import ( - "fmt" - "sig-pub/api/pb" -) - -type Exchange string - -var ( - ExchangeSIG = Exchange(pb.Exchange_SIG.String()) - ExchangeOKX = Exchange(pb.Exchange_OKX.String()) // okx - ExchangeBINANCE = Exchange(pb.Exchange_BINANCE.String()) // 币安 -) - -func (ex Exchange) Exchange2PB() (pb.Exchange, error) { - switch ex { - case ExchangeOKX: - return pb.Exchange_OKX, nil - case ExchangeBINANCE: - return pb.Exchange_BINANCE, nil - default: - return pb.Exchange_SIG, fmt.Errorf("unknown exchange: %v", ex) - } -} - -func ExchangePBParse(pbex pb.Exchange) (Exchange, bool) { - switch pbex { - case pb.Exchange_OKX: - return ExchangeOKX, true - case pb.Exchange_BINANCE: - return ExchangeBINANCE, true - default: - return ExchangeSIG, false - } -} diff --git a/pkg/types/indicator.go b/pkg/types/indicator.go index faac4d5..2b6d412 100644 --- a/pkg/types/indicator.go +++ b/pkg/types/indicator.go @@ -15,10 +15,10 @@ type Indicator interface { Meta() IndicatorMeta // init id, 计算完成 publish 时使用id, // exchange: 封装indicator访问, 封装历史k线访问 - // args: 动态指标参数, 执行时创建 + // args: 动态指标参数定义, 执行时创建 Init(indId int64, exchange any, args map[string]any) Dependes() []IndicatorDef // 需要订阅的指标列表(包括k线, 实时k线/关闭k线) - OnKline(Kline) // 驱动k线数据, 待驱动k线到达后, 再待subscribe计算完成后执行 + OnKline([]Kline) // 驱动k线数据, 待驱动k线到达后, 再待subscribe计算完成后执行 Emit(func(klineTs int64, indicators map[string]decimal.Decimal)) // 指标计算完成后发送, 由指标执行器进行存储或分发 } diff --git a/pkg/types/instance.go b/pkg/types/instance.go index 1279ba2..2e0bbf2 100644 --- a/pkg/types/instance.go +++ b/pkg/types/instance.go @@ -1,11 +1,13 @@ package types +import "sig-pub/api/pb" + // TradeInstance 交易产品 type TradeInstance struct { - InstId string // 交易产品系统id - PriceSz int32 // 价格精度 - QuantitySz int32 // 交易量精度 - Status int32 // 交易所交易产品状态 - ExchangeInstId string // 交易所交易产品id - Exchange Exchange // 当前处理交易产品交易所 + InstId string // 交易产品系统id + PriceSz int32 // 价格精度 + QuantitySz int32 // 交易量精度 + Status int32 // 交易所交易产品状态 + ExchangeInstId string // 交易所交易产品id + Exchange pb.ExchangeType // 当前处理交易产品交易所 } diff --git a/pkg/types/interval.go b/pkg/types/interval.go index 9be8276..70e8c56 100644 --- a/pkg/types/interval.go +++ b/pkg/types/interval.go @@ -1,6 +1,8 @@ package types import ( + "sig-pub/pkg/zlog" + "sort" "time" ) @@ -18,40 +20,7 @@ func (i Interval) AddMul(ts, mul int64) (int64, bool) { return c(ts, mul), true } -// Iota 返回interval唯一索引 -func (i Interval) Iota(ts, mul int64) (int, bool) { - index, ok := IntervalIotas[i] - if !ok { - return 0, false - } - return index, true -} - -// func (i Interval) Minutes() (int64, bool) { -// c, ok := SupportedIntervals[i] -// if !ok || c <= 0 { -// return c, false -// } -// return c / 60, true -// } - -// func (i Interval) Seconds() (int64, bool) { -// m, ok := SupportedIntervals[i] -// if !ok || m <= 0 { -// return m, false -// } -// return m, true -// } - -// func (i Interval) Milliseconds() (int64, bool) { -// m, ok := SupportedIntervals[i] -// if !ok || m <= 0 { -// return m, false -// } -// return m * 1000, true -// } - -var ( +const ( Interval1s = Interval("1s") Interval1m = Interval("1m") Interval3m = Interval("3m") @@ -109,24 +78,65 @@ var SupportedIntervals = IntervalMap{ Interval3mo: func(ts, mul int64) (ret int64) { return time.UnixMilli(ts).AddDate(0, int(3*mul), 0).UnixMilli() }, } -type IntervalIotaMap map[Interval]int - -var IntervalIotas = IntervalIotaMap{ - Interval1m: 2, - Interval3m: 3, - Interval5m: 4, - Interval15m: 5, - Interval30m: 6, - Interval1h: 7, - Interval2h: 8, - Interval4h: 9, - Interval6h: 10, - Interval12h: 11, - Interval1d: 12, - Interval2d: 13, - Interval3d: 14, - Interval5d: 15, - Interval1w: 16, - Interval1mo: 17, - Interval3mo: 18, +var ( + iotasIntervals []Interval + intervalIotaMax int + intervalIotas = map[Interval]int{} +) + +func init() { + iotasIntervals = make([]Interval, len(SupportedIntervals)) + var tss = make([]int64, len(SupportedIntervals)) + var i = 0 + for interval, adder := range SupportedIntervals { + iotasIntervals[i] = interval + tss[i] = adder(0, 1) + i++ + } + sort.Slice(iotasIntervals, func(i, j int) bool { + return tss[i] < tss[j] + }) + for i, interval := range iotasIntervals { + index := i + 1 // 0保留 + intervalIotas[interval] = index + intervalIotaMax = max(intervalIotaMax, index) + } + zlog.Debugf("init intervals: %#v", iotasIntervals) + zlog.Debugf("init intervalsIotas: %#v", intervalIotas) +} + +type IntervalState[T any] struct { + state []T +} + +func NewIntervalState[T any]() *IntervalState[T] { + return &IntervalState[T]{ + state: make([]T, intervalIotaMax+1), + } +} + +func (s *IntervalState[T]) Get(interval Interval) T { + i := intervalIotas[interval] + return s.state[i] +} + +func (s *IntervalState[T]) Set(interval Interval, v T) { + i := intervalIotas[interval] + s.state[i] = v +} + +func (s *IntervalState[T]) Range(f func(i int, interval Interval, v T)) { + for i, interval := range iotasIntervals { + index := i + 1 // 0保留 + v := s.state[index] + f(i, interval, v) + } +} + +func (s *IntervalState[T]) SetIf(interval Interval, v T, cond func(old T) bool) { + i := intervalIotas[interval] + old := s.state[i] + if cond(old) { + s.state[i] = v + } } diff --git a/pkg/types/kline.go b/pkg/types/kline.go index de55cf9..d1fb04f 100644 --- a/pkg/types/kline.go +++ b/pkg/types/kline.go @@ -25,7 +25,7 @@ type Kline struct { // okx.CandleData{[]string{"1746842757000", "0.20545", "0.20545", "0.2054", "0.2054", "7.57", "7570", "1555.2542", "1"}} -func (k *Kline) ParsePBKline(exchange pb.Exchange, kline *pb.Kline) { +func (k *Kline) ParsePBKline(exchange pb.ExchangeType, kline *pb.Kline) { // k.Exchange = exchange.String() k.Interval = Interval(kline.Interval) k.Ts = kline.Ts @@ -55,7 +55,7 @@ func (k *Kline) ToPBKline() (kline *pb.Kline) { // ChannelKline k线订阅消息 type ChannelKline struct { - ExgInstId string `json:"instId"` // 交易所交易产品id，如 BTC_USDT_SWAP - Exchange Exchange `json:"exchange"` // 交易所 - Klines []*Kline `json:"klines"` + ExgInstId string `json:"instId"` // 交易所交易产品id，如 BTC_USDT_SWAP + Exchange pb.ExchangeType `json:"exchange"` // 交易所 + Klines []*Kline `json:"klines"` } diff --git a/pkg/types/series/decimals.go b/pkg/types/series/decimals.go new file mode 100644 index 0000000..d94329c --- /dev/null +++ b/pkg/types/series/decimals.go @@ -0,0 +1,33 @@ +package series + +import "github.com/govalues/decimal" + +// 值列表 +type Decimals []decimal.Decimal + +func NewDecimals(a ...decimal.Decimal) Decimals { + return Decimals(a) +} + +func (s *Decimals) Push(v decimal.Decimal) { + *s = append(*s, v) +} + +func (s *Decimals) Append(vs ...decimal.Decimal) { + *s = append(*s, vs...) +} + +func (s Decimals) Diff() (values Decimals, err error) { + var r decimal.Decimal + for i, v := range s { + if i == 0 { + values.Push(decimal.Zero) + continue + } + if r, err = v.Sub(s[i-1]); err != nil { + return + } + values.Push(r) + } + return +} diff --git a/pkg/types/series/floats.go b/pkg/types/series/floats.go new file mode 100644 index 0000000..fd1a97f --- /dev/null +++ b/pkg/types/series/floats.go @@ -0,0 +1,26 @@ +package series + +type Floats []float64 + +func NewFloats(a ...float64) Floats { + return Floats(a) +} + +func (s *Floats) Push(v float64) { + *s = append(*s, v) +} + +func (s *Floats) Append(vs ...float64) { + *s = append(*s, vs...) +} + +func (s Floats) Diff() (values Floats) { + for i, v := range s { + if i == 0 { + values.Push(0) + continue + } + values.Push(v - s[i-1]) + } + return values +} diff --git a/pkg/types/series/series.go b/pkg/types/series/series.go new file mode 100644 index 0000000..bca78b6 --- /dev/null +++ b/pkg/types/series/series.go @@ -0,0 +1,10 @@ +package series + +type Series interface { + Last(i int) float64 + Index(i int) float64 + Length() int +} + +type Series0 struct { +}