diff --git a/cmd/exchange/main.go b/cmd/exchange/main.go index cadcecf..ec55d33 100644 --- a/cmd/exchange/main.go +++ b/cmd/exchange/main.go @@ -110,7 +110,7 @@ func main() { if deregister, err := dis.Registry(grpcServer, register); err != nil { panic(err) } else { - defer deregister() + exit.AddHook(deregister, exit.WithOrderFront()) } // run grpc server diff --git a/cmd/market/main.go b/cmd/market/main.go index 4af9cc7..6a65932 100644 --- a/cmd/market/main.go +++ b/cmd/market/main.go @@ -71,7 +71,7 @@ func main() { if deregister, err := dis.Registry(grpcServer, register); err != nil { panic(err) } else { - defer deregister() + exit.AddHook(deregister, exit.WithOrderFront()) } // run grpc server diff --git a/internal/trading/backtest/backtest.go b/internal/trading/backtest/backtest.go index 7f49087..66e5202 100644 --- a/internal/trading/backtest/backtest.go +++ b/internal/trading/backtest/backtest.go @@ -40,7 +40,7 @@ func (b *Backtest) RunTradingPlan(ctx context.Context, tradingPlan *sig.TradingP instId := plan.InstId sigStrategy := tradingPlan.GetSigStrategy() - maxWindow := int(sigStrategy.RequiredSeries()) + maxWindow := int(sigStrategy.(strategy.ISingleSigStrategy).RequiredSeries()) if maxWindow < 0 || maxWindow > indicator.MaxWindow { err = fmt.Errorf("invalid window %d 0-%d, planId=%d", maxWindow, indicator.MaxWindow, plan.Id) return diff --git a/internal/trading/sig/indicator_context.go b/internal/trading/sig/indicator_context.go index ce45422..1f3d8a2 100644 --- a/internal/trading/sig/indicator_context.go +++ b/internal/trading/sig/indicator_context.go @@ -50,7 +50,7 @@ func (c *IndicatorContext) Get(offset int16) (kline types.Kline) { k, ok := c.kSeries.Get(offset) if !ok { lastTs := c.kSeries.LastTs() - zlog.Warningf("get kline series offset out of range: offset=%d, lastTs=%d", offset, lastTs) + zlog.Warningf("get kline series offset out of range: offset=%d, length=%d, lastTs=%d", offset, c.kSeries.Length(), lastTs) panic(fmt.Errorf("get kline series offset out of range: offset=%d", offset)) } return k diff --git a/internal/trading/sig/kline_series.go b/internal/trading/sig/kline_series.go index 7bbe94b..beb84e9 100644 --- a/internal/trading/sig/kline_series.go +++ b/internal/trading/sig/kline_series.go @@ -67,7 +67,7 @@ func (s *KlineSeries) Get(offset int16) (k types.Kline, ok bool) { length := len(s.klines) index := (length - 1) - int(offset) - if ok = index > 0 && index < length; !ok { + if ok = index >= 0 && index < length; !ok { return } return *(s.klines[index]), true diff --git a/internal/trading/sig/strategy_context.go b/internal/trading/sig/strategy_context.go index 3825df1..47b8a26 100644 --- a/internal/trading/sig/strategy_context.go +++ b/internal/trading/sig/strategy_context.go @@ -8,13 +8,8 @@ import ( "sig-pub/pkg/types/series" ) -type IOffsetStrategyContext interface { - strategy.ISigStrategyContext - SetOffset(offset int16) -} - type StrategyContext struct { - IOffsetStrategyContext + strategy.ISingleSigStrategyContext indicatorContext IOffsetIndicatorContext indicatorsReg *indicator.IndicatorRegistry @@ -27,10 +22,6 @@ func NewStrategyContext(indicatorContext IOffsetIndicatorContext, indicatorsReg } } -func (c *StrategyContext) SetOffset(offset int16) { - c.indicatorContext.SetOffset(offset) -} - func (c *StrategyContext) Get(offset int16) (kline types.Kline) { return c.indicatorContext.Get(offset) } @@ -48,32 +39,53 @@ func (c *StrategyContext) IndicatorW(name string, window int16) (s indicator.IIn return NewWindowIndicatorSeries(window, indicator, c.indicatorContext) } -type IOffsetIntervalStrategyContext interface { - strategy.IIntervalStrategyContext - SetOffset(offset int16) -} - // IntervalStrategyContext 周期策略上下文 type IntervalStrategyContext struct { - IOffsetIntervalStrategyContext + strategy.IIntervalSigStrategyContext + intervalIndicatorContexts map[types.Interval]*IndicatorContext + intervalKlineSeries *types.IntervalState[*KlineSeries] + indicatorsReg *indicator.IndicatorRegistry +} + +func NewIntervalStrategyContext(intervalKlineSeries *types.IntervalState[*KlineSeries], indicatorsReg *indicator.IndicatorRegistry) *IntervalStrategyContext { + return &IntervalStrategyContext{ + intervalIndicatorContexts: make(map[types.Interval]*IndicatorContext), + intervalKlineSeries: intervalKlineSeries, + indicatorsReg: indicatorsReg, + } +} + +func (c *IntervalStrategyContext) getIndicatorContext(interval types.Interval) *IndicatorContext { + ctx, ok := c.intervalIndicatorContexts[interval] + if !ok { + klineSeries := c.intervalKlineSeries.Get(interval) + if klineSeries == nil { + panic(fmt.Errorf("interval %s kline series is nil", interval)) + } + ctx = NewIndicatorContext(klineSeries) + c.intervalIndicatorContexts[interval] = ctx + } + return ctx } // Get [0]当前k线 func (c *IntervalStrategyContext) Get(interval types.Interval, offset int16) (kline types.Kline) { - return + ctx := c.getIndicatorContext(interval) + return ctx.Get(offset) } // Series [offset...end] func (c *IntervalStrategyContext) Series(interval types.Interval, offset, count int16) (klines series.Klines) { - return + ctx := c.getIndicatorContext(interval) + return ctx.Series(offset, count) } // 获取窗口类型指标 func (c *IntervalStrategyContext) IndicatorW(interval types.Interval, name string, window int16) (series indicator.IIndicatorSeries) { - return + indicatorContext := c.getIndicatorContext(interval) + indicator, ok := c.indicatorsReg.IndicatorW(name) + if !ok { + panic(fmt.Errorf("indicatorW %s not exists", name)) + } + return NewWindowIndicatorSeries(window, indicator, indicatorContext) } - -// 获取其它策略 -// func (c *IntervalStrategyContext) SigStrategy(interval types.Interval, name string, sigParam strategy.SigStrategyParam) (sigStrategy strategy.ISigStrategy) { -// return -// } diff --git a/internal/trading/sig/trading_plan.go b/internal/trading/sig/trading_plan.go index 366cefd..6534813 100644 --- a/internal/trading/sig/trading_plan.go +++ b/internal/trading/sig/trading_plan.go @@ -13,8 +13,9 @@ type TradingPlan struct { Plan entity.TradePlan indicatorReg *indicator.IndicatorRegistry + sigStrategyType strategy.SigStrategyType sigStrategy strategy.ISigStrategy - sigStrategyContext strategy.ISigStrategyContext + sigStrategyContext strategy.ISingleSigStrategyContext } func NewTradingPlan(plan entity.TradePlan, indicatorReg *indicator.IndicatorRegistry) *TradingPlan { @@ -41,10 +42,11 @@ func (r *TradingPlan) Init() (err error) { // initSigStrategy 初始化多空信号策略 // buy/sell -> 过滤/风控 -> tradeStrategy -> closeStrategy -func (r *TradingPlan) InitSigStrategy(sigStrategy strategy.ISigStrategy, param strategy.StrategyParam, sigStrategyContext strategy.ISigStrategyContext) (err error) { +func (r *TradingPlan) InitSigStrategy(sigStrategyType strategy.SigStrategyType, sigStrategy strategy.ISigStrategy, param strategy.StrategyParam, sigStrategyContext strategy.ISingleSigStrategyContext) (err error) { if err = sigStrategy.Init(param); err != nil { return } + r.sigStrategyType = sigStrategyType r.sigStrategy = sigStrategy r.sigStrategyContext = sigStrategyContext return @@ -59,7 +61,12 @@ func (r *TradingPlan) GetSigStrategy() strategy.ISigStrategy { func (r *TradingPlan) Update(signalType strategy.StrategyType) (side types.Side) { switch signalType { case strategy.StrategyTypeSig: - return r.sigStrategy.Update(r.sigStrategyContext) + switch r.sigStrategyType { + case strategy.SigStrategyTypeSingle: + return r.sigStrategy.(strategy.ISingleSigStrategy).Update(r.sigStrategyContext) + case strategy.SigStrategyTypeInterval: + return r.sigStrategy.(strategy.IIntervalSigStrategy).Update(nil) + } } return } diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index ec3ecbf..985145c 100644 --- a/internal/trading/trading_service.go +++ b/internal/trading/trading_service.go @@ -113,7 +113,7 @@ func (svc *TradingService) getTradingPlan(plan *entity.TradePlan, sigKlineSeries if err = sonic.UnmarshalString(plan.SigStrategyParam, &sigStrategyParam); err != nil { return } - sigStrategy, ok := svc.strategyReg.NewSigStrategy(plan.SigStrategy) + sigStrategyType, sigStrategy, ok := svc.strategyReg.NewSigStrategy(plan.SigStrategy) if !ok { err = fmt.Errorf("strategy %s not exists", plan.SigStrategy) return @@ -132,7 +132,7 @@ func (svc *TradingService) getTradingPlan(plan *entity.TradePlan, sigKlineSeries } sigIndicatorCtx := sig.NewIndicatorContext(sigKlineSeries) sigStrategyCtx := sig.NewStrategyContext(sigIndicatorCtx, svc.indicatorReg) - if err = tradingPlan.InitSigStrategy(sigStrategy, sigStrategyParam, sigStrategyCtx); err != nil { + if err = tradingPlan.InitSigStrategy(sigStrategyType, sigStrategy, sigStrategyParam, sigStrategyCtx); err != nil { return } return @@ -192,7 +192,7 @@ func (svc *TradingService) fetchHistoryKlineSeries(ctx context.Context, sr *pb.S } } } - zlog.Debugf("fetch history kline series: recv=%d, total=%d, use %s", recvTimes, recvTotal, watch.ElapsedFmt(".")) + zlog.Debugf("fetch history kline series: inst=%s(%s), interval=%s, recv=%d, total=%d, use %s", sr.InstId, sr.Exchange, sr.Interval, recvTimes, recvTotal, watch.ElapsedFmt(".")) return } @@ -241,7 +241,7 @@ func (svc *TradingService) IndicatorSeries(ctx context.Context, indicatorName st // StrategySeries 简单策略信号测试 func (svc *TradingService) StrategySeries(ctx context.Context, req *pb.ReqStrategySeries, rsp *pb.RspStrategySeries) (err error) { // sigStrategy - sigStrategy, ok := svc.strategyReg.NewSigStrategy(req.SigStrategy) + sigStrategyType, sigStrategy, ok := svc.strategyReg.NewSigStrategy(req.SigStrategy) if !ok { err = fmt.Errorf("strategy %s not exists", req.SigStrategy) return @@ -258,8 +258,153 @@ func (svc *TradingService) StrategySeries(ctx context.Context, req *pb.ReqStrate return } + switch sigStrategyType { + case strategy.SigStrategyTypeSingle: + rsp.Signal, rsp.Times, err = svc.singleStrategySeries(ctx, sigStrategy.(strategy.ISingleSigStrategy), req.Series) + case strategy.SigStrategyTypeInterval: + rsp.Signal, rsp.Times, err = svc.intervalStrategySeries(ctx, sigStrategy.(strategy.IIntervalSigStrategy), req.Series) + } + return +} + +// intervalStrategySeries 多周期策略 +func (svc *TradingService) intervalStrategySeries(ctx context.Context, intervalSigStrategy strategy.IIntervalSigStrategy, sr *pb.SeriesRange) (signals []pb.Side, times []int64, err error) { + driverInterval := types.Interval(sr.Interval) + driverIntervalAdder := types.SupportedIntervals[driverInterval] + // 各周期所需k线数量 + requiredIntervalSeries := intervalSigStrategy.RequiredIntervalSeries() + // 驱动周期外其他周期 + otherIntervals := make([]types.Interval, 0, 3) + requiredIntervalSeries.Range(func(interval types.Interval, series int16) { + if series > 0 { + otherIntervals = append(otherIntervals, interval) + } + }) + otherIntervals = collect.Filter(otherIntervals, func(_ int, interval types.Interval) bool { return interval != driverInterval }) + + // 通知其他周期更新的channel + otherIntervalCh := types.NewIntervalState[[]chan int64]() + // otherIntervalDstCh := types.NewIntervalState[chan int64]() + for _, interval := range otherIntervals { + otherIntervalCh.Set(interval, []chan int64{make(chan int64), make(chan int64)}) + } + // 各周期 series + intervalKlineSeries := types.NewIntervalState[*sig.KlineSeries]() + // 其他周期数据拉取 + stopCh := make(chan struct{}) + for _, interval := range otherIntervals { + kSeries := sig.NewKlineSeries(sr.Exchange, sr.InstId, interval) + intervalKlineSeries.Set(interval, kSeries) + + go func(interval types.Interval, kSeries *sig.KlineSeries) { + ch := otherIntervalCh.Get(interval) + srcCh := ch[0] + dstCh := ch[1] + intervalAdder := types.SupportedIntervals[interval] + driverTs := int64(0) + isr := &pb.SeriesRange{Exchange: sr.Exchange, InstId: sr.InstId, Before: sr.Before, After: sr.After, Count: sr.Count, Open: sr.Open, Live: sr.Live, Desc: sr.Desc, WindowExtra: sr.WindowExtra, Limit: sr.Limit, + Interval: string(interval)} + err1 := svc.fetchHistoryKlineSeries(ctx, isr, func(k *types.Kline) (err error) { + closeTs := intervalAdder(k.Ts, 1) + // 与驱动周期series保持同步更新 + if closeTs > driverTs { + if driverTs != 0 { + dstCh <- 0 // 通知更新完毕 + } + waitLoop: + for { + select { + case <-stopCh: + return io.EOF + case driverTs = <-srcCh: + if closeTs <= driverTs { + break waitLoop + } else { + dstCh <- 0 // 通知更新完毕 + } + } + } + } + if lastTs, serial := kSeries.Update(k); !serial { + err = fmt.Errorf("kline not series: %s(%s), interval=%s, lastTs=%d", sr.InstId, sr.Exchange, interval, lastTs) + return + } + return + }) + + if err1 == nil { + otherIntervalCh.Set(interval, nil) + } else if err1 != io.EOF { + zlog.Errorf("fetch history interval error: inst=%s(%s) interval=%s, err=%v", sr.InstId, sr.Exchange, interval, err1) + err = err1 + close(stopCh) + } + }(interval, kSeries) + } + + // 策略上下文 + intervalStrategyContext := sig.NewIntervalStrategyContext(intervalKlineSeries, svc.indicatorReg) + + // 驱动周期数据拉取 + driverSeries := sig.NewKlineSeries(sr.Exchange, sr.InstId, driverInterval) + intervalKlineSeries.Set(driverInterval, driverSeries) + err = svc.fetchHistoryKlineSeries(ctx, sr, func(k *types.Kline) (err error) { + driverTS := driverIntervalAdder(k.Ts, 1) + // 通知其他周期先更新 + otherIntervalCh.Range(func(interval types.Interval, ch []chan int64) { + if len(ch) == 2 { + select { + case <-stopCh: + err = io.EOF + return + case ch[0] <- driverTS: + <-ch[1] // 等待更新完毕 + } + } + }) + if err != nil { + return + } + + if lastTs, serial := driverSeries.Update(k); !serial { + err = fmt.Errorf("kline not series: %s(%s), interval=%s, lastTs=%d", sr.InstId, sr.Exchange, driverInterval, lastTs) + return + } + // 检查满足策略执行条件 + update := true + requiredIntervalSeries.Range(func(interval types.Interval, require int16) { + if update && require > 0 { + series := intervalKlineSeries.Get(interval) + update = series.Length() >= int(require) + } + }) + if !update { + return + } + intervalKlineSeries.Range(func(interval types.Interval, v *sig.KlineSeries) { + if v != nil { + zlog.Debugf("strategy update: interval series %s, %d", interval, v.Length()) + } + }) + sigSide := intervalSigStrategy.Update(intervalStrategyContext) + if sigSide.IsValid() { + side := lang.Ternary(sigSide == types.SideLong, pb.Side_BUY, pb.Side_SELL) + signals = append(signals, side) + times = append(times, k.Ts) + } + zlog.Debugf("strategy update finish ------------------------------------") + return + }) + if err != io.EOF { + close(stopCh) + } + return +} + +// singleStrategySeries 单周期策略 +func (svc *TradingService) singleStrategySeries(ctx context.Context, sigStrategy strategy.ISingleSigStrategy, sr *pb.SeriesRange) (signals []pb.Side, times []int64, err error) { + interval := types.Interval(sr.Interval) requiredSeries := int(sigStrategy.RequiredSeries()) - sr := req.Series sr.WindowExtra = uint32(requiredSeries) kSeries := sig.NewKlineSeries(sr.Exchange, sr.InstId, interval) indicatorContext := sig.NewIndicatorContext(kSeries) @@ -275,8 +420,8 @@ func (svc *TradingService) StrategySeries(ctx context.Context, req *pb.ReqStrate sigSide := sigStrategy.Update(strategyContext) if sigSide.IsValid() { side := lang.Ternary(sigSide == types.SideLong, pb.Side_BUY, pb.Side_SELL) - rsp.Signal = append(rsp.Signal, side) - rsp.Times = append(rsp.Times, k.Ts) + signals = append(signals, side) + times = append(times, k.Ts) } return }) diff --git a/pkg/config/loader.go b/pkg/config/loader.go index 0e561dd..faf0b9e 100644 --- a/pkg/config/loader.go +++ b/pkg/config/loader.go @@ -34,7 +34,7 @@ func LoadConfig[T any](conf T, confPathArg ...string) error { confPath := "" confName := "config" confType := "toml" - envPrefix := "SO" + envPrefix := "SIG" if len(confPathArg) > 0 { confPath = confPathArg[0] diff --git a/pkg/grpc/discovery/consul/watch.go b/pkg/grpc/discovery/consul/watch.go index c32f648..e890ccf 100644 --- a/pkg/grpc/discovery/consul/watch.go +++ b/pkg/grpc/discovery/consul/watch.go @@ -41,8 +41,8 @@ func (w *Watcher) WatchServices() (err error) { // services case map[string][]string: services := make([]string, 0, len(d)) - for i := range d { - if i == "consul" { + for i, t := range d { + if i == "consul" || collect.NotIn("grpc", t...) { continue } services = append(services, i) @@ -50,7 +50,7 @@ func (w *Watcher) WatchServices() (err error) { w.registerServiceWatcher(i) } } - zlog.Infof("consul services update: %v", services) + zlog.Infof("watch consul grpc services: %v", services) // remove unknown services from watchers // var dels []string diff --git a/pkg/strategy/buy_strategy.go b/pkg/strategy/buy_strategy.go deleted file mode 100644 index fdc23ed..0000000 --- a/pkg/strategy/buy_strategy.go +++ /dev/null @@ -1,12 +0,0 @@ -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/cross_star.go b/pkg/strategy/cross_star.go index 60d7ae7..114ff5e 100644 --- a/pkg/strategy/cross_star.go +++ b/pkg/strategy/cross_star.go @@ -8,7 +8,6 @@ import ( // CrossStar type CrossStar struct { - ISigStrategy IIntervalSigStrategy rate float64 rate2 float64 @@ -39,13 +38,17 @@ func (s *CrossStar) Init(param StrategyParam) (err error) { // 校验参数, 并 return } -func (s *CrossStar) RequiredSeries() int16 { - return 1 +func (s *CrossStar) RequiredIntervalSeries() (iss *types.IntervalState[int16]) { + iss = types.NewIntervalState[int16]() + iss.Set(types.Interval5m, 1) + iss.Set(types.Interval15m, 2) + iss.Set(types.Interval30m, 2) + return } -func (s *CrossStar) Update(ctx ISigStrategyContext) (side types.Side) { +func (s *CrossStar) Update(ctx IIntervalSigStrategyContext) (side types.Side) { // O 109744.8 H 110600 L 109507.5 C 109686.8 - k0 := ctx.Get(0) + k0 := ctx.Get("5m", 0) open, close, high, low := k0.OpenF64(), k0.CloseF64(), k0.HighF64(), k0.LowF64() base := math.Abs(open - close) // 58 rup := (high - max(open, close)) / base // 855.2 / 2 427.6 diff --git a/pkg/strategy/exit_strategy.go b/pkg/strategy/exit_strategy.go deleted file mode 100644 index 93b293f..0000000 --- a/pkg/strategy/exit_strategy.go +++ /dev/null @@ -1,13 +0,0 @@ -package strategy - -import "github.com/govalues/decimal" - -// todo Exit 止盈止损策略(trading service 管理) -type IExitStrategy interface { - Name() string // 获取策略名称，便于日志 - Update(ctx IExitStrategyContext) -} - -type IExitStrategyContext interface { - LastPrice() decimal.Decimal -} diff --git a/pkg/strategy/gold_x.go b/pkg/strategy/gold_x.go index c083801..7e8b05a 100644 --- a/pkg/strategy/gold_x.go +++ b/pkg/strategy/gold_x.go @@ -8,7 +8,6 @@ import ( // GoldX 金叉策略 type GoldX struct { ISigStrategy - IIntervalSigStrategy short, long int16 } @@ -45,7 +44,7 @@ func (s *GoldX) RequiredSeries() int16 { return max(s.long, s.short) + 1 } -func (s *GoldX) Update(ctx ISigStrategyContext) (side types.Side) { +func (s *GoldX) Update(ctx ISingleSigStrategyContext) (side types.Side) { sma14 := ctx.IndicatorW("sma", s.short) sma28 := ctx.IndicatorW("sma", s.long) // 包装方法 crossover/crossunder @@ -61,9 +60,3 @@ func (s *GoldX) Update(ctx ISigStrategyContext) (side types.Side) { } return } - -func (s *GoldX) UpdateByIntervals(ctx IIntervalStrategyContext) (side types.Side) { - series5m := ctx.Series(types.Interval5m, 0, 2) - series5m.Close().Diff() - return -} diff --git a/pkg/strategy/sig_strategy.go b/pkg/strategy/sig_strategy.go index 98eb87a..e6f697e 100644 --- a/pkg/strategy/sig_strategy.go +++ b/pkg/strategy/sig_strategy.go @@ -10,9 +10,7 @@ import ( type ISigStrategy interface { New() ISigStrategy Meta() StrategyMeta - RequiredSeries() int16 // 需要的最小数据k线数, 回测时用, 若不定义则取最大窗口值 Init(param StrategyParam) (err error) // 校验参数, 并根据参数初始化策略 - Update(ctx ISigStrategyContext) (side types.Side) } type StrategyMeta struct { @@ -21,9 +19,15 @@ type StrategyMeta struct { Args []Param `json:"args"` // 参数定义 } -// ISigStrategyContext 策略外部访问能力 -// klineSeries, Indicator -type ISigStrategyContext interface { +// ISingleSigStrategy 单周期单交易所策略 +type ISingleSigStrategy interface { + ISigStrategy + RequiredSeries() int16 // 需要的最小数据k线数, 回测时用, 若不定义则取最大窗口值 + Update(ctx ISingleSigStrategyContext) (side types.Side) +} + +// ISingleSigStrategyContext 策略上下文 +type ISingleSigStrategyContext interface { // Get [0]当前k线 Get(offset int16) types.Kline // Series [offset...end] @@ -31,3 +35,20 @@ type ISigStrategyContext interface { // 获取窗口类型指标 IndicatorW(name string, window int16) indicator.IIndicatorSeries } + +// 多周期k线策略接口 +type IIntervalSigStrategy interface { + ISigStrategy + RequiredIntervalSeries() (iss *types.IntervalState[int16]) // 需要的各周期最小数据k线数, 回测时用, 若不定义则取最大窗口值 + Update(ctx IIntervalSigStrategyContext) (side types.Side) +} + +// IIntervalSigStrategyContext 多周期策略上下文 +type IIntervalSigStrategyContext interface { + // Get [0]当前k线 + Get(interval types.Interval, offset int16) types.Kline + // Series [offset...end] + Series(interval types.Interval, offset, count int16) (klines series.Klines) + // 获取窗口类型指标 + IndicatorW(interval types.Interval, name string, window int16) indicator.IIndicatorSeries +} diff --git a/pkg/strategy/sig_strategy_exchanges.go b/pkg/strategy/sig_strategy_exchanges.go deleted file mode 100644 index 64add10..0000000 --- a/pkg/strategy/sig_strategy_exchanges.go +++ /dev/null @@ -1,31 +0,0 @@ -package strategy - -import ( - "sig-pub/pkg/indicator" - "sig-pub/pkg/types" - "sig-pub/pkg/types/series" -) - -// IExchangesSigStrategy 多交易所策略 -type IExchangesSigStrategy interface { - ISigStrategy -} - -// IExchangesIntervalsStrategy 多交易所多周期策略 -type IExchangesIntervalsStrategy interface { - ISigStrategy -} - -// IExchangesSigStrategyContext 策略外部访问能力 -// klineSeries, Indicator -type IExchangesSigStrategyContext interface { - Buy() // 发出多信号 - Sell() // 发出空信号 - - // Get [0]当前k线 - Get(interval types.Interval, offset int16) types.Kline - // Series [offset...end] - Series(interval types.Interval, offset, count int16) (klines series.Klines) - // 获取窗口类型指标 - IndicatorW(interval types.Interval, name string, window int16) indicator.IIndicatorSeries -} diff --git a/pkg/strategy/sig_strategy_intervals.go b/pkg/strategy/sig_strategy_intervals.go deleted file mode 100644 index 1201700..0000000 --- a/pkg/strategy/sig_strategy_intervals.go +++ /dev/null @@ -1,22 +0,0 @@ -package strategy - -import ( - "sig-pub/pkg/indicator" - "sig-pub/pkg/types" - "sig-pub/pkg/types/series" -) - -// 多周期k线策略接口 -type IIntervalSigStrategy interface { - UpdateByInterval(ctx IIntervalStrategyContext) (side types.Side) -} - -// IIntervalStrategyContext 多周期策略上下文 -type IIntervalStrategyContext interface { - // Get [0]当前k线 - Get(interval types.Interval, offset int16) types.Kline - // Series [offset...end] - Series(interval types.Interval, offset, count int16) (klines series.Klines) - // 获取窗口类型指标 - IndicatorW(interval types.Interval, name string, window int16) indicator.IIndicatorSeries -} diff --git a/pkg/strategy/sig_strategy_registry.go b/pkg/strategy/sig_strategy_registry.go index eaa2729..50973c5 100644 --- a/pkg/strategy/sig_strategy_registry.go +++ b/pkg/strategy/sig_strategy_registry.go @@ -7,13 +7,15 @@ import ( // 指标注册器 type SigStrategyRegistry struct { - sigStrategies *collect.SyncMap[string, ISigStrategy] // 注册信号策略 - intervalSigStrategies *collect.SyncMap[string, IIntervalSigStrategy] // 注册窗口指标 + sigStrategies *collect.SyncMap[string, SigStrategyType] // 注册策略 + singleSigStrategies *collect.SyncMap[string, ISingleSigStrategy] // 单周期策略 + intervalSigStrategies *collect.SyncMap[string, IIntervalSigStrategy] // 多周期策略 } func NewSigStrategyRegistry() *SigStrategyRegistry { return &SigStrategyRegistry{ - sigStrategies: collect.NewSyncMap[string, ISigStrategy](), + sigStrategies: collect.NewSyncMap[string, SigStrategyType](), + singleSigStrategies: collect.NewSyncMap[string, ISingleSigStrategy](), intervalSigStrategies: collect.NewSyncMap[string, IIntervalSigStrategy](), } } @@ -27,25 +29,51 @@ func (r *SigStrategyRegistry) Init() (err error) { } // RegisterStrategy -func (svc *SigStrategyRegistry) RegistStrategy(strategy ISigStrategy) (err error) { +func (r *SigStrategyRegistry) RegistStrategy(strategy ISigStrategy) (err error) { + var sigType SigStrategyType + if _, ok := strategy.(ISingleSigStrategy); ok { + sigType = SigStrategyTypeSingle + } else if _, ok := strategy.(IIntervalSigStrategy); ok { + sigType = SigStrategyTypeInterval + } else { + err = fmt.Errorf("sig strategy must be one of the [ISingleSigStrategy, IIntervalSigStrategy]") + return + } + strategyName := strategy.Meta().Name - _, loaded := svc.sigStrategies.LoadOrStore(strategyName, strategy) + _, loaded := r.sigStrategies.LoadOrStore(strategyName, sigType) if loaded { - err = fmt.Errorf("strategy name %s already duplicated", strategyName) + err = fmt.Errorf("strategy name %s duplicated", strategyName) return } + + switch sigType { + case SigStrategyTypeSingle: + r.singleSigStrategies.Store(strategyName, strategy.(ISingleSigStrategy)) + case SigStrategyTypeInterval: + r.intervalSigStrategies.Store(strategyName, strategy.(IIntervalSigStrategy)) + } return } -func (svc *SigStrategyRegistry) MustRegistStrategy(strategy ISigStrategy) { - if err := svc.RegistStrategy(strategy); err != nil { +func (r *SigStrategyRegistry) MustRegistStrategy(strategy ISigStrategy) { + if err := r.RegistStrategy(strategy); err != nil { panic(err) } } // NewSigStrategy -func (r *SigStrategyRegistry) NewSigStrategy(name string) (strategy ISigStrategy, ok bool) { - strategy, ok = r.sigStrategies.Load(name) +func (r *SigStrategyRegistry) NewSigStrategy(strategyName string) (sigType SigStrategyType, strategy ISigStrategy, ok bool) { + sigType, ok = r.sigStrategies.Load(strategyName) + if !ok { + return + } + switch sigType { + case SigStrategyTypeSingle: + strategy, ok = r.singleSigStrategies.Load(strategyName) + case SigStrategyTypeInterval: + strategy, ok = r.intervalSigStrategies.Load(strategyName) + } if ok { strategy = strategy.New() } diff --git a/pkg/strategy/strategy.go b/pkg/strategy/strategy.go index e3afecf..5003e43 100644 --- a/pkg/strategy/strategy.go +++ b/pkg/strategy/strategy.go @@ -9,6 +9,15 @@ import ( "strings" ) +type SigStrategyType int32 + +const ( + _ SigStrategyType = iota + SigStrategyTypeSingle // 单周期单交易所策略 + SigStrategyTypeInterval // 多周期策略 + SigStrategyTypeExchange // 多交易所策略 +) + type StrategyType int32 const ( diff --git a/pkg/strategy/super_trend.go b/pkg/strategy/super_trend.go index 8f22aed..6b6a590 100644 --- a/pkg/strategy/super_trend.go +++ b/pkg/strategy/super_trend.go @@ -68,7 +68,7 @@ func (s *SupertrendBOSWaves) RequiredSeries() int16 { return s.atrLength + 1 } -func (s *SupertrendBOSWaves) Update(ctx ISigStrategyContext) (side types.Side) { +func (s *SupertrendBOSWaves) Update(ctx ISingleSigStrategyContext) (side types.Side) { k0 := ctx.Get(0) high, low, close := k0.HighF64(), k0.LowF64(), k0.CloseF64() diff --git a/pkg/trade/close_strategy.go b/pkg/trade/close_strategy.go index 9b2f049..c9cbbc2 100644 --- a/pkg/trade/close_strategy.go +++ b/pkg/trade/close_strategy.go @@ -5,6 +5,7 @@ import ( "sig-pub/pkg/types/decimals" ) +// Exit 止盈止损策略(trading service 管理) type ICloseStrategy interface { OnKline(k types.Kline, pos *Position) (closePos bool, cause Cause) OnPrice(price float64, pos *Position) (closePos bool, cause Cause) diff --git a/pkg/trade/trade_strategy.go b/pkg/trade/trade_strategy.go index 23a45ca..9fc8016 100644 --- a/pkg/trade/trade_strategy.go +++ b/pkg/trade/trade_strategy.go @@ -1,13 +1,22 @@ package trade +import "github.com/govalues/decimal" + // ITradeStrategy 下单策略 // 根据购买信号和账户信息生成下单参数 +// TradeStrategy 下单买入策略接口(控制滑点, 仓位管理) type ITradeStrategy interface { - OnSingal() + // 下单, 币种,方向,杠杆 + Trade(arg ...string) + + // 市场价格更新 + Update(ctx ITradeStrategyContext, account ITradeAccount) } -type TradeStrategyParam struct { +type ITradeStrategyContext interface { + // 最新价格 + LastPrice() decimal.Decimal } -type TradeStrategy struct { +type TradeStrategyParam struct { }