open
/
sig-pub
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

kline branch stream

main
strange 10 months ago
parent
e5d051e446
commit
d011868cca
5 changed files with 96 additions and 128 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 19
      internal/exchange/exchange_grpc_server.go
  2. 175
      internal/exchange/exchange_service.go
  3. 6
      pkg/strategy/strategy.go
  4. 14
      pkg/types/series/klines.go
  5. 10
      pkg/utils/collect/collect.go

19
internal/exchange/exchange_grpc_server.go
Unescape View File

@ -133,20 +133,21 @@ func (svr *ExchangeGrpcServer) HistoryKline(ctx context.Context, req *pb.ReqHist
return return
} }
rsp = new(pb.RspHistoryKline)
rsp.Klines = make([]*pb.Kline, 0, total+1)
rsp = &pb.RspHistoryKline{ rsp = &pb.RspHistoryKline{
Exchange: req.Series.Exchange, Exchange: req.Series.Exchange,
InstId: req.Series.InstId, InstId: req.Series.InstId,
Interval: req.Series.Interval, Interval: req.Series.Interval,
} }
live, klines, err := svr.exchangeService.HistoryKline(ctx, req.Series) live, err := svr.exchangeService.HistoryKline(req.Series, DefaultHistoryKlines, func(klines []*pb.Kline) error {
rsp.Klines = append(rsp.Klines, klines...)
return nil
})
if err != nil { if err != nil {
return return
} }
rsp.Live = live rsp.Live = live
rsp.Klines = make([]*pb.Kline, 0, len(klines))
for _, k := range klines {
rsp.Klines = append(rsp.Klines, k.ToPBKline())
}
return return
} }
@ -156,6 +157,12 @@ func (svr *ExchangeGrpcServer) HistoryKlineStream(req *pb.ReqHistoryKlineStream,
err = fmt.Errorf("series arg is required") err = fmt.Errorf("series arg is required")
return return
} }
err = svr.exchangeService.HistoryKlineStream(req.Series, stream) _, err = svr.exchangeService.HistoryKline(req.Series, 200, func(klines []*pb.Kline) error {
rsp := &pb.RspHistoryKlineStream{Klines: klines}
return stream.Send(rsp)
})
if err != nil {
return
}
return return
} }

175
internal/exchange/exchange_service.go
Unescape View File

