diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index 985145c..f736e1b 100644 --- a/internal/trading/trading_service.go +++ b/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, - Interval: string(interval)} + 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} + 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 { - if driverTs != 0 { - dstCh <- 0 // 通知更新完毕 - } waitLoop: for { + if driverTs != 0 { + dstCh <- 0 // 通知更新完毕 + } 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) diff --git a/pkg/indicator/rsi.go b/pkg/indicator/rsi.go index f944a17..6f6df55 100644 --- a/pkg/indicator/rsi.go +++ b/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() diff --git a/pkg/strategy/cross_star.go b/pkg/strategy/cross_star.go index 114ff5e..a734268 100644 --- a/pkg/strategy/cross_star.go +++ b/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 - rup := (high - max(open, close)) / base // 855.2 / 2 427.6 - rdown := (min(open, close) - low) / base // 179.3 / 2 89.65 + base := math.Abs(open - close) + rup := (high - max(open, close)) / base + rdown := (min(open, close) - low) / base if k0.Ts == 1761833700000 { zlog.Debugf("base=%.4f, rup=%.4f, rdown=%.4f", base, rup, rdown)