strategy runner

config/exchange.toml

@@ -18,8 +18,8 @@ receiveBuffer = 4096
 marketSubscribeLimit = 16
 consumeBatch = 1024
 consumeLater = 2000 # 时间到达later或者数据累计到batch触发consume
-httpProxy = "http://192.168.1.5:7890"
-# httpProxy = "http://10.255.183.209:7890"
+# httpProxy = "http://192.168.1.5:7890"
+httpProxy = "http://10.255.183.209:7890"
 
 # 模拟盘API交易地址如下：
 # REST：https://www.okx.com

go.mod

@@ -20,6 +20,7 @@ require (
 	github.com/jhump/protoreflect/v2 v2.0.0-beta.2
 	github.com/klauspost/compress v1.18.0
 	github.com/lib/pq v1.10.9
+	github.com/markcheno/go-talib v0.0.0-20250114000313-ec55a20c902f
 	github.com/mostynb/go-grpc-compression v1.2.3
 	github.com/nats-io/nats.go v1.47.0
 	github.com/redis/go-redis/v9 v9.7.3
@@ -81,7 +82,6 @@ require (
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.10 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
-	github.com/markcheno/go-talib v0.0.0-20250114000313-ec55a20c902f // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect

internal/exchange/exchange_data_persist.go

@@ -28,16 +28,26 @@ func NewExchangeDataService(
 }
 
 func (p *ExchangeDataPersist) Init() (err error) {
-	// victoriametrics 强制刷盘
 	exit.AddHook(func() {
 		zlog.Infof("force flush vmtsdb")
-		if err := p.vmtsdb.ForceFlush(); err != nil {
-			zlog.Infof("force flush vmtsdb error: ", err)
-		}
+		p.Flush()
 	}, exit.WithOrderFront())
 	return
 }
 
+// Flush victoriametrics 强制刷盘
+func (p *ExchangeDataPersist) Flush0() (err error) {
+	err = p.vmtsdb.ForceFlush()
+	return
+}
+
+// Flush victoriametrics 强制刷盘
+func (p *ExchangeDataPersist) Flush() {
+	if err := p.Flush0(); err != nil {
+		zlog.Infof("vmtsdb force flush error: ", err)
+	}
+}
+
 func (p *ExchangeDataPersist) SaveKline(inst types.TradeInstance, klines []*types.Kline) (err error) {
 	err = p.vmtsdb.SaveKlines(inst, klines)
 	if err != nil {

internal/exchange/exchange_service.go

@@ -160,38 +160,10 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch
 		// 升序排序
 		collect.SortAsc(channelK.Klines, func(k *types.Kline) int64 { return k.Ts })
 
-		// k线完整性检查, k线是否连续并补齐
-		padding := false
-		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 {
-							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.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)
-								}
-							}
-						}
-					}
-				}
-				break
-			}
-		}
-
 		// 取出头尾k线
 		lastKline := channelK.Klines[len(channelK.Klines)-1]
+		interval := lastKline.Interval
+		intervalAdder, intervalSupport := types.SupportedIntervals[interval]
 
 		// 记录实时k线
 		exchangeInst.LiveKline.Set(lastKline.Interval, *lastKline)
@@ -204,8 +176,11 @@ 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)
+				if intervalSupport {
+					delay := time.Now().UnixMilli() - intervalAdder(kline.Ts, 1)
+					zlog.Debugf("recv confirm kline: delay=%dms, inst=%s(%s), interval=%s", delay, tradeInst.InstId, tradeInst.Exchange, kline.Interval)
+				}
+
 				confirm = 1
 				confirmKlines = append(confirmKlines, kline)
 			}
@@ -221,6 +196,25 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch
 			msg.Klines = append(msg.Klines, pbk)
 		}
 
