interval strategy

internal/trading/trading_service.go

@@ -213,7 +213,7 @@ func (svc *TradingService) IndicatorSeries(ctx context.Context, indicatorName st
 
 	// 查询历史指标数据
 	requiredSeries := int(indicator.RequiredSeries(int16(window)))
-	sr.WindowExtra = uint32(requiredSeries)
+	sr.WindowExtra = uint32(requiredSeries - 1)
 	kSeries := sig.NewKlineSeries(sr.Exchange, sr.InstId, types.Interval(sr.Interval))
 	indicatorContext := sig.NewIndicatorContext(kSeries)
 
@@ -263,6 +263,38 @@ func (svc *TradingService) StrategySeries(ctx context.Context, req *pb.ReqStrate
 		rsp.Signal, rsp.Times, err = svc.singleStrategySeries(ctx, sigStrategy.(strategy.ISingleSigStrategy), req.Series)
 	case strategy.SigStrategyTypeInterval:
 		rsp.Signal, rsp.Times, err = svc.intervalStrategySeries(ctx, sigStrategy.(strategy.IIntervalSigStrategy), req.Series)
+	default:
+		err = fmt.Errorf("unknown sig strategy type %v", sigStrategyType)
+	}
+	return
+}
+
+// singleStrategySeries 单周期策略
+func (svc *TradingService) singleStrategySeries(ctx context.Context, sigStrategy strategy.ISingleSigStrategy, sr *pb.SeriesRange) (signals []pb.Side, times []int64, err error) {
+	interval := types.Interval(sr.Interval)
+	requiredSeries := int(sigStrategy.RequiredSeries())
+	sr.WindowExtra = uint32(requiredSeries - 1)
+	kSeries := sig.NewKlineSeries(sr.Exchange, sr.InstId, interval)
+	indicatorContext := sig.NewIndicatorContext(kSeries)
+	strategyContext := sig.NewStrategyContext(indicatorContext, svc.indicatorReg)
+	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 kSeries.Length() < requiredSeries {
+			return
+		}
+		sigSide := sigStrategy.Update(strategyContext)
+		if sigSide.IsValid() {
+			side := lang.Ternary(sigSide == types.SideLong, pb.Side_BUY, pb.Side_SELL)
+			signals = append(signals, side)
+			times = append(times, k.Ts)
+		}
+		return
+	})
+	if err != nil {
+		return
 	}
 	return
 }
