trading indicator/strategy series

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;

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则返回错误
+}

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<string,string> sigParam = 15; // 策略参数
+    SeriesRange series = 1;
+    string sigStrategy = 2;
+    map<string,string> sigParam = 3; // 策略参数
 }
 message RspStrategySeries {
     repeated Side signal = 1; // 0.sell,1.buy

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

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
 }

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 arg.Before != 0 {
+			before = min(intervalAdder(before, 1), after)
+			if arg.After == 0 {
+				after = min(intervalAdder(before, 1), lastTs)
 			}
-		if req.Before != 0 {
-			beforeTs = min(intervalAdder(beforeTs, 1), afterTs)
-			if req.After == 0 {
-				afterTs = min(intervalAdder(beforeTs, 1), lastTs)
 		}
 	}
+	// 额外拉取
+	if arg.Window > 0 {
+		before = max(intervalAdder(before, -int64(arg.Window)), KlineBefore0)
 	}
-	if beforeTs > afterTs {
+	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)

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)
 }

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{
+		Series: &pb.SeriesRange{
 			Exchange: exchange,
 			InstId:   instId,
 			Interval: interval,
-		Count:    count,
 			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 {

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) {

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
 }
 

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
 }
 

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 := 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, 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 = intervalAdd(before, int64(-req.Window)) // 多拉取窗口大小的k线数据
+	sr.Window = window
+	indCtx := NewHistoryIndicatorContext(svc.exchangeClient)
 	totalK := 0
-		if totalK, err = ctx.Init(req.Exchange, req.InstId, interval, before, after); err != nil {
+	if totalK, err = indCtx.Init(sr); err != nil {
 		return
 	}
-		count = int32(totalK) - req.Window
-		indCtx = ctx
-	}
+	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)
-	}
-	rsp.Signal = strategyCtx.signal
-	rsp.Times = strategyCtx.signalTimes
+		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
+		}
+	}
 	// 信号点胜率判断
 	wins := collect.Filter(rsp.Wins, func(_ int, win bool) bool { return win })
 	rsp.WinRate = float64(len(wins)) / float64(len(rsp.Wins))

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
 }

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]