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.
556 lines
16 KiB
556 lines
16 KiB
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/collect" |
|
"sig-pub/pkg/utils/times" |
|
"sig-pub/pkg/zlog" |
|
"sync" |
|
"sync/atomic" |
|
"time" |
|
|
|
"google.golang.org/grpc" |
|
) |
|
|
|
// ExchangeService 交易所服务 |
|
type ExchangeService struct { |
|
exchangeMap map[pb.ExchangeType]*Exchange |
|
tradeInstanceAside *aside.TradeInstanceAside |
|
exchangeDataPersist *ExchangeDataPersist |
|
|
|
klinePublisher *Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]] |
|
} |
|
|
|
// exchanges: 支持的数据源交易所 |
|
func NewExchangeService( |
|
tradeInstanceAside *aside.TradeInstanceAside, |
|
exchangeDataPersist *ExchangeDataPersist, |
|
exchanges ...*Exchange, |
|
) *ExchangeService { |
|
exchangeMap := make(map[pb.ExchangeType]*Exchange) |
|
for _, exchange := range exchanges { |
|
exchangeMap[exchange.ExchangeType] = exchange |
|
} |
|
|
|
return &ExchangeService{ |
|
exchangeMap: exchangeMap, |
|
tradeInstanceAside: tradeInstanceAside, |
|
exchangeDataPersist: exchangeDataPersist, |
|
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() { |
|
// 交易所订阅交易产品 |
|
for _, exchange := range svc.exchangeMap { |
|
go func(exchange *Exchange) { |
|
// get exchange all 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: inst.PriceSz, |
|
QuantitySz: inst.QuantitySz, |
|
ExchangeInstId: inst.ExchangeInstId, |
|
Leverages: inst.Leverages, |
|
Exchange: exchange.ExchangeType, |
|
} |
|
// 待初始化币种数据 |
|
processingInsts = append(processingInsts, *tradeInst) |
|
|
|
exchange.TradeInstIds.Store(inst.InstId, inst.ExchangeInstId) |
|
|
|
exchangeInst := &ExchangeTradeInstance{ |
|
Inst: tradeInst, |
|
LiveKline: types.NewIntervalState[types.Kline](), |
|
LiveKStartTs: types.NewIntervalState[int64](), |
|
HistoryMarkTs: types.NewIntervalState[int64](), |
|
} |
|
exchangeInst.Status.Store(int32(data.StatusProcessing)) |
|
exchange.ExchangeInsts.Store(inst.ExchangeInstId, exchangeInst) |
|
} |
|
|
|
// 订阅实时k线数据 |
|
err = exchange.Subscriber.SubscribeKline(exchangeInstIds...) // []string{"BTC-USDT", "DOGE-USDT-SWAP"} |
|
if err != nil { |
|
zlog.Error(err) |
|
return |
|
} |
|
// 消费实时k线数据 |
|
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 |
|
} |
|
// 排序后取出头尾k线 |
|
collect.SortAsc(channelK.Klines, func(k *types.Kline) int64 { return k.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.exchangeDataPersist.SaveKline(*tradeInst, confirmKlines) |
|
if err != nil { |
|
zlog.Errorf("kline save to tsdb error: %v, %#v", err, confirmKlines) |
|
} else { |
|
// 初始化状态完成, 更新k线时间戳标记 |
|
if exchangeInst.Status.Load() == int32(data.StatusOk) { |
|
latestK := collect.MustMax(confirmKlines, func(k *types.Kline) int64 { return k.Ts }) |
|
// 标记确认k线 |
|
tsKey, ex := svc.exchangeDataPersist.SaveHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, latestK.Interval, latestK.Ts) |
|
if ex != nil { |
|
zlog.Errorf("history mark inititaled ts error: key=%s, ts=%d, %v", tsKey, latestK.Ts, ex) |
|
} |
|
// todo k线完整性检查 |
|
} |
|
} |
|
} |
|
|
|
// 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.exchangeDataPersist.SaveHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, interval, ts) |
|
if ex != nil { |
|
zlog.Errorf("history mark inititaled ts error: key=%s, ts=%d, %v", tsKey, ts, ex) |
|
} |
|
}) |
|
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.exchangeDataPersist.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.fetchTaskKlinesToTSDB(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 |
|
}) |
|
} |
|
|
|
// 任务都已执行成功结束 |
|
subs := subTasks.Add(1) |
|
zlog.Debugf("trade instance initial kline tasks processing: %s(%s), pub %d, sub %d, fail %d", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load(), subTasks.Load(), failTasks.Load()) |
|
if pubTaskDone.Load() && subs >= pubTasks.Load() { |
|
zlog.Infof("trade instance initial kline tasks success finished, %s(%s), pub %d, sub %d, fail %d", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load(), subTasks.Load(), failTasks.Load()) |
|
cancel() |
|
return |
|
} |
|
} |
|
} |
|
}() |
|
} |
|
wg.Wait() |
|
return |
|
} |
|
|
|
func (svc *ExchangeService) fetchTaskKlinesToTSDB(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) |
|
zlog.Infof("fetch history interval klines: %s, %d klines, %s~%s", task.logKey(), len(klines), ee, ss) |
|
|
|
// store to tsdb |
|
err = svc.exchangeDataPersist.SaveKline(task.inst, klines) |
|
if err != nil { |
|
zlog.Errorf("save history klines to tsdb error: task -> %s, err -> %v", task.logKey(), err) |
|
return |
|
} |
|
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 |
|
} |
|
|
|
// HistoryKline 获取交易产品历史k线 (before < klines... < after) |
|
func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistoryKline) (klines []*types.Kline, err error) { |
|
// 交易产品参数检查 |
|
exchange, ok := svc.exchangeMap[req.Exchange] |
|
if !ok { |
|
err = fmt.Errorf("exchange not support: %s", req.Exchange) |
|
return |
|
} |
|
exchangeInstId, ok := exchange.TradeInstIds.Load(req.InstId) |
|
if !ok { |
|
err = fmt.Errorf("trade instance not support: %s", req.InstId) |
|
return |
|
} |
|
interval := types.Interval(req.Interval) |
|
intervalAdder, ok := types.SupportedIntervals[interval] |
|
if !ok { |
|
err = fmt.Errorf("interval not support: %s", req.Interval) |
|
return |
|
} |
|
|
|
exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId) |
|
if !ok { |
|
err = fmt.Errorf("trade instance not support for exchange: %s for %s", req.InstId, req.Exchange) |
|
return |
|
} |
|
|
|
// k线长度检查 |
|
afterTs, beforeTs, count := int64(req.After), int64(req.Before), int64(req.Count) |
|
latest := false |
|
if afterTs == 0 && beforeTs == 0 { |
|
afterTs = time.Now().UnixMilli() |
|
latest = true |
|
} |
|
if count == 0 { |
|
count = 120 |
|
} |
|
if afterTs == 0 { |
|
afterTs = intervalAdder(beforeTs, count) |
|
} |
|
if beforeTs == 0 { |
|
beforeTs = intervalAdder(afterTs, -count) |
|
} |
|
if beforeTs > afterTs { |
|
err = fmt.Errorf("time range invalid: before must less then after") |
|
return |
|
} |
|
if beforeTs == afterTs { |
|
return |
|
} |
|
total := (afterTs - beforeTs) / intervalAdder(0, 1) |
|
if total > 1000 { |
|
err = fmt.Errorf("time range too large") |
|
return |
|
} |
|
|
|
klines, err = svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, beforeTs, afterTs) |
|
if err != nil || len(klines) == 0 { |
|
return |
|
} |
|
|
|
// 开闭区间 confirmed |
|
if req.Open { |
|
if klines[0].Ts == afterTs { |
|
klines = klines[1:] |
|
} |
|
if len(klines) > 0 && klines[len(klines)-1].Ts == beforeTs { |
|
klines = klines[:len(klines)-1] |
|
} |
|
} |
|
|
|
if latest && req.Live { |
|
liveK := exchangeInst.LiveKline.Get(interval) |
|
klines = append([]*types.Kline{&liveK}, klines...) |
|
} |
|
return |
|
}
|
|
|