+		padding := true
+		if len(confirmKlines) > 0 {
+			if intervalSupport {
+				// k线完整性检查, k线是否连续并补齐
+				if lastConfirmK := exchangeInst.LastKline.Get(interval); lastConfirmK.Ts != 0 {
+					tempK := make([]*types.Kline, 0, len(confirmKlines)+1)
+					tempK = append(tempK, &lastConfirmK)
+					tempK = append(tempK, confirmKlines...)
+					if err := svc.paddingKlinesIfNotSeries(exchange, tradeInst.InstId, confirmKlines[0].Interval, tempK); err != nil {
+						padding = false
+						zlog.Errorf("try padding klines error: inst=%s(%s), interval=%s, ts=%d~%d, %v", tradeInst.InstId, tradeInst.Exchange, lastKline.Interval, lastKline.Ts, lastConfirmK.Ts, err)
+					}
+				}
+			}
+
+			// 记录最后确认k线
+			exchangeInst.LastKline.Set(interval, *confirmKlines[len(confirmKlines)-1])
+		}
+
 		// 存储到 tsdb
 		if _, ok := types.SupportedIntervals[lastKline.Interval]; ok && len(confirmKlines) > 0 {
 			// tsdb storage todo 异步处理
@@ -230,7 +224,7 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch
 				zlog.Errorf("kline save to tsdb error: %v, %#v", err, confirmKlines)
 			} else {
 				// k线未缺失, 初始化状态完成, 更新k线时间戳标记
-				if !padding && exchangeInst.Status.Load() == int32(data.StatusOk) {
+				if intervalSupport && !padding && exchangeInst.Status.Load() == int32(data.StatusOk) {
 					latestK := collect.MustMax(confirmKlines, func(k *types.Kline) int64 { return k.Ts })
 					// 标记确认k线
 					tsKey, ex := svc.exchangeDataPersist.SaveHistoryKlineMarkTs(tradeInst.Exchange, tradeInst.InstId, latestK.Interval, latestK.Ts)
@@ -329,7 +323,7 @@ func (svc *ExchangeService) paddingTradeInstanceKlines(exchange *Exchange, trade
 
 		// flush vmtsdb to disk
 		retry.DoWithFixDelay(5, time.Second, func(retryTimes uint32) (_ struct{}, err error) {
-			if err = svc.exchangeDataPersist.vmtsdb.ForceFlush(); err != nil {
+			if err = svc.exchangeDataPersist.Flush0(); err != nil {
 				zlog.Errorf("flush vmts db error: ", err)
 			}
 			return
@@ -556,6 +550,46 @@ func (svc *ExchangeService) fetchTaskKlinesToTSDB(exchange *Exchange, task fetch
 	return
 }
 
+// paddingKlinesIfNotSeries 如k线不连续, 从缺失处进行补齐
+func (svc *ExchangeService) paddingKlinesIfNotSeries(exchange *Exchange, instId string, interval types.Interval, klines []*types.Kline) (err error) {
+	// 检查k线是否连续
+	paddingMarkTs := int64(0)
+	for i, k := range klines {
+		if i > 0 && k.Ts != interval.MustAddMul(klines[i-1].Ts, 1) {
+			paddingMarkTs = klines[i-1].Ts
+			break
+		}
+	}
+	if paddingMarkTs == 0 {
+		return
+	}
+	// k线不连续进行补齐
+	exchangeInstId, _ := exchange.TradeInstIds.Load(instId)
+	exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId)
+	if !ok {
+		err = fmt.Errorf("trade instance not support for exchange: %s(%s)", instId, exchange.ExchangeType.String())
+		return
+	}
+	if exchangeInst.Status.CompareAndSwap(int32(data.StatusOk), int32(data.StatusProcessing)) {
+		defer exchangeInst.Status.CompareAndSwap(int32(data.StatusProcessing), int32(data.StatusOk))
+
+		watch := times.NewWatch()
+		zlog.Warningf("vmtsdb kline not series, try padding: instId=%s(%s), interval=%s, ts=%d", instId, exchange.ExchangeType.String(), interval, paddingMarkTs)
+		if _, err = svc.exchangeDataPersist.SaveHistoryKlineMarkTs(exchange.ExchangeType, instId, interval, paddingMarkTs); err != nil {
+			zlog.Errorf("try padding series save markTs error: ", err)
+			return
+		}
+		if err = svc.paddingTradeInstanceIntervalKlines(4, exchange, *exchangeInst.Inst, interval); err != nil {
+			zlog.Errorf("try padding series fetch to vmtsdb error: ", err)
+			return
+		}
+		// flush vmtsdb to disk
+		svc.exchangeDataPersist.Flush()
+		zlog.Debugf("vmtsdb kline not series padding success: instId=%s(%s), interval=%s, ts=%d, use %s", instId, exchange.ExchangeType.String(), interval, paddingMarkTs, watch.ElapsedFmt("."))
+	}
+	return
+}
+
 // Exchanges 支持的交易所列表
 func (svc *ExchangeService) Exchanges() (exchanges []pb.ExchangeType, err error) {
 	svc.exchanges.Range(func(exchange pb.ExchangeType, _ *Exchange) {
@@ -725,7 +759,7 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory
 	return
 }
 
-// 查询历史k线(流式返回)
+// 查询历史k线(按时间升序流式返回)
 func (svc *ExchangeService) HistoryKlineStream(req *pb.ReqHistoryKlineStream, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) {
 	// 交易产品参数检查
 	if !svc.exchanges.IsSupport(req.Exchange) {
@@ -784,7 +818,11 @@ func (svc *ExchangeService) HistoryKlineStream(req *pb.ReqHistoryKlineStream, st
 	if len(klines) == 0 {
 		return
 	}
-	// todo 检查k线是否连续进行补齐
+
+	// 检查k线是否连续进行补齐
+	if err = svc.paddingKlinesIfNotSeries(exchange, req.InstId, interval, klines); err != nil {
+		return
+	}
 
 	lastK := klines[len(klines)-1]
 	// vmtsdb 数据刷盘30s延迟, 使用内存数据替代第一根k线

internal/trading/backtrace/backtrace.go

@@ -3,6 +3,9 @@ package backtrace
 import "sig-pub/pkg/strategy"
 
 // 回测引擎
+// sig strategy
+// trade strategy
+// close strategy
 type BacktraceEngine struct {
 	strategy strategy.ISigStrategy
 }

internal/trading/kline_store.go

@@ -8,6 +8,7 @@ import (
 	"sig-pub/api/pb"
 	"sig-pub/pkg/data"
 	"sig-pub/pkg/mq"
+	"sig-pub/pkg/strategy"
 	"sig-pub/pkg/types"
 	"sig-pub/pkg/utils/collect"
 	"sig-pub/pkg/utils/retry"
@@ -24,10 +25,14 @@ type KlineStore struct {
 	subKlineIntervals []string                                                                         // 订阅的k线的周期列表
 	subKlineInsts     *types.ExchangeState[*collect.SyncMap[string, bool]]                             // 订阅k线中的交易产品列表
 	subKlineStream    grpc.BidiStreamingClient[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline] // 订阅k线的stream
+	klineSignalChan   chan string
 }
 
 func NewKlineSeriesStore(exchangeClient pb.ExchangeServiceClient) (kss *KlineStore) {
-	kss = &KlineStore{exchangeClient: exchangeClient}
+	kss = &KlineStore{
+		exchangeClient:  exchangeClient,
+		klineSignalChan: make(chan string, 128),
+	}
 
 	// 周期列表
 	kss.subKlineIntervals = collect.Map2Slice(types.SupportedIntervals, func(interval types.Interval, _ types.IntervalAdder) string {
@@ -243,6 +248,10 @@ func (s *KlineStore) fetchHistoryKlineToSeries(exchange pb.ExchangeType, instId,
 	return
 }
 
+func (s *KlineStore) ConsumerKlineSignel() <-chan string {
+	return s.klineSignalChan
+}
+
 // Update
 // kline klineStore -> klineSeries -> strategy -> indicator -> klineSeries.Series
 func (s *KlineStore) Update(exchange pb.ExchangeType, instId string, kline *types.Kline) {
@@ -297,10 +306,15 @@ func (s *KlineStore) Update(exchange pb.ExchangeType, instId string, kline *type
 			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
+	// zlog.Debugf("confirm kline intervals: instId=%s(%s), interval=%s, ts=%d, %v", instId, exchange, kline.Interval, kline.Ts, intervals)
+
+	// publish kline update signal
+	pubKey := strategy.DriverIntervalKey(instId, intervals, exchange)
+	select {
+	case s.klineSignalChan <- pubKey:
+	default:
+		zlog.Warningf("publish kline update signal fail: instId=%s(%s), interval=%s, ts=%d, %v", instId, exchange, kline.Interval, kline.Ts, intervals)
+	}
 }
 
 // GetKlineSeires 获取k线序列

internal/trading/libs/account_position.go

@@ -0,0 +1,3 @@
+package libs
+
+// 账户持仓

internal/trading/strategy_context.go

@@ -7,7 +7,6 @@ import (
 	"sig-pub/pkg/strategy"
 	"sig-pub/pkg/types"
 	"sig-pub/pkg/types/series"
-	"sig-pub/pkg/utils/collect"
 	"sig-pub/pkg/zlog"
 )
 
@@ -20,16 +19,16 @@ type StrategyContext struct {
 	IOffsetStrategyContext
 
 	indicatorContext *IndicatorContext
-	indicatorsW      *collect.SyncMap[string, indicator.IWindowIndicator]
+	indicatorsReg    *indicator.IndicatorRegistry
 	signal           []pb.Side // 0.sell,1.buy
 	signalTimes      []int64
 	wins             []bool
 }
 
-func NewStrategyContext(klineSeries *KlineSeries, indicatorsW *collect.SyncMap[string, indicator.IWindowIndicator]) *StrategyContext {
+func NewStrategyContext(klineSeries *KlineSeries, indicatorsReg *indicator.IndicatorRegistry) *StrategyContext {
 	return &StrategyContext{
 		indicatorContext: NewIndicatorContext(klineSeries),
-		indicatorsW:      indicatorsW,
+		indicatorsReg:    indicatorsReg,
 	}
 }
 
@@ -78,7 +77,7 @@ func (c *StrategyContext) Sell() {
 
 // 获取窗口类型指标
 func (c *StrategyContext) IndicatorW(name string, window int16) (s indicator.IIndicatorSeries) {
-	indicator, ok := c.indicatorsW.Load(name)
+	indicator, ok := c.indicatorsReg.IndicatorW(name)
 	if !ok {
 		panic(fmt.Errorf("indicatorW %s not exists", name))
 	}

internal/trading/trading_plan.go

@@ -1,25 +0,0 @@
-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.TradePlan, strategy strategy.ISigStrategy) *TradingPlan {
-
-	return &TradingPlan{}
-}
-
-func (plan *TradingPlan) Update() (err error) {
-	return
-}

internal/trading/trading_plan_runner.go

@@ -0,0 +1,61 @@
+package trading
+
+import (
+	"fmt"
+	"sig-pub/pkg/data/entity"
+	"sig-pub/pkg/indicator"
+	"sig-pub/pkg/publish"
+	"sig-pub/pkg/strategy"
+)
+
+type TradingPlanRunner struct {
+	plan         entity.TradePlan
+	indicatorReg indicator.IndicatorRegistry
+	strategyReg  strategy.SigStrategyRegistry
+	publisher    publish.Publisher[int32, any]
+}
+
+func NewTradingPlan(plan entity.TradePlan,
+	indicatorReg indicator.IndicatorRegistry,
+	strategyReg strategy.SigStrategyRegistry,
+) *TradingPlanRunner {
+	return &TradingPlanRunner{
+		plan:         plan,
+		indicatorReg: indicatorReg,
+		strategyReg:  strategyReg,
+	}
+}
+
+// Init 初始化交易计划
+// subKlineKeys 订阅k线更新, 更新时调用Update方法
+// 交易信号/下单/平仓/风控
+func (r *TradingPlanRunner) Init() (subSignalKeyKeys []string, err error) {
+	// 初始化执行策略
+	_, err = r.initSigStrategy(r.plan.SigStrategy, r.plan.SigStrategyParams)
+	if err != nil {
+		return
+	}
+
+	// sigStrategy 1m
+	// tradeStrategy 1s
+	// closeStrategy 2s
+	return
+}
+
+// initSigStrategy 初始化多空信号策略
+// buy/sell -> 过滤/风控 -> tradeStrategy -> closeStrategy
+func (r *TradingPlanRunner) initSigStrategy(name, params string) (subSignalKeyKeys []string, err error) {
+	strategy, ok := r.strategyReg.NewSigStrategy(name)
+	if !ok {
+		err = fmt.Errorf("strategy not exists: %s", name)
+		return
+	}
+	_ = strategy
+	// strategy.Update()
+	return
+}
+
+// Update 订阅k线更新
+func (r *TradingPlanRunner) Update(signalKey string) {
+
+}

internal/trading/trading_service.go

@@ -4,22 +4,24 @@ import (
 	"fmt"
 	"sig-pub/api/pb"
 	"sig-pub/pkg/client"
+	"sig-pub/pkg/data/entity"
 	"sig-pub/pkg/indicator"
 	"sig-pub/pkg/publish"
 	"sig-pub/pkg/strategy"
 	"sig-pub/pkg/types"
 	"sig-pub/pkg/utils/collect"
+	"sig-pub/pkg/zlog"
 )
 
 type TradingService struct {
 	marketClientAside *client.TradeInstanceAside
 	exchangeClient    pb.ExchangeServiceClient
 
-	klineStore    *KlineStore
-	indicatorsW   *collect.SyncMap[string, indicator.IWindowIndicator] // 注册窗口指标
-	sigStrategies *collect.SyncMap[string, strategy.ISigStrategy]      // 注册信号策略
-	publisher     publish.Publisher[int64, *TradingPlan]
-	tradingPlan   chan *TradingPlan
+	klineStore   *KlineStore
+	indicatorReg *indicator.IndicatorRegistry  // 注册窗口指标
+	strategyReg  *strategy.SigStrategyRegistry // 注册信号策略
+	publisher    *publish.Publisher[int64, *TradingPlanRunner]
+	tradingPlan  chan *TradingPlanRunner
 }
 
 func NewTradingService(
@@ -30,71 +32,48 @@ func NewTradingService(
 		marketClientAside: marketClientAside,
 		exchangeClient:    exchangeClient,
 		klineStore:        NewKlineSeriesStore(exchangeClient),
-		indicatorsW:       collect.NewSyncMap[string, indicator.IWindowIndicator](),
-		sigStrategies:     collect.NewSyncMap[string, strategy.ISigStrategy](),
+		indicatorReg:      indicator.NewIndicatorRegistry(),
+		strategyReg:       strategy.NewSigStrategyRegistry(),
+		publisher:         publish.NewPublisher[int64, *TradingPlanRunner](8),
 	}
 }
 
 // 初始化历史k线, 订阅实时k线
 func (svc *TradingService) Init() (err error) {
-	if err = svc.klineStore.Init(); err != nil {
+	if err = svc.indicatorReg.Init(); err != nil {
 		return
 	}
-	// indicator registry
-	{
-		svc.MustRegisterWindowIndicator(&indicator.RSI{})
-		svc.MustRegisterWindowIndicator(&indicator.SMA{})
-	}
-	// strategy registry
-	{
-		svc.MustRegisterStrategy(&strategy.GoldX{})
-	}
-	// strategy initial
-	// 初始化策略执行器 64
-
-	return
-}
-
-// RegisterWindowIndicator
-func (svc *TradingService) RegisterWindowIndicator(ind indicator.IWindowIndicator) (err error) {
-	indName := ind.Name()
-	_, loaded := svc.indicatorsW.LoadOrStore(indName, ind)
-	if loaded {
-		err = fmt.Errorf("window indicator name %s already duplicated", indName)
+	if err = svc.strategyReg.Init(); err != nil {
 		return
 	}
-	return
-}
-
-func (svc *TradingService) MustRegisterWindowIndicator(ind indicator.IWindowIndicator) {
-	if err := svc.RegisterWindowIndicator(ind); err != nil {
-		panic(err)
-	}
-}
 
-// RegisterStrategy
-func (svc *TradingService) RegisterStrategy(strategy strategy.ISigStrategy) (err error) {
-	strategyName := strategy.Meta().Name
-	_, loaded := svc.sigStrategies.LoadOrStore(strategyName, strategy)
-	if loaded {
-		err = fmt.Errorf("strategy name %s already duplicated", strategyName)
+	if err = svc.klineStore.Init(); err != nil {
 		return
 	}
+
+	go svc.consumerKlineSignal()
 	return
 }
 
-func (svc *TradingService) MustRegisterStrategy(strategy strategy.ISigStrategy) {
-	if err := svc.RegisterStrategy(strategy); err != nil {
-		panic(err)
+// consumerKlineSignal 订阅k线更新
+func (svc *TradingService) consumerKlineSignal() {
+	c := svc.klineStore.ConsumerKlineSignel()
+	for {
+		signalKey := <-c
+		zlog.Debugf("signal: %s", signalKey)
+		_, plans := svc.publisher.Publisher(signalKey)
+		for _, plan := range plans {
+			plan.Update(signalKey)
+		}
 	}
 }
 
 // RunStrategy 运行策略
 // todo 止盈止损...
-func (svc *TradingService) RunQuantPlan(plan TradingPlan) (err error) {
-	strategy, ok := svc.sigStrategies.Load(plan.StrategyName)
+func (svc *TradingService) RunQuantPlan(plan *entity.TradePlan) (err error) {
+	strategy, ok := svc.strategyReg.NewSigStrategy(plan.SigStrategy)
 	if !ok {
-		err = fmt.Errorf("strategy %s not exists", plan.StrategyName)
+		err = fmt.Errorf("strategy %s not exists", plan.SigStrategy)
 		return
 	}
 	runner := strategy.New()
@@ -106,7 +85,7 @@ func (svc *TradingService) RunQuantPlan(plan TradingPlan) (err error) {
 // IndicatorSeries 获取指标实时或历史序列数据, 闭区间
 func (svc *TradingService) IndicatorSeries(req *pb.ReqIndicatorSeries, rsp *pb.RspIndicatorSeries) (err error) {
 	// indicatorName string, exchange pb.ExchangeType, instId string, interval types.Interval, window int
-	indicator, ok := svc.indicatorsW.Load(req.Indicator)
+	indicator, ok := svc.indicatorReg.IndicatorW(req.Indicator)
 	if !ok {
 		err = fmt.Errorf("indicator %s not exists", req.Indicator)
 		return
@@ -167,7 +146,7 @@ func (svc *TradingService) IndicatorSeries(req *pb.ReqIndicatorSeries, rsp *pb.R
 
 // StrategySeries 简单策略信号测试
 func (svc *TradingService) StrategySeries(req *pb.ReqStrategySeries, rsp *pb.RspStrategySeries) (err error) {
-	strategy, ok := svc.sigStrategies.Load(req.Strategy)
+	strategy, ok := svc.strategyReg.NewSigStrategy(req.Strategy)
 	if !ok {
 		err = fmt.Errorf("strategy %s not exists", req.Strategy)
 		return
@@ -186,7 +165,7 @@ func (svc *TradingService) StrategySeries(req *pb.ReqStrategySeries, rsp *pb.Rsp
 		return
 	}
 	// recover todo out of range
-	strategyCtx := NewStrategyContext(klineSeries, svc.indicatorsW)
+	strategyCtx := NewStrategyContext(klineSeries, svc.indicatorReg)
 	for i := range req.Count {
 		strategyCtx.SetOffset(int16(i))
 		strategy.Update(strategyCtx)

pkg/indicator/indicator_registry.go

@@ -0,0 +1,46 @@
+package indicator
+
+import (
+	"fmt"
+	"sig-pub/pkg/utils/collect"
+)
+
+// 指标注册器
+type IndicatorRegistry struct {
+	indicatorsW *collect.SyncMap[string, IWindowIndicator] // 注册窗口指标
+}
+
+func NewIndicatorRegistry() *IndicatorRegistry {
+	return &IndicatorRegistry{
+		indicatorsW: collect.NewSyncMap[string, IWindowIndicator](),
+	}
+}
+
+func (r *IndicatorRegistry) Init() (err error) {
+	// indicator regist
+	r.MustRegistIndicatorW(&RSI{})
+	r.MustRegistIndicatorW(&SMA{})
+	return
+}
+
+// RegistIndicatorW
+func (r *IndicatorRegistry) RegistIndicatorW(ind IWindowIndicator) (err error) {
+	indName := ind.Name()
+	_, loaded := r.indicatorsW.LoadOrStore(indName, ind)
+	if loaded {
+		err = fmt.Errorf("window indicator name %s already duplicated", indName)
+		return
+	}
+	return
+}
+
+func (r *IndicatorRegistry) MustRegistIndicatorW(ind IWindowIndicator) {
+	if err := r.RegistIndicatorW(ind); err != nil {
+		panic(err)
+	}
+}
+
+// IndicatorW
+func (r *IndicatorRegistry) IndicatorW(name string) (indW IWindowIndicator, ok bool) {
+	return r.indicatorsW.Load(name)
+}

pkg/strategy/buy_strategy.go

@@ -1,5 +1,12 @@
 package strategy
 
+import "github.com/govalues/decimal"
+
 // TradeStrategy 下单买入策略接口(控制滑点, 仓位管理)
 type TradeStrategy interface {
+	Update(ctx ITradeStrategyContext)
+}
+
+type ITradeStrategyContext interface {
+	LastPrice() decimal.Decimal
 }

pkg/strategy/exit_strategy.go

@@ -5,7 +5,7 @@ import "github.com/govalues/decimal"
 // todo Exit 止盈止损策略(trading service 管理)
 type IExitStrategy interface {
 	Name() string // 获取策略名称，便于日志
-	Tick(ctx IExitStrategyContext)
+	Update(ctx IExitStrategyContext)
 }
 
 type IExitStrategyContext interface {

pkg/strategy/gold_x.go

@@ -1,8 +1,11 @@
 package strategy
 
+import "fmt"
+
 // GoldX 金叉策略
 type GoldX struct {
 	ISigStrategy
+	short, long int16
 }
 
 func (s *GoldX) New() ISigStrategy {
@@ -13,19 +16,30 @@ func (s *GoldX) Meta() StrategyMeta {
 	return StrategyMeta{
 		Name: "GoldX",
 		Desc: "金叉策略",
+		Args: []Param{
+			{Name: "short", Type: ParamTypeUInt, Desc: "短周期"},
+			{Name: "long", Type: ParamTypeUInt, Desc: "长周期"},
+		},
 	}
 }
 
-func (s *GoldX) Arg() map[string]any {
-	return map[string]any{
-		"short": 14,
-		"long":  28,
+func (s *GoldX) Init(param SigStrategyParam) (err error) { // 校验参数, 并根据参数初始化策略
+	if s.short, err = param.GetInt16E("short"); err != nil {
+		return
+	}
+	if s.long, err = param.GetInt16E("long"); err != nil {
+		return
+	}
+	if s.long <= s.short {
+		err = fmt.Errorf("param short should bigger then short")
+		return
 	}
+	return
 }
 
 func (s *GoldX) Update(ctx ISigStrategyContext) {
-	sma14 := ctx.IndicatorW("sma", 14)
-	sma28 := ctx.IndicatorW("sma", 28)
+	sma14 := ctx.IndicatorW("sma", s.short)
+	sma28 := ctx.IndicatorW("sma", s.long)
 	// 包装方法 crossover/crossunder
 	s14 := sma14.Series(0, 2)
 	s28 := sma28.Series(0, 2)

pkg/strategy/sig_strategy.go

@@ -14,22 +14,14 @@ import (
 type ISigStrategy interface {
 	New() ISigStrategy
 	Meta() StrategyMeta
+	Init(param SigStrategyParam) (err error) // 校验参数, 并根据参数初始化策略
 	Update(ctx ISigStrategyContext)
 }
 
-// todo Meta 策略调参, 回测引擎自动调参回测(最佳参数) argGenerator.next() (arg, ok)
-// argA range -> [1,...,5], argB:=[0.1,...,0.7], argC:=[true,false]
-// 可变参数组合 argValidate(argA, ArgB, argC...) bool(true则使用该组合进行回测，记录组合参数回测结果)
-type ISigStrategyAdjustable interface {
-	ISigStrategy
-	NextParams() map[string]any  // 根据当前策略参数, 返回下一批策略参数(并行回测 stateless)
-	AdjustParams(map[string]any) // 重置策略设置策略参数
-}
-
 type StrategyMeta struct {
-	// Id     string `json:"id"` // 策略注册/执行器系统分配
-	Name string
-	Desc string
+	Name string  `json:"name"`
+	Desc string  `json:"desc"`
+	Args []Param `json:"args"` // 参数定义
 }
 
 // ISigStrategyContext 策略外部访问能力
@@ -50,6 +42,7 @@ type ISigStrategyContext interface {
 // interval/BTC_USDT/OKX,BINANCE/1m,3m,5m
 func DriverIntervalKey(instId string, intervals []types.Interval, exchanges ...pb.ExchangeType) string {
 	types.IntervalsSort(intervals)
+	types.ExchangesSort(exchanges)
 	strIntervals := collect.Mapping(intervals, func(_ int, interval types.Interval) string { return string(interval) })
 	strExchanges := collect.Mapping(exchanges, func(_ int, exchange pb.ExchangeType) string { return exchange.String() })
 	pubKey := fmt.Sprintf("/interval/%s/%s/%s", instId, strings.Join(strExchanges, ","), strings.Join(strIntervals, ","))

pkg/strategy/sig_strategy_params.go

@@ -0,0 +1,139 @@
+package strategy
+
+import (
+	"fmt"
+	"sig-pub/pkg/types"
+
+	"github.com/spf13/cast"
+)
+
+// 参数类型
+type ParamType int8
+
+const (
+	_ ParamType = iota
+	ParamTypeBool
+	ParamTypeFloat
+	ParamTypeUFloat
+	ParamTypeInt
+	ParamTypeUInt
+	ParamTypeString
+	ParamTypeSelect   // 单选
+	ParamTypeCheckBox // 多选
+)
+
+type Param struct {
+	Name    string        `json:"name"`
+	Desc    string        `json:"desc"`
+	Type    ParamType     `json:"type"`    // 参数类型
+	Options []ParamOption `json:"options"` // 单选/多选选项列表
+}
+
+type ParamOption struct {
+	Name string `json:"name"`
+	Desc string `json:"desc"`
+}
+
+// CastValidate 数据类型校验
+func (t Param) TypeValidate(v string) bool {
+	switch t.Type {
+	default:
+		return false
+	case ParamTypeBool:
+		if _, e := cast.ToBoolE(v); e != nil {
+			return false
+		}
+		return true
+	case ParamTypeString:
+		return true
+	case ParamTypeInt:
+		fallthrough
+	case ParamTypeUInt:
+		if r, e := cast.ToIntE(v); e != nil {
+			return false
+		} else if t.Type == ParamTypeUInt {
+			return r >= 0
+		}
+		return true
+	case ParamTypeFloat:
+		fallthrough
+	case ParamTypeUFloat:
+		if r, e := cast.ToFloat64E(v); e != nil {
+			return false
+		} else if t.Type == ParamTypeUFloat {
+			return r >= 0
+		}
+		return true
+	}
+}
+
+// 策略默认参数
+type ISigStrategyDefaultParam interface {
+	DefaultParam() map[string]string
+}
+
+// todo Meta 策略调参, 回测引擎自动调参回测(最佳参数) argGenerator.next() (arg, ok)
+// argA range -> [1,...,5], argB:=[0.1,...,0.7], argC:=[true,false]
+// 可变参数组合 argValidate(argA, ArgB, argC...) bool(true则使用该组合进行回测，记录组合参数回测结果)
+type ISigStrategyParamGenerator interface {
+	NextParam(map[string]string) (map[string]string, bool) // 根据当前策略参数, 返回下一批策略参数(并行回测 stateless)
+}
+
+type SigStrategyParam struct {
+	Interval types.Interval    `json:"interval"` // 策略驱动周期
+	Params   map[string]string `json:"params"`   // 策略执行参数
+}
+
+func (s *SigStrategyParam) Get(key string) (v string, ok bool) {
+	if len(s.Params) == 0 {
+		return
+	}
+	v, ok = s.Params[key]
+	return
+}
+
+func (s *SigStrategyParam) GetInt(key string) (r int, ok bool) {
+	v, ok := s.Get(key)
+	if !ok {
+		return
+	}
+	r, err := cast.ToIntE(v)
+	if ok = err == nil; !ok {
+		return
+	}
+	return
+}
+
+func (s *SigStrategyParam) GetInt16E(key string) (r int16, err error) {
+	v, ok := s.Get(key)
+	if !ok {
+		err = fmt.Errorf("param %s not provided", key)
+		return
+	}
+	r, err = cast.ToInt16E(v)
+	return
+}
+
+func (s *SigStrategyParam) GetFloat64(key string) (r float64, ok bool) {
+	v, ok := s.Get(key)
+	if !ok {
+		return
+	}
+	r, err := cast.ToFloat64E(v)
+	if ok = err == nil; !ok {
+		return
+	}
+	return
+}
+
+func (s *SigStrategyParam) GetBool(key string) (r bool, ok bool) {
+	v, ok := s.Get(key)
+	if !ok {
+		return
+	}
+	r, err := cast.ToBoolE(v)
+	if ok = err == nil; !ok {
+		return
+	}
+	return
+}

pkg/strategy/sig_strategy_registry.go

@@ -0,0 +1,51 @@
+package strategy
+
+import (
+	"fmt"
+	"sig-pub/pkg/utils/collect"
+)
+
+// 指标注册器
+type SigStrategyRegistry struct {
+	sigStrategies         *collect.SyncMap[string, ISigStrategy]          // 注册信号策略
+	intervalSigStrategies *collect.SyncMap[string, IIntervalsSigStrategy] // 注册窗口指标
+}
+
+func NewSigStrategyRegistry() *SigStrategyRegistry {
+	return &SigStrategyRegistry{
+		sigStrategies:         collect.NewSyncMap[string, ISigStrategy](),
+		intervalSigStrategies: collect.NewSyncMap[string, IIntervalsSigStrategy](),
+	}
+}
+
+func (r *SigStrategyRegistry) Init() (err error) {
+	// indicator regist
+	r.MustRegistStrategy(&GoldX{})
+	return
+}
+
+// RegisterStrategy
+func (svc *SigStrategyRegistry) RegistStrategy(strategy ISigStrategy) (err error) {
+	strategyName := strategy.Meta().Name
+	_, loaded := svc.sigStrategies.LoadOrStore(strategyName, strategy)
+	if loaded {
+		err = fmt.Errorf("strategy name %s already duplicated", strategyName)
+		return
+	}
+	return
+}
+
+func (svc *SigStrategyRegistry) MustRegistStrategy(strategy ISigStrategy) {
+	if err := svc.RegistStrategy(strategy); err != nil {
+		panic(err)
+	}
+}
+
+// NewSigStrategy
+func (r *SigStrategyRegistry) NewSigStrategy(name string) (strategy ISigStrategy, ok bool) {
+	strategy, ok = r.sigStrategies.Load(name)
+	if ok {
+		strategy = strategy.New()
+	}
+	return
+}

pkg/types/exchange.go

@@ -3,6 +3,7 @@ package types
 import (
 	"sig-pub/api/pb"
 	"sig-pub/pkg/utils/collect"
+	"sort"
 )
 
 var SupportedExchanges = []pb.ExchangeType{
@@ -18,6 +19,16 @@ func IsSupportExchange(exchange pb.ExchangeType) bool {
 	return false
 }
 
+// 对交易所进行排序
+func ExchangesSort(exchanges []pb.ExchangeType) {
+	if len(exchanges) < 2 {
+		return
+	}
+	sort.SliceStable(exchanges, func(i, j int) bool {
+		return exchanges[i] < exchanges[j]
+	})
+}
+
 type ExchangeState[T any] struct {
 	state []T
 }

pkg/types/interval.go

@@ -121,7 +121,7 @@ func IntervalsSort(intervals []Interval) {
 	if len(intervals) < 2 {
 		return
 	}
-	sort.Slice(intervals, func(i, j int) bool {
+	sort.SliceStable(intervals, func(i, j int) bool {
 		return intervalIotas[intervals[i]] < intervalIotas[intervals[j]]
 	})
 }

pkg/utils/collect/collect.go

@@ -2,6 +2,7 @@ package collect
 
 import (
 	"cmp"
+	"maps"
 	"sort"
 )
 
@@ -177,9 +178,7 @@ func SliceEquals[T comparable](slice1 []T, slice2 []T) bool {
 
 func CopyMap[K comparable, V any](src map[K]V) (dst map[K]V) {
 	dst = make(map[K]V, len(src))
-	for k, v := range src {
-		dst[k] = v
-	}
+	maps.Copy(dst, src)
 	return
 }