trading plan

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: 支持的数据源交易所

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线时间戳标记
+				// k线未缺失, 初始化状态完成, 更新k线时间戳标记
-				if exchangeInst.Status.Load() == int32(data.StatusOk) {
+				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 {
+		// if useMs := time.Now().UnixMilli() - receivedTs; useMs > 10 {
-			zlog.Debugf("handle consume kline use: %dms", useMs)
+		// 	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线数据
+// paddingTradeInstanceKlines 初始化交易产品历史k线数据
-func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, tradeInst types.TradeInstance) (err error) {
+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) {
+		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("."))
-			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, "."))
 
-		// flush vmtsdb to disk
+		markTs := exchangeInst.HistoryMarkTs.Get(interval)
-		retry.DoWithFixDelay(5, time.Second, func(retryTimes uint32) (_ struct{}, err error) {
+		tsKey, ex := svc.exchangeDataPersist.SaveHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, interval, markTs)
-			if err = svc.exchangeDataPersist.vmtsdb.ForceFlush(); err != nil {
+		if ex != nil {
-				zlog.Errorf("flush vmts db error: ", err)
+			zlog.Errorf("save history mark ts error: key=%s, ts=%d, %v", tsKey, markTs, ex)
 		}
-			return
-		})
 	}()
 
-	// 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,16 +414,18 @@ 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 {
+		// history 未补全前, history写 kvdb ts mark, 补全后 ws live cnofirm 写 ts mark
-			// interval := types.Interval1d
+		beforeTs := int64(0)
-			// intervalAdder := types.SupportedIntervals[interval]
+		beforeTs, err := retry.DoWithStepDelay(10, time.Second, func(retryTimes uint32) (markTs int64, err error) {
-			// history 未补全前, history写 kvdb ts mark, 补全后 ws live 写 ts mark
+			markTs, err = svc.exchangeDataPersist.GetHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, interval)
-			beforeTs, ex := svc.exchangeDataPersist.GetHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, interval)
+			if err != nil {
-			if ex != nil {
+				zlog.Error("get history kline mark ts error: ", err)
-				err = ex
+			}
+			return
+		})
+		if err != nil {
 			zlog.Error(err)
 			cancel()
 			return
@@ -393,18 +436,13 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade
 			// 不足100根，向前补齐100根一次拉取过来
 			total := (time.Now().UnixMilli() - beforeTs) / intervalAdder(0, 1)
 			if total < 100 {
-					beforeTs = max(intervalAdder(beforeTs, -100), KlineBefore0)
+				beforeTs = max(intervalAdder(beforeTs, -100), intervalAdder(KlineBefore0, -1))
 			}
 		}
 
-			exchangeInst.HistoryMarkTs.Set(interval, beforeTs)
 		for {
 			// 判定订阅任务发布完成
-				// liveStartTs := exchangeInst.LiveKStartTs.Get(interval)
+			if intervalAdder(beforeTs, 2) > time.Now().UnixMilli() {
-				// if liveStartTs != 0 && beforeTs >= liveStartTs {
-				// 	break
-				// }
-				if beforeTs > time.Now().UnixMilli() {
 				break
 			}
 
@@ -427,7 +465,6 @@ func (svc *ExchangeService) initialTradeInstanceKlines(exchange *Exchange, trade
 
 			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)

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

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

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)
+			total, err := s.fetchHistoryKlineToSeries(exchange, instId, string(kline.Interval), before, after, 0)
-			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
 }

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

internal/trading/trading_service.go

@@ -1,8 +1,13 @@
 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 {
@@ -10,6 +15,10 @@ type TradingService struct {
 	exchangeClient    pb.ExchangeServiceClient
 
 	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
 }

pkg/backtrace/backtrace.go

@@ -4,7 +4,7 @@ import "sig-pub/pkg/strategy"
 
 // 回测引擎
 type BacktraceEngine struct {
-	strategy strategy.Strategy
+	strategy strategy.IStrategy
 }
 
 // 多周期策略回测引擎

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"
+}

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服务提供

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) {
+func (c *RSI) Name() string {
-	c.window = window
+	return "rsi"
 }
 
 // Calculate 计算单根k线rsi指标
-func (c *RSI) Calculate() (vector float64) {
+func (c *RSI) Calculate(kSeries IKlineSeries, window int16) (vector float64) {
-	// kline := c.kliner.Get(0)
-	// if kline.Close == kline.High {
-	// 	// emit buy/sell
-	// }
 	// 读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)
+	avgGain := closeDiff.PositiveValuesOrZero().Abs().Sum() / float64(window)
-	avgLoss := closeDiff.NegativeValuesOrZero().Abs().Sum() / float64(c.window)
+	avgLoss := closeDiff.NegativeValuesOrZero().Abs().Sum() / float64(window)
 
 	rs := avgGain / avgLoss
 	rsi := 100 - (100 / (1 + rs))

pkg/indicator/rsi0.go

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

pkg/publish/publisher.go

@@ -1,4 +1,4 @@
-package exchange
+package publish
 
 import "sig-pub/pkg/utils/collect"
 

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.*

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)
 	// 包装方法

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 IStrategy interface {
-type Strategy interface {
+	New() IStrategy
-	New() Strategy
 	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
+}

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
 }

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

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

pkg/utils/conver/unit_conver.go

@@ -44,8 +44,9 @@ func MustParseDuration(s string) time.Duration {
 	return d
 }
 
-// TimeMilliFormat 时间戳差转换成可视化时间单位 2100 -> 1h-2s-100ms
+// TimeDurationFormat 时间差转换成可视化时间单位 2100 -> 1h.2s.100ms
-func TimeMilliFormat(milli int64, sep string) (r string) {
+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)

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