stateful indicator

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://192.168.1.5:7890"
-httpProxy = "http://10.255.183.209:7890"
+# httpProxy = "http://10.255.183.209:7890"
 
 # 模拟盘API交易地址如下：
 # REST：https://www.okx.com

internal/exchange/exchange_service.go

@@ -734,7 +734,7 @@ func (svc *ExchangeService) CalcSeriesRange(arg *pb.SeriesRange) (after, before,
 	}
 	// 额外拉取
 	if arg.WindowExtra > 0 {
-		if arg.WindowExtra > indicator.MaxWindow {
+		if arg.WindowExtra > indicator.MaxWindow+indicator.ApproCandles {
 			err = fmt.Errorf("series range window extra %d big then %d", arg.WindowExtra, indicator.MaxWindow)
 			return
 		}

internal/trading/backtest/sig_strategy_backtester.go

@@ -141,7 +141,7 @@ func (b *SigStrategyBacktester) multiIntervalSeries(ctx context.Context, sr *pb.
 	if err != nil {
 		return
 	}
-	before, after := rsp.Before, rsp.After
+	driverBefore, driverAfter := rsp.Before, rsp.After
 
 	driverInterval := types.Interval(sr.Interval)
 	driverIntervalAdder := types.SupportedIntervals[driverInterval]
@@ -175,10 +175,10 @@ func (b *SigStrategyBacktester) multiIntervalSeries(ctx context.Context, sr *pb.
 			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(before, -1)
+			isr.Before = intervalAdder(driverBefore, -1)
-			isr.After = after
+			isr.After = driverAfter
 			isr.Interval = string(interval)
-			isr.WindowExtra = uint32(max(0, requiredIntervalSeries.Get(interval)-1))
+			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保持同步更新
@@ -232,7 +232,7 @@ func (b *SigStrategyBacktester) multiIntervalSeries(ctx context.Context, sr *pb.
 	}
 	// 驱动周期数据拉取
 	driverSeries := intervalKlineSeries.ComputeIfAbsent(driverInterval, func() *sig.KlineSeries { return sig.NewKlineSeries(sr.Exchange, sr.InstId, driverInterval) })
-	sr.WindowExtra = max(sr.WindowExtra, uint32(max(0, requiredIntervalSeries.Get(driverInterval)-1)))
+	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)
@@ -257,6 +257,9 @@ func (b *SigStrategyBacktester) multiIntervalSeries(ctx context.Context, sr *pb.
 		if err != nil {
 			return
 		}
+		if k.Ts < driverBefore {
+			return
+		}
 		// intervalKlineSeries.Range(func(interval types.Interval, v *sig.KlineSeries) {
 		// 	if v != nil {
 		// 		zlog.Debugf("interval series update: %s, %d", interval, v.Length())
@@ -324,31 +327,3 @@ 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
 }
-
-// Deprecated
-// _singleStrategySeries 单周期策略
-func (b *SigStrategyBacktester) _singleStrategySeries(ctx context.Context, sigStrategy strategy.ISingleSigStrategy, sr *pb.SeriesRange, recvSignal func(sigSide types.Side, k types.Kline) (err error)) (err error) {
-	interval := types.Interval(sr.Interval)
-	kSeries := sig.NewKlineSeries(sr.Exchange, sr.InstId, interval)
-	strategyContext := sig.NewStrategyContext(nil, kSeries, b.indicatorReg)
-
-	requiredSeries := int(sigStrategy.CandlePeriods(nil))
-	sr.WindowExtra = uint32(max(0, requiredSeries-1))
-	err = b.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() {
-			if err = recvSignal(sigSide, *k); err != nil {
-				return
-			}
-		}
-		return
-	})
-	return
-}

internal/trading/sig/indicator_context.go

@@ -97,27 +97,26 @@ func (c *IndicatorContext) State() indicator.IIndicatorState {
 
 	state, ok := c.indicatorStates[stateKey]
 	if !ok {
-		state = NewIndicatorState(c.kSeries)
+		state = NewIndicatorState(c.kSeries.Interval)
 		c.indicatorStates[stateKey] = state
-		// 从头KlineSeries跑一遍，针对ema,macd等回溯迭代指标
+		// 从头KlineSeries跑一遍, 针对ema,macd等回溯迭代指标, 将state与KlineSeries对齐
 		c.backtrackIndicatorState(c.indicator)
 	}
+	state.SetIndicatorContext(c)
 	return state
 }
 
-// 当首次初始化某个state dea后, 把indicator在klineSeries从头跑一遍
+// 当首次初始化某个state后, 把indicator在klineSeries从头跑一遍
 func (c *IndicatorContext) backtrackIndicatorState(indicator indicator.IIndicator) {
 	_offset := c.offset
 	defer c.SetOffset(_offset)
 
-	previousCandles := indicator.CandlePeriods(c)
+	candleLength := c.kSeries.Length()
-	length := c.kSeries.Length()
+	candlePeriods := indicator.CandlePeriods(c)
-	for i := range length {
+	for i := int(candlePeriods) - 1; i < candleLength; i++ {
-		if i < int(previousCandles) {
+		offset := int16(candleLength - i - 1)
-			continue
-		}
-		offset := int16(length - 1 - i)
 		c.SetOffset(offset)
+		// 计算指标值，指标内部会更新state
 		indicator.Calculate(c)
 	}
 }
@@ -135,7 +134,9 @@ func (c *IndicatorContext) Indicator(name string, args ...any) (series indicator
 	}
 
 	input := matchIndicatorArgs(args...)
+
 	indicatorContext := NewIndicatorContext(indicator, input, c.indicatorStates, c.kSeries, c.indicatorsReg)
+	indicatorContext.offset = c.offset
 	indicatorContext.indicatorTrace = append(c.indicatorTrace, name)
 	return NewWindowIndicatorSeries(indicator, indicatorContext)
 }
@@ -152,7 +153,7 @@ inputLoop:
 windowLoop:
 	for _, arg := range args {
 		switch v := arg.(type) {
-		case int16, int, int32, int64:
+		case int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64:
 			window := cast.ToInt16(v)
 			if input == nil {
 				input = make(types.Input)

internal/trading/sig/indicator_state.go

@@ -1,6 +1,7 @@
 package sig
 
 import (
+	"fmt"
 	"sig-pub/pkg/indicator"
 	"sig-pub/pkg/types"
 	"sig-pub/pkg/types/series"
@@ -10,38 +11,69 @@ import (
 // ema, obv 指标递归计算时的状态存储
 type IndicatorState struct {
 	indicator.IIndicatorState
-	kSeries *KlineSeries
+	indicatorContext indicator.IIndicatorContext
-	state   map[string]*types.RingSeries[float64]
+	interval         types.Interval
+	intervalAdder    types.IntervalAdder
+	state            map[string]*types.RingSeries[float64]
+	lastTs           int64
 }
 
-func NewIndicatorState(kSeries *KlineSeries) *IndicatorState {
+func NewIndicatorState(interval types.Interval) *IndicatorState {
+	intervalAdder, ok := types.SupportedIntervals[interval]
+	if !ok {
+		panic(fmt.Errorf("unsupport interval: %s", interval))
+	}
 	return &IndicatorState{
-		kSeries: kSeries,
+		interval:      interval,
-		state:   make(map[string]*types.RingSeries[float64]),
+		intervalAdder: intervalAdder,
+		state:         make(map[string]*types.RingSeries[float64]),
 	}
 }
 
+func (s *IndicatorState) SetIndicatorContext(indicatorContext indicator.IIndicatorContext) {
+	s.indicatorContext = indicatorContext
+}
+
 func (s *IndicatorState) ring(k string) *types.RingSeries[float64] {
 	ring, ok := s.state[k]
 	if !ok {
 		ring = types.NewRingSeries[float64](indicator.MaxWindow, 8)
 		s.state[k] = ring
-		// 从 kSeries0 开始 calc ind 初始化
-		// 递归初始值
 	}
 	return ring
 }
 
 func (s *IndicatorState) Set(k string, v float64) {
-	s.ring(k).Push(v)
+	ts := s.indicatorContext.Get(0).Ts
+	if s.lastTs < ts {
+		if expectTs := s.intervalAdder(s.lastTs, 1); expectTs != ts && s.lastTs != 0 {
+			panic(fmt.Errorf("state 不连续: lastTs=%d, got=%d, expected=%d", s.lastTs, ts, expectTs))
+		}
+		s.ring(k).Push(v)
+		s.lastTs = ts
+	}
 }
 
 func (s *IndicatorState) Get(k string, offset int16) (v float64, ok bool) {
-	offset -= 1
+	ts := s.indicatorContext.Get(0).Ts
-	return s.ring(k).Get(int(offset))
+	target := s.intervalAdder(ts, -int64(offset))
+	ring := s.ring(k)
+	for i := 0; i < ring.Length(); i++ {
+		if s.intervalAdder(s.lastTs, -int64(i)) == target {
+			return ring.Get(i)
+		}
+	}
+	return
 }
 
 func (s *IndicatorState) Series(k string, offset, count int16) (v series.Floats, ok bool) {
-	offset -= 1
+	ts := s.indicatorContext.Get(0).Ts
-	return s.ring(k).Series(int(offset), int(count))
+	target := s.intervalAdder(ts, -int64(offset))
+	ring := s.ring(k)
+	for i := 0; i < ring.Length(); i++ {
+		if s.intervalAdder(s.lastTs, -int64(i)) == target {
+			return ring.Series(i, int(count))
+		}
+	}
+	return
 }

internal/trading/trading_service.go

@@ -200,6 +200,7 @@ func (svc *TradingService) fetchHistoryKlineSeries(ctx context.Context, sr *pb.S
 
 // IndicatorSeries 获取指标实时或历史序列数据, 闭区间
 func (svc *TradingService) IndicatorSeries(ctx context.Context, indicatorName string, digit int32, input types.Input, sr *pb.SeriesRange) (matrix []float64, times []int64, err error) {
+	appros := indicator.ApproCandles
 	indicator, ok := svc.indicatorReg.Indicator(indicatorName)
 	if !ok {
 		err = fmt.Errorf("indicator %s not exists", indicatorName)
@@ -218,7 +219,14 @@ func (svc *TradingService) IndicatorSeries(ctx context.Context, indicatorName st
 
 	candlePeriods := int(indicator.CandlePeriods(indicatorContext))
 	sr.Desc = false
-	sr.WindowExtra = uint32(max(0, candlePeriods-1))
+
+	// 计算第一个指标值需要多取candlePeriods - 1根k线; 计算ema等递归指标需要多拉取appros根k线逼近值
+	srRsp, err := svc.exchangeClient.SeriesRange(ctx, &pb.ReqSeriesRange{Series: sr})
+	if err != nil {
+		return
+	}
+	srBefore := srRsp.Before
+	sr.WindowExtra = uint32(max(0, candlePeriods-1)) + uint32(appros)
 
 	// 保留小数位数
 	digit = lang.Ternary(digit > 0 && digit <= 10, digit, 6)
@@ -230,6 +238,9 @@ func (svc *TradingService) IndicatorSeries(ctx context.Context, indicatorName st
 			err = fmt.Errorf("kline not series: %s(%s), interval=%s, lastTs=%d", sr.InstId, sr.Exchange, interval, lastTs)
 			return
 		}
+		if k.Ts < srBefore {
+			return
+		}
 		if kSeries.Length() < candlePeriods {
 			return
 		}

pkg/indicator/ema.go

@@ -24,7 +24,7 @@ func (c *EMA) CandlePeriods(ctx IIndicatorContext) int16 {
 // Calculate 计算单根k线sma指标
 func (c *EMA) Calculate(ctx IIndicatorContext) (vector float64) {
 	window := ctx.Input().Int16("window")
-	prevEma, ok := ctx.State().Get("ema", 1)
+	prevEma, ok := ctx.State().Get("vector", 1)
 	if !ok {
 		// 初始值用 sma 替代
 		prevEma = ctx.Series(1, window).Close().Avg()
@@ -35,6 +35,6 @@ func (c *EMA) Calculate(ctx IIndicatorContext) (vector float64) {
 	vector = multiplier*close + (1-multiplier)*prevEma
 	// same: vector = ((close - prevEma) * multiplier) + prevEma
 
-	ctx.State().Set("ema", vector)
+	ctx.State().Set("vector", vector)
 	return
 }

pkg/indicator/indicator.go

@@ -6,7 +6,8 @@ import (
 )
 
 const (
-	MaxWindow = 256
+	MaxWindow    = 256 // 最大窗口引用大小
+	ApproCandles = 64  // 递归指标近似计算k线数
 )
 
 type IndicatorMeta struct {

pkg/indicator/indicator_registry.go

@@ -22,9 +22,9 @@ func (r *IndicatorRegistry) Init() (err error) {
 	r.MustRegistIndicatorW(&SMA{})
 	r.MustRegistIndicatorW(&ATR{})
 	r.MustRegistIndicatorW(&EMA{})
-	r.MustRegistIndicatorW(&MACD{})
+	r.MustRegistIndicatorW(&MacdDIF{})
 	r.MustRegistIndicatorW(&MacdDEA{})
-	r.MustRegistIndicatorW(&MacdHist{})
+	r.MustRegistIndicatorW(&Macd{})
 	r.MustRegistIndicatorW(&OBV{})
 	r.MustRegistIndicatorW(&WOBV{})
 	return

pkg/indicator/macd.go

@@ -4,16 +4,46 @@ import (
 	"sig-pub/pkg/types"
 )
 
-// MACD 拆分成: MACD线, MacdDEA(信号线), MacdHist(柱状图)
+// Macd macd柱状图计算
-// 计算 MACD 线 (DIF): 反映短期趋势与长期趋势的“收敛/散度”
+
-// MACD: https://www.investopedia.com/terms/m/macd.asp
+// Macd 拆分成: Macd(柱状图), MacdDIF线, MacdDEA(信号线)
-type MACD struct {
+// 计算 MacdDIF 线 (DIF): 反映短期趋势与长期趋势的“收敛/散度”
+// Macd: https://www.investopedia.com/terms/m/macd.asp
+type Macd struct {
 }
 
-// indicator interface
+func (c *Macd) Meta() IndicatorMeta {
-func (c *MACD) Meta() IndicatorMeta {
 	return IndicatorMeta{
 		Name: "macd",
+		Input: []types.InputArg{
+			{Name: "fast", Type: types.InputTypeUInt, Desc: "快线周期"},
+			{Name: "slow", Type: types.InputTypeUInt, Desc: "慢线周期"},
+			{Name: "singal", Type: types.InputTypeUInt, Desc: "信号线周期"},
+		},
+	}
+}
+
+func (c *Macd) CandlePeriods(ctx IIndicatorContext) int16 {
+	return max(
+		ctx.Indicator("macd_dif", ctx.Input()).CandlePeriods(),
+		ctx.Indicator("macd_dea", ctx.Input()).CandlePeriods(),
+	)
+}
+
+func (c *Macd) Calculate(ctx IIndicatorContext) (vector float64) {
+	macd_dea := ctx.Indicator("macd_dea", ctx.Input()).Get(0)
+	macd_dif := ctx.Indicator("macd_dif", ctx.Input()).Get(0)
+	vector = (macd_dif - macd_dea) * 2
+	return
+}
+
+type MacdDIF struct {
+}
+
+// indicator interface
+func (c *MacdDIF) Meta() IndicatorMeta {
+	return IndicatorMeta{
+		Name: "macd_dif",
 		Input: []types.InputArg{
 			{Name: "fast", Type: types.InputTypeUInt, Desc: "快线周期"},
 			{Name: "slow", Type: types.InputTypeUInt, Desc: "慢线周期"},
@@ -21,7 +51,7 @@ func (c *MACD) Meta() IndicatorMeta {
 	}
 }
 
-func (c *MACD) CandlePeriods(ctx IIndicatorContext) int16 {
+func (c *MacdDIF) CandlePeriods(ctx IIndicatorContext) int16 {
 	return max(
 		ctx.Indicator("ema", ctx.Input().Int16("fast")).CandlePeriods(),
 		ctx.Indicator("ema", ctx.Input().Int16("slow")).CandlePeriods(),
@@ -29,7 +59,7 @@ func (c *MACD) CandlePeriods(ctx IIndicatorContext) int16 {
 }
 
 // Calculate 计算单根k线sma指标
-func (c *MACD) Calculate(ctx IIndicatorContext) (vector float64) {
+func (c *MacdDIF) Calculate(ctx IIndicatorContext) (vector float64) {
 	fast := ctx.Input().Int16("fast") // 12
 	slow := ctx.Input().Int16("slow") // 26
 
@@ -57,56 +87,24 @@ func (c *MacdDEA) Meta() IndicatorMeta {
 }
 
 func (c *MacdDEA) CandlePeriods(ctx IIndicatorContext) int16 {
-	return max(
+	return ctx.Indicator("macd_dif", ctx.Input()).CandlePeriods() + ctx.Input().Int16("singal") + 1
-		ctx.Input().Int16("singal")+1,
-		ctx.Indicator("macd", ctx.Input()).CandlePeriods(),
-	)
 }
 
 func (c *MacdDEA) Calculate(ctx IIndicatorContext) (vector float64) {
 	singal := ctx.Input().Int16("singal") // 9
 
-	deaPrev, ok := ctx.State().Get("macd_dea", 1)
+	deaPrev, ok := ctx.State().Get("vector", 1)
 	if !ok {
-		// 初始值前9期的 MACD SMA
+		// 初始值前9期的 MACD_DIF SMA
-		macdPrevs := ctx.Indicator("macd", ctx.Input()).Series(1, singal)
+		macdDifs := ctx.Indicator("macd_dif", ctx.Input()).Series(1, singal)
-		deaPrev = macdPrevs.Avg()
+		deaPrev = macdDifs.Avg()
 	}
-	macd := ctx.Indicator("macd", ctx.Input()).Get(0)
+	macd_dif := ctx.Indicator("macd_dif", ctx.Input()).Get(0)
 	// 计算DEA
 	beta := 2 / float64(singal+1)
-	dea := beta*macd + (1-beta)*deaPrev
+	dea := beta*macd_dif + (1-beta)*deaPrev
-	ctx.State().Set("macd_dea", dea)
+	ctx.State().Set("vector", dea)
 
 	vector = dea
 	return
 }
-
-// MacdSingal macd柱状图计算
-type MacdHist struct {
-}
-
-func (c *MacdHist) Meta() IndicatorMeta {
-	return IndicatorMeta{
-		Name: "macd_hist",
-		Input: []types.InputArg{
-			{Name: "fast", Type: types.InputTypeUInt, Desc: "快线周期"},
-			{Name: "slow", Type: types.InputTypeUInt, Desc: "慢线周期"},
-			{Name: "singal", Type: types.InputTypeUInt, Desc: "信号线周期"},
-		},
-	}
-}
-
-func (c *MacdHist) CandlePeriods(ctx IIndicatorContext) int16 {
-	return max(
-		ctx.Indicator("macd", ctx.Input()).CandlePeriods(),
-		ctx.Indicator("macd_dea", ctx.Input()).CandlePeriods(),
-	)
-}
-
-func (c *MacdHist) Calculate(ctx IIndicatorContext) (vector float64) {
-	macd := ctx.Indicator("macd", ctx.Input()).Get(0)
-	macd_dea := ctx.Indicator("macd_dea", ctx.Input()).Get(0)
-	vector = macd - macd_dea
-	return
-}

pkg/strategy/gold_x.go

@@ -33,22 +33,24 @@ func (s *GoldX) Init(input types.Input) (err error) {
 func (s *GoldX) CandlePeriods(ctx ISingleSigStrategyContext) int16 {
 	return max(
 		ctx.Indicator("macd", ctx.Input()).CandlePeriods(),
+		ctx.Indicator("macd_dif", ctx.Input()).CandlePeriods(),
 		ctx.Indicator("macd_dea", ctx.Input()).CandlePeriods(),
-		ctx.Indicator("macd_hist", ctx.Input()).CandlePeriods(),
 	)
 }
 
 func (s *GoldX) Update(ctx ISingleSigStrategyContext) (side types.Side) {
-	macd := ctx.Indicator("macd", ctx.Input()).Series(0, 2)        // macd线
+	macdHist := ctx.Indicator("macd", ctx.Input()).Get(0)          // macd柱状图
-	macdDea := ctx.Indicator("macd_dea", ctx.Input()).Series(0, 2) // macd信号线
+	macdDea := ctx.Indicator("macd_dea", ctx.Input()).Series(0, 2) // macd_dea信号线
-	macdHist := ctx.Indicator("macd_hist", ctx.Input()).Get(0)     // macd柱状图
+	macdDif := ctx.Indicator("macd_dif", ctx.Input()).Series(0, 2) // macd_dif线
 
 	// todo 包装方法 crossover/crossunder
-	// 1.MACD 线接近或上穿零轴（表示整体多头市场）
+	// 1.MACD DIF线接近或上穿零轴（表示整体多头市场）
-	crossover := macd[0] > macdDea[0] && macd[1] < macdDea[1]  // 上穿
+	crossover := macdDif[0] > macdDea[0] && macdDif[1] < macdDea[1]  // 上穿
-	crossunder := macd[0] < macdDea[0] && macd[1] > macdDea[1] // 下穿
+	crossunder := macdDif[0] < macdDea[0] && macdDif[1] > macdDea[1] // 下穿
-
+	ts := ctx.Get(0).Ts
+	_ = ts
 	if crossover {
+		_ = macdHist
 		// 2.附加确认条件: 柱状图从负值转为正值
 		if macdHist > 0 {
 			// todo 3.成交量放大（结合 OBV 等指标验证资金流入）。