You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
483 lines
13 KiB
483 lines
13 KiB
package exchange |
|
|
|
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/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 |
|
|
|
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) |
|
for _, exchange := range exchanges { |
|
exchangeMap[exchange.ExType] = exchange |
|
} |
|
|
|
return &ExchangeGrpcServer{ |
|
exchangeMap: exchangeMap, |
|
tradeInstanceAside: tradeInstanceAside, |
|
exchangeDataService: exchangeDataService, |
|
kvdb: kvdb, |
|
klinePublisher: NewPublisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]](16), |
|
} |
|
} |
|
|
|
func (svc *ExchangeGrpcServer) Init() (err error) { |
|
svc.subscribeExchanges() |
|
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, |
|
HistoryMarkTs: 0, |
|
}) |
|
// 待初始化币种数据 |
|
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 inst.LiveMarkTs == 0 && len(channelK.Klines) > 0 { |
|
inst.LiveMarkTs = channelK.Klines[0].Ts |
|
} |
|
} 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) |
|
// subKey = /kline/exchange/instId/interval/confirm |
|
|
|
// 接收消息的goroutine |
|
recvChan := make(chan *pb.ReqStreamSubscribeKline) |
|
go func() { |
|
for { |
|
msg, err := stream.Recv() |
|
if err == io.EOF { |
|
zlog.Infof("关闭EOF: %v", err) |
|
close(recvChan) |
|
return |
|
} |
|
if err != nil { |
|
zlog.Infof("接收错误: %v", err) |
|
close(recvChan) |
|
return |
|
} |
|
zlog.Infof("recv stream msg: %#v", msg) |
|
recvChan <- msg |
|
} |
|
}() |
|
|
|
// 发送和处理消息 |
|
for { |
|
select { |
|
case <-stream.Context().Done(): |
|
// 客户端断开连接 |
|
svc.klinePublisher.UnsubscribeAll(streamId) |
|
return stream.Context().Err() |
|
case msg, ok := <-recvChan: |
|
if !ok { |
|
// 接收通道关闭，结束流 |
|
svc.klinePublisher.UnsubscribeAll(streamId) |
|
return |
|
} |
|
|
|
if msg.SubType == pb.SubscribeType_UnsubscribeAll { |
|
svc.klinePublisher.UnsubscribeAll(streamId) |
|
continue |
|
} |
|
for _, exchange := range msg.Exchanges { |
|
for _, instId := range msg.InstIds { |
|
for _, interval := range msg.Intervals { |
|
confirms := []int{1} |
|
if !msg.OnlyConfirm { |
|
confirms = append(confirms, 0) |
|
} |
|
for _, confirm := range confirms { |
|
subKey := fmt.Sprintf("/kline/%s/%s/%s/%d", exchange.String(), instId, interval, confirm) |
|
zlog.Infof("stream: %d sub: %s", streamId, subKey) |
|
switch msg.SubType { |
|
case pb.SubscribeType_Subscribe: |
|
svc.klinePublisher.Subscribe(subKey, streamId, stream) |
|
case pb.SubscribeType_Unsubscribe: |
|
svc.klinePublisher.Unsubscribe(subKey, streamId) |
|
} |
|
} |
|
} |
|
} |
|
} |
|
} |
|
} |
|
|
|
// 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 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) |
|
}
|
|
|