sig-pub/internal/exchange/exchange_service.go

package exchange

import (
	"context"
	"errors"
	"fmt"
	"runtime"
	"sig-pub/api/pb"
	"sig-pub/pkg/client"
	"sig-pub/pkg/data"
	"sig-pub/pkg/mq"
	"sig-pub/pkg/publish"
	"sig-pub/pkg/types"
	"sig-pub/pkg/utils/collect"
	"sig-pub/pkg/utils/lang"
	"sig-pub/pkg/utils/retry"
	"sig-pub/pkg/utils/times"
	"sig-pub/pkg/zlog"
	"sync"
	"sync/atomic"
	"time"

	"google.golang.org/grpc"
)

// ExchangeService 交易所服务
type ExchangeService struct {
	// exchanges         map[pb.ExchangeType]*Exchange
	exchanges           *types.ExchangeState[*Exchange]
	tradeInstanceAside  *client.TradeInstanceAside
	exchangeDataPersist *ExchangeDataPersist

	klinePublisher *publish.Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]]
}

// exchanges: 支持的数据源交易所
func NewExchangeService(
	tradeInstanceAside *client.TradeInstanceAside,
	exchangeDataPersist *ExchangeDataPersist,
	exchanges ...*Exchange,
) *ExchangeService {
	exchangeState := types.NewExchangeState[*Exchange]()
	for _, exchange := range exchanges {
		if !exchangeState.IsSupport(exchange.ExchangeType) {
			panic(fmt.Errorf("unsupport exchange: %s", exchange.ExchangeType.String()))
		}
		exchangeState.Set(exchange.ExchangeType, exchange)
	}

	return &ExchangeService{
		exchanges:           exchangeState,
		tradeInstanceAside:  tradeInstanceAside,
		exchangeDataPersist: exchangeDataPersist,
		klinePublisher:      publish.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 *publish.Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]]) {
	subscriber = svc.klinePublisher
	return
}