@@ -295,32 +327,30 @@ func (svc *TradingService) intervalStrategySeries(ctx context.Context, intervalS
 	for _, interval := range otherIntervals {
 		kSeries := sig.NewKlineSeries(sr.Exchange, sr.InstId, interval)
 		intervalKlineSeries.Set(interval, kSeries)
-
 		go func(interval types.Interval, kSeries *sig.KlineSeries) {
 			ch := otherIntervalCh.Get(interval)
 			srcCh := ch[0]
 			dstCh := ch[1]
 			intervalAdder := types.SupportedIntervals[interval]
 			driverTs := int64(0)
-			isr := &pb.SeriesRange{Exchange: sr.Exchange, InstId: sr.InstId, Before: sr.Before, After: sr.After, Count: sr.Count, Open: sr.Open, Live: sr.Live, Desc: sr.Desc, WindowExtra: sr.WindowExtra, Limit: sr.Limit,
+			isr := &pb.SeriesRange{Exchange: sr.Exchange, InstId: sr.InstId, Before: sr.Before, After: sr.After, Count: sr.Count, Open: sr.Open, Live: sr.Live, Desc: sr.Desc, Limit: sr.Limit}
-				Interval: string(interval)}
+			isr.Interval = string(interval)
+			isr.WindowExtra = uint32(requiredIntervalSeries.Get(interval) - 1)
 			err1 := svc.fetchHistoryKlineSeries(ctx, isr, func(k *types.Kline) (err error) {
 				closeTs := intervalAdder(k.Ts, 1)
 				// 与驱动周期series保持同步更新
 				if closeTs > driverTs {
+				waitLoop:
+					for {
 						if driverTs != 0 {
 							dstCh <- 0 // 通知更新完毕
 						}
-				waitLoop:
-					for {
 						select {
 						case <-stopCh:
 							return io.EOF
 						case driverTs = <-srcCh:
 							if closeTs <= driverTs {
 								break waitLoop
-							} else {
-								dstCh <- 0 // 通知更新完毕
 							}
 						}
 					}
@@ -331,9 +361,9 @@ func (svc *TradingService) intervalStrategySeries(ctx context.Context, intervalS
 				}
 				return
 			})
-
 			if err1 == nil {
-				otherIntervalCh.Set(interval, nil)
+				otherIntervalCh.Set(interval, nil) // 该周期数据拉取结束
+				dstCh <- 0                         // 通知更新完毕
 			} else if err1 != io.EOF {
 				zlog.Errorf("fetch history interval error: inst=%s(%s) interval=%s, err=%v", sr.InstId, sr.Exchange, interval, err1)
 				err = err1
@@ -348,17 +378,24 @@ func (svc *TradingService) intervalStrategySeries(ctx context.Context, intervalS
 	// 驱动周期数据拉取
 	driverSeries := sig.NewKlineSeries(sr.Exchange, sr.InstId, driverInterval)
 	intervalKlineSeries.Set(driverInterval, driverSeries)
+	sr.WindowExtra = uint32(requiredIntervalSeries.Get(driverInterval) - 1)
 	err = svc.fetchHistoryKlineSeries(ctx, sr, func(k *types.Kline) (err error) {
 		driverTS := driverIntervalAdder(k.Ts, 1)
-		// 通知其他周期先更新
 		otherIntervalCh.Range(func(interval types.Interval, ch []chan int64) {
 			if len(ch) == 2 {
+				// 通知其它周期先更新
 				select {
 				case <-stopCh:
 					err = io.EOF
 					return
 				case ch[0] <- driverTS:
-					<-ch[1] // 等待更新完毕
+					// 等待其它周期更新完毕
+					select {
+					case <-stopCh:
+						err = io.EOF
+						return
+					case <-ch[1]:
+					}
 				}
 			}
 		})
@@ -366,6 +403,7 @@ func (svc *TradingService) intervalStrategySeries(ctx context.Context, intervalS
 			return
 		}
 
+		zlog.Debugf("driver series update: %s, %d", driverInterval, driverTS)
 		if lastTs, serial := driverSeries.Update(k); !serial {
 			err = fmt.Errorf("kline not series: %s(%s), interval=%s, lastTs=%d", sr.InstId, sr.Exchange, driverInterval, lastTs)
 			return
@@ -392,7 +430,6 @@ func (svc *TradingService) intervalStrategySeries(ctx context.Context, intervalS
 			signals = append(signals, side)
 			times = append(times, k.Ts)
 		}
-		zlog.Debugf("strategy update finish ------------------------------------")
 		return
 	})
 	if err != io.EOF {
@@ -401,36 +438,6 @@ func (svc *TradingService) intervalStrategySeries(ctx context.Context, intervalS
 	return
 }
 
-// singleStrategySeries 单周期策略
-func (svc *TradingService) singleStrategySeries(ctx context.Context, sigStrategy strategy.ISingleSigStrategy, sr *pb.SeriesRange) (signals []pb.Side, times []int64, err error) {
-	interval := types.Interval(sr.Interval)
-	requiredSeries := int(sigStrategy.RequiredSeries())
-	sr.WindowExtra = uint32(requiredSeries)
-	kSeries := sig.NewKlineSeries(sr.Exchange, sr.InstId, interval)
-	indicatorContext := sig.NewIndicatorContext(kSeries)
-	strategyContext := sig.NewStrategyContext(indicatorContext, svc.indicatorReg)
-	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 kSeries.Length() < requiredSeries {
-			return
-		}
-		sigSide := sigStrategy.Update(strategyContext)
-		if sigSide.IsValid() {
-			side := lang.Ternary(sigSide == types.SideLong, pb.Side_BUY, pb.Side_SELL)
-			signals = append(signals, side)
-			times = append(times, k.Ts)
-		}
-		return
-	})
-	if err != nil {
-		return
-	}
-	return
-}
-
 // Backtest 回测交易计划
 func (svc *TradingService) Backtest(planId, stime, etime int64) (err error) {
 	plan, err := svc.tradingDataPersist.GetTradePlanById(planId)

pkg/indicator/rsi.go

@@ -22,7 +22,7 @@ func (c *RSI) RequiredSeries(window int16) int16 {
 // Calculate 计算单根k线rsi指标
 func (c *RSI) Calculate(ctx IIndicatorContext, window int16) (vector float64) {
 	// 读k线, 计算
-	klineSeries := ctx.Series(0, int16(window)) // 7根
+	klineSeries := ctx.Series(0, int16(window))
 	closeSeries := klineSeries.Close()
 	closeDiff := closeSeries.Diff()
 

pkg/strategy/cross_star.go

@@ -50,9 +50,9 @@ func (s *CrossStar) Update(ctx IIntervalSigStrategyContext) (side types.Side) {
 	// O 109744.8 H 110600 L 109507.5 C 109686.8
 	k0 := ctx.Get("5m", 0)
 	open, close, high, low := k0.OpenF64(), k0.CloseF64(), k0.HighF64(), k0.LowF64()
-	base := math.Abs(open - close)           // 58
+	base := math.Abs(open - close)
-	rup := (high - max(open, close)) / base  // 855.2 / 2 427.6
+	rup := (high - max(open, close)) / base
-	rdown := (min(open, close) - low) / base // 179.3 / 2 89.65
+	rdown := (min(open, close) - low) / base
 
 	if k0.Ts == 1761833700000 {
 		zlog.Debugf("base=%.4f, rup=%.4f, rdown=%.4f", base, rup, rdown)