From 51ba35ac7b43b62a450006c1a40acb5cc9792d4d Mon Sep 17 00:00:00 2001 From: strange Date: Sun, 19 Oct 2025 13:27:22 +0800 Subject: [PATCH] trading kline store --- api/exchange.proto | 24 ++- internal/exchange/exchange_grpc_server.go | 4 +- internal/exchange/exchange_service.go | 234 +++++++++++++-------- internal/exchange/okx/okx_fetch.go | 2 +- internal/trading/kline_series.go | 30 ++- internal/trading/kline_series_test.go | 34 --- internal/trading/kline_store.go | 241 +++++++++------------- pkg/backtrace/backtrace.go | 13 ++ pkg/client/trade_instance_client.go | 3 +- pkg/data/common.go | 2 +- pkg/indicator/base.go | 8 +- pkg/strategy/gold_x.go | 31 +++ pkg/strategy/strategy.go | 35 ++++ pkg/strategy/strategy_multi_interval.go | 18 ++ pkg/types/exchange.go | 4 +- pkg/types/interval.go | 2 +- 16 files changed, 407 insertions(+), 278 deletions(-) delete mode 100644 internal/trading/kline_series_test.go create mode 100644 pkg/backtrace/backtrace.go create mode 100644 pkg/strategy/gold_x.go create mode 100644 pkg/strategy/strategy.go create mode 100644 pkg/strategy/strategy_multi_interval.go diff --git a/api/exchange.proto b/api/exchange.proto index 0ccf7a8..f341765 100644 --- a/api/exchange.proto +++ b/api/exchange.proto @@ -19,7 +19,7 @@ service ExchangeService { rpc HistoryKline(ReqHistoryKline) returns (RspHistoryKline); // 查询历史k线(流式返回) - rpc HistoryKlineStream(ReqHistoryKline) returns (stream RspHistoryKlineStream); + rpc HistoryKlineStream(ReqHistoryKlineStream) returns (stream RspHistoryKlineStream); } message ReqStreamSubscribeKline { @@ -45,9 +45,12 @@ message RspExchanges { } message ReqExchangeInstanceState { - repeated string insts = 1; - bool allExchange = 2; - repeated ExchangeType exchanges = 3; // 交易所 + bool allExchange = 1; // 所有交易所 + bool allInsts = 2; // 所有交易产品 + bool allStatus = 3; // 所有状态 + repeated ExchangeType exchanges = 4; // 指定交易所 + repeated string insts = 5; // 指定交易产品 + repeated int32 status = 6; // 指定交易产品状态 } message RspExchangeInstanceState { repeated TradeInstanceState instsState = 1; // 交易产品列表 @@ -69,15 +72,18 @@ message RspHistoryKline { string instId = 2; string interval = 3; bool live = 4; // 第一根是否实时k线 - bool next = 5; // 是否有更多历史数据: 为true时, 可以用最后一根k线的ts作为before继续请求 + // bool next = 5; // 是否有更多历史数据: 为true时, 可以用最后一根k线的ts作为before继续请求 repeated Kline klines = 9; } -message RspHistoryKlineStream { +message ReqHistoryKlineStream { ExchangeType exchange = 1; // 交易所 string instId = 2; string interval = 3; - bool live = 4; // 第一根是否实时k线 - bool next = 5; // 是否有更多历史数据: 为true时, 可以用最后一根k线的ts作为before继续请求 - repeated Kline klines = 9; + int64 before = 4; + int64 after = 5; + uint32 count = 6; // k线条数,before或after其中一个为0时有效 +} +message RspHistoryKlineStream { + repeated Kline klines = 2; } diff --git a/internal/exchange/exchange_grpc_server.go b/internal/exchange/exchange_grpc_server.go index 7483b4a..6b5f2a8 100644 --- a/internal/exchange/exchange_grpc_server.go +++ b/internal/exchange/exchange_grpc_server.go @@ -109,7 +109,7 @@ func (svr *ExchangeGrpcServer) Exchanges(ctx context.Context, req *pb.ReqExchang // ExchangeInstanceState 获取交易产品系统状态 func (svr *ExchangeGrpcServer) ExchangeInstanceState(ctx context.Context, req *pb.ReqExchangeInstanceState) (rsp *pb.RspExchangeInstanceState, err error) { - states, err := svr.exchangeService.ExchangeInstanceState(req.AllExchange, req.Exchanges, req.Insts) + states, err := svr.exchangeService.ExchangeInstanceState(req) if err != nil { return } @@ -138,7 +138,7 @@ func (svr *ExchangeGrpcServer) HistoryKline(ctx context.Context, req *pb.ReqHist } // HistoryKlineStream 获取交易产品历史k线(流式返回) -func (svr *ExchangeGrpcServer) HistoryKlineStream(req *pb.ReqHistoryKline, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) { +func (svr *ExchangeGrpcServer) HistoryKlineStream(req *pb.ReqHistoryKlineStream, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) { err = svr.exchangeService.HistoryKlineStream(req, stream) return } diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index 41676fd..b53a115 100644 --- a/internal/exchange/exchange_service.go +++ b/internal/exchange/exchange_service.go @@ -12,6 +12,7 @@ import ( "sig-pub/pkg/types" "sig-pub/pkg/utils/collect" "sig-pub/pkg/utils/conver" + "sig-pub/pkg/utils/retry" "sig-pub/pkg/utils/times" "sig-pub/pkg/zlog" "sync" @@ -142,10 +143,7 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch 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] - + receivedTs := time.Now().UnixMilli() // 交易所 instid 转 sig-instid var tradeInst *types.TradeInstance exchangeInst, ok := exchange.ExchangeInsts.Load(channelK.ExgInstId) @@ -155,15 +153,47 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch } tradeInst = exchangeInst.Inst + // 升序排序 + collect.SortAsc(channelK.Klines, func(k *types.Kline) int64 { return k.Ts }) + + // 检查已确认k线是否连续并补齐 + for _, kline := range channelK.Klines { + if kline.Confirm { + if kms, ok := kline.Interval.AddMul(kline.Ts, 1); ok { + delay := time.Now().UnixMilli() - kms + zlog.Debugf("recv confirm kline: delay=%dms, inst=%s(%s), interval=%s", delay, channelK.ExgInstId, channelK.Exchange, kline.Interval) + + if lastConfirmK := exchangeInst.LastKline.Get(kline.Interval); lastConfirmK.Ts != 0 { + if expectTs, ok := lastConfirmK.Interval.AddMul(lastConfirmK.Ts, 1); ok && expectTs != kline.Ts { + zlog.Warningf("fetching padding klines: inst=%s(%s), interval=%s, ts=%d~%d", tradeInst.InstId, tradeInst.Exchange, kline.Interval, kline.Ts, lastConfirmK.Ts) + if err := svc.initialTradeInstanceKlines(exchange, *tradeInst); err != nil { + zlog.Errorf("fetch padding kline error: inst=%s(%s), interval=%s, ts=%d~%d, error=%v", tradeInst.InstId, tradeInst.Exchange, kline.Interval, kline.Ts, lastConfirmK.Ts, err) + } + // zlog.Warningf("fetching padding klines: inst=%s(%s), interval=%s, ts=%d~%d", tradeInst.InstId, tradeInst.Exchange, kline.Interval, kline.Ts, lastConfirmK.Ts) + // paddingKlines, err := exchange.Fetcher.FetchHistoryKlines(context.Background(), tradeInst.ExchangeInstId, kline.Interval, kline.Ts, lastConfirmK.Ts) + // if err != nil { + // zlog.Errorf("fetch padding kline error: inst=%s(%s), interval=%s, ts=%d~%d, error=%v", tradeInst.InstId, tradeInst.Exchange, kline.Interval, kline.Ts, lastConfirmK.Ts, err) + // } else { + // zlog.Debugf("fetched padding klines: inst=%s(%s), interval=%s, ts=%d~%d, %#v", tradeInst.InstId, tradeInst.Exchange, kline.Interval, kline.Ts, lastConfirmK.Ts, paddingKlines) + // channelK.Klines = append(paddingKlines, channelK.Klines...) + // collect.SortAsc(channelK.Klines, func(k *types.Kline) int64 { return k.Ts }) + // } + } + } + } + break + } + } + + // 取出头尾k线 + lastKline := channelK.Klines[len(channelK.Klines)-1] + // 标记交易产品开始订阅k线时间 - exchangeInst.LiveKStartTs.SetIf(firstKline.Interval, firstKline.Ts, func(old int64) bool { return old == 0 }) + // exchangeInst.LiveKStartTs.SetIf(lastKline.Interval, lastKline.Ts, func(old int64) bool { return old == 0 }) // 记录实时k线 exchangeInst.LiveKline.Set(lastKline.Interval, *lastKline) - // 记录最后确认k线 - if lastKline.Confirm { - exchangeInst.LastKline.Set(lastKline.Interval, *lastKline) - } + // 记录实时价格 exchangeInst.Last = lastKline.Close @@ -172,6 +202,8 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch // zlog.Infof("recv kline: %#v", kline) confirm := 0 if kline.Confirm { + // 记录最后确认k线 + exchangeInst.LastKline.Set(kline.Interval, *kline) confirm = 1 confirmKlines = append(confirmKlines, kline) } @@ -188,7 +220,7 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch } // 存储到 tsdb - if _, ok := types.SupportedIntervals[firstKline.Interval]; ok && len(confirmKlines) > 0 { + 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) @@ -221,6 +253,10 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch } } } + + if useMs := time.Now().UnixMilli() - receivedTs; useMs > 10 { + zlog.Debugf("handle consume kline use: %dms", useMs) + } } } @@ -304,7 +340,15 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade } }) exchangeInst.Status.Store(int32(data.StatusOk)) - zlog.Infof("initial history kline finish: instId=%s(%s), pub=%d, sub=%d, fail=%d, use %s", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load(), subTasks.Load(), failTimes.Load(), conver.TimeMilliFormat(time.Now().UnixMilli()-startTs, "/")) + zlog.Infof("initial history kline finish: instId=%s(%s), pub=%d, sub=%d, fail=%d, use %s", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load(), subTasks.Load(), failTimes.Load(), conver.TimeMilliFormat(time.Now().UnixMilli()-startTs, ".")) + + // flush vmtsdb to disk + retry.DoWithFixDelay(5, time.Second, func(retryTimes uint32) (_ struct{}, err error) { + if err = svc.exchangeDataPersist.vmtsdb.ForceFlush(); err != nil { + zlog.Errorf("flush vmts db error: ", err) + } + return + }) }() // progress monitor @@ -345,15 +389,21 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade } 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), KlineBefore0) + } } exchangeInst.HistoryMarkTs.Set(interval, beforeTs) for { // 判定订阅任务发布完成 - liveStartTs := exchangeInst.LiveKStartTs.Get(interval) - if liveStartTs != 0 && beforeTs >= liveStartTs { - break - } + // liveStartTs := exchangeInst.LiveKStartTs.Get(interval) + // if liveStartTs != 0 && beforeTs >= liveStartTs { + // break + // } if beforeTs > time.Now().UnixMilli() { break } @@ -478,14 +528,14 @@ func (svc *ExchangeService) Exchanges() (exchanges []pb.ExchangeType, err error) } // ExchangeInstanceState 交易所交易产品状态 -func (svc *ExchangeService) ExchangeInstanceState(allExchange bool, exchangeTypes []pb.ExchangeType, instIds []string) (states []*pb.TradeInstanceState, err error) { +func (svc *ExchangeService) ExchangeInstanceState(req *pb.ReqExchangeInstanceState) (states []*pb.TradeInstanceState, err error) { var exchanges []*Exchange - if allExchange { + if req.AllExchange { svc.exchanges.Range(func(_ pb.ExchangeType, exchange *Exchange) { exchanges = append(exchanges, exchange) }) } else { - for _, exchangeType := range exchangeTypes { + for _, exchangeType := range req.Exchanges { if !svc.exchanges.IsSupport(exchangeType) { err = fmt.Errorf("not support exchange: %v", exchangeType) return @@ -499,20 +549,34 @@ func (svc *ExchangeService) ExchangeInstanceState(allExchange bool, exchangeType } for _, exchange := range exchanges { - for _, instId := range instIds { - // trade instId to exchangeInstId - exchangeInstId, ok := exchange.TradeInstIds.Load(instId) - if !ok { - continue - } - exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId) - if !ok { - continue + 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: instId, + InstId: exchangeInst.Inst.InstId, Status: exchangeInst.Status.Load(), Last: exchangeInst.Last.String(), } @@ -522,6 +586,10 @@ func (svc *ExchangeService) ExchangeInstanceState(allExchange bool, exchangeType return } +const ( + MaxHistoryKlines = 100 +) + // HistoryKline 获取交易产品历史k线 (before < klines... < after) func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistoryKline, rsp *pb.RspHistoryKline) (klines []*types.Kline, err error) { // 交易产品参数检查 @@ -548,59 +616,56 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory // k线长度检查 afterTs, beforeTs, count := int64(req.After), int64(req.Before), int64(req.Count) if count == 0 { - count = 100 + count = MaxHistoryKlines } - nowTime := time.Now().UnixMilli() - if afterTs == 0 && beforeTs == 0 { - // 拉取最新的 + // 拉取最新的 + lastTs := int64(0) + if afterTs == 0 { liveK := exchangeInst.LiveKline.Get(interval) - lastTs := liveK.Ts + lastTs = liveK.Ts if !liveK.Confirm { lastTs = intervalAdder(liveK.Ts, -1) } - if !req.Asc { - afterTs = lastTs - } else { - // 从低到高拉取 - beforeTs = max(intervalAdder(lastTs, -count+1), KlineBefore0) - } + } + if afterTs == 0 && beforeTs == 0 { + afterTs = lastTs } if afterTs == 0 { - afterTs = min(intervalAdder(beforeTs, count), nowTime) + afterTs = min(intervalAdder(beforeTs, count-1), lastTs) } if beforeTs == 0 { - beforeTs = max(intervalAdder(afterTs, -count), KlineBefore0) + beforeTs = max(intervalAdder(afterTs, -count+1), KlineBefore0) } if beforeTs > afterTs { err = fmt.Errorf("time range invalid: before must less then after") return } // 限制最大时间范围 - total := (afterTs - beforeTs) / intervalAdder(0, 1) - if total > 100 { - // err = fmt.Errorf("time range too large max 100") - if req.Asc { - afterTs = min(intervalAdder(beforeTs, 100), nowTime) - } else { - beforeTs = max(intervalAdder(afterTs, -100), KlineBefore0) - } - rsp.Next = true + total := (afterTs-beforeTs)/intervalAdder(0, 1) + 1 + if total > MaxHistoryKlines { + err = fmt.Errorf("time range too large max %d", MaxHistoryKlines) + return } klines, err = svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, beforeTs, afterTs) - if err != nil || len(klines) == 0 { - rsp.Next = false + if err != nil { + zlog.Error("list vmtsdb kline error: ", err) return } - if len(klines) < int(count) { - rsp.Next = false + if len(klines) == 0 { + return } - lastK := klines[len(klines)-1] + lastK := klines[len(klines)-1] // vmtsdb 数据刷盘30s延迟, 使用内存数据替代第一根k线 - if lastConfirmK := exchangeInst.LastKline.Get(interval); lastConfirmK.Ts == lastK.Ts { - klines[len(klines)-1] = &lastConfirmK + 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) } // 降序排序 @@ -624,7 +689,7 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory } // 查询历史k线(流式返回) -func (svc *ExchangeService) HistoryKlineStream(req *pb.ReqHistoryKline, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) { +func (svc *ExchangeService) HistoryKlineStream(req *pb.ReqHistoryKlineStream, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) { // 交易产品参数检查 if !svc.exchanges.IsSupport(req.Exchange) { err = fmt.Errorf("exchange not support: %s", req.Exchange) @@ -674,34 +739,39 @@ func (svc *ExchangeService) HistoryKlineStream(req *pb.ReqHistoryKline, stream g return } - branch := 100 - curBeforeTs, curAfterTs := beforeTs, intervalAdder(beforeTs, int64(branch)-1) - for i := 0; curAfterTs <= afterTs; i++ { - if i > 0 { - curBeforeTs = intervalAdder(curAfterTs, 1) - curAfterTs = min(intervalAdder(curBeforeTs, int64(branch)-1), afterTs) - } + klines, err := svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, beforeTs, afterTs) + if err != nil { + zlog.Error("fetch history kline stream error: ", err) + return + } + if len(klines) == 0 { + return + } + lastK := klines[len(klines)-1] + // vmtsdb 数据刷盘30s延迟, 使用内存数据替代第一根k线 + lastConfirmK := exchangeInst.LastKline.Get(interval) + if lastConfirmK.Ts == lastK.Ts { + klines[len(klines)-1] = &lastConfirmK + lastK = klines[len(klines)-1] + } + if lastConfirmK.Ts == afterTs && intervalAdder(lastK.Ts, 1) == afterTs { + klines = append(klines, &lastConfirmK) + } - klines, kerr := svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, curBeforeTs, curAfterTs) - if kerr != nil { - err = kerr - return - } - if len(klines) == 0 { - break + branch := 100 + length := len(klines) + kBuffer := make([]*pb.Kline, 0, branch) + for i, kline := range klines { + kBuffer = append(kBuffer, kline.ToPBKline()) + if len(kBuffer) < branch && i < length-1 { + continue } - - resp := new(pb.RspHistoryKlineStream) - resp.Klines = collect.Mapping(klines, func(_ int, k *types.Kline) *pb.Kline { return k.ToPBKline() }) - // 是否还有更多 - resp.Next = len(klines) == int(count) - if sendErr := stream.Send(resp); sendErr != nil { + rsp := &pb.RspHistoryKlineStream{Klines: kBuffer} + if sendErr := stream.Send(rsp); sendErr != nil { err = sendErr return } - if !resp.Next { - break - } + kBuffer = kBuffer[:0] } return } diff --git a/internal/exchange/okx/okx_fetch.go b/internal/exchange/okx/okx_fetch.go index 73d3865..3425fbf 100644 --- a/internal/exchange/okx/okx_fetch.go +++ b/internal/exchange/okx/okx_fetch.go @@ -53,7 +53,7 @@ func (okx *OkxFetcher) ExhcangeType() pb.ExchangeType { // FetchHistoryKlines 获取交易产品历史K线数据 // https://my.okx.com/docs-v5/zh/#order-book-trading-market-data-get-candlesticks-history -// 周期区间 after > before, (after, before) +// 开区间降序响应 after > before, (after, before) func (f *OkxFetcher) FetchHistoryKlines(ctx context.Context, okxInstId string, interval types.Interval, after, before int64) (klines []*types.Kline, err error) { if okxInstId == "" { err = errors.New("instid is empty") diff --git a/internal/trading/kline_series.go b/internal/trading/kline_series.go index bf85748..b2f690d 100644 --- a/internal/trading/kline_series.go +++ b/internal/trading/kline_series.go @@ -3,10 +3,12 @@ package trading import ( "fmt" "sig-pub/api/pb" + "sig-pub/pkg/data" "sig-pub/pkg/types" "sig-pub/pkg/types/series" "sig-pub/pkg/zlog" "sync" + "sync/atomic" ) const ( @@ -14,6 +16,23 @@ const ( MaxCapSeriesKlines = 1280 // 最大k线数量, slice扩容12次后cap=1280 ) +// TradeInstanceKlineSeries 单个交易产品所有周期k线 +type TradeInstanceKlineSeries struct { + IntervalKlines *types.IntervalState[*KlineSeries] + Status atomic.Int32 // 交易产品状态 +} + +func NewTradeInstanceKlineSeries(exchange pb.ExchangeType, instId string) *TradeInstanceKlineSeries { + si := new(TradeInstanceKlineSeries) + si.Status.Store(int32(data.StatusNone)) + si.IntervalKlines = types.NewIntervalState[*KlineSeries]() + + for interval := range types.SupportedIntervals { + si.IntervalKlines.Set(interval, NewKlineSeries(exchange, instId, interval)) + } + return si +} + type KlineSeries struct { sync.RWMutex Exchange pb.ExchangeType @@ -38,8 +57,7 @@ func NewKlineSeries(exchange pb.ExchangeType, instId string, interval types.Inte } } -// Get -// [0]当前k线 +// Get [0]当前k线 func (s *KlineSeries) Get(start int16) types.Kline { index := len(s.klines) - 1 - int(start) if index >= 0 && index < len(s.klines)-1 { @@ -59,6 +77,8 @@ func (s *KlineSeries) Get(start int16) types.Kline { // Series [start...end] func (s *KlineSeries) Series(start, end int16) (klines series.Klines) { + s.RLock() + defer s.RUnlock() endTs := s.Interval.MustAddMul(s.lastTs, int64(-start)) startTs := s.Interval.MustAddMul(s.lastTs, int64(-end)) _ = endTs @@ -90,6 +110,12 @@ func (s *KlineSeries) Update(kline *types.Kline) (lastTs int64, serial bool) { if len(s.klines) < MaxCapSeriesKlines { s.klines = append(s.klines, kline) + // 容量超过0.65, 直接扩容到最大 + if capacity := cap(s.klines); capacity < MaxCapSeriesKlines && capacity > MaxCapSeriesKlines*0.65 { + klines := make([]*types.Kline, len(s.klines), MaxCapSeriesKlines) + copy(klines, s.klines) + s.klines = klines + } } else { // 循环复用切片空间，避免扩容 length := len(s.klines) diff --git a/internal/trading/kline_series_test.go b/internal/trading/kline_series_test.go deleted file mode 100644 index 1559a98..0000000 --- a/internal/trading/kline_series_test.go +++ /dev/null @@ -1,34 +0,0 @@ -package trading - -import ( - "fmt" - "testing" -) - -func TestSliceCap(t *testing.T) { - Max, MaxCap := 1000, 1280 - _ = MaxCap - var arr []int - prevCap := cap(arr) - for i := range 1600 { - ele := i - if len(arr) < MaxCap { - arr = append(arr, ele) - } else { - length := len(arr) - copy(arr, arr[length-Max+1:]) - arr[Max-1] = ele - arr = arr[:Max] - fmt.Printf("move=%d, %d, arr=(%d~%d)\n", i, len(arr), arr[0], arr[len(arr)-1]) - } - - c := cap(arr) - if c != prevCap { - // 发生扩容 - prevCap = c - fmt.Printf("i=%d, cap=%d, len=%d\n", i, c, len(arr)) - } - } - fmt.Println("----------------------------------------------------------") - fmt.Println(arr) -} diff --git a/internal/trading/kline_store.go b/internal/trading/kline_store.go index 36c16bf..2526dba 100644 --- a/internal/trading/kline_store.go +++ b/internal/trading/kline_store.go @@ -2,7 +2,6 @@ package trading import ( "context" - "fmt" "io" "math" "sig-pub/api/pb" @@ -12,8 +11,6 @@ import ( "sig-pub/pkg/utils/collect" "sig-pub/pkg/utils/retry" "sig-pub/pkg/zlog" - "sync" - "sync/atomic" "time" "google.golang.org/grpc" @@ -22,7 +19,7 @@ import ( type KlineStore struct { exchangeClient pb.ExchangeServiceClient - store *types.ExchangeState[*collect.ConcurrentMap[string, *KlineStoreInstance]] // K线列表: []exchange + store *types.ExchangeState[*collect.ConcurrentMap[string, *TradeInstanceKlineSeries]] // K线列表: []exchange subKlineIntervals []string // 订阅的k线的周期列表 subKlineInsts *types.ExchangeState[*collect.SyncMap[string, bool]] // 订阅k线中的交易产品列表 subKlineStream grpc.BidiStreamingClient[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline] // 订阅k线的stream @@ -36,12 +33,12 @@ func NewKlineSeriesStore(exchangeClient pb.ExchangeServiceClient) (kss *KlineSto return string(interval) }) // 产品列表 - kss.subKlineInsts = types.NewExchangeState0(func() *collect.SyncMap[string, bool] { + kss.subKlineInsts = types.NewExchangeStateInit(func() *collect.SyncMap[string, bool] { return collect.NewSyncMap[string, bool]() }) // 各交易所 store 初始化 - kss.store = types.NewExchangeState0(func() *collect.ConcurrentMap[string, *KlineStoreInstance] { - return collect.NewConcurrentMap[string, *KlineStoreInstance](64, func(s string) string { + kss.store = types.NewExchangeStateInit(func() *collect.ConcurrentMap[string, *TradeInstanceKlineSeries] { + return collect.NewConcurrentMap[string, *TradeInstanceKlineSeries](64, func(s string) string { return s }) }) @@ -57,11 +54,27 @@ func (s *KlineStore) Init() (err error) { func(msg *mq.PublishExchangeTradeInstanceInited) (err error) { // 初始化k线, 开始订阅k线 zlog.Infof("subscribed TopicExchangeTradeInstanceInited: %#v", msg) - go s.initKlineSeries(msg.Exchange, msg.InstId) + go s.inititalKlineSeries(msg.Exchange, msg.InstId) return }) - // todo 拉取已初始化完成交易产品, 初始化k线, 开始订阅k线 + go func() { + // 拉取已初始化完成交易产品, 初始化k线, 开始订阅k线 + rsp, _ := retry.DoWithFixDelay(math.MaxInt32, time.Second, func(retryTimes uint32) (rsp *pb.RspExchangeInstanceState, err error) { + rsp, err = s.exchangeClient.ExchangeInstanceState(context.Background(), &pb.ReqExchangeInstanceState{ + AllExchange: true, + AllInsts: true, + Status: []int32{int32(data.StatusOk)}, + }) + if err != nil { + zlog.Errorf("ExchangeInstanceState error: retry=%d, %v", retryTimes, err) + } + return + }) + for _, inst := range rsp.InstsState { + s.inititalKlineSeries(inst.Exchange, inst.InstId) + } + }() return } @@ -115,7 +128,10 @@ func (s *KlineStore) connectSubscribeKline(reconnect bool) { for _, k := range msg.Kline.Klines { kline := new(types.Kline) kline.ParsePBKline(msg.Kline.Exchange, k) - zlog.Debugf("recv: streamId=%d, %v, %s, %#v", msg.Kline.StreamId, msg.Kline.Exchange, msg.Kline.InstId, kline) + if kms, ok := kline.Interval.AddMul(kline.Ts, 1); ok { + delay := time.Now().UnixMilli() - kms + zlog.Debugf("recv kline: streamId=%d, delay=%dms, inst=%s(%v), interval=%s, close=%s(%v)", msg.Kline.StreamId, delay, msg.Kline.InstId, msg.Kline.Exchange, kline.Interval, kline.Close.String(), kline.Confirm) + } // kline klineStore -> klineSeries -> strategy -> indicator -> klineSeries.Series s.Update(msg.Kline.Exchange, msg.Kline.InstId, kline) } @@ -146,9 +162,9 @@ func (s *KlineStore) sendSubscribeKline(save bool, exchange pb.ExchangeType, ins if s.subKlineStream == nil { return } - zlog.Debugf("send stream subscribe kline msg: retry=%d, %#v", retry, subMsg) + zlog.Debugf("sending stream subscribe kline: retry=%d, instId=%s%v", retry, exchange, instIds) if err = s.subKlineStream.Send(subMsg); err != nil { - zlog.Errorf("send stream subscribe kline msg error: %v", subMsg, err) + zlog.Errorf("send stream subscribe kline msg error: %#v", subMsg, err) return } return @@ -160,160 +176,107 @@ func (s *KlineStore) sendSubscribeKline(save bool, exchange pb.ExchangeType, ins go retry.DoWithFixDelay(math.MaxInt32, time.Second, doSend) } -func (s *KlineStore) initKlineSeries(exchange pb.ExchangeType, instId string) { +func (s *KlineStore) inititalKlineSeries(exchange pb.ExchangeType, instId string) { if !s.store.IsSupport(exchange) { + zlog.Errorf("unsupport exchange %s", exchange) return } - - storeInst := s.store.Get(exchange).ComputeIfAbsent(instId, func(k string) *KlineStoreInstance { - return NewKlineStoreInstance(exchange, k) + storeInst := s.store.Get(exchange).ComputeIfAbsent(instId, func(k string) *TradeInstanceKlineSeries { + return NewTradeInstanceKlineSeries(exchange, k) }) + // 交易产品已初始化过 + if !storeInst.Status.CompareAndSwap(int32(data.StatusNone), int32(data.StatusProcessing)) { + return + } + zlog.Infof("initial kline series starting: %s(%s),", instId, exchange) // 初始化最新的 klineSeries for _, interval := range s.subKlineIntervals { - for { - before := int64(0) - rsp, err := retry.DoWithFixDelay(math.MaxInt32, 2*time.Second, func(retryTimes uint32) (rsp *pb.RspHistoryKline, err error) { - rsp, err = s.exchangeClient.HistoryKline(context.Background(), &pb.ReqHistoryKline{ - Exchange: exchange, - InstId: instId, - Interval: string(interval), - Count: MaxSeriesKlines, - After: 0, - Before: before, - Live: false, - Asc: true, - }, grpc.UseCompressor("snappy")) - if err != nil { - zlog.Errorf("fetch missing klines error: instId=%s(%s) before=%d, err=%v", instId, exchange, before, err) - } - return - }) - if err != nil { - zlog.Errorf("trade instance initial failed: %s(%s), %v", instId, exchange, err) - return - } - klines := rsp.Klines - if len(klines) == 0 { - break - } - for _, kline := range klines { - k := new(types.Kline) - k.ParsePBKline(exchange, kline) - _, _, err = storeInst.Update(k) - if err != nil { - zlog.Error("update initial kline series error: instId=%s(%s)", instId, exchange, err) - return - } - - s.Update(exchange, instId, k) - } - - before = klines[len(klines)-1].Ts - if !rsp.Next { - break - } - } + retry.DoWithFixDelay(math.MaxInt32, 2*time.Second, func(retryTimes uint32) (_ struct{}, err error) { + err = s.fetchHistoryKlineToSeries(exchange, instId, interval, 0, 0, MaxSeriesKlines) + return + }) } // 初始化历史k线完成, 开始订阅k线 storeInst.Status.Store(int32(data.StatusOk)) s.sendSubscribeKline(true, exchange, instId) + zlog.Infof("initial kline series success: %s(%s),", instId, exchange) } -// Update -// kline klineStore -> klineSeries -> strategy -> indicator -> klineSeries.Series -func (s *KlineStore) Update(exchange pb.ExchangeType, instId string, kline *types.Kline) { - storeInst := s.store.Get(exchange).ComputeIfAbsent(instId, func(k string) *KlineStoreInstance { - return NewKlineStoreInstance(exchange, k) - }) - - // 只处理已初始化完成的交易产品k线 - if storeInst.Status.Load() != int32(data.StatusOk) { - return +// fetchHistoryKlineToSeries 拉去历史k线数据更新series +func (s *KlineStore) fetchHistoryKlineToSeries(exchange pb.ExchangeType, instId, interval string, before, after int64, count uint32) (err error) { + // 拉取最新的1000条k线 + req := &pb.ReqHistoryKlineStream{ + Exchange: exchange, + InstId: instId, + Interval: interval, + Count: count, + Before: before, + After: after, } - before, serial, err := storeInst.Update(kline) + stream, err := s.exchangeClient.HistoryKlineStream(context.Background(), req, grpc.UseCompressor("snappy")) if err != nil { - zlog.Error("update kline series error: instId=%s(%s)", instId, exchange, err) + zlog.Errorf("fetch history kline stream error: instId=%s(%s), interval=%s, %#v, err=%v", instId, exchange, interval, req, err) return } - if !serial { - // 拉取缺失的k线 - s.paddingMissKlines(storeInst, exchange, instId, kline.Interval, kline.Ts, before) - // emit kline event - } -} -// 拉取缺失的k线 -func (s *KlineStore) paddingMissKlines(storeInst *KlineStoreInstance, exchange pb.ExchangeType, instId string, interval types.Interval, after, before int64) { - // 拉取缺失的k线 - rsp, err := retry.DoWithFixDelay(math.MaxInt32, 2*time.Second, func(retryTimes uint32) (rsp *pb.RspHistoryKline, err error) { - rsp, err = s.exchangeClient.HistoryKline(context.Background(), &pb.ReqHistoryKline{ - Exchange: exchange, - InstId: instId, - Interval: string(interval), - After: after, - Before: before, - Live: false, - }, grpc.UseCompressor("snappy")) - if err != nil { - zlog.Errorf("fetch missing klines error: instId=%s(%s) after=%d before=%d, err=%v", instId, exchange, after, before, err) + for { + msg, err0 := stream.Recv() + if err0 == io.EOF { + // zlog.Debugf("fetch kline stream connection server closed") + break } - return - }) - if err != nil { - return - } - - collect.Reverse(rsp.Klines) - for _, kline := range rsp.Klines { - k := new(types.Kline) - k.ParsePBKline(exchange, kline) - zlog.Infof("padding missing kline: instId=%s(%s) %s %#v", instId, exchange, interval, k) - _, _, err := storeInst.Update(k) - if err != nil { - zlog.Error("update missing kline series error: instId=%s(%s)", instId, exchange, err) + if err0 != nil { + err = err0 + zlog.Error("fetch kline stream recv error: ", err0) + return + } + // zlog.Debugf("recv: %s(%s), %s, branch=%d, ts=%d~%d", instId, exchange, interval, len(msg.Klines), msg.Klines[0].Ts, msg.Klines[len(msg.Klines)-1].Ts) + for _, k := range msg.Klines { + kline := new(types.Kline) + kline.ParsePBKline(exchange, k) + s.Update(exchange, instId, kline) } } + return } -// KlineStoreInstance 单个交易产品所有周期k线 -type KlineStoreInstance struct { - sync.RWMutex - Exchange pb.ExchangeType - InstId string - intervalKlines *types.IntervalState[*KlineSeries] - Status atomic.Int32 // 交易产品状态 -} - -func NewKlineStoreInstance(exchange pb.ExchangeType, instId string) *KlineStoreInstance { - si := &KlineStoreInstance{ - Exchange: exchange, - InstId: instId, - intervalKlines: types.NewIntervalState[*KlineSeries](), - } - si.Status.Store(int32(data.StatusProcessing)) - - for interval := range types.SupportedIntervals { - si.intervalKlines.Set(interval, NewKlineSeries(exchange, instId, interval)) - } - return si -} - -// Update 更新k线 -// serial k线是否连续 -func (si *KlineStoreInstance) Update(kline *types.Kline) (before int64, serial bool, err error) { +// Update +// kline klineStore -> klineSeries -> strategy -> indicator -> klineSeries.Series +func (s *KlineStore) Update(exchange pb.ExchangeType, instId string, kline *types.Kline) { if _, ok := types.SupportedIntervals[kline.Interval]; !ok { - err = fmt.Errorf("unsupport interval: %s", kline.Interval) + zlog.Warningf("unsupport interval: %s", kline.Interval) + return + } + if !s.store.IsSupport(exchange) { + zlog.Warningf("unsupport exchange: %v", exchange) return } - si.Lock() - defer si.Unlock() + instSeries := s.store.Get(exchange).ComputeIfAbsent(instId, func(k string) *TradeInstanceKlineSeries { + return NewTradeInstanceKlineSeries(exchange, k) + }) - ks := si.intervalKlines.Get(kline.Interval) - before, serial = ks.Update(kline) - if !serial { - // si.status.Store(int32(data.StatusProcessing)) + before, serial := instSeries.IntervalKlines.Get(kline.Interval).Update(kline) + if !serial && instSeries.Status.CompareAndSwap(int32(data.StatusOk), int32(data.StatusProcessing)) { + func() { + defer instSeries.Status.Store(int32(data.StatusOk)) + // 拉取缺失的k线 + after := kline.Ts + zlog.Debugf("fetching padding kline series: instId=%s(%s), interval=%s, ts=%d~%d", instId, exchange, kline.Interval, before, after) + err := s.fetchHistoryKlineToSeries(exchange, instId, string(kline.Interval), before, after, 0) + if err != nil { + zlog.Error("fetch padding kline series error: instId=%s(%s), interval=%s, ts=%d~%d, err=%v", instId, exchange, kline.Interval, before, after, err) + return + } + }() + } + + if expTs, ok := kline.Interval.AddMul(kline.Ts, 2); ok { + // k线已过期则不执行策略 + if expTs < time.Now().Unix() { + return + } + // todo emit kline update, calc indicator... } - return } diff --git a/pkg/backtrace/backtrace.go b/pkg/backtrace/backtrace.go new file mode 100644 index 0000000..56e75a1 --- /dev/null +++ b/pkg/backtrace/backtrace.go @@ -0,0 +1,13 @@ +package backtrace + +import "sig-pub/pkg/strategy" + +// 回测引擎 +type BacktraceEngine struct { + strategy strategy.Strategy +} + +// 多周期策略回测引擎 +type MultiIntervalBacktraceEngine struct { + strategy strategy.MultiIntervalStrategy +} diff --git a/pkg/client/trade_instance_client.go b/pkg/client/trade_instance_client.go index 39d26bd..6e4c41c 100644 --- a/pkg/client/trade_instance_client.go +++ b/pkg/client/trade_instance_client.go @@ -2,6 +2,7 @@ package client import ( "context" + "math" "sig-pub/api/pb" "sig-pub/pkg/data/entity" "sig-pub/pkg/mapping" @@ -64,7 +65,7 @@ func (c *TradeInstanceAside) getTradeInstance0(ctx context.Context, instId strin // ListExchangeTradeInstance 获取交易所支持的交易实例 func (c *TradeInstanceAside) ListExchangeTradeInstance(ctx context.Context, exchange pb.ExchangeType) (marketInsts []*pb.MarketTradeInstance, err error) { - return retry.DoWithStepDelay(10, time.Second, func(retryTimes uint32) ([]*pb.MarketTradeInstance, error) { + return retry.DoWithFixDelay(math.MaxInt32, time.Second, func(retryTimes uint32) ([]*pb.MarketTradeInstance, error) { rsp, err := c.marketClient.ListMarketTradeInstance(ctx, &pb.ReqListMarketTradeInstance{Exchange: exchange}) if err != nil { zlog.Errorf("list market trade instances error: retry %d times, %v", retryTimes, err) diff --git a/pkg/data/common.go b/pkg/data/common.go index f6dcdc6..072a733 100644 --- a/pkg/data/common.go +++ b/pkg/data/common.go @@ -6,7 +6,7 @@ import "errors" type Status int32 const ( - StatusDisabled Status = 0 + StatusNone Status = 0 StatusOk Status = 1 StatusProcessing Status = 2 StatusDeleted Status = 4 diff --git a/pkg/indicator/base.go b/pkg/indicator/base.go index 8e716b2..2053370 100644 --- a/pkg/indicator/base.go +++ b/pkg/indicator/base.go @@ -12,12 +12,12 @@ type IIndicator interface { // IKlineSeries k线序列, strategy服务提供 type IKlineSeries interface { - Get(start int16) (kline types.Kline) - Series(start, end int16) (klines series.Klines) + Get(offset int16) (kline types.Kline) + Series(offset, count int16) (klines series.Klines) } // IIndicatorSeries 指标序列, 供策略读取, strategy服务提供 type IIndicatorSeries interface { - Get(start int16) (vector float64) - Series(start, end int16) (matrix series.Floats) + Get(offset int16) (vector float64) + Series(offset, count int16) (matrix series.Floats) } diff --git a/pkg/strategy/gold_x.go b/pkg/strategy/gold_x.go new file mode 100644 index 0000000..ea5eb29 --- /dev/null +++ b/pkg/strategy/gold_x.go @@ -0,0 +1,31 @@ +package strategy + +// GoldX 金叉策略 +type GoldX struct { +} + +func (s *GoldX) New() Strategy { + return &GoldX{} +} + +func (s *GoldX) Meta() StrategyMeta { + return StrategyMeta{ + Name: "GoldX", + } +} + +func (s *GoldX) Update(ctx StrategyContext) { + sma14 := ctx.IndicatorW("sma", 14) + sma28 := ctx.IndicatorW("sma", 28) + // 包装方法 + s14 := sma14.Series(0, 2) + s28 := sma28.Series(0, 2) + crossover := s14[0] > s28[0] && s14[1] < s28[1] // 上穿 + crossunder := s14[0] < s28[0] && s14[1] > s28[1] // 下穿 + if crossover { + ctx.Buy() + } + if crossunder { + ctx.Sell() + } +} diff --git a/pkg/strategy/strategy.go b/pkg/strategy/strategy.go new file mode 100644 index 0000000..9ca337a --- /dev/null +++ b/pkg/strategy/strategy.go @@ -0,0 +1,35 @@ +package strategy + +import ( + "sig-pub/pkg/indicator" + "sig-pub/pkg/types" + "sig-pub/pkg/types/series" +) + +// todo Exit 止盈止损策略(trading service 管理) +// todo Meta 策略调参, 回测引擎自动调参回测(最佳参数) argGenerator.next() (arg, ok) +type Strategy interface { + New() Strategy + Meta() StrategyMeta + Update(ctx StrategyContext) +} + +type StrategyMeta struct { + // Id string `json:"id"` // 策略注册/执行器系统分配 + Name string + Desc string +} + +// StrategyContext 策略外部访问能力 +// klineSeries, Indicator +type StrategyContext interface { + Buy() // 发出多信号 + Sell() // 发出空信号 + + // Get [0]当前k线 + Get(offset int16) types.Kline + // Series [offset...end] + Series(offset, count int16) (klines series.Klines) + // 获取窗口类型指标 + IndicatorW(name string, window int) indicator.IIndicatorSeries +} diff --git a/pkg/strategy/strategy_multi_interval.go b/pkg/strategy/strategy_multi_interval.go new file mode 100644 index 0000000..2695801 --- /dev/null +++ b/pkg/strategy/strategy_multi_interval.go @@ -0,0 +1,18 @@ +package strategy + +import "sig-pub/pkg/types" + +// 多k线周期策略 +type MultiIntervalStrategy interface { + Strategy + DriverInterval() types.Interval // 驱动k线周期, 当驱动周期k线更新时则判断调用Update方法 + SubscribeIntervals() []types.Interval // 订阅k线周期, 当同一时间的订阅周期都更新时调用Update方法 +} + +type MultiExchangeStrategy interface { + Strategy +} + +type MultiIntervalExchangeStrategy interface { + Strategy +} diff --git a/pkg/types/exchange.go b/pkg/types/exchange.go index d7d8802..7f35ee1 100644 --- a/pkg/types/exchange.go +++ b/pkg/types/exchange.go @@ -23,10 +23,10 @@ type ExchangeState[T any] struct { } func NewExchangeState[T any]() *ExchangeState[T] { - return NewExchangeState0[T](func() (v T) { return }) + return NewExchangeStateInit(func() (v T) { return }) } -func NewExchangeState0[T any](newer func() T) *ExchangeState[T] { +func NewExchangeStateInit[T any](newer func() T) *ExchangeState[T] { maxExchange := collect.MustMax(SupportedExchanges, func(e pb.ExchangeType) int32 { return int32(e) }) es := &ExchangeState[T]{ state: make([]T, maxExchange+1), diff --git a/pkg/types/interval.go b/pkg/types/interval.go index 66e66ce..daafc80 100644 --- a/pkg/types/interval.go +++ b/pkg/types/interval.go @@ -78,7 +78,7 @@ var SupportedIntervals = IntervalMap{ Interval1h: func(ts, mul int64) (ret int64) { return ts + (60 * 60 * 1000 * mul) }, Interval2h: func(ts, mul int64) (ret int64) { return ts + (2 * 60 * 60 * 1000 * mul) }, Interval4h: func(ts, mul int64) (ret int64) { return ts + (4 * 60 * 60 * 1000 * mul) }, - Interval6h: func(ts, mul int64) (ret int64) { return ts + (4 * 60 * 60 * 1000 * mul) }, + Interval6h: func(ts, mul int64) (ret int64) { return ts + (6 * 60 * 60 * 1000 * mul) }, Interval12h: func(ts, mul int64) (ret int64) { return ts + (12 * 60 * 60 * 1000 * mul) }, Interval1d: func(ts, mul int64) (ret int64) { return ts + (24 * 60 * 60 * 1000 * mul) }, Interval2d: func(ts, mul int64) (ret int64) { return ts + (2 * 24 * 60 * 60 * 1000 * mul) },