From 0777bb1fe2854c8cd3ffbd6bb44cdfddf6e3011d Mon Sep 17 00:00:00 2001 From: strange Date: Mon, 27 Oct 2025 00:39:50 +0800 Subject: [PATCH] fetch rpc --- README.md | 9 +++++++ api/exchange.proto | 4 +-- api/pub.proto | 14 +++++------ api/trading.proto | 1 + config/config.toml | 5 ++-- config/exchange.toml | 4 +-- internal/exchange/exchange_service.go | 31 +++++++++++++++++++----- internal/trading/kline_series.go | 2 +- internal/trading/trading_service.go | 13 ++++++++-- pkg/config/config.go | 1 - pkg/config/grpc_options.go | 9 +++---- pkg/data/entity/trade_order.go | 2 ++ pkg/strategy/sig_strategy_params.go | 6 ++--- pkg/types/decimals/decimal.go | 8 ++++++ pkg/types/kline.go | 35 ++++++++++++++++----------- 15 files changed, 97 insertions(+), 47 deletions(-) diff --git a/README.md b/README.md index 1b98a60..3e87630 100644 --- a/README.md +++ b/README.md @@ -49,6 +49,9 @@ k线推送 价格/交易量精度->tsdb读写存储 kline时间窗口 +Dragonfly redis替换 + +### 绘图框架 tradingview advanced-charts: - https://cn.tradingview.com/advanced-charts/ tradingview lightweight-charts: @@ -56,6 +59,8 @@ tradingview lightweight-charts: - https://github.com/tradingview/lightweight-charts d3js: - https://d3js.org/ +night-vision: + - [night-vision](https://github.com/project-nv/night-vision) DXcharts: - https://devexperts.com/dxcharts/ @@ -71,3 +76,7 @@ strategy0: 趋势追踪，增长趋势， 2. 当前属于波段增长点 3. 买卖点画叉 4. 移动止盈，固定止损，风险评估止损 + +### 量化框架参考 + +[investing-algorithm-framework](https://github.com/coding-kitties/investing-algorithm-framework) \ No newline at end of file diff --git a/api/exchange.proto b/api/exchange.proto index f341765..94ba9e4 100644 --- a/api/exchange.proto +++ b/api/exchange.proto @@ -63,9 +63,9 @@ message ReqHistoryKline { int64 before = 4; int64 after = 5; uint32 count = 6; // k线条数,before或after其中一个为0时有效 - // bool open = 7; // 是否开区间, 不包含 before/after + bool open = 7; // 是否开区间, before/after不为0时不包含 bool live = 8; // 实时k线, before和after为0时是否追加实时k线 - bool asc = 9; // 是否升序, 默认降序 + bool desc = 9; // 是否降序, 默认升序 } message RspHistoryKline { ExchangeType exchange = 1; // 交易所 diff --git a/api/pub.proto b/api/pub.proto index 8c46000..a7a5ca6 100644 --- a/api/pub.proto +++ b/api/pub.proto @@ -106,14 +106,14 @@ message TradeInstanceState { } message Kline { - int64 ts = 2; + int64 time = 2; string interval = 3; // 周期 - string open = 4; - string high = 5; - string low = 6; - string close = 7; - string vol = 8; // 交易量 - string volQuote = 9; // 交易额 + double open = 4; + double high = 5; + double low = 6; + double close = 7; + double vol = 8; // 交易量 + double volQuote = 9; // 交易额 bool confirm = 10; // k线是否完结 } diff --git a/api/trading.proto b/api/trading.proto index 72b174c..8baec15 100644 --- a/api/trading.proto +++ b/api/trading.proto @@ -48,6 +48,7 @@ message ReqStrategySeries { int64 before = 6; // 0表示最新 int64 after = 7; // 0表示最新 int32 count = 8; // k线条数,before或after其中一个为0时有效 + map sigParam = 15; // 策略参数 } message RspStrategySeries { repeated Side signal = 1; // 0.sell,1.buy diff --git a/config/config.toml b/config/config.toml index 081c698..70ebead 100644 --- a/config/config.toml +++ b/config/config.toml @@ -1,13 +1,12 @@ [grpc] -maxSendMsgSize = "16Mi" -maxRecvMsgSize = "16Mi" +maxSendMsgSize = "8Mi" +maxRecvMsgSize = "8Mi" readBufferSize = "8Ki" writeBufferSize = "8Ki" [grpc.keepalive] idleTimeout = "60s" # 空闲连接超时 -forceCloseWait = "20s" keepAliveInterval = "60s" # 发送 ping 的间隔 keepAliveTimeout = "20s" # ping 超时 maxLifeTime = "2h" diff --git a/config/exchange.toml b/config/exchange.toml index 4d567a5..c0a42b7 100644 --- a/config/exchange.toml +++ b/config/exchange.toml @@ -18,8 +18,8 @@ receiveBuffer = 4096 marketSubscribeLimit = 16 consumeBatch = 1024 consumeLater = 2000 # 时间到达later或者数据累计到batch触发consume -# httpProxy = "http://192.168.1.5:7890" -httpProxy = "http://10.255.183.209:7890" +httpProxy = "http://192.168.1.5:7890" +# httpProxy = "http://10.255.183.209:7890" # 模拟盘API交易地址如下： # REST：https://www.okx.com diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index c0882ba..e5ba466 100644 --- a/internal/exchange/exchange_service.go +++ b/internal/exchange/exchange_service.go @@ -658,7 +658,7 @@ func (svc *ExchangeService) ExchangeInstanceState(req *pb.ReqExchangeInstanceSta } const ( - MaxHistoryKlines = 100 + MaxHistoryKlines = 200 ) // HistoryKline 获取交易产品历史k线 (before < klines... < after) @@ -691,7 +691,7 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory } // 拉取最新的 lastTs := int64(0) - if afterTs == 0 { + if req.After == 0 { liveK := exchangeInst.LiveKline.Get(interval) lastTs = liveK.Ts if !liveK.Confirm { @@ -707,6 +707,21 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory if beforeTs == 0 { beforeTs = max(intervalAdder(afterTs, -count+1), KlineBefore0) } + // 开区间 + if req.Open { + if req.After != 0 { + afterTs = max(intervalAdder(afterTs, -1), beforeTs) + if req.Before == 0 { + beforeTs = max(intervalAdder(beforeTs, -1), KlineBefore0) + } + } + if req.Before != 0 { + beforeTs = min(intervalAdder(beforeTs, 1), afterTs) + if req.After == 0 { + afterTs = min(intervalAdder(beforeTs, 1), lastTs) + } + } + } if beforeTs > afterTs { err = fmt.Errorf("time range invalid: before must less then after") return @@ -726,6 +741,10 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory if len(klines) == 0 { return } + // 检查k线是否连续进行补齐 + if err = svc.paddingKlinesIfNotSeries(exchange, req.InstId, interval, klines); err != nil { + return + } lastK := klines[len(klines)-1] // vmtsdb 数据刷盘30s延迟, 使用内存数据替代第一根k线 @@ -740,7 +759,7 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory } // 降序排序 - if !req.Asc { + if req.Desc { collect.Reverse(klines) } @@ -748,10 +767,10 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory if req.Live && len(klines) > 0 { liveK := exchangeInst.LiveKline.Get(interval) if latest := intervalAdder(lastK.Ts, 1) == liveK.Ts; latest { - if req.Asc { - klines = append(klines, &liveK) - } else { + if req.Desc { klines = append([]*types.Kline{&liveK}, klines...) + } else { + klines = append(klines, &liveK) } rsp.Live = true } diff --git a/internal/trading/kline_series.go b/internal/trading/kline_series.go index e42fcfa..37c8083 100644 --- a/internal/trading/kline_series.go +++ b/internal/trading/kline_series.go @@ -80,7 +80,7 @@ func (s *KlineSeries) Series(offset, count int16) (klines series.Klines, ok bool if ok = offset >= 0 && offset < MaxSeriesKlines; !ok { return } - if ok = count >= 0 && offset+count < MaxSeriesKlines; !ok { + if ok = count > 0 && offset+count < MaxSeriesKlines; !ok { return } diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index 222fabc..05810c4 100644 --- a/internal/trading/trading_service.go +++ b/internal/trading/trading_service.go @@ -208,11 +208,20 @@ func (svc *TradingService) IndicatorSeries(req *pb.ReqIndicatorSeries, rsp *pb.R // StrategySeries 简单策略信号测试 func (svc *TradingService) StrategySeries(req *pb.ReqStrategySeries, rsp *pb.RspStrategySeries) (err error) { - strategy, ok := svc.strategyReg.NewSigStrategy(req.Strategy) + // sigStrategy + sigStrategy, ok := svc.strategyReg.NewSigStrategy(req.Strategy) if !ok { err = fmt.Errorf("strategy %s not exists", req.Strategy) return } + err = sigStrategy.Init(strategy.SigStrategyParam{ + Interval: types.Interval(req.Interval), + Param: req.SigParam, + }) + if err != nil { + return + } + interval := types.Interval(req.Interval) intervalAdd, ok := types.SupportedIntervals[interval] if !ok { @@ -230,7 +239,7 @@ func (svc *TradingService) StrategySeries(req *pb.ReqStrategySeries, rsp *pb.Rsp strategyCtx := NewStrategyContext(klineSeries, svc.indicatorReg) for i := range req.Count { strategyCtx.SetOffset(int16(i)) - strategy.Update(strategyCtx) + sigStrategy.Update(strategyCtx) } rsp.Signal = strategyCtx.signal rsp.Times = strategyCtx.signalTimes diff --git a/pkg/config/config.go b/pkg/config/config.go index 5f5788b..e70bf1c 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -35,7 +35,6 @@ type GrpcConfig struct { type GrpcKeepalive struct { IdleTimeout string - ForceCloseWait string KeepAliveInterval string KeepAliveTimeout string MaxLifeTime string diff --git a/pkg/config/grpc_options.go b/pkg/config/grpc_options.go index b95c090..db0780e 100644 --- a/pkg/config/grpc_options.go +++ b/pkg/config/grpc_options.go @@ -12,13 +12,13 @@ func GetGrpcOptions(c GrpcConfig, customOpts ...grpc.ServerOption) (opts []grpc. opts = append(opts, grpc.MaxSendMsgSize(conver.MustParseDataUnitInt(c.MaxSendMsgSize))) } if c.MaxRecvMsgSize != "" { - opts = append(opts, grpc.MaxSendMsgSize(conver.MustParseDataUnitInt(c.MaxRecvMsgSize))) + opts = append(opts, grpc.MaxRecvMsgSize(conver.MustParseDataUnitInt(c.MaxRecvMsgSize))) } if c.ReadBufferSize != "" { - opts = append(opts, grpc.MaxSendMsgSize(conver.MustParseDataUnitInt(c.ReadBufferSize))) + opts = append(opts, grpc.ReadBufferSize(conver.MustParseDataUnitInt(c.ReadBufferSize))) } if c.WriteBufferSize != "" { - opts = append(opts, grpc.MaxSendMsgSize(conver.MustParseDataUnitInt(c.WriteBufferSize))) + opts = append(opts, grpc.WriteBufferSize(conver.MustParseDataUnitInt(c.WriteBufferSize))) } // grpc server keepalive @@ -26,9 +26,6 @@ func GetGrpcOptions(c GrpcConfig, customOpts ...grpc.ServerOption) (opts []grpc. if c.keepalive.IdleTimeout != "" { keep.MaxConnectionIdle = conver.MustParseDuration(c.keepalive.IdleTimeout) } - if c.keepalive.ForceCloseWait != "" { - keep.MaxConnectionAgeGrace = conver.MustParseDuration(c.keepalive.ForceCloseWait) - } if c.keepalive.KeepAliveInterval != "" { keep.Time = conver.MustParseDuration(c.keepalive.KeepAliveInterval) } diff --git a/pkg/data/entity/trade_order.go b/pkg/data/entity/trade_order.go index 4dd226b..89f7141 100644 --- a/pkg/data/entity/trade_order.go +++ b/pkg/data/entity/trade_order.go @@ -25,6 +25,8 @@ type TradeOrder struct { TradeTime int64 `gorm:"column:trade_time" json:"tradeTime"` // 成交时间 CreateTime int64 `gorm:"column:create_time" json:"createTime"` // 创建时间 UpdateTime int64 `gorm:"column:update_time" json:"updateTime"` // 更新时间 + Decision string `gorm:"column:decision" json:"decision"` // 决策过程数据 json map + DecisionM map[string]any `gorm:"-" json:"decisionM"` } func (TradeOrder) TableName() string { diff --git a/pkg/strategy/sig_strategy_params.go b/pkg/strategy/sig_strategy_params.go index a49c532..817c715 100644 --- a/pkg/strategy/sig_strategy_params.go +++ b/pkg/strategy/sig_strategy_params.go @@ -81,14 +81,14 @@ type ISigStrategyParamGenerator interface { type SigStrategyParam struct { Interval types.Interval `json:"interval"` // 策略驱动周期 - Params map[string]string `json:"params"` // 策略执行参数 + Param map[string]string `json:"param"` // 策略执行参数 } func (s *SigStrategyParam) Get(key string) (v string, ok bool) { - if len(s.Params) == 0 { + if len(s.Param) == 0 { return } - v, ok = s.Params[key] + v, ok = s.Param[key] return } diff --git a/pkg/types/decimals/decimal.go b/pkg/types/decimals/decimal.go index ae6ae9a..e4c29ce 100644 --- a/pkg/types/decimals/decimal.go +++ b/pkg/types/decimals/decimal.go @@ -13,3 +13,11 @@ func MustToFloat64(v decimal.Decimal) float64 { } return f } + +func MustFromFloat64(f float64) (v decimal.Decimal) { + v, err := decimal.NewFromFloat64(f) + if err != nil { + panic(fmt.Errorf("decimal from float64 error: %v", err)) + } + return +} diff --git a/pkg/types/kline.go b/pkg/types/kline.go index d1fb04f..0c0032e 100644 --- a/pkg/types/kline.go +++ b/pkg/types/kline.go @@ -2,6 +2,7 @@ package types import ( "sig-pub/api/pb" + "sig-pub/pkg/types/decimals" "github.com/govalues/decimal" ) @@ -27,28 +28,34 @@ type Kline struct { func (k *Kline) ParsePBKline(exchange pb.ExchangeType, kline *pb.Kline) { // k.Exchange = exchange.String() + k.Ts = kline.Time k.Interval = Interval(kline.Interval) - k.Ts = kline.Ts - k.Open = decimal.MustParse(kline.Open) - k.High = decimal.MustParse(kline.High) - k.Low = decimal.MustParse(kline.Low) - k.Close = decimal.MustParse(kline.Close) - k.Vol = decimal.MustParse(kline.Vol) - k.VolQuote = decimal.MustParse(kline.VolQuote) k.Confirm = kline.Confirm + // k.Open = decimal.MustParse(kline.Open) + // k.High = decimal.MustParse(kline.High) + // k.Low = decimal.MustParse(kline.Low) + // k.Close = decimal.MustParse(kline.Close) + // k.Vol = decimal.MustParse(kline.Vol) + // k.VolQuote = decimal.MustParse(kline.VolQuote) + k.Open = decimals.MustFromFloat64(kline.Open) + k.High = decimals.MustFromFloat64(kline.High) + k.Low = decimals.MustFromFloat64(kline.Low) + k.Close = decimals.MustFromFloat64(kline.Close) + k.Vol = decimals.MustFromFloat64(kline.Vol) + k.VolQuote = decimals.MustFromFloat64(kline.VolQuote) } func (k *Kline) ToPBKline() (kline *pb.Kline) { kline = &pb.Kline{ - Ts: k.Ts, + Time: k.Ts, Interval: string(k.Interval), - Open: k.Open.String(), - High: k.High.String(), - Low: k.Low.String(), - Close: k.Close.String(), - Vol: k.Vol.String(), - VolQuote: k.VolQuote.String(), Confirm: k.Confirm, + Open: decimals.MustToFloat64(k.Open), + High: decimals.MustToFloat64(k.High), + Low: decimals.MustToFloat64(k.Low), + Close: decimals.MustToFloat64(k.Close), + Vol: decimals.MustToFloat64(k.Vol), + VolQuote: decimals.MustToFloat64(k.VolQuote), } return }