@ -733,7 +733,7 @@ func (svc *ExchangeService) CalcSeriesRange(arg *pb.SeriesRange) (after, before,
} }
// HistoryKline 获取交易产品历史k线 (before < klines... < after) // HistoryKline 获取交易产品历史k线 (before < klines... < after)
func (svc *ExchangeService) HistoryKline(ctx context.Context, arg *pb.SeriesRange) (live bool, klines []*types.Kline, err error) { 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) exchange := svc.exchanges.Get(arg.Exchange)
exchangeInstId, ok := exchange.TradeInstIds.Load(arg.InstId) exchangeInstId, ok := exchange.TradeInstIds.Load(arg.InstId)
@ -748,46 +748,47 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, arg *pb.SeriesRang
return return
} }
// todo 交易产品初始化完成检查
exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId) exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId)
if !ok { if !ok {
err = fmt.Errorf("trade instance not support for exchange: %s for %s", arg.InstId, arg.Exchange) err = fmt.Errorf("trade instance not support for exchange: %s for %s", arg.InstId, arg.Exchange)
return 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) afterTs, beforeTs, total, err := svc.CalcSeriesRange(arg)
if err != nil { if err != nil {
return return
} }
if total > MaxHistoryKlines { // if total > MaxHistoryKlines {
err = fmt.Errorf("time range too large max %d", MaxHistoryKlines) // err = fmt.Errorf("time range too large max %d", MaxHistoryKlines)
return // return
} // }
if arg.Limit > 0 && total > int64(arg.Limit) { if arg.Limit > 0 && total > int64(arg.Limit) {
err = fmt.Errorf("time range %d out of limit %d", total, arg.Limit) err = fmt.Errorf("time range %d out of limit %d", total, arg.Limit)
return return
} }
// 时间分段查询: 升降序, live, open // 分批查询
branch := int64(10) branch := int64(2000)
before, after := beforeTs, afterTs before, after := beforeTs, afterTs
if arg.Desc { recvBuffer := make([]*pb.Kline, 0, recvBranch)
before = intervalAdder(after, -branch) for range 10000 {
if before > afterTs { // int64 越界 if arg.Desc {
before = beforeTs before = max(intervalAdder(after, -branch+1), beforeTs)
} else {
after = min(intervalAdder(before, branch-1), afterTs)
} }
before = max(before, beforeTs) if after < KlineBefore0 || before > after {
} else { break
after = intervalAdder(before, branch)
if after < beforeTs { // int64 越界
after = afterTs
} }
after = min(after, afterTs)
}
for { klines, errK := svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, before, after)
klines, err = svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, before, after) if errK != nil {
if err != nil { err = errK
zlog.Error("list vmtsdb kline error: ", err) zlog.Error("list vmtsdb kline error: ", err)
return return
} }
@ -795,107 +796,59 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, arg *pb.SeriesRang
break break
} }
if arg.Desc { // 检查k线是否连续进行补齐
after = intervalAdder(klines[0].Ts, -1) if err = svc.paddingKlinesIfNotSeries(exchange, arg.InstId, interval, klines); err != nil {
before = max(intervalAdder(after, -branch), beforeTs) return
if after < KlineBefore0 || before > after { }
break 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
} }
} else { if lastConfirmK.Ts == afterTs && intervalAdder(lastK.Ts, 1) == afterTs {
before = intervalAdder(klines[0].Ts, 1) lastK = &lastConfirmK
after = min(intervalAdder(before, 1000), afterTs) klines = append(klines, &lastConfirmK)
if before > time.Now().UnixMilli() || before > after {
break
} }
} }
if arg.Desc { // 实时k线
collect.Reverse(klines) if !live && arg.Live && lastK.Ts == afterTs {
} liveK := exchangeInst.LiveKline.Get(interval)
tss := collect.Mapping(klines, func(_ int, k *types.Kline) int64 { return k.Ts }) if latest := intervalAdder(lastK.Ts, 1) == liveK.Ts; latest {
zlog.Info(tss)
}
klines, err = svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, beforeTs, afterTs)
if err != nil {
zlog.Error("list vmtsdb kline error: ", err)
return
}
if len(klines) == 0 {
return
}
// 检查k线是否连续进行补齐
if err = svc.paddingKlinesIfNotSeries(exchange, arg.InstId, interval, klines); err != nil {
return
}
lastK := klines[len(klines)-1]
// 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)
}
// 降序排序
if arg.Desc {
collect.Reverse(klines)
}
// 实时k线
if arg.Live && len(klines) > 0 {
liveK := exchangeInst.LiveKline.Get(interval)
if latest := intervalAdder(lastK.Ts, 1) == liveK.Ts; latest {
if arg.Desc {
klines = append([]*types.Kline{&liveK}, klines...)
} else {
klines = append(klines, &liveK) klines = append(klines, &liveK)
live = true
} }
live = true
} }
}
return
}
// 查询历史k线(按时间升序流式返回) // next loop
func (svc *ExchangeService) HistoryKlineStream(arg *pb.SeriesRange, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) { if arg.Desc {
ctx := context.Background() after = intervalAdder(klines[0].Ts, -1)
_, klines, err := svc.HistoryKline(ctx, arg) } else {
if err != nil { before = intervalAdder(klines[len(klines)-1].Ts, 1)
return }
}
// 交易产品参数检查 // 降序排序
if !svc.exchanges.IsSupport(arg.Exchange) { if arg.Desc {
err = fmt.Errorf("exchange not support: %s", arg.Exchange) collect.Reverse(klines)
return }
}
exchange := svc.exchanges.Get(arg.Exchange)
interval := types.Interval(arg.Interval)
// 检查k线是否连续进行补齐 zlog.Infof("krange: %s, total=%d, %d~%d", arg.InstId, len(klines), klines[0].Ts, klines[len(klines)-1].Ts)
if err = svc.paddingKlinesIfNotSeries(exchange, arg.InstId, interval, klines); err != nil {
return
}
branch := 100 // 分成小批量recv
length := len(klines) length := len(klines)
kBuffer := make([]*pb.Kline, 0, branch) for i, kline := range klines {
for i, kline := range klines { recvBuffer = append(recvBuffer, kline.ToPBKline())
kBuffer = append(kBuffer, kline.ToPBKline()) if len(recvBuffer) < recvBranch && i < length-1 {
if len(kBuffer) < branch && i < length-1 { continue
continue }
} if err = recvKline(recvBuffer); err != nil {
rsp := &pb.RspHistoryKlineStream{Klines: kBuffer} break
if sendErr := stream.Send(rsp); sendErr != nil { }
err = sendErr recvBuffer = recvBuffer[:0]
return
} }
kBuffer = kBuffer[:0]
} }
return return
} }

6
pkg/strategy/strategy.go
Unescape View File

