diff --git a/config/exchange.toml b/config/exchange.toml index 662a8bd..07f7bca 100644 --- a/config/exchange.toml +++ b/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 diff --git a/internal/exchange/exchange_service.go b/internal/exchange/exchange_service.go index 7f4a432..d45c98e 100644 --- a/internal/exchange/exchange_service.go +++ b/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 } diff --git a/internal/trading/backtest/sig_strategy_backtester.go b/internal/trading/backtest/sig_strategy_backtester.go index e3324e3..0ddd3cf 100644 --- a/internal/trading/backtest/sig_strategy_backtester.go +++ b/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.After = after + isr.Before = intervalAdder(driverBefore, -1) + 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 -} diff --git a/internal/trading/sig/indicator_context.go b/internal/trading/sig/indicator_context.go index 35f1b44..75695c1 100644 --- a/internal/trading/sig/indicator_context.go +++ b/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) - length := c.kSeries.Length() - for i := range length { - if i < int(previousCandles) { - continue - } - offset := int16(length - 1 - i) + candleLength := c.kSeries.Length() + candlePeriods := indicator.CandlePeriods(c) + for i := int(candlePeriods) - 1; i < candleLength; i++ { + offset := int16(candleLength - i - 1) 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) diff --git a/internal/trading/sig/indicator_state.go b/internal/trading/sig/indicator_state.go index 0bb0bcd..ca72d36 100644 --- a/internal/trading/sig/indicator_state.go +++ b/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 - state map[string]*types.RingSeries[float64] + indicatorContext indicator.IIndicatorContext + 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, - state: make(map[string]*types.RingSeries[float64]), + interval: interval, + 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 - return s.ring(k).Get(int(offset)) + ts := s.indicatorContext.Get(0).Ts + 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 - return s.ring(k).Series(int(offset), int(count)) + ts := s.indicatorContext.Get(0).Ts + 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 } diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index 06bd314..9a491a2 100644 --- a/internal/trading/trading_service.go +++ b/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 } diff --git a/pkg/indicator/ema.go b/pkg/indicator/ema.go index d937c27..181e16c 100644 --- a/pkg/indicator/ema.go +++ b/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 } diff --git a/pkg/indicator/indicator.go b/pkg/indicator/indicator.go index 28f3e36..c0725e5 100644 --- a/pkg/indicator/indicator.go +++ b/pkg/indicator/indicator.go @@ -6,7 +6,8 @@ import ( ) const ( - MaxWindow = 256 + MaxWindow = 256 // 最大窗口引用大小 + ApproCandles = 64 // 递归指标近似计算k线数 ) type IndicatorMeta struct { diff --git a/pkg/indicator/indicator_registry.go b/pkg/indicator/indicator_registry.go index 22abe0e..ac53b08 100644 --- a/pkg/indicator/indicator_registry.go +++ b/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 diff --git a/pkg/indicator/macd.go b/pkg/indicator/macd.go index 21f0a05..8df476d 100644 --- a/pkg/indicator/macd.go +++ b/pkg/indicator/macd.go @@ -4,16 +4,46 @@ import ( "sig-pub/pkg/types" ) -// MACD 拆分成: MACD线, MacdDEA(信号线), MacdHist(柱状图) -// 计算 MACD 线 (DIF): 反映短期趋势与长期趋势的“收敛/散度” -// MACD: https://www.investopedia.com/terms/m/macd.asp -type MACD struct { +// Macd macd柱状图计算 + +// Macd 拆分成: Macd(柱状图), MacdDIF线, MacdDEA(信号线) +// 计算 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( - ctx.Input().Int16("singal")+1, - ctx.Indicator("macd", ctx.Input()).CandlePeriods(), - ) + return ctx.Indicator("macd_dif", ctx.Input()).CandlePeriods() + ctx.Input().Int16("singal") + 1 } 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 - macdPrevs := ctx.Indicator("macd", ctx.Input()).Series(1, singal) - deaPrev = macdPrevs.Avg() + // 初始值前9期的 MACD_DIF SMA + macdDifs := ctx.Indicator("macd_dif", ctx.Input()).Series(1, singal) + 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 - ctx.State().Set("macd_dea", dea) + dea := beta*macd_dif + (1-beta)*deaPrev + 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 -} diff --git a/pkg/strategy/gold_x.go b/pkg/strategy/gold_x.go index 93838ef..bb06b70 100644 --- a/pkg/strategy/gold_x.go +++ b/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线 - macdDea := ctx.Indicator("macd_dea", ctx.Input()).Series(0, 2) // macd信号线 - macdHist := ctx.Indicator("macd_hist", ctx.Input()).Get(0) // macd柱状图 + macdHist := ctx.Indicator("macd", ctx.Input()).Get(0) // macd柱状图 + macdDea := ctx.Indicator("macd_dea", ctx.Input()).Series(0, 2) // macd_dea信号线 + macdDif := ctx.Indicator("macd_dif", ctx.Input()).Series(0, 2) // macd_dif线 // todo 包装方法 crossover/crossunder - // 1.MACD 线接近或上穿零轴（表示整体多头市场） - crossover := macd[0] > macdDea[0] && macd[1] < macdDea[1] // 上穿 - crossunder := macd[0] < macdDea[0] && macd[1] > macdDea[1] // 下穿 - + // 1.MACD DIF线接近或上穿零轴（表示整体多头市场） + crossover := macdDif[0] > macdDea[0] && macdDif[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 等指标验证资金流入）。