From 6c0a0f8fbcc37471b6b1c5446919de544d30551e Mon Sep 17 00:00:00 2001 From: strange Date: Tue, 2 Dec 2025 18:31:24 +0800 Subject: [PATCH] kline series nocopy --- README.md | 4 +- api/pub.proto | 18 ++++ api/trading.proto | 7 ++ config/exchange.toml | 4 +- internal/trading/sig/kline_series.go | 125 ++++++++++++++++++++++-- internal/trading/trading_grpc_server.go | 11 +++ internal/trading/trading_service.go | 14 +++ pkg/indicator/indicator_plot.go | 31 +++--- pkg/indicator/super_trend.go | 9 +- pkg/strategy/super_trend2_macd.go | 4 +- pkg/strategy/super_trend_macd_rsi.go | 12 +++ pkg/types/ring_series.go | 3 +- 12 files changed, 212 insertions(+), 30 deletions(-) diff --git a/README.md b/README.md index c654452..b2922b0 100644 --- a/README.md +++ b/README.md @@ -94,11 +94,13 @@ strategy0: 趋势追踪，增长趋势， - python, pine script - Wasm(WebAssembly), 接口类型WIT - go plugin插件化(指标/策略, 安全问题?), 调度docker容器运行回测, grpc资源访问 + - 自定义脚本语言（偏配置内容） -> 编译为 go plugin (安全检查) -> 直接运行 收益方向: - trade strategy - - 自动测试(因子挖掘): + - 自动测试(因子挖掘), 因子失效检测: 1. 指标分类 2. 指标参数组合迭代 (确定潜在的可调参数)!!! + 3. 绘图能力(plot), 回测能力 KlineSeries 替换, copy性能消耗 macd dif/dea -> state -> plot diff --git a/api/pub.proto b/api/pub.proto index 6e69062..b83febc 100644 --- a/api/pub.proto +++ b/api/pub.proto @@ -1,5 +1,7 @@ syntax = "proto3"; +import "google/protobuf/struct.proto"; + option go_package = "./pb"; enum SubscribeType { @@ -156,3 +158,19 @@ message Paging { int32 size = 2; bool asc = 3; // 升序排序 } + +message IndicatorPlot { + string indicator = 1; + IndicatorPlotSeries series = 2; + repeated IndicatorPlotSeries stateSeries = 3; +} +message IndicatorPlotSeries { + string state = 1; + int32 type = 2; + google.protobuf.Struct props = 3; + repeated IndicatorStateToPlotProps state2Props = 4; +} +message IndicatorStateToPlotProps { + string state = 1; + map enumProps = 2; +} diff --git a/api/trading.proto b/api/trading.proto index f2467cb..8fda022 100644 --- a/api/trading.proto +++ b/api/trading.proto @@ -6,12 +6,19 @@ import "api/pub.proto"; option go_package = "./pb"; service TradingService { + rpc IndicatorPlots(ReqIndicatorPlots) returns (RspIndicatorPlots); // 获取指标绘图属性 rpc IndicatorSeries(ReqIndicatorSeries) returns (RspIndicatorSeries); // 获取指标序列 rpc StrategySeries(ReqStrategySeries) returns (RspStrategySeries); // 获取指标序列 rpc Backtest(ReqBacktest) returns (RspBacktest); // 交易计划回测 rpc BacktestLog(ReqBacktestLog) returns (RspBacktestLog); // 交易计划回测记录 rpc BacktestLogTrades(ReqBacktestLogTrades) returns (RspBacktestLogTrades); // 交易计划回测交易单详情 } +message ReqIndicatorPlots { + repeated string indicators = 1; +} +message RspIndicatorPlots{ + repeated IndicatorPlot plots = 1; +} message ReqIndicatorSeries { string indicator = 1; diff --git a/config/exchange.toml b/config/exchange.toml index d5bc355..e65fb93 100644 --- a/config/exchange.toml +++ b/config/exchange.toml @@ -18,8 +18,8 @@ receiveBuffer = 4096 marketSubscribeLimit = 16 consumeBatch = 1024 consumeLater = 2000 # 时间到达later或者数据累计到batch触发consume -# httpProxy = "" -httpProxy = "http://192.168.1.5:7890" +httpProxy = "" +# httpProxy = "http://192.168.1.5:7890" # httpProxy = "http://10.255.183.209:7890" # 模拟盘API交易地址如下： diff --git a/internal/trading/sig/kline_series.go b/internal/trading/sig/kline_series.go index 40829b7..31b9631 100644 --- a/internal/trading/sig/kline_series.go +++ b/internal/trading/sig/kline_series.go @@ -40,7 +40,7 @@ type KlineSeries struct { Interval types.Interval IntervalAdder types.IntervalAdder lastTs int64 - klines []*types.Kline + ringSeries *types.RingSeries[*types.Kline] } func NewKlineSeries(exchange pb.ExchangeType, instId string, interval types.Interval) *KlineSeries { @@ -53,7 +53,7 @@ func NewKlineSeries(exchange pb.ExchangeType, instId string, interval types.Inte InstId: instId, Interval: interval, IntervalAdder: intervalAdder, - klines: make([]*types.Kline, 0, MaxSeriesKlines/10), + ringSeries: types.NewRingSeries[*types.Kline](MaxSeriesKlines, MaxSeriesKlines/10), } } @@ -66,8 +66,117 @@ func (s *KlineSeries) MustGet(offset int16) (k types.Kline) { return } -// GetE [0]当前k线 +// Get [0]当前k线 func (s *KlineSeries) Get(offset int16) (k types.Kline, err error) { + s.mu.RLock() + defer s.mu.RUnlock() + v, ok := s.ringSeries.Get(int(offset)) + if !ok { + err = fmt.Errorf("get kline series offset out of range: offset=%d, length=%d", offset, s.ringSeries.Length()) + return + } + k = *v + return +} + +func (s *KlineSeries) MustSeries(offset, count int16) (klines series.Klines) { + klines, err := s.Series(offset, count) + if err != nil { + panic(err) + } + return +} + +// Series 时间降序序列[count...offset] +// offset: 从序列尾部开始偏移量 +// count: 从offset位置开始向序列头部k线条数 +func (s *KlineSeries) Series(offset, count int16) (klines series.Klines, err error) { + s.mu.RLock() + defer s.mu.RUnlock() + series, ok := s.ringSeries.Series(int(offset), int(count)) + if !ok { + err = fmt.Errorf("get kline series offset out of range: offset=%d, count=%d", offset, count) + return + } + for _, si := range series { + klines = append(klines, *si) + } + return +} + +func (s *KlineSeries) Length() int { + s.mu.RLock() + defer s.mu.RUnlock() + return s.ringSeries.Length() +} + +func (s *KlineSeries) LastTs() int64 { + return s.lastTs +} + +// 检查k线序列完整 +func (s *KlineSeries) Update(kline *types.Kline) (lastTs int64, serial bool) { + s.mu.Lock() + defer s.mu.Unlock() + + serial = true + lastTs = s.lastTs + if kline.Ts <= s.lastTs { + return + } + + // 检查k线是否连续 + if s.lastTs != 0 { + expectTs := s.IntervalAdder(s.lastTs, 1) + if kline.Ts != expectTs { + serial = false + miss := (kline.Ts-s.lastTs)/s.IntervalAdder(0, 1) - 1 + zlog.Warningf("k线不连续: instId=%s(%s), interval=%s, miss=%d, lastTs=%d, got=%d, expected=%d", s.InstId, s.Exchange, s.Interval, miss, s.lastTs, kline.Ts, expectTs) + return + } + } + + s.ringSeries.Push(kline) + s.lastTs = kline.Ts + return +} + +// Deprecated: use KlineSeries +type KlineSeries0 struct { + mu sync.RWMutex + Exchange pb.ExchangeType + InstId string + Interval types.Interval + IntervalAdder types.IntervalAdder + lastTs int64 + klines []*types.Kline +} + +func NewKlineSeries0(exchange pb.ExchangeType, instId string, interval types.Interval) *KlineSeries0 { + intervalAdder, ok := types.SupportedIntervals[interval] + if !ok { + panic(fmt.Errorf("unsupport interval: %s", interval)) + } + return &KlineSeries0{ + Exchange: exchange, + InstId: instId, + Interval: interval, + IntervalAdder: intervalAdder, + klines: make([]*types.Kline, 0, MaxSeriesKlines/10), + } +} + +// Get [0]当前k线 +func (s *KlineSeries0) MustGet(offset int16) (k types.Kline) { + k, err := s.Get(offset) + if err != nil { + panic(err) + } + return +} + +// GetE [0]当前k线 +func (s *KlineSeries0) Get(offset int16) (k types.Kline, err error) { if ok := offset >= 0 && offset < MaxSeriesKlines; !ok { err = fmt.Errorf("get kline series offset out of range: offset=%d", offset) return @@ -84,7 +193,7 @@ func (s *KlineSeries) Get(offset int16) (k types.Kline, err error) { return *(s.klines[index]), nil } -func (s *KlineSeries) MustSeries(offset, count int16) (klines series.Klines) { +func (s *KlineSeries0) MustSeries(offset, count int16) (klines series.Klines) { klines, err := s.Series(offset, count) if err != nil { panic(err) @@ -95,7 +204,7 @@ func (s *KlineSeries) MustSeries(offset, count int16) (klines series.Klines) { // Series 时间降序序列[count...offset] // offset: 从序列尾部开始偏移量 // count: 从offset位置开始向序列头部k线条数 -func (s *KlineSeries) Series(offset, count int16) (klines series.Klines, err error) { +func (s *KlineSeries0) Series(offset, count int16) (klines series.Klines, err error) { if ok := offset >= 0 && offset < MaxSeriesKlines; !ok { err = fmt.Errorf("get kline series offset out of range: offset=%d, count=%d", offset, count) return @@ -124,18 +233,18 @@ func (s *KlineSeries) Series(offset, count int16) (klines series.Klines, err err return klines, nil } -func (s *KlineSeries) Length() int { +func (s *KlineSeries0) Length() int { s.mu.RLock() defer s.mu.RUnlock() return len(s.klines) } -func (s *KlineSeries) LastTs() int64 { +func (s *KlineSeries0) LastTs() int64 { return s.lastTs } // 检查k线序列完整 -func (s *KlineSeries) Update(kline *types.Kline) (lastTs int64, serial bool) { +func (s *KlineSeries0) Update(kline *types.Kline) (lastTs int64, serial bool) { s.mu.Lock() defer s.mu.Unlock() diff --git a/internal/trading/trading_grpc_server.go b/internal/trading/trading_grpc_server.go index 4ec0131..1ea5f4c 100644 --- a/internal/trading/trading_grpc_server.go +++ b/internal/trading/trading_grpc_server.go @@ -22,6 +22,17 @@ func (svr *TradingGrpcServer) Init() (err error) { return } +func (svr *TradingGrpcServer) IndicatorPlots(ctx context.Context, req *pb.ReqIndicatorPlots) (rsp *pb.RspIndicatorPlots, err error) { + plots, err := svr.tradingService.IndicatorPlots(req.Indicators...) + if err != nil { + return + } + for _, plot := range plots { + _ = plot + } + return +} + 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() diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index 9f1d5dd..6560d0d 100644 --- a/internal/trading/trading_service.go +++ b/internal/trading/trading_service.go @@ -198,6 +198,20 @@ func (svc *TradingService) fetchHistoryKlineSeries(ctx context.Context, sr *pb.S return } +// IndicatorPlots +func (svc *TradingService) IndicatorPlots(indicatorNames ...string) (plots map[string]indicator.Plot, err error) { + plots = make(map[string]indicator.Plot, len(indicatorNames)) + for _, indicatorName := range indicatorNames { + indicator, ok := svc.indicatorReg.Indicator(indicatorName) + if !ok { + err = fmt.Errorf("indicator %s not exists", indicatorName) + return + } + plots[indicatorName] = indicator.Meta().Plot + } + return +} + // IndicatorSeries 获取指标实时或历史序列数据, 闭区间 func (svc *TradingService) IndicatorSeries(ctx context.Context, indicatorName string, digit int32, input types.Input, sr *pb.SeriesRange) (matrix []float64, times []int64, states map[string][]float64, err error) { appros := indicator.ApproCandles diff --git a/pkg/indicator/indicator_plot.go b/pkg/indicator/indicator_plot.go index 66aac6a..c8481ac 100644 --- a/pkg/indicator/indicator_plot.go +++ b/pkg/indicator/indicator_plot.go @@ -1,26 +1,31 @@ package indicator -// 绘图属性 +// 指标绘图 type Plot struct { - Props map[string]any // 长度颜色 - Series []PlotSeries // 需要绘图的时间序列列表 + Props PlotProps // 长度颜色等属性 + Series PlotSeries + StateSeries []PlotSeries // 需要绘图的时间序列列表 } +// 指标绘图: 逻辑化 -> 配置化 type PlotSeries struct { - State string `json:"state"` // vector, stateName - Series SeriesType `json:"series"` // 绘图类型 线/柱 - Props map[string]any `json:"props"` // 绘图属性 - SeriesFn func(vector float64, states map[string]float64) map[string]any + State string `json:"state"` // vector, stateName + Type PlotSeriesType `json:"series"` // 绘图类型 线/柱 + Props PlotProps `json:"props"` // 绘图属性 + State2Props map[string]map[float64]PlotProps `json:"state2Props"` // state转绘图属性 } -// SeriesType 序列值绘图类型 -type SeriesType int32 +// 指标绘图 +type PlotProps map[string]any + +// PlotSeriesType 序列值绘图类型 +type PlotSeriesType int32 const ( - SeriesNone SeriesType = iota - SeriesLine - SeriesHistogram - SeriesArea + PlotSeriesNone PlotSeriesType = iota + PlotSeriesLine + PlotSeriesHistogram + PlotSeriesArea ) // Series 绘图颜色 diff --git a/pkg/indicator/super_trend.go b/pkg/indicator/super_trend.go index 526aaee..d7e6e99 100644 --- a/pkg/indicator/super_trend.go +++ b/pkg/indicator/super_trend.go @@ -15,9 +15,12 @@ func (c SuperTrend) Meta() IndicatorMeta { }, State: []string{"direction"}, Plot: Plot{ - Series: []PlotSeries{ - {State: "", Series: SeriesLine, Props: map[string]any{"color": "red"}}, // hist - }, + Series: PlotSeries{Type: PlotSeriesLine, Props: PlotProps{"color": ColorGreen}, State2Props: map[string]map[float64]PlotProps{ + "direction": { + -1: {"color": ColorRed}, + 1: {"color": ColorGreen}, + }, + }}, }, } } diff --git a/pkg/strategy/super_trend2_macd.go b/pkg/strategy/super_trend2_macd.go index ca36010..aeec604 100644 --- a/pkg/strategy/super_trend2_macd.go +++ b/pkg/strategy/super_trend2_macd.go @@ -1,6 +1,8 @@ package strategy -import "sig-pub/pkg/types" +import ( + "sig-pub/pkg/types" +) // SuperTrend2Macd 结合super trend和rsi指标策略 type SuperTrend2Macd struct { diff --git a/pkg/strategy/super_trend_macd_rsi.go b/pkg/strategy/super_trend_macd_rsi.go index a1f68dc..fec299a 100644 --- a/pkg/strategy/super_trend_macd_rsi.go +++ b/pkg/strategy/super_trend_macd_rsi.go @@ -68,5 +68,17 @@ func (s *SuperTrendMacdRSI) Update(ctx ISingleSigStrategyContext) (side types.Si } } } + + _ = ` + // 策略算子脚本DST, 优化golang底层不影响策略语法 + st1 = sig.SuperTrend(window=10, mul=3) + rsi = sig.RSI(window=14) + closeAvg = close[1:10].avg() + st1[0] + st1[0:10] + st1.direction[0] + if rsi[0] > 50 + ... + ` return } diff --git a/pkg/types/ring_series.go b/pkg/types/ring_series.go index f61d37d..b74694d 100644 --- a/pkg/types/ring_series.go +++ b/pkg/types/ring_series.go @@ -27,7 +27,7 @@ func (r *RingSeries[T]) Length() int { return r.length } -func (r *RingSeries[T]) Push(v T) (ok bool) { +func (r *RingSeries[T]) Push(v T) { if !r.full { r.values = append(r.values, v) r.head++ @@ -38,7 +38,6 @@ func (r *RingSeries[T]) Push(v T) (ok bool) { r.values[r.tail] = v r.head = r.tail r.tail = (r.tail + 1) % r.capacity - return } // Get 0当前, 1前一个