@ -23,7 +23,7 @@ const (
// completed 同一时刻的所有其他周期都完成 // completed 同一时刻的所有其他周期都完成
func DriverIntervalKey(instId string, exchange pb.ExchangeType, completed bool, intervals ...types.Interval) string { func DriverIntervalKey(instId string, exchange pb.ExchangeType, completed bool, intervals ...types.Interval) string {
types.IntervalsSort(intervals) types.IntervalsSort(intervals)
strIntervals := collect.Mapping(intervals, func(_ int, interval types.Interval) string { return string(interval) }) strIntervals := collect.Mapping0(intervals, func(_ int, interval types.Interval) string { return string(interval) })
pubKey := fmt.Sprintf("/interval/%s/%s/%s/%d", instId, exchange.String(), strings.Join(strIntervals, ","), lang.Ternary(completed, 1, 0)) pubKey := fmt.Sprintf("/interval/%s/%s/%s/%d", instId, exchange.String(), strings.Join(strIntervals, ","), lang.Ternary(completed, 1, 0))
return pubKey return pubKey
} }
@ -33,8 +33,8 @@ func DriverIntervalKey(instId string, exchange pb.ExchangeType, completed bool,
func MulExchangeDriverIntervalKey(instId string, exchanges []pb.ExchangeType, intervals ...types.Interval) string { func MulExchangeDriverIntervalKey(instId string, exchanges []pb.ExchangeType, intervals ...types.Interval) string {
types.IntervalsSort(intervals) types.IntervalsSort(intervals)
types.ExchangesSort(exchanges) types.ExchangesSort(exchanges)
strIntervals := collect.Mapping(intervals, func(_ int, interval types.Interval) string { return string(interval) }) strIntervals := collect.Mapping0(intervals, func(_ int, interval types.Interval) string { return string(interval) })
strExchanges := collect.Mapping(exchanges, func(_ int, exchange pb.ExchangeType) string { return exchange.String() }) strExchanges := collect.Mapping0(exchanges, func(_ int, exchange pb.ExchangeType) string { return exchange.String() })
pubKey := fmt.Sprintf("/interval/%s/%s/%s", instId, strings.Join(strExchanges, ","), strings.Join(strIntervals, ",")) pubKey := fmt.Sprintf("/interval/%s/%s/%s", instId, strings.Join(strExchanges, ","), strings.Join(strIntervals, ","))
return pubKey return pubKey
} }

14
pkg/types/series/klines.go
Unescape View File

@ -9,29 +9,29 @@ import (
type Klines []types.Kline type Klines []types.Kline
func (s Klines) Times() []int64 { func (s Klines) Times() []int64 {
return collect.Mapping(s, func(_ int, k types.Kline) int64 { return k.Ts }) return collect.Mapping0(s, func(_ int, k types.Kline) int64 { return k.Ts })
} }
func (s Klines) Open() Floats { func (s Klines) Open() Floats {
return collect.Mapping(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Open) }) return collect.Mapping0(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Open) })
} }
func (s Klines) Close() Floats { func (s Klines) Close() Floats {
return collect.Mapping(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Close) }) return collect.Mapping0(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Close) })
} }
func (s Klines) High() Floats { func (s Klines) High() Floats {
return collect.Mapping(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.High) }) return collect.Mapping0(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.High) })
} }
func (s Klines) Low() Floats { func (s Klines) Low() Floats {
return collect.Mapping(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Low) }) return collect.Mapping0(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Low) })
} }
func (s Klines) Vol() Floats { func (s Klines) Vol() Floats {
return collect.Mapping(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Vol) }) return collect.Mapping0(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.Vol) })
} }
func (s Klines) VolQuote() Floats { func (s Klines) VolQuote() Floats {
return collect.Mapping(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.VolQuote) }) return collect.Mapping0(s, func(_ int, k types.Kline) float64 { return decimals.MustToFloat64(k.VolQuote) })
} }

10
pkg/utils/collect/collect.go
Unescape View File

@ -155,7 +155,15 @@ func Slice2MapKv[T any, K comparable, V any](slice []T, mapping func(int, T) (K,
return m return m
} }
func Mapping[T any, V any](slice []T, mapping func(int, T) V) (vs []V) { func Mapping[T any, V any](slice []T, mapping func(T) V) (vs []V) {
vs = make([]V, 0, len(slice))
for _, item := range slice {
vs = append(vs, mapping(item))
}
return
}
func Mapping0[T any, V any](slice []T, mapping func(int, T) V) (vs []V) {
vs = make([]V, 0, len(slice)) vs = make([]V, 0, len(slice))
for i, item := range slice { for i, item := range slice {
vs = append(vs, mapping(i, item)) vs = append(vs, mapping(i, item))

Write Preview
Loading…
Cancel
Save