diff --git a/internal/trading/backtest/sig_strategy_backtester.go b/internal/trading/backtest/sig_strategy_backtester.go index fa5d93d..17aea33 100644 --- a/internal/trading/backtest/sig_strategy_backtester.go +++ b/internal/trading/backtest/sig_strategy_backtester.go @@ -2,6 +2,7 @@ package backtest import ( "context" + "errors" "fmt" "io" "sig-pub/api/pb" @@ -12,7 +13,6 @@ import ( "sig-pub/pkg/utils/collect" "sig-pub/pkg/utils/times" "sig-pub/pkg/zlog" - "sync/atomic" "google.golang.org/grpc" ) @@ -66,7 +66,7 @@ func (b *SigStrategyBacktester) Backtest(ctx context.Context, sigStrategyInput t case strategy.SigStrategyTypeSingle: err = b.singleStrategySeries(ctx, b.sigStrategy.(strategy.ISingleSigStrategy), sigStrategyInput, sr, iiks, recvSignal) case strategy.SigStrategyTypeInterval: - err = b.intervalStrategySeries(ctx, b.sigStrategy.(strategy.IIntervalSigStrategy), sigStrategyInput, sr, iiks.GetIntervalKlineSeries(sr.InstId), recvSignal) + err = b.intervalStrategySeries(ctx, b.sigStrategy.(strategy.IIntervalSigStrategy), sigStrategyInput, sr, iiks, recvSignal) case strategy.SigStrategyTypeInstanceInterval: // todo default: @@ -77,21 +77,20 @@ func (b *SigStrategyBacktester) Backtest(ctx context.Context, sigStrategyInput t // singleStrategySeries 单周期策略 func (b *SigStrategyBacktester) singleStrategySeries(ctx context.Context, sigStrategy strategy.ISingleSigStrategy, sigStrategyInput types.Input, sr *pb.SeriesRange, iiks *types.InstanceIntervalKlineSeries, recvSignal func(sigSide types.Side, k types.Kline) (err error)) (err error) { - interval := types.Interval(sr.Interval) - intervalKlineSeries := iiks.GetIntervalKlineSeries(sr.InstId) - kSeries := intervalKlineSeries.ComputeIfAbsent(interval, func() *types.KlineSeries { return types.NewKlineSeries(sr.Exchange, sr.InstId, interval) }) - strategyContext := sig.NewStrategyContext(sigStrategyInput, kSeries, b.indicatorReg) - requiredSeries := int(sigStrategy.CandlePeriods(strategyContext)) + driverInstId := sr.InstId + driverInterval := types.Interval(sr.Interval) + driverSeries := iiks.Get(driverInstId, driverInterval) + strategyContext := sig.NewStrategyContext(sigStrategyInput, driverSeries, b.indicatorReg) + requiredPeriods := int(sigStrategy.CandlePeriods(strategyContext)) - requiredIntervalSeries := types.NewIntervalState[int16]() - requiredIntervalSeries.Set(interval, int16(max(1, requiredSeries))) + intervalCandlePeriods := types.NewIntervalState[int16]() + intervalCandlePeriods.Set(driverInterval, int16(max(1, requiredPeriods))) - err = b.multiIntervalSeries(ctx, sr, requiredIntervalSeries, intervalKlineSeries, func(driver bool, interval types.Interval, k *types.Kline) (err error) { - if !driver { + err = b.multiInstanceIntervalSeries(ctx, sr, []string{sr.InstId}, intervalCandlePeriods, iiks, func(driver bool, instId string, interval types.Interval, k *types.Kline) (err error) { + if !driver || instId != driverInstId || interval != driverInterval { return } - kSeries := intervalKlineSeries.Get(interval) - if kSeries.Length() < requiredSeries { + if driverSeries.Length() < requiredPeriods { return } sigSide := sigStrategy.Update(strategyContext) @@ -106,17 +105,21 @@ func (b *SigStrategyBacktester) singleStrategySeries(ctx context.Context, sigStr } // intervalStrategySeries 多周期策略 -func (b *SigStrategyBacktester) intervalStrategySeries(ctx context.Context, intervalSigStrategy strategy.IIntervalSigStrategy, sigStrategyInput types.Input, sr *pb.SeriesRange, intervalKlineSeries *types.IntervalState[*types.KlineSeries], recvSignal func(sigSide types.Side, k types.Kline) (err error)) (err error) { +func (b *SigStrategyBacktester) intervalStrategySeries(ctx context.Context, intervalSigStrategy strategy.IIntervalSigStrategy, sigStrategyInput types.Input, sr *pb.SeriesRange, iiks *types.InstanceIntervalKlineSeries, recvSignal func(sigSide types.Side, k types.Kline) (err error)) (err error) { + driverInstId := sr.InstId + driverInterval := types.Interval(sr.Interval) + intervalKlineSeries := iiks.GetIntervalKlineSeries(driverInstId) // 策略上下文 intervalStrategyContext := sig.NewIntervalStrategyContext(sigStrategyInput, intervalKlineSeries, b.indicatorReg) // 各周期所需k线数量 - requiredIntervalSeries := intervalSigStrategy.CandlePeriods(intervalStrategyContext) - err = b.multiIntervalSeries(ctx, sr, requiredIntervalSeries, intervalKlineSeries, func(driver bool, interval types.Interval, k *types.Kline) (err error) { - if !driver { + intervalCandlePeriods := intervalSigStrategy.CandlePeriods(intervalStrategyContext) + + err = b.multiInstanceIntervalSeries(ctx, sr, []string{sr.InstId}, intervalCandlePeriods, iiks, func(driver bool, instId string, interval types.Interval, k *types.Kline) (err error) { + if !driver || instId != driverInstId || interval != driverInterval { return } update := true - requiredIntervalSeries.Range(func(interval types.Interval, require int16) { + intervalCandlePeriods.Range(func(interval types.Interval, require int16) { if update && require > 0 { series := intervalKlineSeries.Get(interval) update = series.Length() >= int(require) @@ -133,200 +136,33 @@ func (b *SigStrategyBacktester) intervalStrategySeries(ctx context.Context, inte } return }) - return -} - -// multiIntervalSeries 多周期k线数据拉取 -// intervalKlineSeries: 各周期 series 从外部传入方便外部处理逻辑 -func (b *SigStrategyBacktester) multiIntervalSeries(ctx context.Context, sr *pb.SeriesRange, - requiredIntervalSeries *types.IntervalState[int16], - intervalKlineSeries *types.IntervalState[*types.KlineSeries], - recvFn func(driver bool, interval types.Interval, k *types.Kline) (err error), -) (err error) { - // 查询主周期时间范围 - rsp, err := b.exchangeClient.SeriesRange(ctx, &pb.ReqSeriesRange{Series: sr}) - if err != nil { - return - } - driverBefore, driverAfter := rsp.Before, rsp.After - - driverInstId := sr.InstId - driverInterval := types.Interval(sr.Interval) - driverIntervalAdder := types.SupportedIntervals[driverInterval] - - var otherIntervals []types.Interval - // 运行时周期 - requiredIntervalSeries.Range(func(interval types.Interval, window int16) { - if interval != driverInterval && window > 0 { - otherIntervals = append(otherIntervals, interval) - } - }) - // 运行时订阅周期 - for interval := range b.intervalSubscribe { - if interval != driverInterval && collect.NotIn(interval, otherIntervals...) { - otherIntervals = append(otherIntervals, interval) - } - } - // otherIntervals = collect.Filter(otherIntervals, func(_ int, interval types.Interval) bool { return interval != driverInterval }) - - // 通知其他周期更新的channel - otherIntervalSyncCh := types.NewIntervalState[chan int64]() - for _, interval := range otherIntervals { - otherIntervalSyncCh.Set(interval, make(chan int64)) - } - stopCh := make(chan struct{}) - stopChClosed := atomic.Bool{} - for _, interval := range otherIntervals { - kSeries := intervalKlineSeries.ComputeIfAbsent(interval, func() *types.KlineSeries { return types.NewKlineSeries(sr.Exchange, sr.InstId, interval) }) - go func(interval types.Interval, kSeries *types.KlineSeries) { - syncCh := otherIntervalSyncCh.Get(interval) - intervalAdder := types.SupportedIntervals[interval] - driverTs := int64(0) - isr := &pb.SeriesRange{Exchange: sr.Exchange, InstId: sr.InstId, Open: false, Live: sr.Live, Desc: sr.Desc} - isr.Before = intervalAdder(driverBefore, -1) - isr.After = driverAfter - isr.Interval = string(interval) - isr.WindowExtra = uint32(max(0, requiredIntervalSeries.Get(interval)-1)) + indicator.ApproCandles - err1 := b.fetchHistoryKlineSeries(ctx, isr, func(k *types.Kline) (err error) { - closeTs := intervalAdder(k.Ts, 1) - // 与驱动周期series保持同步更新 - if closeTs > driverTs { - waitLoop: - for { - if driverTs != 0 { - syncCh <- 0 // 通知更新完毕 - } - select { - case <-ctx.Done(): - return fmt.Errorf("kline series canceled") - case <-stopCh: - return io.EOF - case driverTs = <-syncCh: // 等待主周期通知更新 - if closeTs <= driverTs { - break waitLoop - } - } - } - } - 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 - } - // 回调周期订阅 - for _, subFn := range b.intervalSubscribe[interval] { - if err = subFn(driverInstId, interval, *k); err != nil { - return - } - } - // 运行时周期 - if requiredIntervalSeries.Get(interval) > 0 { - return recvFn(false, interval, k) - } - return - }) - if err1 == nil { - otherIntervalSyncCh.Set(interval, nil) // 该周期数据拉取结束 - syncCh <- 0 // 通知更新完毕 - } else if err1 != io.EOF { - zlog.Errorf("fetch interval history error: inst=%s(%s) interval=%s, err=%v", sr.InstId, sr.Exchange, interval, err1) - err = err1 - if stopChClosed.CompareAndSwap(false, true) { - close(stopCh) - } - } - }(interval, kSeries) - } - // 驱动周期数据拉取 - driverSeries := intervalKlineSeries.ComputeIfAbsent(driverInterval, func() *types.KlineSeries { return types.NewKlineSeries(sr.Exchange, sr.InstId, driverInterval) }) - sr.WindowExtra = max(sr.WindowExtra, uint32(max(0, requiredIntervalSeries.Get(driverInterval)-1))) + indicator.ApproCandles - err0 := b.fetchHistoryKlineSeries(ctx, sr, func(k *types.Kline) (err error) { - 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 - } - driverTS := driverIntervalAdder(k.Ts, 1) - otherIntervalSyncCh.RangeBreak(func(_ types.Interval, syncCh chan int64) bool { - if syncCh != nil { - syncCh <- driverTS // 通知其他周期更新到主周期时间 - select { - case <-syncCh: // 等待其它周期更新完毕 - case <-ctx.Done(): - err = fmt.Errorf("kline series canceled") - return false - case <-stopCh: - err = io.EOF - return false - } - } - return true - }) - if err != nil { - return - } - if k.Ts < driverBefore { - return - } - // 回调周期订阅 - for _, subFn := range b.intervalSubscribe[driverInterval] { - if err = subFn(driverInstId, driverInterval, *k); err != nil { - return - } - } - return recvFn(true, driverInterval, k) - }) - if err0 != io.EOF { - if stopChClosed.CompareAndSwap(false, true) { - close(stopCh) - } - if err0 != nil { - err = err0 - zlog.Errorf("fetch driver interval history error: inst=%s(%s) interval=%s, err=%v", sr.InstId, sr.Exchange, driverInterval, err0) - } - } + // err = b.multiIntervalSeries(ctx, sr, intervalCandlePeriods, intervalKlineSeries, func(driver bool, interval types.Interval, k *types.Kline) (err error) { + // if !driver { + // return + // } + // update := true + // intervalCandlePeriods.Range(func(interval types.Interval, require int16) { + // if update && require > 0 { + // series := intervalKlineSeries.Get(interval) + // update = series.Length() >= int(require) + // } + // }) + // if !update { + // return + // } + // sigSide := intervalSigStrategy.Update(intervalStrategyContext) + // if sigSide.IsValid() { + // if err = recvSignal(sigSide, *k); err != nil { + // return + // } + // } + // return + // }) return } -// fetchHistoryKlineSeries 请求k线数据流式处理 -func (b *SigStrategyBacktester) fetchHistoryKlineSeries(ctx context.Context, sr *pb.SeriesRange, recvFn func(k *types.Kline) error) (err error) { - // fetch history klines via stream - req := &pb.ReqHistoryKlineStream{Series: sr} - stream, err := b.exchangeClient.HistoryKlineStream(ctx, req, grpc.UseCompressor("snappy")) - if err != nil { - return - } - var msg *pb.RspHistoryKlineStream - recvTimes, recvTotal := 0, 0 - watch := times.NewWatch() -recvLoop: - for { - select { - case <-ctx.Done(): - err = ctx.Err() - return - default: - } - msg, err = stream.Recv() - if err == io.EOF { - err = nil - break - } - if err != nil { - break - } - recvTimes++ - recvTotal += len(msg.Klines) - for _, k := range msg.Klines { - kline := new(types.Kline) - kline.ParsePBKline(sr.Exchange, k) - if err = recvFn(kline); err != nil { - break recvLoop - } - } - } - zlog.Debugf("fetch history kline series: inst=%s(%s), interval=%s, recv=%d, total=%d, use %s", sr.InstId, sr.Exchange, sr.Interval, recvTimes, recvTotal, watch.ElapsedFmt(".")) - return -} +var errStop = errors.New("stop") // multiInstanceIntervalSeries 多币种多周期数据拉取 func (b *SigStrategyBacktester) multiInstanceIntervalSeries(ctx context.Context, sr *pb.SeriesRange, @@ -334,16 +170,16 @@ func (b *SigStrategyBacktester) multiInstanceIntervalSeries(ctx context.Context, iiks *types.InstanceIntervalKlineSeries, recvFn func(driver bool, instId string, interval types.Interval, k *types.Kline) (err error), ) (err error) { + driverInstId := sr.InstId + driverInterval := types.Interval(sr.Interval) + driverIntervalAdder := types.SupportedIntervals[driverInterval] + // 查询主周期时间范围 rsp, err := b.exchangeClient.SeriesRange(ctx, &pb.ReqSeriesRange{Series: sr}) if err != nil { return } - driverBefore, driverAfter := rsp.Before, rsp.After - - driverInstId := sr.InstId - driverInterval := types.Interval(sr.Interval) - driverIntervalAdder := types.SupportedIntervals[driverInterval] + driverBefore, driverAfter := rsp.Before, driverIntervalAdder(rsp.After, 1) // 运行时周期 fetchIntervals := []types.Interval{driverInterval} @@ -362,7 +198,7 @@ func (b *SigStrategyBacktester) multiInstanceIntervalSeries(ctx context.Context, fetchInsts := append([]string{driverInstId}, tradeInsts...) fetchInsts = collect.Uniq(fetchInsts) - // channels + // fetch kline series var otherSrs []*pb.SeriesRange for _, instId := range fetchInsts { for _, interval := range fetchIntervals { @@ -403,9 +239,9 @@ func (b *SigStrategyBacktester) multiInstanceIntervalSeries(ctx context.Context, } select { case <-ctx.Done(): - return io.EOF + return errStop case <-stopCh: - return io.EOF + return errStop case driverTs = <-syncCh: // 等待主周期通知更新 if closeTs <= driverTs { break waitLoop @@ -425,15 +261,15 @@ func (b *SigStrategyBacktester) multiInstanceIntervalSeries(ctx context.Context, } } // 运行时周期 - // if requiredIntervalSeries.Get(interval) > 0 { - // return recvFn(false, isr.InstId, interval, k) - // } + if intervalCandlePeriods.Get(interval) > 0 { + return recvFn(false, isr.InstId, interval, k) + } return }) + zlog.Debugf("other sr finish with: %s(%s), %v", isr.InstId, isr.Interval, err1) if err1 == nil { - syncCh <- 0 // 通知更新完毕 - // todo 后续不再更新 - } else if err1 != io.EOF { + syncCh <- -1 // 通知更新完毕, 后续不再更新 + } else if err1 != errStop { zlog.Errorf("fetch interval history error: inst=%s(%s) interval=%s, err=%v", sr.InstId, sr.Exchange, interval, err1) err = err1 close(stopCh) @@ -453,14 +289,21 @@ func (b *SigStrategyBacktester) multiInstanceIntervalSeries(ctx context.Context, return } driverTS := driverIntervalAdder(k.Ts, 1) - for _, syncCh := range syncChans { + for i, syncCh := range syncChans { + if syncCh == nil { + continue + } syncCh <- driverTS // 通知其他周期更新到主周期时间 select { - case <-syncCh: // 等待该周期更新完毕 + case sig := <-syncCh: // 等待该周期更新完毕 + if sig == -1 { + // 后续不再更新 + syncChans[i] = nil + } case <-ctx.Done(): - return io.EOF + return errStop case <-stopCh: - return io.EOF + return errStop } } if k.Ts < driverBefore { @@ -474,9 +317,10 @@ func (b *SigStrategyBacktester) multiInstanceIntervalSeries(ctx context.Context, } return recvFn(true, driverInstId, driverInterval, k) }) + zlog.Debugf("driver sr finish with: %s(%s), %v,", sr.InstId, sr.Interval, err0) if err0 == nil { close(stopCh) - } else if err0 != io.EOF { + } else if err0 != errStop { zlog.Errorf("fetch driver interval history error: inst=%s(%s) interval=%s, err=%v", sr.InstId, sr.Exchange, driverInterval, err) err = err0 close(stopCh) @@ -484,3 +328,44 @@ func (b *SigStrategyBacktester) multiInstanceIntervalSeries(ctx context.Context, } return } + +// fetchHistoryKlineSeries 请求k线数据流式处理 +func (b *SigStrategyBacktester) fetchHistoryKlineSeries(ctx context.Context, sr *pb.SeriesRange, recvFn func(k *types.Kline) error) (err error) { + // fetch history klines via stream + req := &pb.ReqHistoryKlineStream{Series: sr} + stream, err := b.exchangeClient.HistoryKlineStream(ctx, req, grpc.UseCompressor("snappy")) + if err != nil { + return + } + var msg *pb.RspHistoryKlineStream + recvTimes, recvTotal := 0, 0 + watch := times.NewWatch() +recvLoop: + for { + select { + case <-ctx.Done(): + err = ctx.Err() + return + default: + } + msg, err = stream.Recv() + if err == io.EOF { + err = nil + break + } + if err != nil { + break + } + recvTimes++ + recvTotal += len(msg.Klines) + for _, k := range msg.Klines { + kline := new(types.Kline) + kline.ParsePBKline(sr.Exchange, k) + if err = recvFn(kline); err != nil { + break recvLoop + } + } + } + zlog.Debugf("fetch history kline series: inst=%s(%s), interval=%s, recv=%d, total=%d, use %s", sr.InstId, sr.Exchange, sr.Interval, recvTimes, recvTotal, watch.ElapsedFmt(".")) + return +} diff --git a/internal/trading/sig/indicator_context.go b/internal/trading/sig/indicator_context.go index d0d87ee..dad7059 100644 --- a/internal/trading/sig/indicator_context.go +++ b/internal/trading/sig/indicator_context.go @@ -28,23 +28,30 @@ type IOffsetIndicatorContext interface { // IndicatorContext 指标上下文, 提供k线序列给指标计算使用 type IndicatorContext struct { IOffsetIndicatorContext - indicator indicator.IIndicator - indicatorsReg *indicator.IndicatorRegistry - input types.Input - indicatorStates map[string]*IndicatorState // {macd{window:0,fast:9,slow:21,single:10}: state} 初始化时与KlineSeries周期同步 - kSeries *types.KlineSeries - offset int16 - indicatorTrace []string // 指标调用链避免指标循环引用 + indicator indicator.IIndicator + indicatorStateKey string + indicatorsReg *indicator.IndicatorRegistry + input types.Input + indicatorStates map[string]*IndicatorState // {macd{window:0,fast:9,slow:21,single:10}: state} 初始化时与KlineSeries周期同步 + kSeries *types.KlineSeries + offset int16 + indicatorTrace []string // 指标调用链避免指标循环引用 } func NewIndicatorContext(indicator indicator.IIndicator, input types.Input, indicatorStates IndicatorStates, kSeries *types.KlineSeries, indicatorsReg *indicator.IndicatorRegistry) *IndicatorContext { + inputs := collect.Mapping(indicator.Meta().Input, func(in types.InputArg) string { + return input.String(in.Name) + }) + stateKey := fmt.Sprintf("%s{%s}", indicator.Meta().Name, strings.Join(inputs, ",")) + return &IndicatorContext{ - indicator: indicator, - input: input, - indicatorStates: indicatorStates, - kSeries: kSeries, - indicatorsReg: indicatorsReg, - indicatorTrace: []string{indicator.Meta().Name}, + indicator: indicator, + indicatorStateKey: stateKey, + input: input, + indicatorStates: indicatorStates, + kSeries: kSeries, + indicatorsReg: indicatorsReg, + indicatorTrace: []string{indicator.Meta().Name}, } } @@ -89,15 +96,10 @@ func (c *IndicatorContext) Series(offset, count int16) (klines types.Klines) { // 窗口/参数 // tradingPlan -> interval -> context -> {macd{window:0,fast:9,slow:21,single:10}: state, ema21:state} -> state[[ema]{1.1, 1.2}, [ema]{1.3, 1.4}] func (c *IndicatorContext) State() indicator.IIndicatorState { - inputs := collect.Mapping(c.indicator.Meta().Input, func(in types.InputArg) string { - return c.Input().String(in.Name) - }) - stateKey := fmt.Sprintf("%s{%s}", c.indicator.Meta().Name, strings.Join(inputs, ",")) - - state, ok := c.indicatorStates[stateKey] + state, ok := c.indicatorStates[c.indicatorStateKey] if !ok { state = NewIndicatorState(c.kSeries.Interval) - c.indicatorStates[stateKey] = state + c.indicatorStates[c.indicatorStateKey] = state // 从头KlineSeries跑一遍, 针对ema,macd等回溯迭代指标, 将state与KlineSeries对齐 c.backtrackIndicatorState(c.indicator) } diff --git a/internal/trading/sig/indicator_state.go b/internal/trading/sig/indicator_state.go index f8559af..c5d41e1 100644 --- a/internal/trading/sig/indicator_state.go +++ b/internal/trading/sig/indicator_state.go @@ -46,10 +46,10 @@ func (s *IndicatorState) ring(k string) *types.RingSeries[float64] { func (s *IndicatorState) Set(k string, v float64) { ts := s.indicatorContext.Get(0).Ts - if s.lastTs[k] < ts { + if lastTs := s.lastTs[k]; lastTs < ts { // panic可替换为丢失指标用前一个值填充类似vmtsdb - if expectTs := s.intervalAdder(s.lastTs[k], 1); expectTs != ts && s.lastTs[k] != 0 { - panic(fmt.Errorf("state 不连续: lastTs=%d, got=%d, expected=%d", s.lastTs[k], ts, expectTs)) + if expectTs := s.intervalAdder(lastTs, 1); expectTs != ts && lastTs != 0 { + panic(fmt.Errorf("state 不连续: lastTs=%d, got=%d, expected=%d", lastTs, ts, expectTs)) } s.ring(k).Push(v) s.lastTs[k] = ts diff --git a/pkg/strategy/super_trend_macd_rsi.go b/pkg/strategy/super_trend_macd_rsi.go index fec299a..1cd07bf 100644 --- a/pkg/strategy/super_trend_macd_rsi.go +++ b/pkg/strategy/super_trend_macd_rsi.go @@ -41,9 +41,13 @@ func (s *SuperTrendMacdRSI) CandlePeriods(ctx ISingleSigStrategyContext) int16 { func (s *SuperTrendMacdRSI) Update(ctx ISingleSigStrategyContext) (side types.Side) { superTrend := ctx.Indicator("SuperTrend", types.Input{"window": 10, "mul": 3}) rsi := ctx.Indicator("RSI", 14).Get(0) - macdHist := ctx.Indicator("Macd", types.Input{"fast": 12, "slow": 26, "singal": 9}).Get(0) // macd柱 - macdDea := ctx.Indicator("MacdDEA", types.Input{"fast": 12, "slow": 26, "singal": 9}).Series(0, 2) // macd_dea信号线 - macdDif := ctx.Indicator("MacdDIF", types.Input{"fast": 12, "slow": 26, "singal": 9}).Series(0, 2) // macd_dif线 + macd := ctx.Indicator("MACD", types.Input{"fast": 12, "slow": 26, "singal": 9}) + macdHist := macd.Get(0) + macdDea := macd.StateSeries("dea", 0, 2) // macd_dea信号线 + macdDif := macd.StateSeries("dif", 0, 2) // macd_dif线 + // macdHist := ctx.Indicator("Macd", types.Input{"fast": 12, "slow": 26, "singal": 9}).Get(0) // macd柱 + // macdDea := ctx.Indicator("MacdDEA", types.Input{"fast": 12, "slow": 26, "singal": 9}).Series(0, 2) // macd_dea信号线 + // macdDif := ctx.Indicator("MacdDIF", types.Input{"fast": 12, "slow": 26, "singal": 9}).Series(0, 2) // macd_dif线 crossover := macdDif[0] > macdDea[0] && macdDif[1] < macdDea[1] // 金叉 // crossunder := macdDif[0] < macdDea[0] && macdDif[1] > macdDea[1] // 死叉