From 684cee1bd2308b214386b53aae3feafd4c4380b3 Mon Sep 17 00:00:00 2001 From: strange Date: Mon, 20 Oct 2025 00:31:11 +0800 Subject: [PATCH] trading plan --- internal/exchange/exchange_grpc_server.go | 3 +- internal/exchange/exchange_service.go | 231 ++++++++++-------- internal/exchange/okx/ws_channel_subscribe.go | 1 + internal/trading/kline_series.go | 9 +- internal/trading/kline_store.go | 41 +++- internal/trading/trading_plan.go | 25 ++ internal/trading/trading_service.go | 71 +++++- pkg/backtrace/backtrace.go | 2 +- pkg/data/entity/trade_plan.go | 21 ++ pkg/indicator/base.go | 9 +- pkg/indicator/rsi.go | 29 +-- pkg/indicator/rsi0.go | 61 ----- .../exchange => pkg/publish}/publisher.go | 2 +- pkg/{stream => publish}/stream.go | 2 +- pkg/strategy/gold_x.go | 5 +- pkg/strategy/strategy.go | 32 ++- pkg/strategy/strategy_multi_interval.go | 6 +- pkg/types/interval.go | 14 +- pkg/utils/collect/sync_map.go | 8 + pkg/utils/conver/unit_conver.go | 5 +- pkg/utils/times/watch.go | 33 +++ 21 files changed, 398 insertions(+), 212 deletions(-) create mode 100644 internal/trading/trading_plan.go create mode 100644 pkg/data/entity/trade_plan.go delete mode 100644 pkg/indicator/rsi0.go rename {internal/exchange => pkg/publish}/publisher.go (98%) rename pkg/{stream => publish}/stream.go (96%) create mode 100644 pkg/utils/times/watch.go diff --git a/internal/exchange/exchange_grpc_server.go b/internal/exchange/exchange_grpc_server.go index 6b5f2a8..fb8902f 100644 --- a/internal/exchange/exchange_grpc_server.go +++ b/internal/exchange/exchange_grpc_server.go @@ -5,6 +5,7 @@ import ( "fmt" "io" "sig-pub/api/pb" + "sig-pub/pkg/publish" "sig-pub/pkg/zlog" "sync/atomic" @@ -16,7 +17,7 @@ type ExchangeGrpcServer struct { exchangeService *ExchangeService klineStreamId int64 - klineSubscriber *Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]] + klineSubscriber *publish.Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]] } // exchanges: 支持的数据源交易所 diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index b53a115..cb3c074 100644 --- a/internal/exchange/exchange_service.go +++ b/internal/exchange/exchange_service.go @@ -9,9 +9,9 @@ import ( "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/conver" "sig-pub/pkg/utils/retry" "sig-pub/pkg/utils/times" "sig-pub/pkg/zlog" @@ -29,7 +29,7 @@ type ExchangeService struct { tradeInstanceAside *client.TradeInstanceAside exchangeDataPersist *ExchangeDataPersist - klinePublisher *Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]] + klinePublisher *publish.Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]] } // exchanges: 支持的数据源交易所 @@ -50,7 +50,7 @@ func NewExchangeService( exchanges: exchangeState, tradeInstanceAside: tradeInstanceAside, exchangeDataPersist: exchangeDataPersist, - klinePublisher: NewPublisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]](16), + klinePublisher: publish.NewPublisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]](16), } } @@ -60,7 +60,7 @@ func (svc *ExchangeService) Init() (err error) { } // GetKlineSubscriber 订阅k线订阅器 -func (svc *ExchangeService) GetKlineSubscriber() (subscriber *Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]]) { +func (svc *ExchangeService) GetKlineSubscriber() (subscriber *publish.Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]]) { subscriber = svc.klinePublisher return } @@ -143,7 +143,11 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch if len(channelK.Klines) == 0 { continue } - receivedTs := time.Now().UnixMilli() + // 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) @@ -156,7 +160,8 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch // 升序排序 collect.SortAsc(channelK.Klines, func(k *types.Kline) int64 { return k.Ts }) - // 检查已确认k线是否连续并补齐 + // k线完整性检查, k线是否连续并补齐 + padding := false for _, kline := range channelK.Klines { if kline.Confirm { if kms, ok := kline.Interval.AddMul(kline.Ts, 1); ok { @@ -165,19 +170,19 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch if lastConfirmK := exchangeInst.LastKline.Get(kline.Interval); lastConfirmK.Ts != 0 { if expectTs, ok := lastConfirmK.Interval.AddMul(lastConfirmK.Ts, 1); ok && expectTs != kline.Ts { + padding = true + startTs := time.Now().UnixMilli() 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 { + if err := svc.paddingTradeInstanceIntervalKlines(4, exchange, *tradeInst, kline.Interval); 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 { + padding = false + zlog.Infof("fetched padding klines: inst=%s(%s), interval=%s, ts=%d~%d, use=%dms", tradeInst.InstId, tradeInst.Exchange, kline.Interval, kline.Ts, lastConfirmK.Ts, time.Now().UnixMilli()-startTs) + // flush vmtsdb to disk + if err = svc.exchangeDataPersist.vmtsdb.ForceFlush(); err != nil { + zlog.Errorf("flush vmts db error: ", 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 }) - // } } } } @@ -188,9 +193,6 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch // 取出头尾k线 lastKline := channelK.Klines[len(channelK.Klines)-1] - // 标记交易产品开始订阅k线时间 - // exchangeInst.LiveKStartTs.SetIf(lastKline.Interval, lastKline.Ts, func(old int64) bool { return old == 0 }) - // 记录实时k线 exchangeInst.LiveKline.Set(lastKline.Interval, *lastKline) @@ -227,15 +229,14 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch if err != nil { zlog.Errorf("kline save to tsdb error: %v, %#v", err, confirmKlines) } else { - // 初始化状态完成, 更新k线时间戳标记 - if exchangeInst.Status.Load() == int32(data.StatusOk) { + // k线未缺失, 初始化状态完成, 更新k线时间戳标记 + if !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) } - // todo k线完整性检查 } } } @@ -254,9 +255,9 @@ 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) - } + // if useMs := time.Now().UnixMilli() - receivedTs; useMs > 10 { + // zlog.Debugf("handle consume kline use: %dms", useMs) + // } } } @@ -284,10 +285,10 @@ func (svc *ExchangeService) initialKlines(exchange *Exchange, insts []types.Trad var success, failed []types.TradeInstance for _, inst := range insts { - err := svc.initialTradeInstanceKlines(exchange, inst) + err := svc.paddingTradeInstanceKlines(exchange, inst) status := data.StatusFailed if err != nil { - zlog.Errorf("initial fetch trade instance error: %s(%s), err=%v", inst.InstId, inst.Exchange, err) + 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) @@ -307,16 +308,61 @@ func (svc *ExchangeService) initialKlines(exchange *Exchange, insts []types.Trad 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) { +// 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.vmtsdb.ForceFlush(); 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) @@ -324,35 +370,30 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade // 发布任务数, 成功任务数, 失败任务次数 var pubTasks, subTasks, failTimes atomic.Int32 var pubTaskDone atomic.Bool // 所有任务已发布 - startTs := time.Now().UnixMilli() + watch := times.NewWatch() 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)) - 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("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(".")) - // 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 - }) + 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) + } }() - // progress monitor + // 任务进度日志(执行超过3s打印进度) go func() { + select { + case <-ctx.Done(): + return + case <-time.After(3 * time.Second): + } + ticker := time.NewTicker(time.Second) for { select { @@ -360,12 +401,12 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade ticker.Stop() return case <-ticker.C: - zlog.Debugf("processing initial history kline tasks: %s(%s), pub %d, sub %d, fail %d", tradeInst.InstId, tradeInst.Exchange, pubTasks.Load(), subTasks.Load(), failTimes.Load()) + 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()) } } }() - // fetch kline history task publish + // 任务发布器 go func() { defer func() { pubTaskDone.Store(true) @@ -373,60 +414,56 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade if pubTasks.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 + // 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) } - 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) - } + 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)) } + } - 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, - } + for { + // 判定订阅任务发布完成 + if intervalAdder(beforeTs, 2) > time.Now().UnixMilli() { + break + } - // 发布任务 - select { - case taskCh <- task: - pubTasks.Add(1) - case <-ctx.Done(): - return - } + afterTs := intervalAdder(beforeTs, 101) + task := fetchKlineTask{ + inst: tradeInst, + interval: interval, + afterTs: afterTs, + beforeTs: beforeTs, + times: 0, + } - beforeTs = intervalAdder(afterTs, -1) + // 发布任务 + select { + case taskCh <- task: + pubTasks.Add(1) + case <-ctx.Done(): + return } + + beforeTs = intervalAdder(afterTs, -1) } }() @@ -747,6 +784,8 @@ func (svc *ExchangeService) HistoryKlineStream(req *pb.ReqHistoryKlineStream, st if len(klines) == 0 { return } + // todo 检查k线是否连续进行补齐 + lastK := klines[len(klines)-1] // vmtsdb 数据刷盘30s延迟, 使用内存数据替代第一根k线 lastConfirmK := exchangeInst.LastKline.Get(interval) diff --git a/internal/exchange/okx/ws_channel_subscribe.go b/internal/exchange/okx/ws_channel_subscribe.go index e49bed8..b8635f8 100644 --- a/internal/exchange/okx/ws_channel_subscribe.go +++ b/internal/exchange/okx/ws_channel_subscribe.go @@ -11,6 +11,7 @@ func (c *wsChannel[T, R]) Consumer() <-chan R { return c.dataC } +// todo 每个交易产品单独一个ws connect, 避免同一时刻被排队推送(影响交易滑点价格) func (c *wsChannel[T, R]) Subscribe(instIds ...string) (err error) { c.Lock() defer c.Unlock() diff --git a/internal/trading/kline_series.go b/internal/trading/kline_series.go index b2f690d..d2fb0c9 100644 --- a/internal/trading/kline_series.go +++ b/internal/trading/kline_series.go @@ -88,6 +88,10 @@ func (s *KlineSeries) Series(start, end int16) (klines series.Klines) { return } +func (s *KlineSeries) LastTs() int64 { + return s.lastTs +} + // 检查k线序列完整 func (s *KlineSeries) Update(kline *types.Kline) (lastTs int64, serial bool) { s.Lock() @@ -100,10 +104,11 @@ func (s *KlineSeries) Update(kline *types.Kline) (lastTs int64, serial bool) { } // 检查k线是否连续 if len(s.klines) > 0 { - expectTs := s.Interval.MustAddMul(s.lastTs, 1) + expectTs := s.IntervalAdder(s.lastTs, 1) if kline.Ts != expectTs { serial = false - zlog.Warningf("k线不连续: instId=%s(%s), interval=%s, lastTs=%d, expected=%d, got=%d", s.InstId, s.Exchange, s.Interval, s.lastTs, expectTs, kline.Ts) + miss := (kline.Ts-s.lastTs)/s.IntervalAdder(0, 1) - 1 + zlog.Warningf("k线不连续: instId=%s(%s), interval=%s, miss=%d, lastTs=%d, got=%d, expected=%d", s.InstId, s.Exchange, s.Interval, miss, s.lastTs, kline.Ts, expectTs) return } } diff --git a/internal/trading/kline_store.go b/internal/trading/kline_store.go index 2526dba..b1b47ce 100644 --- a/internal/trading/kline_store.go +++ b/internal/trading/kline_store.go @@ -193,18 +193,18 @@ func (s *KlineStore) inititalKlineSeries(exchange pb.ExchangeType, instId string // 初始化最新的 klineSeries for _, interval := range s.subKlineIntervals { retry.DoWithFixDelay(math.MaxInt32, 2*time.Second, func(retryTimes uint32) (_ struct{}, err error) { - err = s.fetchHistoryKlineToSeries(exchange, instId, interval, 0, 0, MaxSeriesKlines) + _, 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) + zlog.Infof("initial kline series success: %s(%s)", instId, exchange) } // fetchHistoryKlineToSeries 拉去历史k线数据更新series -func (s *KlineStore) fetchHistoryKlineToSeries(exchange pb.ExchangeType, instId, interval string, before, after int64, count uint32) (err error) { +func (s *KlineStore) fetchHistoryKlineToSeries(exchange pb.ExchangeType, instId, interval string, before, after int64, count uint32) (total int, err error) { // 拉取最新的1000条k线 req := &pb.ReqHistoryKlineStream{ Exchange: exchange, @@ -231,6 +231,7 @@ func (s *KlineStore) fetchHistoryKlineToSeries(exchange pb.ExchangeType, instId, zlog.Error("fetch kline stream recv error: ", err0) return } + total += len(msg.Klines) // 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) @@ -258,25 +259,45 @@ func (s *KlineStore) Update(exchange pb.ExchangeType, instId string, kline *type }) before, serial := instSeries.IntervalKlines.Get(kline.Interval).Update(kline) - if !serial && instSeries.Status.CompareAndSwap(int32(data.StatusOk), int32(data.StatusProcessing)) { + if !serial { + inprocessing := instSeries.Status.CompareAndSwap(int32(data.StatusOk), int32(data.StatusProcessing)) + if !inprocessing { + return + } + // 拉取缺失的k线 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) + total, 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 + } else { + zlog.Debugf("fetched padding kline series: instId=%s(%s), interval=%s, total=%d, ts=%d~%d", instId, exchange, kline.Interval, total, before, after) } }() } - if expTs, ok := kline.Interval.AddMul(kline.Ts, 2); ok { + // 发布k线时间驱动策略执行 + if instSeries.Status.Load() != int32(data.StatusOk) { + return + } + if nts, ok := kline.Interval.AddMul(kline.Ts, 2); ok { // k线已过期则不执行策略 - if expTs < time.Now().Unix() { + if nts < time.Now().UnixMilli() { return } - // todo emit kline update, calc indicator... } + // 判断同一时刻k线 + var intervals []types.Interval + endTs := kline.Interval.MustAddMul(kline.Ts, 1) + instSeries.IntervalKlines.Range(func(interval types.Interval, v *KlineSeries) { + if endTs == interval.MustAddMul(v.LastTs(), 1) { + intervals = append(intervals, interval) + } + }) + zlog.Debugf("confirm kline intervals: instId=%s(%s), interval=%s, ts=%d, %v", instId, exchange, kline.Interval, kline.Ts, intervals) + // todo emit kline update, calc indicator... + // pubKey := fmt.Sprintf("/kline/%s/%s/%s/%d", exchangeType, tradeInst.InstId, kline.Interval, confirm) + // interval/okx/BTC_USDT/1m,3m,5m } diff --git a/internal/trading/trading_plan.go b/internal/trading/trading_plan.go new file mode 100644 index 0000000..cb428ab --- /dev/null +++ b/internal/trading/trading_plan.go @@ -0,0 +1,25 @@ +package trading + +import ( + "sig-pub/api/pb" + "sig-pub/pkg/data/entity" + "sig-pub/pkg/strategy" + "sig-pub/pkg/types" +) + +type TradingPlan struct { + PlanId int64 `json:"planId"` + Exchange pb.ExchangeType `json:"exchange"` + InstId string `json:"instId"` + Interval types.Interval `json:"interval"` + StrategyName string `json:"strategyName"` +} + +func NewTradingPlan(plan entity.TradingPlan, strategy strategy.IStrategy) *TradingPlan { + + return &TradingPlan{} +} + +func (plan *TradingPlan) Update() (err error) { + return +} diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index 79e8430..1184824 100644 --- a/internal/trading/trading_service.go +++ b/internal/trading/trading_service.go @@ -1,15 +1,24 @@ package trading import ( + "fmt" "sig-pub/api/pb" "sig-pub/pkg/client" + "sig-pub/pkg/indicator" + "sig-pub/pkg/publish" + "sig-pub/pkg/strategy" + "sig-pub/pkg/utils/collect" ) type TradingService struct { marketClientAside *client.TradeInstanceAside exchangeClient pb.ExchangeServiceClient - klineStore *KlineStore + klineStore *KlineStore + windowIndicators *collect.SyncMap[string, indicator.IWindowIndicator] + strategies *collect.SyncMap[string, strategy.IStrategy] + publisher publish.Publisher[int64, *TradingPlan] + tradingPlan chan *TradingPlan } func NewTradingService( @@ -20,6 +29,7 @@ func NewTradingService( marketClientAside: marketClientAside, exchangeClient: exchangeClient, klineStore: NewKlineSeriesStore(exchangeClient), + windowIndicators: collect.NewSyncMap[string, indicator.IWindowIndicator](), } } @@ -28,5 +38,64 @@ func (svr *TradingService) Init() (err error) { if err = svr.klineStore.Init(); err != nil { return } + // indicator registry + { + svr.MustRegisterWindowIndicator(&indicator.RSI{}) + } + // strategy registry + { + svr.MustRegisterStrategy(&strategy.GoldX{}) + } + // strategy initial + // 初始化策略执行器 64 + + return +} + +// RegisterWindowIndicator +func (svr *TradingService) RegisterWindowIndicator(ind indicator.IWindowIndicator) (err error) { + indName := ind.Name() + _, loaded := svr.windowIndicators.LoadOrStore(indName, ind) + if loaded { + err = fmt.Errorf("window indicator name %s already duplicated", indName) + return + } + return +} + +func (svr *TradingService) MustRegisterWindowIndicator(ind indicator.IWindowIndicator) { + if err := svr.RegisterWindowIndicator(ind); err != nil { + panic(err) + } +} + +// RegisterStrategy +func (svr *TradingService) RegisterStrategy(strategy strategy.IStrategy) (err error) { + strategyName := strategy.Meta().Name + _, loaded := svr.strategies.LoadOrStore(strategyName, strategy) + if loaded { + err = fmt.Errorf("strategy name %s already duplicated", strategyName) + return + } + return +} + +func (svr *TradingService) MustRegisterStrategy(strategy strategy.IStrategy) { + if err := svr.RegisterStrategy(strategy); err != nil { + panic(err) + } +} + +// RunStrategy 运行策略 +// todo 止盈止损... +func (svr *TradingService) RunQuantPlan(plan TradingPlan) (err error) { + strategy, ok := svr.strategies.Load(plan.StrategyName) + if !ok { + err = fmt.Errorf("strategy %s not exists", plan.StrategyName) + return + } + runner := strategy.New() + _ = runner + runner.Update(nil) return } diff --git a/pkg/backtrace/backtrace.go b/pkg/backtrace/backtrace.go index 56e75a1..5f3f4b6 100644 --- a/pkg/backtrace/backtrace.go +++ b/pkg/backtrace/backtrace.go @@ -4,7 +4,7 @@ import "sig-pub/pkg/strategy" // 回测引擎 type BacktraceEngine struct { - strategy strategy.Strategy + strategy strategy.IStrategy } // 多周期策略回测引擎 diff --git a/pkg/data/entity/trade_plan.go b/pkg/data/entity/trade_plan.go new file mode 100644 index 0000000..0c5e30e --- /dev/null +++ b/pkg/data/entity/trade_plan.go @@ -0,0 +1,21 @@ +package entity + +import ( + "sig-pub/api/pb" + "sig-pub/pkg/types" +) + +type TradingPlan struct { + Id int64 `gorm:"column:id;primaryKey" json:"id"` // 交易计划id + Userid string `gorm:"column:userid" json:"userid"` // 所属用户id + Exchange pb.ExchangeType `gorm:"column:exchange" json:"exchange"` // 交易所 + InstId string `gorm:"column:instId" json:"instId"` // 交易产品 + Interval types.Interval `gorm:"column:interval" json:"interval"` // 交易周期 + StrategyName string `gorm:"column:strategy_name" json:"strategyName"` // 策略名称 + UpdateBy string `gorm:"column:update_by" json:"updateBy"` // 更新人 + UpdateTime int64 `gorm:"column:update_time" json:"updateTime"` // 更新时间戳毫秒 +} + +func (TradingPlan) TableName() string { + return "t_trading_plan" +} diff --git a/pkg/indicator/base.go b/pkg/indicator/base.go index 2053370..d43eeb8 100644 --- a/pkg/indicator/base.go +++ b/pkg/indicator/base.go @@ -7,7 +7,14 @@ import ( // IIndicator 指标基础计算接口 type IIndicator interface { - Calculate() (vector float64) + Name() string + Calculate(kSeries IKlineSeries) (vector float64) +} + +// IIndicator 窗口指标基础计算接口 +type IWindowIndicator interface { + Name() string + Calculate(kSeries IKlineSeries, window int16) (vector float64) } // IKlineSeries k线序列, strategy服务提供 diff --git a/pkg/indicator/rsi.go b/pkg/indicator/rsi.go index 57e1b20..c4dc62f 100644 --- a/pkg/indicator/rsi.go +++ b/pkg/indicator/rsi.go @@ -7,39 +7,24 @@ import ( ) // RSI stateless indicator +// 相对强弱指数 (RSI) rsi define: https://www.investopedia.com/terms/r/rsi.asp type RSI struct { - // types.IntervalWindow - window int // 窗口大小 - kliner IKlineSeries -} - -// New indicator interface -// 依赖注入 -// 指标参数注入 -func (c RSI) New(klineAccesser IKlineSeries) *RSI { - return &RSI{ - kliner: klineAccesser, - } } // indicator interface -func (c *RSI) IntervalWindow(window int) { - c.window = window +func (c *RSI) Name() string { + return "rsi" } // Calculate 计算单根k线rsi指标 -func (c *RSI) Calculate() (vector float64) { - // kline := c.kliner.Get(0) - // if kline.Close == kline.High { - // // emit buy/sell - // } +func (c *RSI) Calculate(kSeries IKlineSeries, window int16) (vector float64) { // 读k线, 计算 - klineSeries := c.kliner.Series(0, int16(c.window)) // 7根 + klineSeries := kSeries.Series(0, int16(window)) // 7根 closeSeries := klineSeries.Close() closeDiff := closeSeries.Diff() - avgGain := closeDiff.PositiveValuesOrZero().Abs().Sum() / float64(c.window) - avgLoss := closeDiff.NegativeValuesOrZero().Abs().Sum() / float64(c.window) + avgGain := closeDiff.PositiveValuesOrZero().Abs().Sum() / float64(window) + avgLoss := closeDiff.NegativeValuesOrZero().Abs().Sum() / float64(window) rs := avgGain / avgLoss rsi := 100 - (100 / (1 + rs)) diff --git a/pkg/indicator/rsi0.go b/pkg/indicator/rsi0.go deleted file mode 100644 index e6e2faa..0000000 --- a/pkg/indicator/rsi0.go +++ /dev/null @@ -1,61 +0,0 @@ -package indicator - -import ( - "fmt" - "sig-pub/pkg/trader" - "sig-pub/pkg/types" - "sig-pub/pkg/types/series" - - "github.com/spf13/cast" -) - -// RSI0: 相对强弱指数 (RSI0) -// rsi define: https://www.investopedia.com/terms/r/rsi.asp -type RSI0 struct { - trader.Indicator - series.Series - values series.Floats - prices series.Floats - - argBaseDay int32 -} - -func NewRSI() *RSI0 { - return &RSI0{} -} - -func (ind RSI0) Meta() trader.IndicatorMeta { - return trader.IndicatorMeta{ - Name: "RSI", - Desc: "", - Args: []trader.Arg{ - {Name: "基准天数", Desc: "", ArgType: trader.ArgTypeUInt}, - }, - } -} - -func (ind *RSI0) Init(indId int64, exchange any, args []string) (code trader.ErrorCode, err error) { - arg0, err := cast.ToInt32E(args[0]) - if err != nil { - return - } - ind.argBaseDay = arg0 - - cast.ToIntE("1") - return -} - -func (ind *RSI0) Update(klines []types.Kline) (err error) { - for _, kline := range klines { - c, ok := kline.Close.Float64() - if !ok { - err = fmt.Errorf("kline close to float64 error: %s", kline.Close.String()) - return - } - ind.prices.Push(c) - } - - diff := ind.prices.Diff() - _ = diff - return -} diff --git a/internal/exchange/publisher.go b/pkg/publish/publisher.go similarity index 98% rename from internal/exchange/publisher.go rename to pkg/publish/publisher.go index aea4451..c7edcf5 100644 --- a/internal/exchange/publisher.go +++ b/pkg/publish/publisher.go @@ -1,4 +1,4 @@ -package exchange +package publish import "sig-pub/pkg/utils/collect" diff --git a/pkg/stream/stream.go b/pkg/publish/stream.go similarity index 96% rename from pkg/stream/stream.go rename to pkg/publish/stream.go index 93535ac..5b9ea1d 100644 --- a/pkg/stream/stream.go +++ b/pkg/publish/stream.go @@ -1,4 +1,4 @@ -package stream +package publish const ( StreamKlineLive = "/stream/kline/live/:bar" // 实时k线推送, 1s,5s,1m... stream.kline.live.* diff --git a/pkg/strategy/gold_x.go b/pkg/strategy/gold_x.go index ea5eb29..8817222 100644 --- a/pkg/strategy/gold_x.go +++ b/pkg/strategy/gold_x.go @@ -4,17 +4,18 @@ package strategy type GoldX struct { } -func (s *GoldX) New() Strategy { +func (s *GoldX) New() IStrategy { return &GoldX{} } func (s *GoldX) Meta() StrategyMeta { return StrategyMeta{ Name: "GoldX", + Desc: "金叉策略", } } -func (s *GoldX) Update(ctx StrategyContext) { +func (s *GoldX) Update(ctx IStrategyContext) { sma14 := ctx.IndicatorW("sma", 14) sma28 := ctx.IndicatorW("sma", 28) // 包装方法 diff --git a/pkg/strategy/strategy.go b/pkg/strategy/strategy.go index 9ca337a..172f9f9 100644 --- a/pkg/strategy/strategy.go +++ b/pkg/strategy/strategy.go @@ -1,17 +1,27 @@ package strategy import ( + "fmt" + "sig-pub/api/pb" "sig-pub/pkg/indicator" "sig-pub/pkg/types" "sig-pub/pkg/types/series" + "sig-pub/pkg/utils/collect" + "strings" ) // todo Exit 止盈止损策略(trading service 管理) -// todo Meta 策略调参, 回测引擎自动调参回测(最佳参数) argGenerator.next() (arg, ok) -type Strategy interface { - New() Strategy +type IStrategy interface { + New() IStrategy Meta() StrategyMeta - Update(ctx StrategyContext) + Update(ctx IStrategyContext) +} + +// todo Meta 策略调参, 回测引擎自动调参回测(最佳参数) argGenerator.next() (arg, ok) +type IStrategyAdjustable interface { + IStrategy + NextParams() map[string]any // 根据当前策略参数, 返回下一批策略参数(并行回测 stateless) + AdjustParams(map[string]any) // 重置策略设置策略参数 } type StrategyMeta struct { @@ -20,9 +30,9 @@ type StrategyMeta struct { Desc string } -// StrategyContext 策略外部访问能力 +// IStrategyContext 策略外部访问能力 // klineSeries, Indicator -type StrategyContext interface { +type IStrategyContext interface { Buy() // 发出多信号 Sell() // 发出空信号 @@ -33,3 +43,13 @@ type StrategyContext interface { // 获取窗口类型指标 IndicatorW(name string, window int) indicator.IIndicatorSeries } + +// DriverIntervalKey 生成周期驱动事件key +// interval/okx/BTC_USDT/1m,3m,5m +func DriverIntervalKey(exchangeType pb.ExchangeType, instId string, intervals ...types.Interval) string { + types.IntervalsSort(intervals) + strIntervals := collect.Mapping(intervals, func(_ int, interval types.Interval) string { return string(interval) }) + + pubKey := fmt.Sprintf("/interval/%s/%s/%s", exchangeType.String(), instId, strings.Join(strIntervals, ",")) + return pubKey +} diff --git a/pkg/strategy/strategy_multi_interval.go b/pkg/strategy/strategy_multi_interval.go index 2695801..58dc952 100644 --- a/pkg/strategy/strategy_multi_interval.go +++ b/pkg/strategy/strategy_multi_interval.go @@ -4,15 +4,15 @@ import "sig-pub/pkg/types" // 多k线周期策略 type MultiIntervalStrategy interface { - Strategy + IStrategy DriverInterval() types.Interval // 驱动k线周期, 当驱动周期k线更新时则判断调用Update方法 SubscribeIntervals() []types.Interval // 订阅k线周期, 当同一时间的订阅周期都更新时调用Update方法 } type MultiExchangeStrategy interface { - Strategy + IStrategy } type MultiIntervalExchangeStrategy interface { - Strategy + IStrategy } diff --git a/pkg/types/interval.go b/pkg/types/interval.go index daafc80..c873eff 100644 --- a/pkg/types/interval.go +++ b/pkg/types/interval.go @@ -116,6 +116,16 @@ func init() { // zlog.Debugf("init intervalsIotas: %#v", intervalIotas) } +// 对周期进行排序 +func IntervalsSort(intervals []Interval) { + if len(intervals) < 2 { + return + } + sort.Slice(intervals, func(i, j int) bool { + return intervalIotas[intervals[i]] < intervalIotas[intervals[j]] + }) +} + type IntervalState[T any] struct { state []T } @@ -136,11 +146,11 @@ func (s *IntervalState[T]) Set(interval Interval, v T) { s.state[i] = v } -func (s *IntervalState[T]) Range(f func(i int, interval Interval, v T)) { +func (s *IntervalState[T]) Range(f func(interval Interval, v T)) { for i, interval := range iotasIntervals { index := i + 1 // 0保留 v := s.state[index] - f(i, interval, v) + f(interval, v) } } diff --git a/pkg/utils/collect/sync_map.go b/pkg/utils/collect/sync_map.go index ef517d8..7cf2e4d 100644 --- a/pkg/utils/collect/sync_map.go +++ b/pkg/utils/collect/sync_map.go @@ -32,3 +32,11 @@ func (m *SyncMap[K, V]) Range(f func(k K, v V) bool) { return f(key.(K), value.(V)) }) } + +func (m *SyncMap[K, V]) CompareAndSwap(k K, old V, new V) (swapped bool) { + return m.m.CompareAndSwap(k, old, new) +} + +func (m *SyncMap[K, V]) LoadOrStore(k K, v V) (actual any, loaded bool) { + return m.m.LoadOrStore(k, v) +} diff --git a/pkg/utils/conver/unit_conver.go b/pkg/utils/conver/unit_conver.go index af9f3f3..152864c 100644 --- a/pkg/utils/conver/unit_conver.go +++ b/pkg/utils/conver/unit_conver.go @@ -44,8 +44,9 @@ func MustParseDuration(s string) time.Duration { return d } -// TimeMilliFormat 时间戳差转换成可视化时间单位 2100 -> 1h-2s-100ms -func TimeMilliFormat(milli int64, sep string) (r string) { +// TimeDurationFormat 时间差转换成可视化时间单位 2100 -> 1h.2s.100ms +func TimeDurationFormat(du time.Duration, sep string) (r string) { + milli := du.Milliseconds() day := milli / (1000 * 60 * 60 * 24) milli -= day * (1000 * 60 * 60 * 24) hour := milli / (1000 * 60 * 60) diff --git a/pkg/utils/times/watch.go b/pkg/utils/times/watch.go new file mode 100644 index 0000000..4ea4ad0 --- /dev/null +++ b/pkg/utils/times/watch.go @@ -0,0 +1,33 @@ +package times + +import ( + "sig-pub/pkg/utils/conver" + "time" +) + +// Watch 计时器 +type Watch struct { + t time.Time +} + +func NewWatch() *Watch { + return &Watch{ + t: time.Now(), + } +} + +// Elapsed 返回从计时开始到现在的所用时间。 +func (w *Watch) Elapsed() time.Duration { + return time.Since(w.t) +} + +// Elapsed 返回从计时开始到现在的所用时间。 +func (w *Watch) ElapsedFmt(sep string) string { + return conver.TimeDurationFormat(w.Elapsed(), sep) +} + +// Watch 重置计时器 +func (w *Watch) Reset() *Watch { + w.t = time.Now() + return w +}