diff --git a/api/trading.proto b/api/trading.proto index f3e2d47..5c88608 100644 --- a/api/trading.proto +++ b/api/trading.proto @@ -6,34 +6,18 @@ import "api/pub.proto"; option go_package = "./pb"; service TradingService { - rpc SubIndicator(IndicatorSubReq) returns (stream Indicator); // 订阅指标 rpc IndicatorSeries(ReqIndicatorSeries) returns (RspIndicatorSeries); // 获取指标序列 rpc StrategySeries(ReqStrategySeries) returns (RspStrategySeries); // 获取指标序列 rpc Backtest(ReqBacktest) returns (RspBacktest); // 交易计划回测 } -message IndicatorSubReq { - string topic = 1; - string instId = 2; - int32 window = 3; -} - -message Indicator { - ExchangeType exhcange = 1; - string instId = 2; - string indicator = 3; - string sub = 4; // 指标子标题, MA5, MA10, MA20 - int64 Ts = 5; - bytes payload = 8; -} - message ReqIndicatorSeries { string indicator = 1; - uint32 window = 2; // 指标窗口大小 + int32 digit = 2; // 结果小数位数 SeriesRange series = 9; google.protobuf.Struct input = 10; // 指标参数 } -message RspIndicatorSeries{ +message RspIndicatorSeries { repeated double matrix = 1; repeated int64 times = 2; } @@ -53,6 +37,6 @@ message ReqBacktest { string stime = 2; string etime = 3; } -message RspBacktest{ +message RspBacktest { } diff --git a/internal/trading/sig/indicator_context.go b/internal/trading/sig/indicator_context.go index d82966a..35f1b44 100644 --- a/internal/trading/sig/indicator_context.go +++ b/internal/trading/sig/indicator_context.go @@ -1,12 +1,24 @@ package sig import ( + "fmt" "sig-pub/pkg/indicator" "sig-pub/pkg/types" "sig-pub/pkg/types/series" + "sig-pub/pkg/utils/collect" "sig-pub/pkg/zlog" + "slices" + "strings" + + "github.com/spf13/cast" ) +type IndicatorStates map[string]*IndicatorState + +func NewIndicatorStates() IndicatorStates { + return make(IndicatorStates) +} + type IOffsetIndicatorContext interface { indicator.IIndicatorContext SetOffset(offset int16) @@ -17,15 +29,23 @@ type IOffsetIndicatorContext interface { // IndicatorContext 指标上下文, 提供k线序列给指标计算使用 type IndicatorContext struct { IOffsetIndicatorContext - input types.Input - kSeries *KlineSeries - offset int16 + indicator indicator.IIndicator + indicatorsReg *indicator.IndicatorRegistry + input types.Input + indicatorStates map[string]*IndicatorState // {macd{window:0,fast:9,slow:21,single:10}: state} 初始化时与KlineSeries周期同步 + kSeries *KlineSeries + offset int16 + indicatorTrace []string // 指标调用链避免指标循环引用 } -func NewIndicatorContext(input types.Input, kSeries *KlineSeries) *IndicatorContext { +func NewIndicatorContext(indicator indicator.IIndicator, input types.Input, indicatorStates IndicatorStates, kSeries *KlineSeries, indicatorsReg *indicator.IndicatorRegistry) *IndicatorContext { return &IndicatorContext{ - input: input, - kSeries: kSeries, + indicator: indicator, + input: input, + indicatorStates: indicatorStates, + kSeries: kSeries, + indicatorsReg: indicatorsReg, + indicatorTrace: []string{indicator.Meta().Name}, } } @@ -65,3 +85,81 @@ func (c *IndicatorContext) Series(offset, count int16) (klines series.Klines) { } return ks } + +// func (c *IndicatorContext) State(effects ...any) +// 窗口/参数 +// tradingPlan -> interval -> context -> {macd{window:0,fast:9,slow:21,single:10}: state, ema21:state} -> state[[ema]{1.1, 1.2}, [ema]{1.3, 1.4}] +func (c *IndicatorContext) State() indicator.IIndicatorState { + inputs := collect.Mapping(c.indicator.Meta().Input, func(in types.InputArg) string { + return c.Input().String(in.Name) + }) + stateKey := fmt.Sprintf("%s{%s}", c.indicator.Meta().Name, strings.Join(inputs, ",")) + + state, ok := c.indicatorStates[stateKey] + if !ok { + state = NewIndicatorState(c.kSeries) + c.indicatorStates[stateKey] = state + // 从头KlineSeries跑一遍,针对ema,macd等回溯迭代指标 + c.backtrackIndicatorState(c.indicator) + } + return state +} + +// 当首次初始化某个state dea后, 把indicator在klineSeries从头跑一遍 +func (c *IndicatorContext) backtrackIndicatorState(indicator indicator.IIndicator) { + _offset := c.offset + defer c.SetOffset(_offset) + + previousCandles := indicator.CandlePeriods(c) + length := c.kSeries.Length() + for i := range length { + if i < int(previousCandles) { + continue + } + offset := int16(length - 1 - i) + c.SetOffset(offset) + indicator.Calculate(c) + } +} + +// 获取窗口类型指标 +func (c *IndicatorContext) Indicator(name string, args ...any) (series indicator.IIndicatorSeries) { + indicator, ok := c.indicatorsReg.Indicator(name) + if !ok { + panic(fmt.Errorf("indicator %s not exists", name)) + } + + // 避免循环依赖 + if slices.Contains(c.indicatorTrace, name) { + panic(fmt.Errorf("indicator %s recursive call", name)) + } + + input := matchIndicatorArgs(args...) + indicatorContext := NewIndicatorContext(indicator, input, c.indicatorStates, c.kSeries, c.indicatorsReg) + indicatorContext.indicatorTrace = append(c.indicatorTrace, name) + return NewWindowIndicatorSeries(indicator, indicatorContext) +} + +func matchIndicatorArgs(args ...any) (input types.Input) { +inputLoop: + for _, arg := range args { + switch v := arg.(type) { + case types.Input: + input = v + break inputLoop + } + } +windowLoop: + for _, arg := range args { + switch v := arg.(type) { + case int16, int, int32, int64: + window := cast.ToInt16(v) + if input == nil { + input = make(types.Input) + } + input["window"] = window + break windowLoop + } + } + return +} diff --git a/internal/trading/sig/indicator_series.go b/internal/trading/sig/indicator_series.go index 6ba7e3b..394fa9a 100644 --- a/internal/trading/sig/indicator_series.go +++ b/internal/trading/sig/indicator_series.go @@ -8,23 +8,25 @@ import ( // WindowIndicatorSeries 封装 type WindowIndicatorSeries struct { indicator.IIndicatorSeries - window int16 - indicator indicator.IWindowIndicator + indicator indicator.IIndicator indicatorContext IOffsetIndicatorContext } -func NewWindowIndicatorSeries(window int16, indicator indicator.IWindowIndicator, indicatorContext IOffsetIndicatorContext) *WindowIndicatorSeries { +func NewWindowIndicatorSeries(indicator indicator.IIndicator, indicatorContext IOffsetIndicatorContext) *WindowIndicatorSeries { return &WindowIndicatorSeries{ - window: window, indicator: indicator, indicatorContext: indicatorContext, } } +func (s *WindowIndicatorSeries) CandlePeriods() int16 { + return s.indicator.CandlePeriods(s.indicatorContext) +} + func (s *WindowIndicatorSeries) Get(offset int16) (vector float64) { // 根据当前相对offset s.indicatorContext.AddOffset(offset) - vector = s.indicator.Calculate(s.indicatorContext, s.window) + vector = s.indicator.Calculate(s.indicatorContext) // 计算结束后还原 s.indicatorContext.AddOffset(-offset) return @@ -35,7 +37,7 @@ func (s *WindowIndicatorSeries) Series(offset, count int16) (matrix series.Float // 设置当前相对offset s.indicatorContext.AddOffset(offset) for range count { - vector := s.indicator.Calculate(s.indicatorContext, s.window) + vector := s.indicator.Calculate(s.indicatorContext) matrix.Push(vector) offset++ diff --git a/internal/trading/sig/indicator_state.go b/internal/trading/sig/indicator_state.go new file mode 100644 index 0000000..0bb0bcd --- /dev/null +++ b/internal/trading/sig/indicator_state.go @@ -0,0 +1,47 @@ +package sig + +import ( + "sig-pub/pkg/indicator" + "sig-pub/pkg/types" + "sig-pub/pkg/types/series" +) + +// IndicatorState +// ema, obv 指标递归计算时的状态存储 +type IndicatorState struct { + indicator.IIndicatorState + kSeries *KlineSeries + state map[string]*types.RingSeries[float64] +} + +func NewIndicatorState(kSeries *KlineSeries) *IndicatorState { + return &IndicatorState{ + kSeries: kSeries, + state: make(map[string]*types.RingSeries[float64]), + } +} + +func (s *IndicatorState) ring(k string) *types.RingSeries[float64] { + ring, ok := s.state[k] + if !ok { + ring = types.NewRingSeries[float64](indicator.MaxWindow, 8) + s.state[k] = ring + // 从 kSeries0 开始 calc ind 初始化 + // 递归初始值 + } + return ring +} + +func (s *IndicatorState) Set(k string, v float64) { + s.ring(k).Push(v) +} + +func (s *IndicatorState) Get(k string, offset int16) (v float64, ok bool) { + offset -= 1 + return s.ring(k).Get(int(offset)) +} + +func (s *IndicatorState) Series(k string, offset, count int16) (v series.Floats, ok bool) { + offset -= 1 + return s.ring(k).Series(int(offset), int(count)) +} diff --git a/internal/trading/sig/strategy_context.go b/internal/trading/sig/strategy_context.go index 389837f..fb562fe 100644 --- a/internal/trading/sig/strategy_context.go +++ b/internal/trading/sig/strategy_context.go @@ -11,16 +11,18 @@ import ( type StrategyContext struct { strategy.ISingleSigStrategyContext - input types.Input - kSeries *KlineSeries - indicatorsReg *indicator.IndicatorRegistry + input types.Input + kSeries *KlineSeries + indicatorsReg *indicator.IndicatorRegistry + indicatorContextStates IndicatorStates } func NewStrategyContext(input types.Input, kSeries *KlineSeries, indicatorsReg *indicator.IndicatorRegistry) *StrategyContext { return &StrategyContext{ - input: input, - kSeries: kSeries, - indicatorsReg: indicatorsReg, + input: input, + kSeries: kSeries, + indicatorsReg: indicatorsReg, + indicatorContextStates: NewIndicatorStates(), } } @@ -38,35 +40,32 @@ func (c *StrategyContext) Series(offset, count int16) (klines series.Klines) { } // 获取窗口类型指标 -func (c *StrategyContext) IndicatorW(name string, window int16, args ...any) (s indicator.IIndicatorSeries) { - indicator, ok := c.indicatorsReg.IndicatorW(name) +func (c *StrategyContext) Indicator(name string, args ...any) (s indicator.IIndicatorSeries) { + indicator, ok := c.indicatorsReg.Indicator(name) if !ok { - panic(fmt.Errorf("indicatorW %s not exists", name)) + panic(fmt.Errorf("indicator %s not exists", name)) } - var input types.Input - if len(args) > 0 { - if in, ok := args[0].(types.Input); ok { - input = in - } - } - indicatorContext := NewIndicatorContext(input, c.kSeries) - return NewWindowIndicatorSeries(window, indicator, indicatorContext) + input := matchIndicatorArgs(args...) + indicatorContext := NewIndicatorContext(indicator, input, c.indicatorContextStates, c.kSeries, c.indicatorsReg) + return NewWindowIndicatorSeries(indicator, indicatorContext) } // IntervalStrategyContext 周期策略上下文 type IntervalStrategyContext struct { strategy.IIntervalSigStrategyContext - input types.Input - intervalKlineSeries *types.IntervalState[*KlineSeries] - indicatorsReg *indicator.IndicatorRegistry + input types.Input + intervalKlineSeries *types.IntervalState[*KlineSeries] + indicatorsReg *indicator.IndicatorRegistry + intervalIndicatorContextStates map[types.Interval]IndicatorStates } func NewIntervalStrategyContext(input types.Input, intervalKlineSeries *types.IntervalState[*KlineSeries], indicatorsReg *indicator.IndicatorRegistry) *IntervalStrategyContext { return &IntervalStrategyContext{ - input: input, - intervalKlineSeries: intervalKlineSeries, - indicatorsReg: indicatorsReg, + input: input, + intervalKlineSeries: intervalKlineSeries, + indicatorsReg: indicatorsReg, + intervalIndicatorContextStates: make(map[types.Interval]IndicatorStates), } } @@ -96,20 +95,20 @@ func (c *IntervalStrategyContext) Series(interval types.Interval, offset, count } // 获取窗口类型指标 -func (c *IntervalStrategyContext) IndicatorW(interval types.Interval, name string, window int16, args ...any) (series indicator.IIndicatorSeries) { - indicator, ok := c.indicatorsReg.IndicatorW(name) +func (c *IntervalStrategyContext) Indicator(interval types.Interval, name string, args ...any) (series indicator.IIndicatorSeries) { + indicator, ok := c.indicatorsReg.Indicator(name) if !ok { - panic(fmt.Errorf("indicatorW %s not exists", name)) + panic(fmt.Errorf("indicator %s not exists", name)) } - var input types.Input - if len(args) > 0 { - if in, ok := args[0].(types.Input); ok { - input = in - } + input := matchIndicatorArgs(args...) + kSeries := c.getCandleSeries(interval) + // 状态传递 + state, ok := c.intervalIndicatorContextStates[interval] + if !ok { + state = NewIndicatorStates() + c.intervalIndicatorContextStates[interval] = state } - cs := c.getCandleSeries(interval) - indicatorContext := NewIndicatorContext(input, cs) - - return NewWindowIndicatorSeries(window, indicator, indicatorContext) + indicatorContext := NewIndicatorContext(indicator, input, state, kSeries, c.indicatorsReg) + return NewWindowIndicatorSeries(indicator, indicatorContext) } diff --git a/internal/trading/trading_grpc_server.go b/internal/trading/trading_grpc_server.go index bba5f91..9b4fdeb 100644 --- a/internal/trading/trading_grpc_server.go +++ b/internal/trading/trading_grpc_server.go @@ -25,7 +25,7 @@ func (svr *TradingGrpcServer) Init() (err error) { func (svr *TradingGrpcServer) IndicatorSeries(ctx context.Context, req *pb.ReqIndicatorSeries) (rsp *pb.RspIndicatorSeries, err error) { // s, err := structpb.NewStruct(map[string]any{}) input := req.Input.AsMap() - matrix, times, err := svr.tradingService.IndicatorSeries(ctx, req.Indicator, req.Window, types.Input(input), req.Series) + matrix, times, err := svr.tradingService.IndicatorSeries(ctx, req.Indicator, req.Digit, types.Input(input), req.Series) if err != nil { return } diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index 9a6ba27..06bd314 100644 --- a/internal/trading/trading_service.go +++ b/internal/trading/trading_service.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "io" + "math" "sig-pub/api/pb" "sig-pub/pkg/client" "sig-pub/pkg/data" @@ -198,8 +199,8 @@ func (svc *TradingService) fetchHistoryKlineSeries(ctx context.Context, sr *pb.S } // IndicatorSeries 获取指标实时或历史序列数据, 闭区间 -func (svc *TradingService) IndicatorSeries(ctx context.Context, indicatorName string, window uint32, input types.Input, sr *pb.SeriesRange) (matrix []float64, times []int64, err error) { - indicator, ok := svc.indicatorReg.IndicatorW(indicatorName) +func (svc *TradingService) IndicatorSeries(ctx context.Context, indicatorName string, digit int32, input types.Input, sr *pb.SeriesRange) (matrix []float64, times []int64, err error) { + indicator, ok := svc.indicatorReg.Indicator(indicatorName) if !ok { err = fmt.Errorf("indicator %s not exists", indicatorName) return @@ -212,11 +213,16 @@ func (svc *TradingService) IndicatorSeries(ctx context.Context, indicatorName st } // 查询历史指标数据 - requiredSeries := int(indicator.RequiredSeries(int16(window), input)) - sr.WindowExtra = uint32(max(0, requiredSeries-1)) kSeries := sig.NewKlineSeries(sr.Exchange, sr.InstId, types.Interval(sr.Interval)) - indicatorContext := sig.NewIndicatorContext(input, kSeries) + indicatorContext := sig.NewIndicatorContext(indicator, input, sig.NewIndicatorStates(), kSeries, svc.indicatorReg) + candlePeriods := int(indicator.CandlePeriods(indicatorContext)) + sr.Desc = false + sr.WindowExtra = uint32(max(0, candlePeriods-1)) + + // 保留小数位数 + digit = lang.Ternary(digit > 0 && digit <= 10, digit, 6) + pow := math.Pow(10, float64(digit)) matrix = make([]float64, 0, 200) times = make([]int64, 0, 200) err = svc.fetchHistoryKlineSeries(ctx, sr, func(k *types.Kline) (err error) { @@ -224,11 +230,11 @@ func (svc *TradingService) IndicatorSeries(ctx context.Context, indicatorName st err = fmt.Errorf("kline not series: %s(%s), interval=%s, lastTs=%d", sr.InstId, sr.Exchange, interval, lastTs) return } - if kSeries.Length() < requiredSeries { + if kSeries.Length() < candlePeriods { return } - vector := indicator.Calculate(indicatorContext, int16(window)) - matrix = append(matrix, vector) + vector := indicator.Calculate(indicatorContext) + matrix = append(matrix, math.Round(vector*pow)/pow) times = append(times, indicatorContext.Get(0).Ts) return }) diff --git a/pkg/indicator/atr.go b/pkg/indicator/atr.go index ad10318..7297f16 100644 --- a/pkg/indicator/atr.go +++ b/pkg/indicator/atr.go @@ -11,16 +11,22 @@ type ATR struct { } // indicator interface -func (c *ATR) Name() string { - return "atr" +func (c *ATR) Meta() IndicatorMeta { + return IndicatorMeta{ + Name: "atr", + Input: []types.InputArg{ + {Name: "window", Type: types.InputTypeUInt, Desc: "窗口大小"}, + }, + } } -func (c *ATR) RequiredSeries(window int16, in types.Input) int16 { - return window + 1 +func (c *ATR) CandlePeriods(ctx IIndicatorContext) int16 { + return ctx.Input().Int16("window") + 1 } // Calculate 计算单根k线rsi指标 -func (c *ATR) Calculate(ctx IIndicatorContext, window int16) (vector float64) { +func (c *ATR) Calculate(ctx IIndicatorContext) (vector float64) { + window := ctx.Input().Int16("window") klineSeries := ctx.Series(0, int16(window)+1) highs := klineSeries.High() lows := klineSeries.Low() diff --git a/pkg/indicator/ema.go b/pkg/indicator/ema.go index 36d7a9d..d937c27 100644 --- a/pkg/indicator/ema.go +++ b/pkg/indicator/ema.go @@ -8,28 +8,33 @@ import ( type EMA struct { } -func (c *EMA) Name() string { - return "ema" +func (c *EMA) Meta() IndicatorMeta { + return IndicatorMeta{ + Name: "ema", + Input: []types.InputArg{ + {Name: "window", Type: types.InputTypeUInt, Desc: "窗口大小"}, + }, + } } -func (c *EMA) RequiredSeries(window int16, in types.Input) int16 { - return window + 1 +func (c *EMA) CandlePeriods(ctx IIndicatorContext) int16 { + return ctx.Input().Int16("window") + 1 } // Calculate 计算单根k线sma指标 -func (c *EMA) Calculate(ctx IIndicatorContext, window int16) (vector float64) { - alpha := 2.0 / float64(window+1) - +func (c *EMA) Calculate(ctx IIndicatorContext) (vector float64) { + window := ctx.Input().Int16("window") + prevEma, ok := ctx.State().Get("ema", 1) + if !ok { + // 初始值用 sma 替代 + prevEma = ctx.Series(1, window).Close().Avg() + } + + multiplier := 2.0 / float64(window+1) close := ctx.Get(0).CloseF64() - prevSMA := ctx.Series(1, window).Close().Avg() - - vector = ((close - prevSMA) * alpha) + prevSMA - - // ctx.GetSelf(0) // 自己计算的上一个值 + vector = multiplier*close + (1-multiplier)*prevEma + // same: vector = ((close - prevEma) * multiplier) + prevEma - // 计算eam - // closeSeries := ctx.Series(0, window).Close().Reverse() - // ema := talib.Ema(closeSeries, int(window)) - // vector = ema[len(ema)-1] + ctx.State().Set("ema", vector) return } diff --git a/pkg/indicator/indicator.go b/pkg/indicator/indicator.go index 1711fb9..28f3e36 100644 --- a/pkg/indicator/indicator.go +++ b/pkg/indicator/indicator.go @@ -6,35 +6,49 @@ import ( ) const ( - MaxWindow = 128 + MaxWindow = 256 ) -// IIndicator 指标基础计算接口 -type IIndicator interface { - Name() string - Calculate(kSeries IIndicatorContext) (vector float64) +type IndicatorMeta struct { + Name string `json:"name"` // 指标名称 + Desc string `json:"desc"` // 指标描述 + Input []types.InputArg `json:"input"` // 输入参数 } -// IIndicator 窗口指标基础计算接口 -type IWindowIndicator interface { - // Name 指标名称 - Name() string - // RequiredSeries 计算窗口大小的指标值需要的K线数量 - RequiredSeries(window int16, in types.Input) int16 +// IIndicator 指标基础计算接口 +type IIndicator interface { + // Meta 指标元信息 + Meta() IndicatorMeta + // CandlePeriods 计算窗口大小的指标值需要的K线数量 + CandlePeriods(ctx IIndicatorContext) int16 // Calculate 计算窗口大小的指标值 - Calculate(ctx IIndicatorContext, window int16) (vector float64) + Calculate(ctx IIndicatorContext) (vector float64) } // IIndicatorContext k线序列, trading服务提供 type IIndicatorContext interface { - // Input 获取输入参数 - Input() types.Input Get(offset int16) (kline types.Kline) Series(offset, count int16) (klines series.Klines) + // Input 获取输入参数 + Input() types.Input + // State 存储指标运行中状态 + State() IIndicatorState + // IndicatorW 获取其他指标 + Indicator(name string, args ...any) (series IIndicatorSeries) } // IIndicatorSeries 指标序列, 供策略读取, trading服务提供 type IIndicatorSeries interface { + CandlePeriods() int16 Get(offset int16) (vector float64) Series(offset, count int16) (matrix series.Floats) } + +type IIndicatorState interface { + // Set 存储指标当前状态 + Set(k string, v float64) + // Get 获取指标之前存储的状态 offset >= 1 + Get(k string, offset int16) (v float64, ok bool) + // Series 获取指标之前存储的状态序列 offset >= 1, count >= 1 + Series(k string, offset, count int16) (v series.Floats, ok bool) +} diff --git a/pkg/indicator/indicator_registry.go b/pkg/indicator/indicator_registry.go index e55fbb8..22abe0e 100644 --- a/pkg/indicator/indicator_registry.go +++ b/pkg/indicator/indicator_registry.go @@ -7,12 +7,12 @@ import ( // 指标注册器 type IndicatorRegistry struct { - indicatorsW *collect.SyncMap[string, IWindowIndicator] // 注册窗口指标 + indicators *collect.SyncMap[string, IIndicator] // 注册窗口指标 } func NewIndicatorRegistry() *IndicatorRegistry { return &IndicatorRegistry{ - indicatorsW: collect.NewSyncMap[string, IWindowIndicator](), + indicators: collect.NewSyncMap[string, IIndicator](), } } @@ -23,13 +23,17 @@ func (r *IndicatorRegistry) Init() (err error) { r.MustRegistIndicatorW(&ATR{}) r.MustRegistIndicatorW(&EMA{}) r.MustRegistIndicatorW(&MACD{}) + r.MustRegistIndicatorW(&MacdDEA{}) + r.MustRegistIndicatorW(&MacdHist{}) + r.MustRegistIndicatorW(&OBV{}) + r.MustRegistIndicatorW(&WOBV{}) return } // RegistIndicatorW -func (r *IndicatorRegistry) RegistIndicatorW(ind IWindowIndicator) (err error) { - indName := ind.Name() - _, loaded := r.indicatorsW.LoadOrStore(indName, ind) +func (r *IndicatorRegistry) RegistIndicatorW(ind IIndicator) (err error) { + indName := ind.Meta().Name + _, loaded := r.indicators.LoadOrStore(indName, ind) if loaded { err = fmt.Errorf("window indicator name %s already duplicated", indName) return @@ -37,13 +41,13 @@ func (r *IndicatorRegistry) RegistIndicatorW(ind IWindowIndicator) (err error) { return } -func (r *IndicatorRegistry) MustRegistIndicatorW(ind IWindowIndicator) { +func (r *IndicatorRegistry) MustRegistIndicatorW(ind IIndicator) { 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) +// Indicator +func (r *IndicatorRegistry) Indicator(name string) (indW IIndicator, ok bool) { + return r.indicators.Load(name) } diff --git a/pkg/indicator/macd.go b/pkg/indicator/macd.go index 40f117f..21f0a05 100644 --- a/pkg/indicator/macd.go +++ b/pkg/indicator/macd.go @@ -2,33 +2,111 @@ package indicator import ( "sig-pub/pkg/types" - - "github.com/markcheno/go-talib" ) -// todo macdSignal(信号线) macdHist(柱状图) +// MACD 拆分成: MACD线, MacdDEA(信号线), MacdHist(柱状图) +// 计算 MACD 线 (DIF): 反映短期趋势与长期趋势的“收敛/散度” +// MACD: https://www.investopedia.com/terms/m/macd.asp type MACD struct { } -func (c *MACD) Name() string { - return "macd" +// indicator interface +func (c *MACD) Meta() IndicatorMeta { + return IndicatorMeta{ + Name: "macd", + Input: []types.InputArg{ + {Name: "fast", Type: types.InputTypeUInt, Desc: "快线周期"}, + {Name: "slow", Type: types.InputTypeUInt, Desc: "慢线周期"}, + }, + } } -func (c *MACD) RequiredSeries(window int16, in types.Input) int16 { - return window +func (c *MACD) CandlePeriods(ctx IIndicatorContext) int16 { + return max( + ctx.Indicator("ema", ctx.Input().Int16("fast")).CandlePeriods(), + ctx.Indicator("ema", ctx.Input().Int16("slow")).CandlePeriods(), + ) } // Calculate 计算单根k线sma指标 -func (c *MACD) Calculate(ctx IIndicatorContext, window int16) (vector float64) { - fast := ctx.Input().Int("fast") - slow := ctx.Input().Int("slow") +func (c *MACD) Calculate(ctx IIndicatorContext) (vector float64) { + fast := ctx.Input().Int16("fast") // 12 + slow := ctx.Input().Int16("slow") // 26 + + // macd计算从第max(fast, slow)期开始稳定 + fastEma := ctx.Indicator("ema", fast).Get(0) + slowEma := ctx.Indicator("ema", slow).Get(0) + macd := fastEma - slowEma + vector = macd + return +} + +// MacdDEA macd信号线计算 +type MacdDEA struct { +} + +func (c *MacdDEA) Meta() IndicatorMeta { + return IndicatorMeta{ + Name: "macd_dea", + Input: []types.InputArg{ + {Name: "fast", Type: types.InputTypeUInt, Desc: "快线周期"}, + {Name: "slow", Type: types.InputTypeUInt, Desc: "慢线周期"}, + {Name: "singal", Type: types.InputTypeUInt, Desc: "信号线周期"}, + }, + } +} + +func (c *MacdDEA) CandlePeriods(ctx IIndicatorContext) int16 { + return max( + ctx.Input().Int16("singal")+1, + ctx.Indicator("macd", ctx.Input()).CandlePeriods(), + ) +} - // 计算eam - closeSeries := ctx.Series(0, window).Close().Reverse() +func (c *MacdDEA) Calculate(ctx IIndicatorContext) (vector float64) { + singal := ctx.Input().Int16("singal") // 9 - aa, bb, cc := talib.Macd(closeSeries, fast, slow, int(window)) - _, _, _ = aa, bb, cc + deaPrev, ok := ctx.State().Get("macd_dea", 1) + if !ok { + // 初始值前9期的 MACD SMA + macdPrevs := ctx.Indicator("macd", ctx.Input()).Series(1, singal) + deaPrev = macdPrevs.Avg() + } + macd := ctx.Indicator("macd", ctx.Input()).Get(0) + // 计算DEA + beta := 2 / float64(singal+1) + dea := beta*macd + (1-beta)*deaPrev + ctx.State().Set("macd_dea", dea) + + vector = dea + return +} + +// MacdSingal macd柱状图计算 +type MacdHist struct { +} + +func (c *MacdHist) Meta() IndicatorMeta { + return IndicatorMeta{ + Name: "macd_hist", + Input: []types.InputArg{ + {Name: "fast", Type: types.InputTypeUInt, Desc: "快线周期"}, + {Name: "slow", Type: types.InputTypeUInt, Desc: "慢线周期"}, + {Name: "singal", Type: types.InputTypeUInt, Desc: "信号线周期"}, + }, + } +} + +func (c *MacdHist) CandlePeriods(ctx IIndicatorContext) int16 { + return max( + ctx.Indicator("macd", ctx.Input()).CandlePeriods(), + ctx.Indicator("macd_dea", ctx.Input()).CandlePeriods(), + ) +} - vector = 0 +func (c *MacdHist) Calculate(ctx IIndicatorContext) (vector float64) { + macd := ctx.Indicator("macd", ctx.Input()).Get(0) + macd_dea := ctx.Indicator("macd_dea", ctx.Input()).Get(0) + vector = macd - macd_dea return } diff --git a/pkg/indicator/obv.go b/pkg/indicator/obv.go new file mode 100644 index 0000000..d9c5ff7 --- /dev/null +++ b/pkg/indicator/obv.go @@ -0,0 +1,41 @@ +package indicator + +import "sig-pub/pkg/types" + +// OBV 成交量平衡指标 +// 1. 初始状态:OBV_0 = 0。 +// 2. 若 Close_t > Close_{t-1}:OBV_t = OBV_{t-1} + Volume_t。 +// 3. 若 Close_t < Close_{t-1}:OBV_t = OBV_{t-1} - Volume_t。 +// 4. 平盘:OBV_t = OBV_{t-1}。 +// 状态:前一 OBV 值。用于判断资金流入/流出 +type OBV struct { +} + +func (c *OBV) Meta() IndicatorMeta { + return IndicatorMeta{ + Name: "obv", + Input: []types.InputArg{}, // todo 无参指标tsdb存储 + } +} + +func (c *OBV) CandlePeriods(ctx IIndicatorContext) int16 { + return 2 +} + +func (c *OBV) Calculate(ctx IIndicatorContext) (vector float64) { + obvPrev, ok := ctx.State().Get("obv", 1) + if !ok { + obvPrev = 0 + } + k := ctx.Get(0) + cmp := k.Close.Cmp(ctx.Get(1).Close) + if cmp > 0 { + vector = obvPrev + k.VolF64() + } else if cmp < 0 { + vector = obvPrev - k.VolF64() + } else { + vector = obvPrev + } + ctx.State().Set("obv", vector) + return +} diff --git a/pkg/indicator/rsi.go b/pkg/indicator/rsi.go index 086892c..5c33c7c 100644 --- a/pkg/indicator/rsi.go +++ b/pkg/indicator/rsi.go @@ -11,16 +11,23 @@ import ( type RSI struct { } -func (c *RSI) Name() string { - return "rsi" +// indicator interface +func (c *RSI) Meta() IndicatorMeta { + return IndicatorMeta{ + Name: "rsi", + Input: []types.InputArg{ + {Name: "window", Type: types.InputTypeUInt, Desc: "窗口大小"}, + }, + } } -func (c *RSI) RequiredSeries(window int16, in types.Input) int16 { - return window +func (c *RSI) CandlePeriods(ctx IIndicatorContext) int16 { + return ctx.Input().Int16("window") } // Calculate 计算单根k线rsi指标 -func (c *RSI) Calculate(ctx IIndicatorContext, window int16) (vector float64) { +func (c *RSI) Calculate(ctx IIndicatorContext) (vector float64) { + window := ctx.Input().Int16("window") // 读k线, 计算 klineSeries := ctx.Series(0, int16(window)) closeSeries := klineSeries.Close() diff --git a/pkg/indicator/sam.go b/pkg/indicator/sam.go index 510d95f..083c139 100644 --- a/pkg/indicator/sam.go +++ b/pkg/indicator/sam.go @@ -12,16 +12,22 @@ type SMA struct { } // indicator interface -func (c *SMA) Name() string { - return "sma" +func (c *SMA) Meta() IndicatorMeta { + return IndicatorMeta{ + Name: "sma", + Input: []types.InputArg{ + {Name: "window", Type: types.InputTypeUInt, Desc: "窗口大小"}, + }, + } } -func (c *SMA) RequiredSeries(window int16, in types.Input) int16 { - return window +func (c *SMA) CandlePeriods(ctx IIndicatorContext) int16 { + return ctx.Input().Int16("window") } // Calculate 计算单根k线sma指标 -func (c *SMA) Calculate(ctx IIndicatorContext, window int16) (vector float64) { +func (c *SMA) Calculate(ctx IIndicatorContext) (vector float64) { + window := ctx.Input().Int16("window") closeSeries := ctx.Series(0, window).Close() sma := talib.Sma(closeSeries, int(window)) _ = sma[len(sma)-1] diff --git a/pkg/indicator/wobv.go b/pkg/indicator/wobv.go new file mode 100644 index 0000000..91652c4 --- /dev/null +++ b/pkg/indicator/wobv.go @@ -0,0 +1,37 @@ +package indicator + +import "sig-pub/pkg/types" + +// WOBV 波动加权 OBV +// 1. 状态:WOBV_{t-1}。 +// 2. 更新:WOBV_t = WOBV_{t-1} + [ (Close - Open) / (High - Low) × Volume_t ]。 +// https://www.95sca.cn/archives/76688 +// WOBV小策略: https://zhuanlan.zhihu.com/p/422341694 +type WOBV struct { +} + +func (c *WOBV) Meta() IndicatorMeta { + return IndicatorMeta{ + Name: "wobv", + Input: []types.InputArg{}, // todo 无参指标tsdb存储 + } +} + +func (c *WOBV) CandlePeriods(ctx IIndicatorContext) int16 { + return 2 +} + +func (c *WOBV) Calculate(ctx IIndicatorContext) (vector float64) { + wobvPrev, ok := ctx.State().Get("wobv", 1) + if !ok { + wobvPrev = 0 + } + + k := ctx.Get(0) + wf := (k.CloseF64() - k.OpenF64()) / (k.HighF64() - k.LowF64()) + wobv := wobvPrev + wf*k.VolF64() + ctx.State().Set("wobv", wobv) + + vector = wobv + return +} diff --git a/pkg/strategy/cross_star.go b/pkg/strategy/cross_star.go index 0d1bc17..b7baf9f 100644 --- a/pkg/strategy/cross_star.go +++ b/pkg/strategy/cross_star.go @@ -21,9 +21,9 @@ func (s *CrossStar) Meta() StrategyMeta { return StrategyMeta{ Name: "CrossStar", Desc: "十字星策略", - Args: []Param{ - {Name: "rate", Type: ParamTypeUFloat, Desc: "上线影线与基线比例"}, - {Name: "rate2", Type: ParamTypeUFloat, Desc: "上线影线之间比例"}, + Input: []types.InputArg{ + {Name: "rate", Type: types.InputTypeUFloat, Desc: "上线影线与基线比例"}, + {Name: "rate2", Type: types.InputTypeUFloat, Desc: "上线影线之间比例"}, }, } } diff --git a/pkg/strategy/gold_x.go b/pkg/strategy/gold_x.go index cd76263..2680151 100644 --- a/pkg/strategy/gold_x.go +++ b/pkg/strategy/gold_x.go @@ -19,9 +19,9 @@ func (s *GoldX) Meta() StrategyMeta { return StrategyMeta{ Name: "GoldX", Desc: "金叉策略", - Args: []Param{ - {Name: "short", Type: ParamTypeUInt, Desc: "短周期"}, - {Name: "long", Type: ParamTypeUInt, Desc: "长周期"}, + Input: []types.InputArg{ + {Name: "short", Type: types.InputTypeUInt, Desc: "短周期"}, + {Name: "long", Type: types.InputTypeUInt, Desc: "长周期"}, }, } } @@ -41,8 +41,8 @@ func (s *GoldX) RequiredSeries(input types.Input) int16 { } func (s *GoldX) Update(ctx ISingleSigStrategyContext) (side types.Side) { - sma14 := ctx.IndicatorW("sma", s.short) - sma28 := ctx.IndicatorW("sma", s.long) + sma14 := ctx.Indicator("sma", s.short) + sma28 := ctx.Indicator("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 6b8495e..19c2806 100644 --- a/pkg/strategy/sig_strategy.go +++ b/pkg/strategy/sig_strategy.go @@ -14,9 +14,9 @@ type ISigStrategy interface { } type StrategyMeta struct { - Name string `json:"name"` - Desc string `json:"desc"` - Args []Param `json:"args"` // 参数定义 + Name string `json:"name"` + Desc string `json:"desc"` + Input []types.InputArg `json:"input"` // 参数定义 } // ISingleSigStrategy 单周期单交易所策略 @@ -34,8 +34,8 @@ type ISingleSigStrategyContext interface { Get(offset int16) types.Kline // Series [offset...end] Series(offset, count int16) (klines series.Klines) - // IndicatorW 获取窗口类型指标 - IndicatorW(name string, window int16, args ...any) indicator.IIndicatorSeries + // Indicator 获取窗口类型指标 + Indicator(name string, args ...any) indicator.IIndicatorSeries } // 多周期k线策略接口 @@ -54,5 +54,5 @@ type IIntervalSigStrategyContext interface { // Series [offset...end] Series(interval types.Interval, offset, count int16) (klines series.Klines) // 获取窗口类型指标 - IndicatorW(interval types.Interval, name string, window int16, args ...any) indicator.IIndicatorSeries + Indicator(interval types.Interval, name string, args ...any) indicator.IIndicatorSeries } diff --git a/pkg/strategy/sig_strategy_params.go b/pkg/strategy/sig_strategy_input_adjust.go similarity index 61% rename from pkg/strategy/sig_strategy_params.go rename to pkg/strategy/sig_strategy_input_adjust.go index ca35abc..5806fc7 100644 --- a/pkg/strategy/sig_strategy_params.go +++ b/pkg/strategy/sig_strategy_input_adjust.go @@ -6,66 +6,6 @@ import ( "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 diff --git a/pkg/strategy/super_trend.go b/pkg/strategy/super_trend.go index d8aa83d..fe22428 100644 --- a/pkg/strategy/super_trend.go +++ b/pkg/strategy/super_trend.go @@ -28,10 +28,10 @@ func (s *SupertrendBOSWaves) Meta() StrategyMeta { return StrategyMeta{ Name: "SupertrendBOSWaves", Desc: "曲线半径超级趋势 [BOSWaves] https://www.tradingview.com/script/v0Fr7PAb-Curved-Radius-Supertrend-BOSWaves/", - Args: []Param{ - {Name: "atrLength", Type: ParamTypeUInt, Desc: "atr指标长度,14"}, - {Name: "atrMult", Type: ParamTypeUFloat, Desc: "atr倍数,2"}, - {Name: "radiusStrength", Type: ParamTypeUFloat, Desc: ` + Input: []types.InputArg{ + {Name: "atrLength", Type: types.InputTypeUInt, Desc: "atr指标长度,14"}, + {Name: "atrMult", Type: types.InputTypeUFloat, Desc: "atr倍数,2"}, + {Name: "radiusStrength", Type: types.InputTypeUFloat, Desc: ` Controls curve acceleration strength.\n\n" + "Recommended values by timeframe:\n" + "• 1-5min (Scalping): 0.08-0.12\n" + @@ -43,7 +43,7 @@ func (s *SupertrendBOSWaves) Meta() StrategyMeta { "Lower = Tighter curves (responsive)\n" + "Higher = Wider curves (smoother) `}, - {Name: "smoothness", Type: ParamTypeUInt, Desc: "Smoothing applied to curved band. Higher = smoother curves, less noise."}, + {Name: "smoothness", Type: types.InputTypeUInt, Desc: "Smoothing applied to curved band. Higher = smoother curves, less noise."}, }, } } @@ -64,7 +64,7 @@ func (s *SupertrendBOSWaves) Update(ctx ISingleSigStrategyContext) (side types.S k0 := ctx.Get(0) high, low, close := k0.HighF64(), k0.LowF64(), k0.CloseF64() - atr := ctx.IndicatorW("atr", s.atrLength).Get(0) + atr := ctx.Indicator("atr", s.atrLength).Get(0) src := (high + low) / 2 // src := k0.HL2() diff --git a/pkg/types/input.go b/pkg/types/input.go index 8ef3215..91c8296 100644 --- a/pkg/types/input.go +++ b/pkg/types/input.go @@ -50,7 +50,9 @@ func (in Input) get(k string, t string) (r any) { func (in Input) Float(k string) (v float64) { if r, ok := in.getCache(k); ok { - return r.(float64) + if v, ok = r.(float64); ok { + return + } } v, err := cast.ToFloat64E(in.get(k, "float")) if err != nil { @@ -62,7 +64,9 @@ func (in Input) Float(k string) (v float64) { func (in Input) Int(k string) (v int) { if r, ok := in.getCache(k); ok { - return r.(int) + if v, ok = r.(int); ok { + return + } } v, err := cast.ToIntE(in.get(k, "int")) if err != nil { @@ -74,7 +78,9 @@ func (in Input) Int(k string) (v int) { func (in Input) Int16(k string) (v int16) { if r, ok := in.getCache(k); ok { - return r.(int16) + if v, ok = r.(int16); ok { + return + } } v, err := cast.ToInt16E(in.get(k, "int16")) if err != nil { @@ -83,3 +89,44 @@ func (in Input) Int16(k string) (v int16) { in.setCache(k, v) return } + +func (in Input) String(k string) (v string) { + if r, ok := in.getCache(k); ok { + if v, ok = r.(string); ok { + return + } + } + v, err := cast.ToStringE(in.get(k, "string")) + if err != nil { + panic(fmt.Errorf("input string parse error: %s", k)) + } + in.setCache(k, v) + return +} + +// 参数类型 +type InputType int8 + +const ( + _ InputType = iota + InputTypeBool + InputTypeFloat + InputTypeUFloat + InputTypeInt + InputTypeUInt + InputTypeString + InputTypeSelect // 单选 + InputTypeCheckBox // 多选 +) + +type InputArg struct { + Name string `json:"name"` + Desc string `json:"desc"` + Type InputType `json:"type"` // 参数类型 + Options []InputOption `json:"options"` // 单选/多选选项列表 +} + +type InputOption struct { + Name string `json:"name"` + Desc string `json:"desc"` +} diff --git a/pkg/types/ring_series.go b/pkg/types/ring_series.go new file mode 100644 index 0000000..f61d37d --- /dev/null +++ b/pkg/types/ring_series.go @@ -0,0 +1,69 @@ +package types + +import "fmt" + +// RingSeries 环形数组, 后入先出 +type RingSeries[T any] struct { + values []T + capacity int // 总长度 + length int // 当前长度 + head int // 下一个读取位置 + tail int // 下一个丢弃位置 + full bool // 环形数组是否已满 +} + +func NewRingSeries[T any](capacity, init int) *RingSeries[T] { + if capacity <= 0 { + panic(fmt.Errorf("ring series capacity must > 0")) + } + return &RingSeries[T]{ + capacity: capacity, + values: make([]T, 0, init), + head: -1, + } +} + +func (r *RingSeries[T]) Length() int { + return r.length +} + +func (r *RingSeries[T]) Push(v T) (ok bool) { + if !r.full { + r.values = append(r.values, v) + r.head++ + r.length++ + r.full = r.length == r.capacity + return + } + r.values[r.tail] = v + r.head = r.tail + r.tail = (r.tail + 1) % r.capacity + return +} + +// Get 0当前, 1前一个 +func (r *RingSeries[T]) Get(offset int) (v T, ok bool) { + if offset < 0 || offset >= r.length || r.length == 0 { + return + } + i := r.head - offset + if i < 0 { + i = i + r.capacity + } + return r.values[i], true +} + +func (r *RingSeries[T]) Series(offset, count int) (v []T, ok bool) { + if offset < 0 || offset >= r.length || count <= 0 || count > r.length { + return + } + v = make([]T, count) + for i := range count { + v[i], ok = r.Get(offset) + if !ok { + return + } + offset++ + } + return +} diff --git a/pkg/types/ring_series_test.go b/pkg/types/ring_series_test.go new file mode 100644 index 0000000..4c328b7 --- /dev/null +++ b/pkg/types/ring_series_test.go @@ -0,0 +1,30 @@ +package types + +import ( + "fmt" + "testing" +) + +func TestRingSeries(t *testing.T) { + rs1 := NewRingSeries[int](1, 1) + rs1.Push(1) + r1, ok := rs1.Get(0) + if !ok { + t.Error(ok) + return + } + if r1 != 1 { + t.Error(r1) + return + } + + rs2 := NewRingSeries[int](10, 3) + for i := range 11 { + rs2.Push(i) + } + for i := range 10 { + fmt.Println(rs2.Get(i)) + } + fmt.Println("--------------------") + fmt.Println(rs2.Series(0, 1)) +}