// 订阅交易所推送行情
func (svc *ExchangeService) subscribeExchanges() {
	// 交易所订阅交易产品
	svc.exchanges.Range(func(_ pb.ExchangeType, exchange *Exchange) {
		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](),
					LastKline:     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
		}
		// receivedTs := time.Now().UnixMilli()
		// if channelK.Klines[0].Interval == types.Interval1s {
		// 	zlog.Debugf("tick delay: %dms", receivedTs-channelK.Klines[0].Ts-1000)
		// }

		// 交易所 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

		// 升序排序
		collect.SortAsc(channelK.Klines, func(k *types.Kline) int64 { return k.Ts })

		// 取出头尾k线
		lastKline := channelK.Klines[len(channelK.Klines)-1]
		interval := lastKline.Interval
		intervalAdder, intervalSupport := types.SupportedIntervals[interval]

		// 记录实时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 {
				if intervalSupport {
					delay := time.Now().UnixMilli() - intervalAdder(kline.Ts, 1)
					zlog.Debugf("recv confirm kline: delay=%dms, inst=%s(%s), interval=%s", delay, tradeInst.InstId, tradeInst.Exchange, kline.Interval)
				}

				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)
		}

		padding := true
		if len(confirmKlines) > 0 {
			if intervalSupport {
				// k线完整性检查, k线是否连续并补齐
				if lastConfirmK := exchangeInst.LastKline.Get(interval); lastConfirmK.Ts != 0 {
					tempK := make([]*types.Kline, 0, len(confirmKlines)+1)
					tempK = append(tempK, &lastConfirmK)
					tempK = append(tempK, confirmKlines...)
					if err := svc.paddingKlinesIfNotSeries(exchange, tradeInst.InstId, confirmKlines[0].Interval, tempK, nil, nil); err != nil {
						padding = false
						zlog.Errorf("try padding klines error: inst=%s(%s), interval=%s, ts=%d~%d, %v", tradeInst.InstId, tradeInst.Exchange, lastKline.Interval, lastKline.Ts, lastConfirmK.Ts, err)
					}
				}
			}

			// 记录最后确认k线
			exchangeInst.LastKline.Set(interval, *confirmKlines[len(confirmKlines)-1])
		}

		// 存储到 tsdb
		if _, ok := types.SupportedIntervals[lastKline.Interval]; ok && len(confirmKlines) > 0 {
			// tsdb storage todo 异步处理
			err := svc.exchangeDataPersist.SaveKline(*tradeInst, confirmKlines)
			// zlog.Infof("save confirm klines: instId=%s(%s), interval=%s, ts=%d", tradeInst.InstId, tradeInst.Exchange, firstKline.Interval, firstKline.Ts)
			if err != nil {
				zlog.Errorf("kline save to tsdb error: %v, %#v", err, confirmKlines)
			} else {
				// k线未缺失, 初始化状态完成, 更新k线时间戳标记
				if intervalSupport && !padding && 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)
					}
				}
			}
		}

		// publish grpc stream klines
		for pubKey, kline := range pubStreamKlineMap {
			if len(kline.Klines) == 0 {
				continue
			}
			tids, subs := svc.klinePublisher.Publisher(pubKey)
			for i, sub := range subs {
				kline.StreamId = tids[i]
				if err := sub.Send(&pb.RspStreamSubscribeKline{Kline: kline}); err != nil {
					zlog.Error(err)
				}
			}
		}

		// if useMs := time.Now().UnixMilli() - receivedTs; useMs > 10 {
		// 	zlog.Debugf("handle consume kline use: %dms", useMs)
		// }
	}
}

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.paddingTradeInstanceKlines(exchange, inst)
		status := data.StatusFailed
		if err != nil {
			zlog.Errorf("padding trade instance klines error: %s(%s), err=%v", inst.InstId, inst.Exchange, err)
			failed = append(failed, inst)
		} else {
			success = append(success, inst)
			status = data.StatusOk
		}
		// 发布交易产品初始化完成事件
		publish := &mq.PublishExchangeTradeInstanceInited{
			Exchange: inst.Exchange,
			InstId:   inst.InstId,
			Status:   status,
		}
		if err := mq.NatsPublish(mq.TopicExchangeTradeInstanceInited, publish); err != nil {
			zlog.Errorf("publish trade instance inited error: %s(%s), err=%v", inst.InstId, inst.Exchange, err)
		}
	}

	zlog.Infof("%d insts initial finished, success %d, failed %d", len(insts), len(success), len(failed))
}

// paddingTradeInstanceKlines 初始化交易产品历史k线数据
func (svc *ExchangeService) paddingTradeInstanceKlines(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
	}

	watch := times.NewWatch()
	defer func() {
		if err != nil {
			// 交易所k线初始化失败
			exchangeInst.Status.Store(int32(data.StatusFailed))
			return
		}
		// 初始化成功
		exchangeInst.Status.Store(int32(data.StatusOk))
		zlog.Infof("padding history kline finish: instId=%s(%s), use %s", tradeInst.InstId, tradeInst.Exchange, watch.ElapsedFmt("."))

		// flush vmtsdb to disk
		retry.DoWithFixDelay(5, time.Second, func(retryTimes uint32) (_ struct{}, err error) {
			if err = svc.exchangeDataPersist.Flush0(); err != nil {
				zlog.Errorf("flush vmts db error: ", err)
			}
			return
		})
	}()

	// 并发数
	concurrent := max(8, runtime.NumCPU()*2)
	// 按周期分割成小任务
	for interval := range types.SupportedIntervals {
		err = svc.paddingTradeInstanceIntervalKlines(concurrent, exchange, tradeInst, interval)
		if err != nil {
			zlog.Errorf("padding trade instance interval error: instId=%s(%s), interval=%s", tradeInst.InstId, tradeInst.Exchange, interval, err)
			return
		}
	}
	return
}

