From 63eff0d641163e75138f225a0ba109f804abd7c0 Mon Sep 17 00:00:00 2001 From: strange Date: Thu, 12 Feb 2026 18:21:52 +0800 Subject: [PATCH] summary indicator VRVP --- api/trading.proto | 16 +- internal/gateway/fast_gateway.go | 28 ++- internal/trading/sig/indicator_context.go | 31 ++- internal/trading/sig/indicator_series.go | 2 +- internal/trading/sig/indicator_summary.go | 58 ++++- internal/trading/sig/strategy_context.go | 57 +++++ internal/trading/trading_grpc_server.go | 19 ++ internal/trading/trading_service.go | 75 +++++++ pkg/indicator/indicator.go | 13 +- pkg/indicator/indicator_registry.go | 41 +++- pkg/indicator/vrvp.go | 256 ++++++++++++++++++++-- pkg/strategy/bollgrid.go | 12 +- pkg/strategy/sig_strategy.go | 2 + pkg/types/indicator.go | 16 ++ 14 files changed, 564 insertions(+), 62 deletions(-) create mode 100644 pkg/types/indicator.go diff --git a/api/trading.proto b/api/trading.proto index aa1a673..c0602e9 100644 --- a/api/trading.proto +++ b/api/trading.proto @@ -19,7 +19,10 @@ service TradingService { // 获取指标序列 rpc IndicatorSeries(ReqIndicatorSeries) returns (RspIndicatorSeries); - // 获取指标序列 + // 获取summary指标结果 + rpc IndicatorSummary(ReqIndicatorSummary) returns (RspIndicatorSummary); + + // 获取策略序列 rpc StrategySeries(ReqStrategySeries) returns (RspStrategySeries); // 交易计划回测 @@ -68,6 +71,17 @@ message IndicatorState { repeated double matrix = 2; } +message ReqIndicatorSummary { + string indicator = 1; + int32 digit = 2; // 结果小数位数 + SeriesRange series = 9; + google.protobuf.Struct input = 10; // 指标参数 +} +message RspIndicatorSummary { + string indicator = 1; + bytes summary = 2; // json格式结果 +} + message ReqStrategySeries { SeriesRange series = 1; string sig_strategy = 2; diff --git a/internal/gateway/fast_gateway.go b/internal/gateway/fast_gateway.go index eb9a7d2..1406d66 100644 --- a/internal/gateway/fast_gateway.go +++ b/internal/gateway/fast_gateway.go @@ -5,6 +5,7 @@ import ( "encoding/json" "net/http" "net/url" + "sig-pub/api/pb" "sig-pub/pkg/grpc/generic" "sig-pub/pkg/grpc/session" "sig-pub/pkg/resp" @@ -159,16 +160,25 @@ func (s *FastGatewayServer) reverseProxyGrpcGenericCall(c *fasthttp.RequestCtx) return } - // encode response - bytes, err := protojson.MarshalOptions{ - UseProtoNames: false, // false:lowerCamelCase, true:snake_case - EmitUnpopulated: false, // 是否包含默认值 - }.Marshal(rsp) - if err != nil { - responseJSON(c, http.StatusInternalServerError, resp.Error(err.Error())) - return + var r any + switch rsp := rsp.(type) { + case *pb.RspIndicatorSummary: + r = resp.H{ + "indicator": rsp.Indicator, + "summary": json.RawMessage(rsp.Summary), + } + default: + // marshal json response + bytes, err := protojson.MarshalOptions{ + UseProtoNames: false, // false:lowerCamelCase, true:snake_case + EmitUnpopulated: false, // 是否包含默认值 + }.Marshal(rsp) + if err != nil { + responseJSON(c, http.StatusInternalServerError, resp.Error(err.Error())) + return + } + r = json.RawMessage(bytes) } - r := json.RawMessage(bytes) responseJSON(c, http.StatusOK, resp.Success(r)) } diff --git a/internal/trading/sig/indicator_context.go b/internal/trading/sig/indicator_context.go index 039048e..c4270e3 100644 --- a/internal/trading/sig/indicator_context.go +++ b/internal/trading/sig/indicator_context.go @@ -28,14 +28,15 @@ type IOffsetIndicatorContext interface { // IndicatorContext 指标上下文, 提供k线序列给指标计算使用 type IndicatorContext struct { IOffsetIndicatorContext - indicator indicator.IIndicatorMetable - indicatorKey string - indicatorsReg *indicator.IndicatorRegistry - input types.Input - indicatorStates map[string]*IndicatorState // {macd{window:0,fast:9,slow:21,single:10}: state} 初始化时与KlineSeries周期同步 - kSeries *types.KlineSeries - offset int16 - indicatorTrace []string // 指标调用链避免指标循环引用 + indicator indicator.IIndicatorMetable + indicatorKey string + indicatorsReg *indicator.IndicatorRegistry + input types.Input + indicatorStates map[string]*IndicatorState // {macd{window:0,fast:9,slow:21,single:10}: state} 初始化时与KlineSeries周期同步 + indicatorSummary map[string]*IndicatorSummary // {vrvp{buckets:48}: summary} + kSeries *types.KlineSeries + offset int16 + indicatorTrace []string // 指标调用链避免指标循环引用 } func NewIndicatorContext(indicator indicator.IIndicatorMetable, input types.Input, indicatorStates IndicatorStates, kSeries *types.KlineSeries, indicatorsReg *indicator.IndicatorRegistry) *IndicatorContext { @@ -152,13 +153,23 @@ func (c *IndicatorContext) Indicator(name string, args ...any) (series indicator } func (c *IndicatorContext) SummaryIndicator(name string, args ...any) (summary indicator.IIndicatorSummary) { + summaryIndicatorNewer, ok := c.indicatorsReg.SummaryIndicatorNewer(name) + if !ok { + panic(fmt.Errorf("summary indicator %s not exists", name)) + } + + // 避免循环依赖 + if slices.Contains(c.indicatorTrace, name) { + panic(fmt.Errorf("summary indicator %s recursive call", name)) + } + input := matchIndicatorArgs(args...) - indicatorContext := NewIndicatorContext(nil, input, c.indicatorStates, c.kSeries, c.indicatorsReg) + indicatorContext := NewIndicatorContext(summaryIndicatorNewer(), input, c.indicatorStates, c.kSeries, c.indicatorsReg) indicatorContext.offset = c.offset indicatorContext.indicatorTrace = append(c.indicatorTrace, name) - return NewIndicatorSummary(nil, indicatorContext) + return NewIndicatorSummary(summaryIndicatorNewer, indicatorContext) } func matchIndicatorArgs(args ...any) (input types.Input) { diff --git a/internal/trading/sig/indicator_series.go b/internal/trading/sig/indicator_series.go index 27a9135..0ea086d 100644 --- a/internal/trading/sig/indicator_series.go +++ b/internal/trading/sig/indicator_series.go @@ -34,7 +34,7 @@ func (s *WindowIndicatorSeries) Get(offset int16) (vector float64) { return } -// Series 返回指标值序列降序 +// Series 返回指标值序列[时间降序] func (s *WindowIndicatorSeries) Series(offset, count int16) (matrix series.Floats) { // 设置当前相对offset s.indicatorContext.AddOffset(offset) diff --git a/internal/trading/sig/indicator_summary.go b/internal/trading/sig/indicator_summary.go index 61d3ef7..5338602 100644 --- a/internal/trading/sig/indicator_summary.go +++ b/internal/trading/sig/indicator_summary.go @@ -1,21 +1,65 @@ package sig -import "sig-pub/pkg/indicator" +import ( + "fmt" + "sig-pub/pkg/indicator" +) type IndicatorSummary struct { indicator.IIndicatorSummary - indicatorContext IOffsetIndicatorContext - summaryIndicator indicator.ISummaryIndicator + indicatorContext IOffsetIndicatorContext + summaryIndicatorNewer func() indicator.ISummaryIndicator + cachedSummaryIndicators map[string]indicator.ISummaryIndicator } -func NewIndicatorSummary(summaryIndicator indicator.ISummaryIndicator, indicatorContext IOffsetIndicatorContext) *IndicatorSummary { +func NewIndicatorSummary(summaryIndicatorNewer func() indicator.ISummaryIndicator, indicatorContext IOffsetIndicatorContext) *IndicatorSummary { return &IndicatorSummary{ - summaryIndicator: summaryIndicator, - indicatorContext: indicatorContext, + summaryIndicatorNewer: summaryIndicatorNewer, + indicatorContext: indicatorContext, + cachedSummaryIndicators: make(map[string]indicator.ISummaryIndicator), } } -func (s *IndicatorSummary) Summary(offset, count int16) (summary any, ok bool) { +// Summary 0 10, 0 20, 1 30 +func (s *IndicatorSummary) Summary(offset, count int16) (summary any, rok bool) { + key := fmt.Sprintf("%d,%d", offset, count) + summaryIndicator, ok := s.cachedSummaryIndicators[key] + if ok { + s.indicatorContext.AddOffset(offset + count + 1) + eliminater, ok := summaryIndicator.(indicator.ISummaryIndicatorEliminater) + if ok { + eliminater.Eliminate(s.indicatorContext) + } + s.indicatorContext.AddOffset(-(offset + count + 1)) + + s.indicatorContext.AddOffset(offset) + summaryIndicator.Accumulate(s.indicatorContext) + summary, ok = summaryIndicator.Summary(s.indicatorContext) + s.indicatorContext.AddOffset(-offset) + return + } + + // 初始化, 计算全量数据 + summaryIndicator = s.summaryIndicatorNewer() + if err := summaryIndicator.Init(s.indicatorContext.Input()); err != nil { + panic(err) + } + s.cachedSummaryIndicators[key] = summaryIndicator + + // 设置当前相对offset + offset = offset + count + s.indicatorContext.AddOffset(offset) + for range count { + summaryIndicator.Accumulate(s.indicatorContext) + + offset-- + s.indicatorContext.AddOffset(-1) + } + + // 获取结果 + summary, rok = summaryIndicator.Summary(s.indicatorContext) + // 计算结束后还原offset + s.indicatorContext.AddOffset(-offset) return } diff --git a/internal/trading/sig/strategy_context.go b/internal/trading/sig/strategy_context.go index 308ddc8..d5bd1e9 100644 --- a/internal/trading/sig/strategy_context.go +++ b/internal/trading/sig/strategy_context.go @@ -15,6 +15,7 @@ type StrategyContext struct { kSeries *types.KlineSeries indicatorsReg *indicator.IndicatorRegistry indicatorContextStates IndicatorStates + summaryIndicators map[string]indicator.IIndicatorSummary } func NewStrategyContext(input types.Input, kSeries *types.KlineSeries, indicatorsReg *indicator.IndicatorRegistry) *StrategyContext { @@ -23,6 +24,7 @@ func NewStrategyContext(input types.Input, kSeries *types.KlineSeries, indicator kSeries: kSeries, indicatorsReg: indicatorsReg, indicatorContextStates: NewIndicatorStates(), + summaryIndicators: make(map[string]indicator.IIndicatorSummary), } } @@ -50,6 +52,21 @@ func (c *StrategyContext) Indicator(name string, args ...any) (s indicator.IIndi return NewWindowIndicatorSeries(indicator, indicatorContext) } +func (c *StrategyContext) SummaryIndicator(name string, args ...any) (summary indicator.IIndicatorSummary) { + indicatorNewer, ok := c.indicatorsReg.SummaryIndicatorNewer(name) + if !ok { + panic(fmt.Errorf("indicator %s not exists", name)) + } + input := matchIndicatorArgs(args...) + indicatorContext := NewIndicatorContext(indicatorNewer(), input, c.indicatorContextStates, c.kSeries, c.indicatorsReg) + indKey := indicatorContext.indicatorKey + if summary, ok = c.summaryIndicators[indKey]; !ok { + summary = NewIndicatorSummary(indicatorNewer, indicatorContext) + c.summaryIndicators[indKey] = summary + } + return +} + // IntervalStrategyContext 周期策略上下文 type IntervalStrategyContext struct { strategy.IIntervalSigStrategyContext @@ -113,6 +130,24 @@ func (c *IntervalStrategyContext) Indicator(interval types.Interval, name string return NewWindowIndicatorSeries(indicator, indicatorContext) } +func (c *IntervalStrategyContext) SummaryIndicator(interval types.Interval, name string, args ...any) (summary indicator.IIndicatorSummary) { + indicatorNewer, ok := c.indicatorsReg.SummaryIndicatorNewer(name) + if !ok { + panic(fmt.Errorf("indicator %s not exists", name)) + } + + input := matchIndicatorArgs(args...) + kSeries := c.getCandleSeries(interval) + // 状态传递 + state, ok := c.intervalIndicatorContextStates[interval] + if !ok { + state = NewIndicatorStates() + c.intervalIndicatorContextStates[interval] = state + } + indicatorContext := NewIndicatorContext(indicatorNewer(), input, state, kSeries, c.indicatorsReg) + return NewIndicatorSummary(indicatorNewer, indicatorContext) +} + // InstIntervalStrategyContext 多币种多周期策略上下文 type InstanceIntervalSigStrategyContext struct { strategy.IInstanceIntervalSigStrategyContext @@ -169,3 +204,25 @@ func (c *InstanceIntervalSigStrategyContext) Indicator(instId string, interval t indicatorContext := NewIndicatorContext(indicator, input, indicatorStates, kSeries, c.indicatorsReg) return NewWindowIndicatorSeries(indicator, indicatorContext) } + +// 获取Summary类型指标 +func (c *InstanceIntervalSigStrategyContext) SummaryIndicator(instId string, interval types.Interval, name string, args ...any) indicator.IIndicatorSummary { + summaryIndicatorNewer, ok := c.indicatorsReg.SummaryIndicatorNewer(name) + if !ok { + panic(fmt.Errorf("summary indicator %s not exists", name)) + } + + input := matchIndicatorArgs(args...) + kSeries := c.iiks.Get(instId, interval) + // 状态传递 + intervalIndicatorStates := c.instanceIntervalIndicatorStates.ComputeIfAbsent(instId, func(k string) (iss *types.IntervalState[IndicatorStates]) { + iss = types.NewIntervalState[IndicatorStates]() + for _, interval := range c.iiks.GetScopeIntervals() { + iss.Set(interval, NewIndicatorStates()) + } + return + }) + indicatorStates := intervalIndicatorStates.Get(interval) + indicatorContext := NewIndicatorContext(summaryIndicatorNewer(), input, indicatorStates, kSeries, c.indicatorsReg) + return NewIndicatorSummary(summaryIndicatorNewer, indicatorContext) +} diff --git a/internal/trading/trading_grpc_server.go b/internal/trading/trading_grpc_server.go index e06ac25..10f0c56 100644 --- a/internal/trading/trading_grpc_server.go +++ b/internal/trading/trading_grpc_server.go @@ -10,6 +10,7 @@ import ( "sig-pub/pkg/utils/times" "sig-pub/pkg/zlog" + "github.com/bytedance/sonic" "google.golang.org/protobuf/types/known/structpb" ) @@ -146,6 +147,24 @@ func (svr *TradingGrpcServer) IndicatorSeries(ctx context.Context, req *pb.ReqIn return } +func (svr *TradingGrpcServer) IndicatorSummary(ctx context.Context, req *pb.ReqIndicatorSummary) (rsp *pb.RspIndicatorSummary, err error) { + input := req.Input.AsMap() + summary, err := svr.tradingService.IndicatorSummary(ctx, req.Indicator, req.Digit, types.Input(input), req.Series) + if err != nil { + return + } + summaryJson, err := sonic.Marshal(summary) + if err != nil { + err = fmt.Errorf("marshal indicator summary error: %w", err) + return + } + rsp = &pb.RspIndicatorSummary{ + Indicator: req.Indicator, + Summary: summaryJson, + } + return +} + func (svr *TradingGrpcServer) StrategySeries(ctx context.Context, req *pb.ReqStrategySeries) (rsp *pb.RspStrategySeries, err error) { rsp = &pb.RspStrategySeries{} err = svr.tradingService.StrategySeries(ctx, req, rsp) diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index 6a1adf1..b62e819 100644 --- a/internal/trading/trading_service.go +++ b/internal/trading/trading_service.go @@ -317,6 +317,81 @@ func (svc *TradingService) IndicatorSeries(ctx context.Context, indicatorName st return } +func (svc *TradingService) IndicatorSummary(ctx context.Context, indicatorName string, digit int32, input types.Input, sr *pb.SeriesRange) (summary any, err error) { + indicator, ok := svc.indicatorReg.SummaryIndicator(indicatorName) + if !ok { + err = fmt.Errorf("indicator %s not exists", indicatorName) + return + } + interval := types.Interval(sr.Interval) + _, ok = types.SupportedIntervals[interval] + if !ok { + err = fmt.Errorf("unsupport interval %s", interval) + return + } + + // 查询历史指标数据 + kSeries := types.NewKlineSeries(sr.Exchange, sr.InstId, types.Interval(sr.Interval)) + indicatorContext := sig.NewIndicatorContext(indicator, input, sig.NewIndicatorStates(), kSeries, svc.indicatorReg) + + // 初始化指标 + err = indicator.Init(input) + if err != nil { + return + } + + candlePeriods := int(indicator.CandlePeriods(indicatorContext)) + sr.Desc = false + + // 计算第一个指标值需要多取candlePeriods - 1根k线; 计算ema等递归指标需要多拉取appros根k线逼近值 + srRsp, err := svc.exchangeClient.SeriesRange(ctx, &pb.ReqSeriesRange{Series: sr}) + if err != nil { + return + } + srBefore := srRsp.Before + sr.WindowExtra = uint32(max(0, candlePeriods-1)) + + err = svc.fetchHistoryKlineSeries(ctx, sr, func(k *types.Kline) (err error) { + 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 + } + if k.Ts < srBefore { + return + } + if kSeries.Length() < candlePeriods { + return + } + indicator.Accumulate(indicatorContext) + return + }) + if err != nil { + return + } + + // 计算指标汇总值 + summary, ok = indicator.Summary(indicatorContext) + if !ok { + err = fmt.Errorf("indicator summary result error: %s", indicatorName) + return + } + + // 保留小数位数 + digit = misc.Ternary(digit > 0 && digit <= 10, digit, 6) + pow := math.Pow(10, float64(digit)) + switch sm := summary.(type) { + case *types.VRVPSummary: + sm.Step = math.Round(sm.Step*pow) / pow + for i := range len(sm.Buckets) { + sm.Buckets[i].Price = math.Round(sm.Buckets[i].Price*pow) / pow + sm.Buckets[i].Volume = math.Round(sm.Buckets[i].Volume*pow) / pow + sm.Buckets[i].BuyVol = math.Round(sm.Buckets[i].BuyVol*pow) / pow + sm.Buckets[i].SellVol = math.Round(sm.Buckets[i].SellVol*pow) / pow + } + } + return +} + // StrategySeries 信号策略回测 func (svc *TradingService) StrategySeries(ctx context.Context, req *pb.ReqStrategySeries, rsp *pb.RspStrategySeries) (err error) { // sigStrategy diff --git a/pkg/indicator/indicator.go b/pkg/indicator/indicator.go index 74050b9..f676fb5 100644 --- a/pkg/indicator/indicator.go +++ b/pkg/indicator/indicator.go @@ -6,8 +6,9 @@ import ( ) const ( - MaxWindow = 256 // 最大窗口引用大小 - ApproCandles = 64 // 递归指标近似计算k线数 + // MaxWindow = 256 // 最大窗口引用大小 + MaxWindow = 1140 // 最大窗口引用大小 + ApproCandles = 64 // 递归指标近似计算k线数 ) type IndicatorMeta struct { @@ -37,6 +38,8 @@ type IIndicator interface { type ISummaryIndicator interface { // Meta 指标元信息 Meta() IndicatorMeta + // Init 初始化指标, 校验参数并初始化指标状态 + Init(input types.Input) (err error) // CandlePeriods 计算窗口大小的指标值需要的K线数量 CandlePeriods(ctx IIndicatorContext) int16 // Accumulate 计算并累加值 @@ -45,6 +48,12 @@ type ISummaryIndicator interface { Summary(ctx IIndicatorContext) (summary any, ok bool) } +// ISummaryIndicator 累加型计算指标清除接口 +type ISummaryIndicatorEliminater interface { + // Eliminate 清除这根K线的数据 + Eliminate(ctx IIndicatorContext) +} + // IIndicatorContext k线序列, trading服务提供 type IIndicatorContext interface { Get(offset int16) (kline types.Kline) diff --git a/pkg/indicator/indicator_registry.go b/pkg/indicator/indicator_registry.go index a59cc15..c88e636 100644 --- a/pkg/indicator/indicator_registry.go +++ b/pkg/indicator/indicator_registry.go @@ -7,12 +7,14 @@ import ( // 指标注册器 type IndicatorRegistry struct { - indicators *collect.SyncMap[string, IIndicator] // 注册窗口指标 + indicators *collect.SyncMap[string, IIndicator] // 注册窗口指标 + summaryIndicators *collect.SyncMap[string, func() ISummaryIndicator] // 注册汇总指标 } func NewIndicatorRegistry() *IndicatorRegistry { return &IndicatorRegistry{ - indicators: collect.NewSyncMap[string, IIndicator](), + indicators: collect.NewSyncMap[string, IIndicator](), + summaryIndicators: collect.NewSyncMap[string, func() ISummaryIndicator](), } } @@ -30,6 +32,8 @@ func (r *IndicatorRegistry) Init() (err error) { r.MustRegistIndicator(&ADX{}) r.MustRegistIndicator(&KDJ{}) r.MustRegistIndicator(&RVI{}) + + r.MustRegistSummaryIndicator(func() ISummaryIndicator { return &VRVP{} }) return } @@ -50,11 +54,44 @@ func (r *IndicatorRegistry) MustRegistIndicator(ind IIndicator) { } } +// RegistSummaryIndicator +func (r *IndicatorRegistry) RegistSummaryIndicator(indNewer func() ISummaryIndicator) (err error) { + indName := indNewer().Meta().Name + _, loaded := r.summaryIndicators.LoadOrStore(indName, indNewer) + if loaded { + err = fmt.Errorf("summary indicator name %s already duplicated", indName) + return + } + return +} + +func (r *IndicatorRegistry) MustRegistSummaryIndicator(indNewer func() ISummaryIndicator) { + if err := r.RegistSummaryIndicator(indNewer); err != nil { + panic(err) + } +} + // Indicator func (r *IndicatorRegistry) Indicator(name string) (indW IIndicator, ok bool) { return r.indicators.Load(name) } +// SummaryIndicator +func (r *IndicatorRegistry) SummaryIndicatorNewer(name string) (indNewer func() ISummaryIndicator, ok bool) { + indNewer, ok = r.summaryIndicators.Load(name) + return +} + +// SummaryIndicator +func (r *IndicatorRegistry) SummaryIndicator(name string) (indW ISummaryIndicator, ok bool) { + indNewer, ok := r.SummaryIndicatorNewer(name) + if !ok { + return + } + indW = indNewer() + return +} + func (r *IndicatorRegistry) RangeIndicators(fn func(k string, v IIndicator) bool) { r.indicators.Range(fn) } diff --git a/pkg/indicator/vrvp.go b/pkg/indicator/vrvp.go index aae7387..c831354 100644 --- a/pkg/indicator/vrvp.go +++ b/pkg/indicator/vrvp.go @@ -1,47 +1,259 @@ package indicator -import "sig-pub/pkg/types" +import ( + "math" + "sig-pub/pkg/types" + "sig-pub/pkg/zlog" +) -// VRVP 成交量分布图 +const ( + MaxRawBuckets = 2000 // 最大原始分桶数量，超过此数量将进行合并 +) + +// VRVP Volume Profile (Visible Range Volume Profile) +// 成交量分布图: 显示特定时间段内各价格水平的成交量分布 type VRVP struct { + buckets int // 价格行数 + minPrice float64 // 最低价 + maxPrice float64 // 最高价 + + // 流式计算状态 + baseStep float64 // 当前的基础步长 + rawBuckets map[int64]*types.VRVPBucket // 原始分桶数据, key = int64(price / baseStep) + klines int64 // 累计K线数量 + skts, ekts int64 // 开始时间戳, 结束时间戳 } func (c *VRVP) Meta() IndicatorMeta { return IndicatorMeta{ Name: "VRVP", + Desc: "成交量分布图", Input: []types.InputArg{ - {Name: "window", Type: types.InputTypeUInt, Desc: "窗口大小", Default: 10}, - }, - State: []string{"sig"}, - Plots: []Plot{ - {Name: "RVI", State: "vector", Type: PlotLine, Props: PlotProps{"color": ColorGreen}}, - {Name: "Signal", State: "sig", Type: PlotLine, Props: PlotProps{"color": ColorRed}}, + {Name: "buckets", Type: types.InputTypeUInt, Desc: "价格分桶数量", Default: 24}, }, } } -func (s *VRVP) Init(input types.Input) (err error) { // 校验参数并初始化 - // s.rate = input.Float("rate") - // s.rate2 = input.Float("rate2") +// Init 初始化参数 +func (v *VRVP) Init(input types.Input) (err error) { + v.buckets = input.Int("buckets") + if v.buckets <= 0 { + v.buckets = 24 + } + v.minPrice = math.MaxFloat64 + v.maxPrice = -math.MaxFloat64 + v.rawBuckets = make(map[int64]*types.VRVPBucket) + v.baseStep = 0 return } -// Accumulate 计算累加值 -func (v *VRVP) Accumulate(ctx IIndicatorContext) { +func (v *VRVP) CandlePeriods(ctx IIndicatorContext) int16 { + return 1 } -// Summary 计算完毕获取计算结果 -func (v *VRVP) Summary(ctx IIndicatorContext) (summary any, ok bool) { +// Eliminate 清除这根K线的数据 +func (v *VRVP) Eliminate(ctx IIndicatorContext) { + k := ctx.Get(0) + high := k.HighF64() + low := k.LowF64() + vol := k.VolF64() - return + if vol <= 0 { + return + } + + v.klines-- + + // 限制说明 ：在流式计算模式下，无法精确回滚 minPrice 、 maxPrice 以及 baseStep 的历史变化。 + // 这意味着在长期运行后，分布图的统计范围可能会比实际存在的 K 线范围略大，但这不影响分布形状的准确性。 + + if v.baseStep <= 0 { + return + } + + // 将成交量从原始桶中移除 + startIdx := int64(math.Floor(low / v.baseStep)) + endIdx := int64(math.Floor(high / v.baseStep)) + + coveredBuckets := float64(endIdx - startIdx + 1) + volPerBucket := vol / coveredBuckets + + isUp := k.CloseF64() >= k.OpenF64() + + for i := startIdx; i <= endIdx; i++ { + bucket, exists := v.rawBuckets[i] + if !exists { + continue + } + + bucket.Volume -= volPerBucket + if isUp { + bucket.BuyVol -= volPerBucket + } else { + bucket.SellVol -= volPerBucket + } + + // 如果桶的成交量归零（考虑浮点误差），则移除该桶以节省内存 + if bucket.Volume < 1e-8 { + delete(v.rawBuckets, i) + } + } +} + +// Accumulate 累积每根K线的数据 +func (v *VRVP) Accumulate(ctx IIndicatorContext) { + k := ctx.Get(0) + high := k.HighF64() + low := k.LowF64() + vol := k.VolF64() + v.klines++ + if v.skts == 0 { + v.skts = k.Ts + } + v.ekts = k.Ts + + if vol <= 0 { + return + } + + // 更新全局极值 + if high > v.maxPrice { + v.maxPrice = high + } + if low < v.minPrice { + v.minPrice = low + } + + // 初始化 baseStep + if v.baseStep == 0 { + // 如果 K 线有波动，使用波动的一小部分作为初始精度 + // 如果无波动(High==Low)，使用价格的万分之一 + if high > low { + v.baseStep = (high - low) / 100.0 + } else { + if low > 0 { + v.baseStep = low * 0.0001 + } else { + v.baseStep = 0.01 // 默认值 + } + } + } + + // 将成交量分配到原始桶中 + startIdx := int64(math.Floor(low / v.baseStep)) + endIdx := int64(math.Floor(high / v.baseStep)) + + coveredBuckets := float64(endIdx - startIdx + 1) + volPerBucket := vol / coveredBuckets + + isUp := k.CloseF64() >= k.OpenF64() + + for i := startIdx; i <= endIdx; i++ { + bucket, exists := v.rawBuckets[i] + if !exists { + bucket = &types.VRVPBucket{ + Price: float64(i)*v.baseStep + v.baseStep/2, // 暂存中心价 + } + v.rawBuckets[i] = bucket + } + + bucket.Volume += volPerBucket + if isUp { + bucket.BuyVol += volPerBucket + } else { + bucket.SellVol += volPerBucket + } + } + + // 检查是否需要合并桶 + if len(v.rawBuckets) > MaxRawBuckets { + v.halveResolution() + } } -type RVVPSummary struct { - TotalVolume float64 - Buckets []PriceBucket +// halveResolution 将分辨率减半（步长翻倍） +func (v *VRVP) halveResolution() { + newBaseStep := v.baseStep * 2 + newBuckets := make(map[int64]*types.VRVPBucket, len(v.rawBuckets)/2+1) + + for key, bucket := range v.rawBuckets { + // 计算新的 key + // oldPrice ~= key * oldStep + // newKey = floor(oldPrice / newStep) = floor(key * oldStep / (2 * oldStep)) = floor(key / 2) + newKey := key >> 1 // key / 2 + + newBucket, exists := newBuckets[newKey] + if !exists { + newBucket = &types.VRVPBucket{ + Price: float64(newKey)*newBaseStep + newBaseStep/2, + } + newBuckets[newKey] = newBucket + } + + newBucket.Volume += bucket.Volume + newBucket.BuyVol += bucket.BuyVol + newBucket.SellVol += bucket.SellVol + } + + v.baseStep = newBaseStep + v.rawBuckets = newBuckets } -type PriceBucket struct { - Price float64 - Volume float64 +// Summary 计算最终的成交量分布结果 +// 在所有K线Accumulate完成后调用 +func (v *VRVP) Summary(ctx IIndicatorContext) (summary any, ok bool) { + zlog.Debugf("VRVP Summary: klines=%d, skts=%d, ekts=%d, minPrice=%.2f, maxPrice=%.2f, baseStep=%.2f", + v.klines, v.skts, v.ekts, v.minPrice, v.maxPrice, v.baseStep) + if v.minPrice >= v.maxPrice { + return nil, false + } + + // 计算最终的目标步长 + rangeHeight := v.maxPrice - v.minPrice + finalStep := rangeHeight / float64(v.buckets) + + // 如果 finalStep 小于当前的 baseStep，说明数据太稀疏，无法满足 rowCount 的精度要求 + // 但通常情况下，由于我们只在桶过多时才合并，baseStep 应该相对较小 + + // 初始化最终分桶 + buckets := make([]types.VRVPBucket, v.buckets) + for i := 0; i < v.buckets; i++ { + buckets[i].Price = v.minPrice + float64(i)*finalStep + finalStep/2 + } + + // 将 rawBuckets 聚合到最终 buckets + for _, rawB := range v.rawBuckets { + // 计算原始桶对应的价格范围中心 + // 注意：rawB.Price 在合并过程中可能不再准确，重新计算更稳妥，或者在合并时更新 Price + // 这里我们使用 key 重新计算，更准确 + // 但由于 map key 不在 value 中，我们无法直接获取 key + // 所以我们需要遍历 map 的 key + // 修正：在上面的循环中我们无法获得 key，所以需要修改遍历方式 + // 或者我们在 PriceBucket 中存储准确的 Price + // 在 halveResolution 中，我们更新了 Price，所以 rawB.Price 是当前 baseStep 下的中心价 + + price := rawB.Price + + // 找到对应的最终桶索引 + idx := int((price - v.minPrice) / finalStep) + + if idx < 0 { + idx = 0 + } else if idx >= v.buckets { + idx = v.buckets - 1 + } + + buckets[idx].Volume += rawB.Volume + buckets[idx].BuyVol += rawB.BuyVol + buckets[idx].SellVol += rawB.SellVol + } + + ok, summary = true, &types.VRVPSummary{ + Klines: v.klines, + Step: finalStep, + MinPrice: v.minPrice, + MaxPrice: v.maxPrice, + Buckets: buckets, + } + return } diff --git a/pkg/strategy/bollgrid.go b/pkg/strategy/bollgrid.go index 0b78bd1..7d867ff 100644 --- a/pkg/strategy/bollgrid.go +++ b/pkg/strategy/bollgrid.go @@ -42,18 +42,14 @@ func (s *BollGrid) CandlePeriods(ctx ISingleSigStrategyContext) int16 { return max( ctx.Indicator("BOLL", s.period).CandlePeriods(), 2, // 需要前一根K线判断交叉 + 128, ) } func (s *BollGrid) Update(ctx ISingleSigStrategyContext) (side types.Side) { - // ind := ctx.Indicator("abc", "", "") - // ind.Summary() - // var ind indicator.IIndicatorSummary - // r, ok := ind.Summary(0, 100) // 100根k线的成交量分布图 - // if !ok { - // return types.SideNone - // } - // _ = r + // 128 根k线的成交量分布图 + summary, ok := ctx.SummaryIndicator("VRVP", types.Input{"buckets": 48}).Summary(0, 128) + _, _ = summary, ok // 获取指标数据 // BOLL指标 Calculate 返回值为 mb (中轨) diff --git a/pkg/strategy/sig_strategy.go b/pkg/strategy/sig_strategy.go index 89f5492..9f47a3a 100644 --- a/pkg/strategy/sig_strategy.go +++ b/pkg/strategy/sig_strategy.go @@ -36,6 +36,8 @@ type ISingleSigStrategyContext interface { Series(offset, count int16) (klines types.Klines) // Indicator 获取窗口类型指标 Indicator(name string, args ...any) indicator.IIndicatorSeries + // SummaryIndicator 获取Summary类型指标 + SummaryIndicator(name string, args ...any) indicator.IIndicatorSummary } // 多周期k线策略接口 diff --git a/pkg/types/indicator.go b/pkg/types/indicator.go new file mode 100644 index 0000000..49f5fe7 --- /dev/null +++ b/pkg/types/indicator.go @@ -0,0 +1,16 @@ +package types + +// VRVPSummary 成交量分布图指标计算结果 +type VRVPSummary struct { + Klines int64 `json:"klines"` + Step float64 `json:"step"` // 每个桶的价格高度 + MinPrice float64 `json:"minPrice"` // 统计范围最低价 + MaxPrice float64 `json:"maxPrice"` // 统计范围最高价 + Buckets []VRVPBucket `json:"buckets"` +} +type VRVPBucket struct { + Price float64 `json:"price"` // 桶代表价格 + Volume float64 `json:"volume"` // 总成交量 + BuyVol float64 `json:"buyVol"` // 主动买入量(近似) + SellVol float64 `json:"sellVol"` // 主动卖出量(近似) +}