|
|
|
|
@ -4,49 +4,54 @@ 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/data/entity" |
|
|
|
|
"sig-pub/pkg/types" |
|
|
|
|
"sig-pub/pkg/utils/times" |
|
|
|
|
"sig-pub/pkg/zlog" |
|
|
|
|
"sync" |
|
|
|
|
"sync/atomic" |
|
|
|
|
"time" |
|
|
|
|
|
|
|
|
|
"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[types.Exchange]*Exchange |
|
|
|
|
exchangeMap map[pb.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, |
|
|
|
|
kvdb *kvrocks.KVRocksDB, |
|
|
|
|
exchanges ...*Exchange, |
|
|
|
|
) *ExchangeGrpcServer { |
|
|
|
|
exchangeMap := make(map[types.Exchange]*Exchange) |
|
|
|
|
func NewExchangeGrpcServer(tradeInstanceAside *aside.TradeInstanceAside, exchangeDataService *ExchangeDataService, exchanges ...ExchangeSubscriber) *ExchangeGrpcServer { |
|
|
|
|
exchangeMap := make(map[pb.Exchange]*Exchange) |
|
|
|
|
for _, exchange := range exchanges { |
|
|
|
|
exchangeMap[exchange.ExType] = exchange |
|
|
|
|
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), |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
return &ExchangeGrpcServer{ |
|
|
|
|
exchangeMap: exchangeMap, |
|
|
|
|
tradeInstanceAside: tradeInstanceAside, |
|
|
|
|
exchangeDataService: exchangeDataService, |
|
|
|
|
kvdb: kvdb, |
|
|
|
|
klinePublisher: NewPublisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]](16), |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
@ -63,35 +68,18 @@ 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.ExType) |
|
|
|
|
insts, err := svc.tradeInstanceAside.ListExchangeTradeInstance(context.Background(), exchange.Type) |
|
|
|
|
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) |
|
|
|
|
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, |
|
|
|
|
HistoryMarkTs: 0, |
|
|
|
|
}) |
|
|
|
|
// 待初始化币种数据
|
|
|
|
|
if inst.Status == data.StatusProcessing { |
|
|
|
|
processingInsts = append(processingInsts, *tradeInst) |
|
|
|
|
} |
|
|
|
|
exchange.Insts[inst.ExchangeInstId] = inst |
|
|
|
|
} |
|
|
|
|
exchange.Unlock() |
|
|
|
|
|
|
|
|
|
// instIds := []string{"BTC-USDT", "DOGE-USDT-SWAP"}
|
|
|
|
|
err = exchange.Subscriber.SubscribeKline(exchangeInstIds...) |
|
|
|
|
@ -99,24 +87,16 @@ func (svc *ExchangeGrpcServer) subscribeExchanges() {
|
|
|
|
|
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) |
|
|
|
|
c := exchange.Subscriber.ConsumerKline() |
|
|
|
|
svc.consumerKline(exchange, c) |
|
|
|
|
// todo subscribe books 订单簿
|
|
|
|
|
zlog.Infof("unsubscribe exchange: %s", exchange.Type.String()) |
|
|
|
|
}(exchange) |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// consumerKline 消费交易所k线数据
|
|
|
|
|
func (svc *ExchangeGrpcServer) consumerKline(exchange *Exchange, c <-chan *types.ChannelKline) { |
|
|
|
|
exchangeType := exchange.ExType |
|
|
|
|
|
|
|
|
|
for { |
|
|
|
|
channelK, ok := <-c |
|
|
|
|
if !ok { |
|
|
|
|
@ -124,43 +104,53 @@ func (svc *ExchangeGrpcServer) consumerKline(exchange *Exchange, c <-chan *types
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 交易所 instid 转 sig-instid
|
|
|
|
|
var exInst *types.TradeInstance |
|
|
|
|
if inst, ok := exchange.Insts.Load(channelK.ExgInstId); ok && inst != nil { |
|
|
|
|
exInst = inst.Inst |
|
|
|
|
// 标记交易产品开始订阅k线时间
|
|
|
|
|
if inst.LiveMarkTs == 0 && len(channelK.Klines) > 0 { |
|
|
|
|
inst.LiveMarkTs = channelK.Klines[0].Ts |
|
|
|
|
} |
|
|
|
|
var exInst *entity.TradeInstanceExchange |
|
|
|
|
exchange.RLock() |
|
|
|
|
if inst, ok := exchange.Insts[channelK.InstId]; ok && inst != nil { |
|
|
|
|
exchange.RUnlock() |
|
|
|
|
exInst = inst |
|
|
|
|
} else { |
|
|
|
|
zlog.Errorf("unknown exchange instId: %v, %s", channelK.Exchange, channelK.ExgInstId) |
|
|
|
|
exchange.RUnlock() |
|
|
|
|
zlog.Errorf("unknown exchange instId: %v, %s", channelK.Exchange, channelK.InstId) |
|
|
|
|
continue |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// tsdb storage
|
|
|
|
|
typeInst := types.TradeInstance{ |
|
|
|
|
InstId: exInst.InstId, // channelK.InstId
|
|
|
|
|
TickSz: 0, |
|
|
|
|
MinSz: 0, |
|
|
|
|
} |
|
|
|
|
// todo 异步处理
|
|
|
|
|
err := svc.exchangeDataService.SaveKlines(typeInst, channelK.Klines) |
|
|
|
|
if err != nil { |
|
|
|
|
zlog.Errorf("kline save to tsdb error: ", err) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// publish to subscribers
|
|
|
|
|
pubMsgMap := make(map[string]*pb.StreamKline) |
|
|
|
|
|
|
|
|
|
// instId := channelK.InstId
|
|
|
|
|
pbExType, err := channelK.Exchange.Exchange2PB() |
|
|
|
|
if err != nil { |
|
|
|
|
zlog.Error(err) |
|
|
|
|
exchange, ok := channelK.Exchange.Exchange2PB() |
|
|
|
|
if !ok { |
|
|
|
|
zlog.Errorf("unknown exchange kline: %v", channelK.Exchange) |
|
|
|
|
continue |
|
|
|
|
} |
|
|
|
|
exchangeName := exchange.String() |
|
|
|
|
|
|
|
|
|
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) |
|
|
|
|
pubKey := fmt.Sprintf("/kline/%s/%s/%s/%d", exchangeName, exInst.InstId, kline.Interval, confirm) |
|
|
|
|
// todo 优化没有订阅者就跳过
|
|
|
|
|
msg, ok := pubMsgMap[pubKey] |
|
|
|
|
if !ok { |
|
|
|
|
msg = new(pb.StreamKline) |
|
|
|
|
msg.InstId = exInst.InstId |
|
|
|
|
msg.Exchange = pbExType |
|
|
|
|
msg.Exchange = exchange |
|
|
|
|
pubMsgMap[pubKey] = msg |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -168,15 +158,6 @@ func (svc *ExchangeGrpcServer) consumerKline(exchange *Exchange, c <-chan *types
|
|
|
|
|
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 |
|
|
|
|
@ -288,196 +269,3 @@ 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
|
|
|
|
|
) |
|
|
|
|
|
|
|
|
|
type initialKlineTask struct { |
|
|
|
|
inst types.TradeInstance |
|
|
|
|
interval types.Interval |
|
|
|
|
afterTs int64 |
|
|
|
|
beforeTs int64 |
|
|
|
|
times int32 // 重试次数
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
func (t initialKlineTask) 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) { |
|
|
|
|
concurrent := max(8, runtime.NumCPU()*2) |
|
|
|
|
|
|
|
|
|
// 任务 channel
|
|
|
|
|
taskCh := make(chan initialKlineTask, concurrent) |
|
|
|
|
|
|
|
|
|
// 任务生成
|
|
|
|
|
go func() { |
|
|
|
|
for _, inst := range insts { |
|
|
|
|
// 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, err := svc.kvdb.GetI64(context.Background(), tsKey) |
|
|
|
|
if err != nil { |
|
|
|
|
zlog.Error(err) |
|
|
|
|
panic(err) |
|
|
|
|
} |
|
|
|
|
if beforeTs == 0 { |
|
|
|
|
beforeTs = intervalAdder(KlineBefore0, -1) |
|
|
|
|
} |
|
|
|
|
for { |
|
|
|
|
afterTs := intervalAdder(beforeTs, 101) |
|
|
|
|
taskCh <- initialKlineTask{ |
|
|
|
|
inst: inst, |
|
|
|
|
interval: interval, |
|
|
|
|
afterTs: afterTs, |
|
|
|
|
beforeTs: beforeTs, |
|
|
|
|
times: 0, |
|
|
|
|
} |
|
|
|
|
beforeTs = intervalAdder(afterTs, -1) |
|
|
|
|
|
|
|
|
|
// 对比 ws 获取的实时k线
|
|
|
|
|
inst, ok := exchange.Insts.Load(inst.ExchangeInstId) |
|
|
|
|
if !ok { |
|
|
|
|
break |
|
|
|
|
} |
|
|
|
|
// 订阅完成
|
|
|
|
|
if inst.LiveMarkTs != 0 && beforeTs > inst.LiveMarkTs { |
|
|
|
|
// history status -> ok
|
|
|
|
|
break |
|
|
|
|
} |
|
|
|
|
if beforeTs > time.Now().UnixMilli() { |
|
|
|
|
if inst.LiveMarkTs != 0 { |
|
|
|
|
// history status -> ok
|
|
|
|
|
} else { |
|
|
|
|
// live status -> not ok
|
|
|
|
|
} |
|
|
|
|
break |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
// }
|
|
|
|
|
} |
|
|
|
|
close(taskCh) |
|
|
|
|
}() |
|
|
|
|
|
|
|
|
|
loc, _ := time.LoadLocation("Asia/Shanghai") |
|
|
|
|
|
|
|
|
|
// 任务消费器 8协程并行
|
|
|
|
|
wg := new(sync.WaitGroup) |
|
|
|
|
for range concurrent { |
|
|
|
|
wg.Add(1) |
|
|
|
|
go func() { |
|
|
|
|
defer wg.Done() |
|
|
|
|
for { |
|
|
|
|
task, ok := <-taskCh |
|
|
|
|
if !ok { |
|
|
|
|
break |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if task.times > 0 { |
|
|
|
|
zlog.Infof("retry fetch history kline task %d times: task -> %s", task.times, task.logKey()) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
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) |
|
|
|
|
// retry task
|
|
|
|
|
task.times++ |
|
|
|
|
taskCh <- task |
|
|
|
|
return |
|
|
|
|
} |
|
|
|
|
if len(klines) == 0 { |
|
|
|
|
continue |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
sts, ets := klines[0].Ts, klines[len(klines)-1].Ts |
|
|
|
|
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, afterTs, beforeTs, sts, ets, len(klines), ss, ee) |
|
|
|
|
|
|
|
|
|
// 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) |
|
|
|
|
// retry task
|
|
|
|
|
task.times++ |
|
|
|
|
taskCh <- task |
|
|
|
|
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)
|
|
|
|
|
} |
|
|
|
|
|