// paddingTradeInstanceIntervalKlines 初始化交易产品指定周期历史k线数据
func (svc *ExchangeService) paddingTradeInstanceIntervalKlines(concurrent int, exchange *Exchange, tradeInst types.TradeInstance, interval types.Interval) (err error) {
	intervalAdder, ok := types.SupportedIntervals[interval]
	if !ok {
		err = fmt.Errorf("unsupport interval %s", interval)
		return
	}

	exchangeInst, ok := exchange.ExchangeInsts.Load(tradeInst.ExchangeInstId)
	if !ok {
		err = fmt.Errorf("unsupport interval exchange trade instance: %s(%s)", tradeInst.ExchangeInstId, exchange.ExchangeType)
		return
	}

	// 任务 channel
	taskCh := make(chan fetchKlineTask, concurrent)
	retryTaskCh := make(chan fetchKlineTask, concurrent)

	// 发布任务数, 成功任务数, 失败任务次数
	var pubTasks, subTasks, failTimes atomic.Int32
	var pubTaskDone atomic.Bool // 所有任务已发布
	watch := times.NewWatch()
	ctx, cancel := context.WithCancel(context.Background())

	defer func() {
		if err != nil {
			return
		}
		zlog.Infof("padding history kline finish: instId=%s(%s), interval=%s, pub=%d, sub=%d, fail=%d, use %s", tradeInst.InstId, tradeInst.Exchange, interval, pubTasks.Load(), subTasks.Load(), failTimes.Load(), watch.ElapsedFmt("."))

		markTs := exchangeInst.HistoryMarkTs.Get(interval)
		tsKey, ex := svc.exchangeDataPersist.SaveHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, interval, markTs)
		if ex != nil {
			zlog.Errorf("save history mark ts error: key=%s, ts=%d, %v", tsKey, markTs, ex)
		}
	}()

	// 任务进度日志(执行超过3s打印进度)
	go func() {
		select {
		case <-ctx.Done():
			return
		case <-time.After(3 * time.Second):
		}

		ticker := time.NewTicker(time.Second)
		for {
			select {
			case <-ctx.Done():
				ticker.Stop()
				return
			case <-ticker.C:
				zlog.Debugf("processing padding history kline tasks: %s(%s), interval=%s, pub %d, sub %d, fail %d", tradeInst.InstId, tradeInst.Exchange, interval, pubTasks.Load(), subTasks.Load(), failTimes.Load())
			}
		}
	}()

	// 任务发布器
	go func() {
		defer func() {
			pubTaskDone.Store(true)
			// 无任务处理
			if pubTasks.Load() == 0 {
				cancel()
			}
		}()

		// history 未补全前, history写 kvdb ts mark, 补全后 ws live cnofirm 写 ts mark
		beforeTs := int64(0)
		beforeTs, err := retry.DoWithStepDelay(10, time.Second, func(retryTimes uint32) (markTs int64, err error) {
			markTs, err = svc.exchangeDataPersist.GetHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, interval)
			if err != nil {
				zlog.Error("get history kline mark ts error: ", err)
			}
			return
		})
		if err != nil {
			zlog.Error(err)
			cancel()
			return
		}
		if beforeTs == 0 {
			beforeTs = intervalAdder(KlineBefore0, -1)
		} else {
			// 不足100根，向前补齐100根一次拉取过来
			total := (time.Now().UnixMilli() - beforeTs) / intervalAdder(0, 1)
			if total < 100 {
				beforeTs = max(intervalAdder(beforeTs, -100), intervalAdder(KlineBefore0, -1))
			}
		}

		for {
			// 判定订阅任务发布完成
			if intervalAdder(beforeTs, 2) > 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())
				}

				// fetch history kline
				if lastKlineTs, ex := svc.fetchTaskKlinesToTSDB(exchange, task); ex != nil {
					failTimes.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)
					if pubTaskDone.Load() && subs >= pubTasks.Load() {
						// zlog.Infof("initial history kline tasks finished success, %s(%s), pub %d, sub %d, fail %d", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load(), subTasks.Load(), failTimes.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
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	klines, err := exchange.Fetcher.FetchHistoryKlines(ctx, 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
}

// paddingKlinesIfNotSeries 如k线不连续, 从缺失处进行补齐
func (svc *ExchangeService) paddingKlinesIfNotSeries(exchange *Exchange, instId string, interval types.Interval, klines []*types.Kline, firstKlinePrev, lastKlineNext *types.Kline) (err error) {
	if len(klines) == 0 {
		return
	}
	// 检查k线是否连续
	paddingMarkTs := int64(0)
	if firstKlinePrev != nil && interval.MustAddMul(firstKlinePrev.Ts, 1) != klines[0].Ts {
		paddingMarkTs = firstKlinePrev.Ts
	}
	if paddingMarkTs == 0 && lastKlineNext != nil && lastKlineNext.Ts != interval.MustAddMul(klines[len(klines)-1].Ts, 1) {
		paddingMarkTs = klines[len(klines)-1].Ts
	}
	if paddingMarkTs == 0 {
		for i, k := range klines {
			if i > 0 && k.Ts != interval.MustAddMul(klines[i-1].Ts, 1) {
				paddingMarkTs = klines[i-1].Ts
				break
			}
		}
	}
	if paddingMarkTs == 0 {
		return
	}
	// k线不连续进行补齐
	exchangeInstId, _ := exchange.TradeInstIds.Load(instId)
	exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId)
	if !ok {
		err = fmt.Errorf("trade instance not support for exchange: %s(%s)", instId, exchange.ExchangeType.String())
		return
	}
	if exchangeInst.Status.CompareAndSwap(int32(data.StatusOk), int32(data.StatusProcessing)) {
		defer exchangeInst.Status.CompareAndSwap(int32(data.StatusProcessing), int32(data.StatusOk))

		watch := times.NewWatch()
		zlog.Warningf("vmtsdb kline not series, try padding: instId=%s(%s), interval=%s, ts=%d", instId, exchange.ExchangeType.String(), interval, paddingMarkTs)
		if _, err = svc.exchangeDataPersist.SaveHistoryKlineMarkTs(exchange.ExchangeType, instId, interval, paddingMarkTs); err != nil {
			zlog.Errorf("try padding series save markTs error: ", err)
			return
		}
		if err = svc.paddingTradeInstanceIntervalKlines(4, exchange, *exchangeInst.Inst, interval); err != nil {
			zlog.Errorf("try padding series fetch to vmtsdb error: ", err)
			return
		}
		// flush vmtsdb to disk
		svc.exchangeDataPersist.Flush()
		zlog.Debugf("vmtsdb kline not series padding success: instId=%s(%s), interval=%s, ts=%d, use %s", instId, exchange.ExchangeType.String(), interval, paddingMarkTs, watch.ElapsedFmt("."))
	}
	return
}

