From 3ea82351cd47b2fd413a94e43ee8eac384a192ac Mon Sep 17 00:00:00 2001 From: strange Date: Thu, 23 Oct 2025 18:30:39 +0800 Subject: [PATCH] strategy runner --- config/exchange.toml | 4 +- go.mod | 2 +- internal/exchange/exchange_data_persist.go | 18 ++- internal/exchange/exchange_service.go | 110 ++++++++++------ internal/trading/backtrace/backtrace.go | 3 + internal/trading/kline_store.go | 24 +++- internal/trading/libs/account_position.go | 3 + internal/trading/strategy_context.go | 9 +- internal/trading/trading_plan.go | 25 ---- internal/trading/trading_plan_runner.go | 61 +++++++++ internal/trading/trading_service.go | 83 +++++------- pkg/indicator/indicator_registry.go | 46 +++++++ pkg/strategy/buy_strategy.go | 7 ++ pkg/strategy/exit_strategy.go | 2 +- pkg/strategy/gold_x.go | 26 +++- pkg/strategy/sig_strategy.go | 17 +-- pkg/strategy/sig_strategy_params.go | 139 +++++++++++++++++++++ pkg/strategy/sig_strategy_registry.go | 51 ++++++++ pkg/types/exchange.go | 11 ++ pkg/types/interval.go | 2 +- pkg/utils/collect/collect.go | 5 +- 21 files changed, 495 insertions(+), 153 deletions(-) create mode 100644 internal/trading/libs/account_position.go delete mode 100644 internal/trading/trading_plan.go create mode 100644 internal/trading/trading_plan_runner.go create mode 100644 pkg/indicator/indicator_registry.go create mode 100644 pkg/strategy/sig_strategy_params.go create mode 100644 pkg/strategy/sig_strategy_registry.go diff --git a/config/exchange.toml b/config/exchange.toml index c0a42b7..4d567a5 100644 --- a/config/exchange.toml +++ b/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 diff --git a/go.mod b/go.mod index fa1d500..7916719 100644 --- a/go.mod +++ b/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 diff --git a/internal/exchange/exchange_data_persist.go b/internal/exchange/exchange_data_persist.go index 6fc0b52..240f0c0 100644 --- a/internal/exchange/exchange_data_persist.go +++ b/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 { diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index cb3c074..c0882ba 100644 --- a/internal/exchange/exchange_service.go +++ b/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线 diff --git a/internal/trading/backtrace/backtrace.go b/internal/trading/backtrace/backtrace.go index 8478223..aad92ed 100644 --- a/internal/trading/backtrace/backtrace.go +++ b/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 } diff --git a/internal/trading/kline_store.go b/internal/trading/kline_store.go index 3bd85d6..5227661 100644 --- a/internal/trading/kline_store.go +++ b/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线序列 diff --git a/internal/trading/libs/account_position.go b/internal/trading/libs/account_position.go new file mode 100644 index 0000000..a8bfd5a --- /dev/null +++ b/internal/trading/libs/account_position.go @@ -0,0 +1,3 @@ +package libs + +// 账户持仓 diff --git a/internal/trading/strategy_context.go b/internal/trading/strategy_context.go index f07496a..27d816e 100644 --- a/internal/trading/strategy_context.go +++ b/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)) } diff --git a/internal/trading/trading_plan.go b/internal/trading/trading_plan.go deleted file mode 100644 index ced6596..0000000 --- a/internal/trading/trading_plan.go +++ /dev/null @@ -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 -} diff --git a/internal/trading/trading_plan_runner.go b/internal/trading/trading_plan_runner.go new file mode 100644 index 0000000..e3a49fb --- /dev/null +++ b/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) { + +} diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index a48814e..cbfa38e 100644 --- a/internal/trading/trading_service.go +++ b/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) diff --git a/pkg/indicator/indicator_registry.go b/pkg/indicator/indicator_registry.go new file mode 100644 index 0000000..122c671 --- /dev/null +++ b/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) +} diff --git a/pkg/strategy/buy_strategy.go b/pkg/strategy/buy_strategy.go index bec1336..fdc23ed 100644 --- a/pkg/strategy/buy_strategy.go +++ b/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 } diff --git a/pkg/strategy/exit_strategy.go b/pkg/strategy/exit_strategy.go index e2572a6..93b293f 100644 --- a/pkg/strategy/exit_strategy.go +++ b/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 { diff --git a/pkg/strategy/gold_x.go b/pkg/strategy/gold_x.go index 9482b5a..621fb9f 100644 --- a/pkg/strategy/gold_x.go +++ b/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) diff --git a/pkg/strategy/sig_strategy.go b/pkg/strategy/sig_strategy.go index 6760145..10c2f0e 100644 --- a/pkg/strategy/sig_strategy.go +++ b/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, ",")) diff --git a/pkg/strategy/sig_strategy_params.go b/pkg/strategy/sig_strategy_params.go new file mode 100644 index 0000000..a49c532 --- /dev/null +++ b/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 +} diff --git a/pkg/strategy/sig_strategy_registry.go b/pkg/strategy/sig_strategy_registry.go new file mode 100644 index 0000000..52ea496 --- /dev/null +++ b/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 +} diff --git a/pkg/types/exchange.go b/pkg/types/exchange.go index 7f35ee1..a4a808e 100644 --- a/pkg/types/exchange.go +++ b/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 } diff --git a/pkg/types/interval.go b/pkg/types/interval.go index c873eff..38ba50f 100644 --- a/pkg/types/interval.go +++ b/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]] }) } diff --git a/pkg/utils/collect/collect.go b/pkg/utils/collect/collect.go index 69cf3a7..afcb314 100644 --- a/pkg/utils/collect/collect.go +++ b/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 }