From 6aea2f69d88922a0ccdba49ed4244692f254465d Mon Sep 17 00:00:00 2001 From: strange Date: Mon, 27 Oct 2025 18:35:46 +0800 Subject: [PATCH] trading indicator/strategy series --- api/exchange.proto | 17 +- api/pub.proto | 18 ++- api/trading.proto | 20 +-- config/exchange.toml | 4 +- internal/exchange/exchange_grpc_server.go | 32 +++- internal/exchange/exchange_service.go | 188 ++++++++++------------ internal/trading/indicator_context.go | 30 ++-- internal/trading/kline_store.go | 17 +- internal/trading/strategy_context.go | 67 ++++---- internal/trading/trading_grpc_server.go | 7 +- internal/trading/trading_plan_runner.go | 4 +- internal/trading/trading_service.go | 129 ++++++++------- pkg/strategy/gold_x.go | 12 +- pkg/strategy/sig_strategy.go | 5 +- 14 files changed, 281 insertions(+), 269 deletions(-) diff --git a/api/exchange.proto b/api/exchange.proto index 94ba9e4..68715d5 100644 --- a/api/exchange.proto +++ b/api/exchange.proto @@ -57,15 +57,7 @@ message RspExchangeInstanceState { } message ReqHistoryKline { - ExchangeType exchange = 1; // 交易所 - string instId = 2; - string interval = 3; - int64 before = 4; - int64 after = 5; - uint32 count = 6; // k线条数,before或after其中一个为0时有效 - bool open = 7; // 是否开区间, before/after不为0时不包含 - bool live = 8; // 实时k线, before和after为0时是否追加实时k线 - bool desc = 9; // 是否降序, 默认升序 + SeriesRange series = 1; } message RspHistoryKline { ExchangeType exchange = 1; // 交易所 @@ -77,12 +69,7 @@ message RspHistoryKline { } message ReqHistoryKlineStream { - ExchangeType exchange = 1; // 交易所 - string instId = 2; - string interval = 3; - int64 before = 4; - int64 after = 5; - uint32 count = 6; // k线条数,before或after其中一个为0时有效 + SeriesRange series = 1; } message RspHistoryKlineStream { repeated Kline klines = 2; diff --git a/api/pub.proto b/api/pub.proto index a7a5ca6..c4c4354 100644 --- a/api/pub.proto +++ b/api/pub.proto @@ -40,8 +40,9 @@ enum Channel { } enum Side { - SELL = 0; + None = 0; BUY = 1; + SELL = 2; } enum OrderType { @@ -133,3 +134,18 @@ message Order { int64 group_id = 15; int64 created_at = 10; } + +message SeriesRange { + ExchangeType exchange = 1; // 交易所 + string instId = 2; + string interval = 3; + int64 before = 4; + int64 after = 5; + uint32 count = 6; // k线条数,before或after其中一个为0时有效 + bool open = 7; // 是否开区间, before/after不为0时不包含 + bool live = 8; // 实时k线, before和after为0时是否追加实时k线 + bool desc = 9; // 是否降序, 默认升序 + + uint32 window = 10; // 需要额外拉取更早的k线条数 + uint32 limit = 11; // 大于0时检查, 数据长度超过limit则返回错误 +} diff --git a/api/trading.proto b/api/trading.proto index 8baec15..7e11076 100644 --- a/api/trading.proto +++ b/api/trading.proto @@ -27,13 +27,8 @@ message Indicator { message ReqIndicatorSeries { string indicator = 1; - int32 window = 2; // 指标窗口大小 - ExchangeType exchange = 3; - string instId = 4; - string interval = 5; - int64 before = 6; // 0表示最新 - int64 after = 7; // 0表示最新 - int32 count = 8; // k线条数,before或after其中一个为0时有效 + uint32 window = 2; // 指标窗口大小 + SeriesRange series = 9; } message RspIndicatorSeries{ repeated double matrix = 1; @@ -41,14 +36,9 @@ message RspIndicatorSeries{ } message ReqStrategySeries { - string strategy = 1; - ExchangeType exchange = 3; - string instId = 4; - string interval = 5; - int64 before = 6; // 0表示最新 - int64 after = 7; // 0表示最新 - int32 count = 8; // k线条数,before或after其中一个为0时有效 - map sigParam = 15; // 策略参数 + SeriesRange series = 1; + string sigStrategy = 2; + map sigParam = 3; // 策略参数 } message RspStrategySeries { repeated Side signal = 1; // 0.sell,1.buy diff --git a/config/exchange.toml b/config/exchange.toml index c0a42b7..4d567a5 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_grpc_server.go b/internal/exchange/exchange_grpc_server.go index fb8902f..724605b 100644 --- a/internal/exchange/exchange_grpc_server.go +++ b/internal/exchange/exchange_grpc_server.go @@ -119,18 +119,30 @@ func (svr *ExchangeGrpcServer) ExchangeInstanceState(ctx context.Context, req *p // HistoryKline 获取交易产品历史k线 (before < klines... < after) func (svr *ExchangeGrpcServer) HistoryKline(ctx context.Context, req *pb.ReqHistoryKline) (rsp *pb.RspHistoryKline, err error) { - rsp = &pb.RspHistoryKline{ - Exchange: req.Exchange, - InstId: req.InstId, - Interval: req.Interval, + if req.Series == nil { + err = fmt.Errorf("series arg is required") + return } - // for before := req.Before; ; { - // } - klines, err := svr.exchangeService.HistoryKline(ctx, req, rsp) + // 查询范围数据条数检查 + _, _, total, err := svr.exchangeService.CalcSeriesRange(req.Series) if err != nil { return } + if total > DefaultHistoryKlines { + err = fmt.Errorf("time range too large max %d", DefaultHistoryKlines) + return + } + rsp = &pb.RspHistoryKline{ + Exchange: req.Series.Exchange, + InstId: req.Series.InstId, + Interval: req.Series.Interval, + } + live, klines, err := svr.exchangeService.HistoryKline(ctx, req.Series) + if err != nil { + return + } + rsp.Live = live rsp.Klines = make([]*pb.Kline, 0, len(klines)) for _, k := range klines { rsp.Klines = append(rsp.Klines, k.ToPBKline()) @@ -140,6 +152,10 @@ func (svr *ExchangeGrpcServer) HistoryKline(ctx context.Context, req *pb.ReqHist // HistoryKlineStream 获取交易产品历史k线(流式返回) func (svr *ExchangeGrpcServer) HistoryKlineStream(req *pb.ReqHistoryKlineStream, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) { - err = svr.exchangeService.HistoryKlineStream(req, stream) + if req.Series == nil { + err = fmt.Errorf("series arg is required") + return + } + err = svr.exchangeService.HistoryKlineStream(req.Series, stream) return } diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index e5ba466..de1cb99 100644 --- a/internal/exchange/exchange_service.go +++ b/internal/exchange/exchange_service.go @@ -658,80 +658,115 @@ func (svc *ExchangeService) ExchangeInstanceState(req *pb.ReqExchangeInstanceSta } const ( - MaxHistoryKlines = 200 + DefaultHistoryKlines = 200 + MaxHistoryKlines = 4096 ) -// HistoryKline 获取交易产品历史k线 (before < klines... < after) -func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistoryKline, rsp *pb.RspHistoryKline) (klines []*types.Kline, err error) { +func (svc *ExchangeService) CalcSeriesRange(arg *pb.SeriesRange) (after, before, total int64, err error) { // 交易产品参数检查 - exchange := svc.exchanges.Get(req.Exchange) - exchangeInstId, ok := exchange.TradeInstIds.Load(req.InstId) + exchange := svc.exchanges.Get(arg.Exchange) + exchangeInstId, ok := exchange.TradeInstIds.Load(arg.InstId) if !ok { - err = fmt.Errorf("trade instance not support: %s", req.InstId) + err = fmt.Errorf("trade instance not support: %s", arg.InstId) return } - interval := types.Interval(req.Interval) - intervalAdder, ok := types.SupportedIntervals[interval] + exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId) if !ok { - err = fmt.Errorf("interval not support: %s", req.Interval) + err = fmt.Errorf("trade instance not support for exchange: %s for %s", arg.InstId, arg.Exchange) return } - - // todo 交易产品初始化完成检查 - exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId) + interval := types.Interval(arg.Interval) + intervalAdder, ok := types.SupportedIntervals[interval] if !ok { - err = fmt.Errorf("trade instance not support for exchange: %s for %s", req.InstId, req.Exchange) + err = fmt.Errorf("interval not support: %s", arg.Interval) return } - // k线长度检查 - afterTs, beforeTs, count := int64(req.After), int64(req.Before), int64(req.Count) + after, before, count := int64(arg.After), int64(arg.Before), int64(arg.Count) if count == 0 { - count = MaxHistoryKlines + count = DefaultHistoryKlines } // 拉取最新的 lastTs := int64(0) - if req.After == 0 { + if arg.After == 0 { liveK := exchangeInst.LiveKline.Get(interval) lastTs = liveK.Ts if !liveK.Confirm { lastTs = intervalAdder(liveK.Ts, -1) } } - if afterTs == 0 && beforeTs == 0 { - afterTs = lastTs + if after == 0 && before == 0 { + after = lastTs } - if afterTs == 0 { - afterTs = min(intervalAdder(beforeTs, count-1), lastTs) + if after == 0 { + after = min(intervalAdder(before, count-1), lastTs) } - if beforeTs == 0 { - beforeTs = max(intervalAdder(afterTs, -count+1), KlineBefore0) + if before == 0 { + before = max(intervalAdder(after, -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 arg.Open { + if arg.After != 0 { + after = max(intervalAdder(after, -1), before) + if arg.Before == 0 { + before = max(intervalAdder(before, -1), KlineBefore0) } } - if req.Before != 0 { - beforeTs = min(intervalAdder(beforeTs, 1), afterTs) - if req.After == 0 { - afterTs = min(intervalAdder(beforeTs, 1), lastTs) + if arg.Before != 0 { + before = min(intervalAdder(before, 1), after) + if arg.After == 0 { + after = min(intervalAdder(before, 1), lastTs) } } } - if beforeTs > afterTs { + // 额外拉取 + if arg.Window > 0 { + before = max(intervalAdder(before, -int64(arg.Window)), KlineBefore0) + } + if before > after { err = fmt.Errorf("time range invalid: before must less then after") return } + // 拉取范围总条数 + total = (after-before)/intervalAdder(0, 1) + 1 + return +} + +// HistoryKline 获取交易产品历史k线 (before < klines... < after) +func (svc *ExchangeService) HistoryKline(ctx context.Context, arg *pb.SeriesRange) (live bool, klines []*types.Kline, err error) { + // 交易产品参数检查 + exchange := svc.exchanges.Get(arg.Exchange) + exchangeInstId, ok := exchange.TradeInstIds.Load(arg.InstId) + if !ok { + err = fmt.Errorf("trade instance not support: %s", arg.InstId) + return + } + interval := types.Interval(arg.Interval) + intervalAdder, ok := types.SupportedIntervals[interval] + if !ok { + err = fmt.Errorf("interval not support: %s", arg.Interval) + return + } + + // todo 交易产品初始化完成检查 + exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId) + if !ok { + err = fmt.Errorf("trade instance not support for exchange: %s for %s", arg.InstId, arg.Exchange) + return + } // 限制最大时间范围 - total := (afterTs-beforeTs)/intervalAdder(0, 1) + 1 + afterTs, beforeTs, total, err := svc.CalcSeriesRange(arg) + if err != nil { + return + } if total > MaxHistoryKlines { err = fmt.Errorf("time range too large max %d", MaxHistoryKlines) return } + if arg.Limit > 0 && total > int64(arg.Limit) { + err = fmt.Errorf("time range %d out of limit %d", total, arg.Limit) + return + } klines, err = svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, beforeTs, afterTs) if err != nil { @@ -742,7 +777,7 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory return } // 检查k线是否连续进行补齐 - if err = svc.paddingKlinesIfNotSeries(exchange, req.InstId, interval, klines); err != nil { + if err = svc.paddingKlinesIfNotSeries(exchange, arg.InstId, interval, klines); err != nil { return } @@ -759,101 +794,46 @@ func (svc *ExchangeService) HistoryKline(ctx context.Context, req *pb.ReqHistory } // 降序排序 - if req.Desc { + if arg.Desc { collect.Reverse(klines) } // 实时k线 - if req.Live && len(klines) > 0 { + if arg.Live && len(klines) > 0 { liveK := exchangeInst.LiveKline.Get(interval) if latest := intervalAdder(lastK.Ts, 1) == liveK.Ts; latest { - if req.Desc { + if arg.Desc { klines = append([]*types.Kline{&liveK}, klines...) } else { klines = append(klines, &liveK) } - rsp.Live = true + live = true } } return } // 查询历史k线(按时间升序流式返回) -func (svc *ExchangeService) HistoryKlineStream(req *pb.ReqHistoryKlineStream, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) { - // 交易产品参数检查 - if !svc.exchanges.IsSupport(req.Exchange) { - err = fmt.Errorf("exchange not support: %s", req.Exchange) - return - } - exchange := svc.exchanges.Get(req.Exchange) - exchangeInstId, ok := exchange.TradeInstIds.Load(req.InstId) - if !ok { - err = fmt.Errorf("trade instance not support: %s", req.InstId) - return - } - interval := types.Interval(req.Interval) - intervalAdder, ok := types.SupportedIntervals[interval] - if !ok { - err = fmt.Errorf("interval not support: %s", req.Interval) - return - } - // todo 交易产品初始化完成检查 - exchangeInst, ok := exchange.ExchangeInsts.Load(exchangeInstId) - if !ok { - err = fmt.Errorf("trade instance not support for exchange: %s for %s", req.InstId, req.Exchange) - return - } - - // k线长度检查 - afterTs, beforeTs, count := int64(req.After), int64(req.Before), int64(req.Count) - if count == 0 { - count = 100 - } - liveK := exchangeInst.LiveKline.Get(interval) - lastTs := liveK.Ts - if !liveK.Confirm { - lastTs = intervalAdder(liveK.Ts, -1) - } - if afterTs == 0 && beforeTs == 0 { - beforeTs = max(intervalAdder(lastTs, -count+1), KlineBefore0) - afterTs = lastTs - } - if afterTs == 0 { - afterTs = min(intervalAdder(beforeTs, count-1), lastTs) - } - if beforeTs == 0 { - beforeTs = max(intervalAdder(afterTs, -count+1), KlineBefore0) - } - if beforeTs > afterTs { - err = fmt.Errorf("time range invalid: before must less then after") - return - } - - klines, err := svc.exchangeDataPersist.ListKline(*exchangeInst.Inst, interval, beforeTs, afterTs) +func (svc *ExchangeService) HistoryKlineStream(arg *pb.SeriesRange, stream grpc.ServerStreamingServer[pb.RspHistoryKlineStream]) (err error) { + ctx := context.Background() + _, klines, err := svc.HistoryKline(ctx, arg) if err != nil { - zlog.Error("fetch history kline stream error: ", err) return } - if len(klines) == 0 { + + // 交易产品参数检查 + if !svc.exchanges.IsSupport(arg.Exchange) { + err = fmt.Errorf("exchange not support: %s", arg.Exchange) return } + exchange := svc.exchanges.Get(arg.Exchange) + interval := types.Interval(arg.Interval) // 检查k线是否连续进行补齐 - if err = svc.paddingKlinesIfNotSeries(exchange, req.InstId, interval, klines); err != nil { + if err = svc.paddingKlinesIfNotSeries(exchange, arg.InstId, interval, klines); err != nil { return } - lastK := klines[len(klines)-1] - // vmtsdb 数据刷盘30s延迟, 使用内存数据替代第一根k线 - lastConfirmK := exchangeInst.LastKline.Get(interval) - if lastConfirmK.Ts == lastK.Ts { - klines[len(klines)-1] = &lastConfirmK - lastK = klines[len(klines)-1] - } - if lastConfirmK.Ts == afterTs && intervalAdder(lastK.Ts, 1) == afterTs { - klines = append(klines, &lastConfirmK) - } - branch := 100 length := len(klines) kBuffer := make([]*pb.Kline, 0, branch) diff --git a/internal/trading/indicator_context.go b/internal/trading/indicator_context.go index 55816a7..1ccd98b 100644 --- a/internal/trading/indicator_context.go +++ b/internal/trading/indicator_context.go @@ -17,6 +17,7 @@ type IOffsetIndicatorContext interface { indicator.IIndicatorContext SetOffset(offset int16) AddOffset(offset int16) + GetOffset() (offset int16) } // IndicatorContext 指标上下文, 提供k线序列给指标计算使用 @@ -40,6 +41,10 @@ func (c *IndicatorContext) AddOffset(offset int16) { c.offset += offset } +func (c *IndicatorContext) GetOffset() (offset int16) { + return c.offset +} + func (c *IndicatorContext) Get(offset int16) (kline types.Kline) { offset += c.offset k, ok := c.kSeries.Get(offset) @@ -72,23 +77,18 @@ func NewHistoryIndicatorContext(exchangeClient pb.ExchangeServiceClient) *Histor exchangeClient: exchangeClient, } } -func (c *HistoryIndicatorContext) Init(exchange pb.ExchangeType, instId string, interval types.Interval, before, after int64) (totalK int, err error) { +func (c *HistoryIndicatorContext) Init(sr *pb.SeriesRange) (totalK int, err error) { // fetch history series req := &pb.ReqHistoryKlineStream{ - Exchange: exchange, - InstId: instId, - Interval: string(interval), - Count: 0, - Before: before, - After: after, + Series: sr, } stream, err := c.exchangeClient.HistoryKlineStream(context.Background(), req, grpc.UseCompressor("snappy")) if err != nil { - zlog.Errorf("fetch history kline stream error: instId=%s(%s), interval=%s, %#v, err=%v", instId, exchange, interval, req, err) + zlog.Errorf("fetch history kline stream error: instId=%s(%s), interval=%s, %#v, err=%v", sr.InstId, sr.Exchange, sr.Interval, req, err) return } - - klineSeries := NewKlineSeries(exchange, instId, interval) + interval := types.Interval(sr.Interval) + klineSeries := NewKlineSeries(sr.Exchange, sr.InstId, interval) for { msg, err0 := stream.Recv() if err0 == io.EOF { @@ -103,7 +103,7 @@ func (c *HistoryIndicatorContext) Init(exchange pb.ExchangeType, instId string, totalK += len(msg.Klines) for _, k := range msg.Klines { kline := new(types.Kline) - kline.ParsePBKline(exchange, k) + kline.ParsePBKline(sr.Exchange, k) if lastTs, ok := klineSeries.Update(kline); !ok { err = fmt.Errorf("history stream kline not series: last=%d", lastTs) return @@ -118,6 +118,14 @@ func (c *HistoryIndicatorContext) SetOffset(offset int16) { c.context.SetOffset(offset) } +func (c *HistoryIndicatorContext) AddOffset(offset int16) { + c.context.AddOffset(offset) +} + +func (c *HistoryIndicatorContext) GetOffset() (offset int16) { + return c.context.GetOffset() +} + func (c *HistoryIndicatorContext) Get(offset int16) (kline types.Kline) { return c.context.Get(offset) } diff --git a/internal/trading/kline_store.go b/internal/trading/kline_store.go index 5227661..b3e6454 100644 --- a/internal/trading/kline_store.go +++ b/internal/trading/kline_store.go @@ -213,12 +213,17 @@ func (s *KlineStore) inititalKlineSeries(exchange pb.ExchangeType, instId string func (s *KlineStore) fetchHistoryKlineToSeries(exchange pb.ExchangeType, instId, interval string, before, after int64, count uint32) (total int, err error) { // 拉取最新的1000条k线 req := &pb.ReqHistoryKlineStream{ - Exchange: exchange, - InstId: instId, - Interval: interval, - Count: count, - Before: before, - After: after, + Series: &pb.SeriesRange{ + Exchange: exchange, + InstId: instId, + Interval: interval, + Before: before, + After: after, + Count: count, + Open: false, + Live: false, + Desc: false, + }, } stream, err := s.exchangeClient.HistoryKlineStream(context.Background(), req, grpc.UseCompressor("snappy")) if err != nil { diff --git a/internal/trading/strategy_context.go b/internal/trading/strategy_context.go index 27d816e..85c6da1 100644 --- a/internal/trading/strategy_context.go +++ b/internal/trading/strategy_context.go @@ -2,12 +2,10 @@ package trading import ( "fmt" - "sig-pub/api/pb" "sig-pub/pkg/indicator" "sig-pub/pkg/strategy" "sig-pub/pkg/types" "sig-pub/pkg/types/series" - "sig-pub/pkg/zlog" ) type IOffsetStrategyContext interface { @@ -18,16 +16,13 @@ type IOffsetStrategyContext interface { type StrategyContext struct { IOffsetStrategyContext - indicatorContext *IndicatorContext + indicatorContext IOffsetIndicatorContext indicatorsReg *indicator.IndicatorRegistry - signal []pb.Side // 0.sell,1.buy - signalTimes []int64 - wins []bool } -func NewStrategyContext(klineSeries *KlineSeries, indicatorsReg *indicator.IndicatorRegistry) *StrategyContext { +func NewStrategyContext(indicatorContext IOffsetIndicatorContext, indicatorsReg *indicator.IndicatorRegistry) *StrategyContext { return &StrategyContext{ - indicatorContext: NewIndicatorContext(klineSeries), + indicatorContext: indicatorContext, indicatorsReg: indicatorsReg, } } @@ -44,36 +39,36 @@ func (c *StrategyContext) Series(offset, count int16) (klines series.Klines) { return c.indicatorContext.Series(offset, count) } -// Buy 发出多信号 -func (c *StrategyContext) Buy() { - zlog.Infof("signal buy: %d", c.Get(0).Ts) +// // Buy 发出多信号 +// func (c *StrategyContext) Buy() { +// zlog.Infof("signal buy: %d", c.Get(0).Ts) - c.signal = append(c.signal, pb.Side_BUY) - c.signalTimes = append(c.signalTimes, c.Get(0).Ts) - win := false - signalPrice := c.Get(0).Close - if c.indicatorContext.offset > 0 { - c.indicatorContext.AddOffset(-1) - win = c.Get(0).Close.Cmp(signalPrice) > 0 - c.indicatorContext.AddOffset(1) - } - c.wins = append(c.wins, win) -} +// c.signal = append(c.signal, pb.Side_BUY) +// c.signalTimes = append(c.signalTimes, c.Get(0).Ts) +// win := false +// signalPrice := c.Get(0).Close +// if c.indicatorContext.GetOffset() > 0 { +// c.indicatorContext.AddOffset(-1) +// win = c.Get(0).Close.Cmp(signalPrice) > 0 +// c.indicatorContext.AddOffset(1) +// } +// c.wins = append(c.wins, win) +// } -// Sell 发出空信号 -func (c *StrategyContext) Sell() { - zlog.Infof("signal sell: %d", c.Get(0).Ts) - c.signal = append(c.signal, pb.Side_SELL) - c.signalTimes = append(c.signalTimes, c.Get(0).Ts) - win := false - signalPrice := c.Get(0).Close - if c.indicatorContext.offset > 0 { - c.indicatorContext.AddOffset(-1) - win = c.Get(0).Close.Cmp(signalPrice) < 0 - c.indicatorContext.AddOffset(1) - } - c.wins = append(c.wins, win) -} +// // Sell 发出空信号 +// func (c *StrategyContext) Sell() { +// zlog.Infof("signal sell: %d", c.Get(0).Ts) +// c.signal = append(c.signal, pb.Side_SELL) +// c.signalTimes = append(c.signalTimes, c.Get(0).Ts) +// win := false +// signalPrice := c.Get(0).Close +// if c.indicatorContext.GetOffset() > 0 { +// c.indicatorContext.AddOffset(-1) +// win = c.Get(0).Close.Cmp(signalPrice) < 0 +// c.indicatorContext.AddOffset(1) +// } +// c.wins = append(c.wins, win) +// } // 获取窗口类型指标 func (c *StrategyContext) IndicatorW(name string, window int16) (s indicator.IIndicatorSeries) { diff --git a/internal/trading/trading_grpc_server.go b/internal/trading/trading_grpc_server.go index c170109..e3edc37 100644 --- a/internal/trading/trading_grpc_server.go +++ b/internal/trading/trading_grpc_server.go @@ -21,8 +21,13 @@ func (svr *TradingGrpcServer) Init() (err error) { } func (svr *TradingGrpcServer) IndicatorSeries(ctx context.Context, req *pb.ReqIndicatorSeries) (rsp *pb.RspIndicatorSeries, err error) { + matrix, times, err := svr.tradingService.IndicatorSeries(req.Indicator, req.Window, req.Series) + if err != nil { + return + } rsp = &pb.RspIndicatorSeries{} - err = svr.tradingService.IndicatorSeries(req, rsp) + rsp.Matrix = matrix + rsp.Times = times return } diff --git a/internal/trading/trading_plan_runner.go b/internal/trading/trading_plan_runner.go index 8086946..e501d9c 100644 --- a/internal/trading/trading_plan_runner.go +++ b/internal/trading/trading_plan_runner.go @@ -42,12 +42,12 @@ func (r *TradingPlan) Init() (err error) { // initSigStrategy 初始化多空信号策略 // buy/sell -> 过滤/风控 -> tradeStrategy -> closeStrategy -func (r *TradingPlan) InitSigStrategy(sigStrategy strategy.ISigStrategy, params strategy.SigStrategyParam, klineSeries *KlineSeries) (err error) { +func (r *TradingPlan) InitSigStrategy(sigStrategy strategy.ISigStrategy, params strategy.SigStrategyParam, sigIndCtx IOffsetIndicatorContext) (err error) { if err = sigStrategy.Init(params); err != nil { return } r.sigStrategy = sigStrategy - r.sigStrategyContext = NewStrategyContext(klineSeries, r.indicatorReg) + r.sigStrategyContext = NewStrategyContext(sigIndCtx, r.indicatorReg) return } diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index 05810c4..e4d80a6 100644 --- a/internal/trading/trading_service.go +++ b/internal/trading/trading_service.go @@ -138,112 +138,121 @@ func (svc *TradingService) runTradingPlan(plan *entity.TradePlan) (err error) { if err = tradingPlan.Init(); err != nil { return } - if err = tradingPlan.InitSigStrategy(sigStrategy, *sigStrategyParam, sigKlineSeries); err != nil { + sigIndCtx := NewIndicatorContext(sigKlineSeries) + if err = tradingPlan.InitSigStrategy(sigStrategy, *sigStrategyParam, sigIndCtx); err != nil { return } return } // IndicatorSeries 获取指标实时或历史序列数据, 闭区间 -func (svc *TradingService) IndicatorSeries(req *pb.ReqIndicatorSeries, rsp *pb.RspIndicatorSeries) (err error) { +func (svc *TradingService) IndicatorSeries(indicatorName string, window uint32, sr *pb.SeriesRange) (matrix []float64, times []int64, err error) { // indicatorName string, exchange pb.ExchangeType, instId string, interval types.Interval, window int - indicator, ok := svc.indicatorReg.IndicatorW(req.Indicator) + indicator, ok := svc.indicatorReg.IndicatorW(indicatorName) if !ok { - err = fmt.Errorf("indicator %s not exists", req.Indicator) + err = fmt.Errorf("indicator %s not exists", indicatorName) return } - interval := types.Interval(req.Interval) - intervalAdd, ok := types.SupportedIntervals[interval] + interval := types.Interval(sr.Interval) + _, ok = types.SupportedIntervals[interval] if !ok { err = fmt.Errorf("unsupport interval %s", interval) return } - if req.Count <= 0 { - req.Count = 100 - } - if req.Count > 0 { - // ... - } - - // todo trade instance status check - - before, after, count := req.Before, req.After, req.Count - var indCtx IOffsetIndicatorContext - // 查询实时指标数据 - if before == 0 && after == 0 { - klineSeries, err1 := svc.klineStore.GetKlineSeires(req.Exchange, req.InstId, interval) - if err1 != nil { - err = err1 - return - } - // recover todo out of range - indCtx = NewIndicatorContext(klineSeries) - } else { - // 查询历史指标数据 - // todo calc before after... - ctx := NewHistoryIndicatorContext(svc.exchangeClient) - if count := int32((after-before)/intervalAdd(0, 1) + 1); count > 100 { - } + // before, after, count := sr.Before, sr.After, sr.Count + // var indCtx IOffsetIndicatorContext + // // 查询实时指标数据 + // if before == 0 && after == 0 { + // klineSeries, err1 := svc.klineStore.GetKlineSeires(sr.Exchange, sr.InstId, interval) + // if err1 != nil { + // err = err1 + // return + // } + // // recover todo out of range + // indCtx = NewIndicatorContext(klineSeries) + // } - before = intervalAdd(before, int64(-req.Window)) // 多拉取窗口大小的k线数据 - totalK := 0 - if totalK, err = ctx.Init(req.Exchange, req.InstId, interval, before, after); err != nil { - return - } - count = int32(totalK) - req.Window - indCtx = ctx + // 查询历史指标数据 + sr.Window = window + indCtx := NewHistoryIndicatorContext(svc.exchangeClient) + totalK := 0 + if totalK, err = indCtx.Init(sr); err != nil { + return } + count := uint32(totalK) - window - rsp.Matrix = make([]float64, 0, req.Count) - rsp.Times = make([]int64, 0, req.Count) + matrix = make([]float64, 0, sr.Count) + times = make([]int64, 0, sr.Count) for i := range count { indCtx.SetOffset(int16(i)) - vector := indicator.Calculate(indCtx, int16(req.Window)) - rsp.Matrix = append(rsp.Matrix, vector) - rsp.Times = append(rsp.Times, indCtx.Get(0).Ts) + vector := indicator.Calculate(indCtx, int16(window)) + matrix = append(matrix, vector) + times = append(times, indCtx.Get(0).Ts) } return } +const ( + MaxIndicatorWindow = 128 +) + // StrategySeries 简单策略信号测试 func (svc *TradingService) StrategySeries(req *pb.ReqStrategySeries, rsp *pb.RspStrategySeries) (err error) { // sigStrategy - sigStrategy, ok := svc.strategyReg.NewSigStrategy(req.Strategy) + sigStrategy, ok := svc.strategyReg.NewSigStrategy(req.SigStrategy) if !ok { - err = fmt.Errorf("strategy %s not exists", req.Strategy) + err = fmt.Errorf("strategy %s not exists", req.SigStrategy) return } err = sigStrategy.Init(strategy.SigStrategyParam{ - Interval: types.Interval(req.Interval), + Interval: types.Interval(req.Series.Interval), Param: req.SigParam, }) if err != nil { return } - interval := types.Interval(req.Interval) - intervalAdd, ok := types.SupportedIntervals[interval] + interval := types.Interval(req.Series.Interval) + _, ok = types.SupportedIntervals[interval] if !ok { err = fmt.Errorf("unsupport interval %s", interval) return } - _ = intervalAdd - klineSeries, err1 := svc.klineStore.GetKlineSeires(req.Exchange, req.InstId, interval) - if err1 != nil { - err = err1 + // klineSeries, err1 := svc.klineStore.GetKlineSeires(req.Exchange, req.InstId, interval) + // if err1 != nil { + // err = err1 + // return + // } + // recover todo out of range + count, totalK := 0, 0 + indCtx := NewHistoryIndicatorContext(svc.exchangeClient) + req.Series.Window += MaxIndicatorWindow + if totalK, err = indCtx.Init(req.Series); err != nil { return } - // recover todo out of range - strategyCtx := NewStrategyContext(klineSeries, svc.indicatorReg) - for i := range req.Count { + count = totalK - MaxIndicatorWindow + + strategyCtx := NewStrategyContext(indCtx, svc.indicatorReg) + for i := range count { strategyCtx.SetOffset(int16(i)) - sigStrategy.Update(strategyCtx) + side := sigStrategy.Update(strategyCtx) + if side == pb.Side_BUY || side == pb.Side_SELL { + k := strategyCtx.Get(0) + rsp.Signal = append(rsp.Signal, side) + rsp.Times = append(rsp.Times, k.Ts) + + strategyCtx.indicatorContext.AddOffset(1) + nextK := strategyCtx.Get(0) + strategyCtx.indicatorContext.AddOffset(-1) + // win = c.Get(0).Close.Cmp(signalPrice) < 0 + rsp.Wins = strategyCtx.wins + // signal []pb.Side // 0.sell,1.buy + // signalTimes []int64 + // wins []bool + } } - rsp.Signal = strategyCtx.signal - rsp.Times = strategyCtx.signalTimes - rsp.Wins = strategyCtx.wins // 信号点胜率判断 wins := collect.Filter(rsp.Wins, func(_ int, win bool) bool { return win }) rsp.WinRate = float64(len(wins)) / float64(len(rsp.Wins)) diff --git a/pkg/strategy/gold_x.go b/pkg/strategy/gold_x.go index 621fb9f..d945680 100644 --- a/pkg/strategy/gold_x.go +++ b/pkg/strategy/gold_x.go @@ -1,6 +1,9 @@ package strategy -import "fmt" +import ( + "fmt" + "sig-pub/api/pb" +) // GoldX 金叉策略 type GoldX struct { @@ -37,7 +40,7 @@ func (s *GoldX) Init(param SigStrategyParam) (err error) { // 校验参数, 并 return } -func (s *GoldX) Update(ctx ISigStrategyContext) { +func (s *GoldX) Update(ctx ISigStrategyContext) (side pb.Side) { sma14 := ctx.IndicatorW("sma", s.short) sma28 := ctx.IndicatorW("sma", s.long) // 包装方法 crossover/crossunder @@ -46,9 +49,10 @@ func (s *GoldX) Update(ctx ISigStrategyContext) { crossover := s14[0] > s28[0] && s14[1] < s28[1] // 上穿 crossunder := s14[0] < s28[0] && s14[1] > s28[1] // 下穿 if crossover { - ctx.Buy() + return pb.Side_BUY } if crossunder { - ctx.Sell() + return pb.Side_SELL } + return } diff --git a/pkg/strategy/sig_strategy.go b/pkg/strategy/sig_strategy.go index 10c2f0e..a6e2dcf 100644 --- a/pkg/strategy/sig_strategy.go +++ b/pkg/strategy/sig_strategy.go @@ -15,7 +15,7 @@ type ISigStrategy interface { New() ISigStrategy Meta() StrategyMeta Init(param SigStrategyParam) (err error) // 校验参数, 并根据参数初始化策略 - Update(ctx ISigStrategyContext) + Update(ctx ISigStrategyContext) (side pb.Side) } type StrategyMeta struct { @@ -27,9 +27,6 @@ type StrategyMeta struct { // ISigStrategyContext 策略外部访问能力 // klineSeries, Indicator type ISigStrategyContext interface { - Buy() // 发出多信号 - Sell() // 发出空信号 - // Get [0]当前k线 Get(offset int16) types.Kline // Series [offset...end]