// Exchanges 支持的交易所列表
func (svc *ExchangeService) Exchanges() (exchanges []pb.ExchangeType, err error) {
	svc.exchanges.Range(func(exchange pb.ExchangeType, _ *Exchange) {
		exchanges = append(exchanges, exchange)
	})
	return
}

// ExchangeInstanceState 交易所交易产品状态
func (svc *ExchangeService) ExchangeInstanceState(req *pb.ReqExchangeInstanceState) (states []*pb.TradeInstanceState, err error) {
	var exchanges []*Exchange
	if req.AllExchange {
		svc.exchanges.Range(func(_ pb.ExchangeType, exchange *Exchange) {
			exchanges = append(exchanges, exchange)
		})
	} else {
		for _, exchangeType := range req.Exchanges {
			if !svc.exchanges.IsSupport(exchangeType) {
				err = fmt.Errorf("not support exchange: %v", exchangeType)
				return
			}
			exchanges = append(exchanges, svc.exchanges.Get(exchangeType))
		}
	}
	if len(exchanges) == 0 {
		err = errors.New("no support exchanges")
		return
	}

	for _, exchange := range exchanges {
		insts := make([]*ExchangeTradeInstance, 0, 8)
		if req.AllInsts {
			exchange.ExchangeInsts.Range(func(_ string, inst *ExchangeTradeInstance) bool {
				if req.AllStatus || collect.In(inst.Status.Load(), req.Status...) {
					insts = append(insts, inst)
				}
				return true
			})
		} else {
			for _, instId := range req.Insts {
				// trade instId 转 exchangeInstId
				exchangeInstId, ok := exchange.TradeInstIds.Load(instId)
				if !ok {
					continue
				}
				inst, ok := exchange.ExchangeInsts.Load(exchangeInstId)
				if !ok {
					continue
				}
				if req.AllStatus || collect.In(inst.Status.Load(), req.Status...) {
					insts = append(insts, inst)
				}
			}
		}
		for _, exchangeInst := range insts {
			state := &pb.TradeInstanceState{
				Exchange: exchange.ExchangeType,
				InstId:   exchangeInst.Inst.InstId,
				Status:   exchangeInst.Status.Load(),
				Last:     exchangeInst.Last.String(),
			}
			states = append(states, state)
		}
	}
	return
}

const (
	DefaultHistoryKlines = 200
	MaxHistoryKlines     = 4096
)

func (svc *ExchangeService) CalcSeriesRange(arg *pb.SeriesRange) (after, before, total int64, err error) {
	// 交易产品参数检查
	exchange := svc.exchanges.Get(arg.Exchange)
	exchangeInstId, ok := exchange.TradeInstIds.Load(arg.InstId)
	if !ok {
		err = fmt.Errorf("trade instance not support: %s", arg.InstId)
		return
	}
	exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId)
	if !ok {
		err = fmt.Errorf("trade instance not support for exchange: %s for %s", arg.InstId, arg.Exchange)
		return
	}
	interval := types.Interval(arg.Interval)
	intervalAdder, ok := types.SupportedIntervals[interval]
	if !ok {
		err = fmt.Errorf("interval not support: %s", arg.Interval)
		return
	}
	// k线长度检查
	after, before, count := int64(arg.After), int64(arg.Before), int64(arg.Count)
	if count == 0 {
		count = DefaultHistoryKlines
	}
	// 拉取最新的
	lastTs := int64(0)
	if arg.After == 0 {
		liveK := exchangeInst.LiveKline.Get(interval)
		lastTs = liveK.Ts
		if !liveK.Confirm {
			lastTs = intervalAdder(liveK.Ts, -1)
		}
	}
	if after == 0 && before == 0 {
		after = lastTs
	}
	if after == 0 {
		after = min(intervalAdder(before, count-1), lastTs)
	}
	if before == 0 {
		before = max(intervalAdder(after, -count+1), KlineBefore0)
	}
	// 开区间
	if arg.Open {
		if arg.After != 0 {
			after = max(intervalAdder(after, -1), before)
			if arg.Before == 0 {
				before = max(intervalAdder(before, -1), KlineBefore0)
			}
		}
		if arg.Before != 0 {
			before = min(intervalAdder(before, 1), after)
			if arg.After == 0 {
				after = min(intervalAdder(before, 1), lastTs)
			}
		}
	}
	// 额外拉取
	if arg.Window > 0 {
		before = max(intervalAdder(before, -int64(arg.Window)), KlineBefore0)
	}
	if before > after {
		err = fmt.Errorf("time range invalid: before must less then after")
		return
	}
	// 拉取范围总条数
	total = (after-before)/intervalAdder(0, 1) + 1
	return
}

// HistoryKline 获取交易产品历史k线 (before < klines... < after)
func (svc *ExchangeService) HistoryKline(arg *pb.SeriesRange, recvBranch int, recvKline func(klines []*pb.Kline) error) (live bool, err error) {
	// 交易产品参数检查
	exchange := svc.exchanges.Get(arg.Exchange)
	exchangeInstId, ok := exchange.TradeInstIds.Load(arg.InstId)
	if !ok {
		err = fmt.Errorf("trade instance not support: %s", arg.InstId)
		return
	}
	interval := types.Interval(arg.Interval)
	intervalAdder, ok := types.SupportedIntervals[interval]
	if !ok {
		err = fmt.Errorf("interval not support: %s", arg.Interval)
		return
	}

	exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId)
	if !ok {
		err = fmt.Errorf("trade instance not support for exchange: %s for %s", arg.InstId, arg.Exchange)
		return
	}
	// 交易产品初始化完成检查
	if status := exchangeInst.Status.Load(); status != int32(data.StatusOk) {
		err = fmt.Errorf("trade instance not ready: %s(%s) for %d", arg.InstId, arg.Exchange, status)
		return
	}
	// 限制最大时间范围
	afterTs, beforeTs, total, err := svc.CalcSeriesRange(arg)
	if err != nil {
		return
	}
	// if total > MaxHistoryKlines {
	// 	err = fmt.Errorf("time range too large max %d", MaxHistoryKlines)
	// 	return
	// }
	if arg.Limit > 0 && total > int64(arg.Limit) {
		err = fmt.Errorf("time range %d out of limit %d", total, arg.Limit)
		return
	}

	// 分批查询
	branch := int64(2000)
	before, after := beforeTs, afterTs
	recvBuffer := make([]*pb.Kline, 0, recvBranch)
	var prevFirstK, prevLastK *types.Kline
	for range 10000 {
		if arg.Desc {
			before = max(intervalAdder(after, -branch+1), beforeTs)
		} else {
			after = min(intervalAdder(before, branch-1), afterTs)
		}
		if after < KlineBefore0 || before > after {
			break
		}

		klines, errK := svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, before, after)
		if errK != nil {
			err = errK
			zlog.Error("list vmtsdb kline error: ", err)
			return
		}
		if len(klines) == 0 {
			break
		}

		// 检查k线是否连续进行补齐
		if err = svc.paddingKlinesIfNotSeries(exchange, arg.InstId, interval, klines,
			lang.Ternary(arg.Desc, nil, prevLastK),
			lang.Ternary(arg.Desc, prevFirstK, nil),
		); err != nil {
			return
		}
		prevFirstK, prevLastK = klines[0], klines[len(klines)-1]
		lastK := klines[len(klines)-1]
		if after == afterTs && lastK.Ts != afterTs {
			// vmtsdb 数据落盘30s延迟, 使用内存数据替代最新的一根k线
			lastConfirmK := exchangeInst.LastKline.Get(interval)
			if lastConfirmK.Ts == lastK.Ts {
				lastK = &lastConfirmK
				klines[len(klines)-1] = lastK
			}
			if lastConfirmK.Ts == afterTs && intervalAdder(lastK.Ts, 1) == afterTs {
				lastK = &lastConfirmK
				klines = append(klines, &lastConfirmK)
			}
		}

		// 实时k线
		if !live && arg.Live && lastK.Ts == afterTs {
			liveK := exchangeInst.LiveKline.Get(interval)
			if latest := intervalAdder(lastK.Ts, 1) == liveK.Ts; latest {
				klines = append(klines, &liveK)
				live = true
			}
		}

		// next loop
		if arg.Desc {
			after = intervalAdder(klines[0].Ts, -1)
		} else {
			before = intervalAdder(klines[len(klines)-1].Ts, 1)
		}

		// 降序排序
		if arg.Desc {
			collect.Reverse(klines)
		}

		// zlog.Debugf("krange: %s(%s), interval=%s, total=%d, %d~%d", arg.InstId, arg.Exchange, interval, len(klines), klines[0].Ts, klines[len(klines)-1].Ts)

		// 分成小批量recv
		length := len(klines)
		for i, kline := range klines {
			recvBuffer = append(recvBuffer, kline.ToPBKline())
			if len(recvBuffer) < recvBranch && i < length-1 {
				continue
			}
			if err = recvKline(recvBuffer); err != nil {
				break
			}
			recvBuffer = recvBuffer[:0]
		}
	}
	return
}