diff --git a/README.md b/README.md index 6f73cfc..4dc0883 100644 --- a/README.md +++ b/README.md @@ -132,3 +132,4 @@ RSI[1,2,3,4] -> RSI[0] [Alphalens 因子评估库](https://github.com/quantopian/alphalens) +[词法解析](https://github.com/alecthomas/participle) diff --git a/config/exchange.toml b/config/exchange.toml index e65fb93..823264d 100644 --- a/config/exchange.toml +++ b/config/exchange.toml @@ -18,9 +18,9 @@ receiveBuffer = 4096 marketSubscribeLimit = 16 consumeBatch = 1024 consumeLater = 2000 # 时间到达later或者数据累计到batch触发consume -httpProxy = "" +# httpProxy = "" # 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 diff --git a/internal/trading/backtest/sig_strategy_backtester.go b/internal/trading/backtest/sig_strategy_backtester.go index 17aea33..2c4e69d 100644 --- a/internal/trading/backtest/sig_strategy_backtester.go +++ b/internal/trading/backtest/sig_strategy_backtester.go @@ -49,10 +49,11 @@ func (b *SigStrategyBacktester) SubKline(instId string, interval types.Interval, } // Backtest 基于历史数据回测信号策略 -func (b *SigStrategyBacktester) Backtest(ctx context.Context, sigStrategyInput types.Input, sr *pb.SeriesRange, - // cleanIntervalSeries *types.IntervalState[*types.KlineSeries], +func (b *SigStrategyBacktester) Backtest(ctx context.Context, + sigStrategyInput types.Input, + sr *pb.SeriesRange, iiks *types.InstanceIntervalKlineSeries, - recvSignal func(sigSide types.Side, k types.Kline) (err error), + recvSignal func(instId string, sigSide types.Side, k types.Kline) (err error), ) (err error) { if iiks == nil { iiks = types.NewInstanceIntervalKlineSeries() @@ -68,7 +69,7 @@ func (b *SigStrategyBacktester) Backtest(ctx context.Context, sigStrategyInput t case strategy.SigStrategyTypeInterval: err = b.intervalStrategySeries(ctx, b.sigStrategy.(strategy.IIntervalSigStrategy), sigStrategyInput, sr, iiks, recvSignal) case strategy.SigStrategyTypeInstanceInterval: - // todo + err = b.instanceIntervalStrategySeries(ctx, b.sigStrategy.(strategy.IInstanceIntervalSigStrategy), sigStrategyInput, sr, iiks, recvSignal) default: err = fmt.Errorf("unknown sig strategy type %v", b.sigStrategyType) } @@ -76,7 +77,13 @@ 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) { +func (b *SigStrategyBacktester) singleStrategySeries(ctx context.Context, + sigStrategy strategy.ISingleSigStrategy, + sigStrategyInput types.Input, + sr *pb.SeriesRange, + iiks *types.InstanceIntervalKlineSeries, + recvSignal func(instId string, sigSide types.Side, k types.Kline) (err error), +) (err error) { driverInstId := sr.InstId driverInterval := types.Interval(sr.Interval) driverSeries := iiks.Get(driverInstId, driverInterval) @@ -95,7 +102,7 @@ func (b *SigStrategyBacktester) singleStrategySeries(ctx context.Context, sigStr } sigSide := sigStrategy.Update(strategyContext) if sigSide.IsValid() { - if err = recvSignal(sigSide, *k); err != nil { + if err = recvSignal(driverInstId, sigSide, *k); err != nil { return } } @@ -105,7 +112,13 @@ 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, iiks *types.InstanceIntervalKlineSeries, 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(instId string, sigSide types.Side, k types.Kline) (err error), +) (err error) { driverInstId := sr.InstId driverInterval := types.Interval(sr.Interval) intervalKlineSeries := iiks.GetIntervalKlineSeries(driverInstId) @@ -114,51 +127,81 @@ func (b *SigStrategyBacktester) intervalStrategySeries(ctx context.Context, inte // 各周期所需k线数量 intervalCandlePeriods := intervalSigStrategy.CandlePeriods(intervalStrategyContext) + periodsChecked := false 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 - intervalCandlePeriods.Range(func(interval types.Interval, require int16) { - if update && require > 0 { - series := intervalKlineSeries.Get(interval) - update = series.Length() >= int(require) + // 策略所需周期窗口是否满足检查 + if !periodsChecked { + 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 } - }) - if !update { - return + periodsChecked = true } sigSide := intervalSigStrategy.Update(intervalStrategyContext) if sigSide.IsValid() { - if err = recvSignal(sigSide, *k); err != nil { + if err = recvSignal(driverInstId, sigSide, *k); err != nil { return } } return }) + return +} + +// instanceIntervalStrategySeries 多币种多周期策略 +func (b *SigStrategyBacktester) instanceIntervalStrategySeries(ctx context.Context, + intervalSigStrategy strategy.IInstanceIntervalSigStrategy, + sigStrategyInput types.Input, + sr *pb.SeriesRange, + iiks *types.InstanceIntervalKlineSeries, + recvSignal func(instId string, sigSide types.Side, k types.Kline) (err error), +) (err error) { + driverInstId := sr.InstId + driverInterval := types.Interval(sr.Interval) - // 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 - // }) + // 策略上下文 + strategyContext := sig.NewInstanceIntervalSigStrategyContext(sigStrategyInput, iiks, b.indicatorReg) + // 各周期所需k线数量 + tradeInsts, intervalCandlePeriods := intervalSigStrategy.CandlePeriods(strategyContext) + + periodsChecked := false + err = b.multiInstanceIntervalSeries(ctx, sr, tradeInsts, intervalCandlePeriods, iiks, func(driver bool, instId string, interval types.Interval, k *types.Kline) (err error) { + if !driver || instId != driverInstId || interval != driverInterval { + return + } + // 策略所需周期窗口是否满足检查 + if !periodsChecked { + update := true + for _, instId := range tradeInsts { + intervalCandlePeriods.Range(func(interval types.Interval, periods int16) { + if update && periods > 0 { + series := iiks.Get(instId, interval) + update = series.Length() >= int(periods) + } + }) + if !update { + return + } + } + periodsChecked = true + } + sigSideInsts := intervalSigStrategy.Update(strategyContext) + for _, si := range sigSideInsts { + if err = recvSignal(si.InstId, si.Side, *k); err != nil { + return + } + } + return + }) return } @@ -266,7 +309,7 @@ func (b *SigStrategyBacktester) multiInstanceIntervalSeries(ctx context.Context, } return }) - zlog.Debugf("other sr finish with: %s(%s), %v", isr.InstId, isr.Interval, err1) + zlog.Debugf("other sr finish with: %s(%s), %v, last=%d", isr.InstId, isr.Interval, err1, intervalAdder(kSeries.MustGet(0).Ts, 1)) if err1 == nil { syncCh <- -1 // 通知更新完毕, 后续不再更新 } else if err1 != errStop { @@ -317,7 +360,7 @@ 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) + zlog.Debugf("driver sr finish with: %s(%s), %v, last=%d", sr.InstId, sr.Interval, err0, driverIntervalAdder(driverSeries.MustGet(0).Ts, 1)) if err0 == nil { close(stopCh) } else if err0 != errStop { diff --git a/internal/trading/backtest/trading_plan_backtester.go b/internal/trading/backtest/trading_plan_backtester.go index 692bf5d..96d45d3 100644 --- a/internal/trading/backtest/trading_plan_backtester.go +++ b/internal/trading/backtest/trading_plan_backtester.go @@ -112,7 +112,7 @@ func (b *TradingPlanBacktester) Backtest(ctx context.Context, sr *pb.SeriesRange }) iiks := types.NewInstanceIntervalKlineSeries() - err = sigStrategyBacktester.Backtest(ctx, b.sigStrategyInput, sr, iiks, func(sigSide types.Side, k types.Kline) (err error) { + err = sigStrategyBacktester.Backtest(ctx, b.sigStrategyInput, sr, iiks, func(instId string, sigSide types.Side, k types.Kline) (err error) { test.Singals++ // 根据交易信号检查仓位平仓 if err = b.closeBySigSingal(sigSide, k); err != nil { @@ -213,7 +213,8 @@ func (b *TradingPlanBacktester) closeBySigSingal(sigSide types.Side, kline types // onSideSingal 出现买卖信号 func (b *TradingPlanBacktester) onSideSingal(sigSide types.Side, k types.Kline) (err error) { // 买卖信号交易风险分析 - doTrade, causes, err := b.riskStrategy.SigRiskAnalyze(sigSide) + + doTrade, causes, err := b.riskStrategy.SigRiskAnalyze(b.account, sigSide) if err != nil { return } diff --git a/internal/trading/sig/indicator_context.go b/internal/trading/sig/indicator_context.go index dad7059..a56534a 100644 --- a/internal/trading/sig/indicator_context.go +++ b/internal/trading/sig/indicator_context.go @@ -28,33 +28,37 @@ type IOffsetIndicatorContext interface { // IndicatorContext 指标上下文, 提供k线序列给指标计算使用 type IndicatorContext struct { IOffsetIndicatorContext - 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 // 指标调用链避免指标循环引用 + indicator indicator.IIndicator + indicatorKey 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, ",")) + indicatorKey := fmt.Sprintf("%s{%s}", indicator.Meta().Name, strings.Join(inputs, ",")) return &IndicatorContext{ - indicator: indicator, - indicatorStateKey: stateKey, - input: input, - indicatorStates: indicatorStates, - kSeries: kSeries, - indicatorsReg: indicatorsReg, - indicatorTrace: []string{indicator.Meta().Name}, + indicator: indicator, + indicatorKey: indicatorKey, + input: input, + indicatorStates: indicatorStates, + kSeries: kSeries, + indicatorsReg: indicatorsReg, + indicatorTrace: []string{indicator.Meta().Name}, } } +func (c *IndicatorContext) getIndicatorKey() string { + return c.indicatorKey +} + func (c *IndicatorContext) SetOffset(offset int16) { c.offset = offset } @@ -96,10 +100,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 { - state, ok := c.indicatorStates[c.indicatorStateKey] + state, ok := c.indicatorStates[c.indicatorKey] if !ok { state = NewIndicatorState(c.kSeries.Interval) - c.indicatorStates[c.indicatorStateKey] = state + c.indicatorStates[c.indicatorKey] = state // 从头KlineSeries跑一遍, 针对ema,macd等回溯迭代指标, 将state与KlineSeries对齐 c.backtrackIndicatorState(c.indicator) } diff --git a/internal/trading/sig/strategy_context.go b/internal/trading/sig/strategy_context.go index 1d8ce3e..308ddc8 100644 --- a/internal/trading/sig/strategy_context.go +++ b/internal/trading/sig/strategy_context.go @@ -124,10 +124,10 @@ type InstanceIntervalSigStrategyContext struct { instanceIntervalIndicatorStates *collect.ConcurrentMap[string, *types.IntervalState[IndicatorStates]] } -func NewInstanceIntervalSigStrategyContext(input types.Input, iis *types.InstanceIntervalKlineSeries, indicatorsReg *indicator.IndicatorRegistry) *InstanceIntervalSigStrategyContext { +func NewInstanceIntervalSigStrategyContext(input types.Input, iiks *types.InstanceIntervalKlineSeries, indicatorsReg *indicator.IndicatorRegistry) *InstanceIntervalSigStrategyContext { return &InstanceIntervalSigStrategyContext{ input: input, - iiks: iis, + iiks: iiks, indicatorsReg: indicatorsReg, instanceIntervalIndicatorStates: collect.NewConcurrentMap[string, *types.IntervalState[IndicatorStates]](4, func(s string) string { return s }), } diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index d5507e6..080755e 100644 --- a/internal/trading/trading_service.go +++ b/internal/trading/trading_service.go @@ -293,10 +293,14 @@ func (svc *TradingService) StrategySeries(ctx context.Context, req *pb.ReqStrate // 信号策略参数 sigStrategyInput := types.Input(req.Input.AsMap()) - - // 使用回测器回测信号 + // 使用回测器回测信号策略 + driverInstId := req.Series.InstId backtester := backtest.NewSigStrategyBacktester(sigStrategyType, sigStrategy, svc.indicatorReg, svc.exchangeClient) - err = backtester.Backtest(ctx, sigStrategyInput, req.Series, nil, func(sigSide types.Side, k types.Kline) (err error) { + err = backtester.Backtest(ctx, sigStrategyInput, req.Series, nil, func(instId string, sigSide types.Side, k types.Kline) (err error) { + if instId != driverInstId { + // todo 多币种回测 + return + } side := lang.Ternary(sigSide == types.SideLong, pb.Side_BUY, pb.Side_SELL) rsp.Signal = append(rsp.Signal, side) rsp.Times = append(rsp.Times, k.Ts) diff --git a/pkg/strategy/mul_inst_rank.go b/pkg/strategy/mul_inst_rank.go index 6fa17a9..5bfaf33 100644 --- a/pkg/strategy/mul_inst_rank.go +++ b/pkg/strategy/mul_inst_rank.go @@ -32,7 +32,7 @@ func (s *MultiInstanceRank) Init(input types.Input) (err error) { // 校验参 return } -func (s *MultiInstanceRank) CandlePeriods(ctx IIntervalSigStrategyContext) (insts []string, iss *types.IntervalState[int16]) { +func (s *MultiInstanceRank) CandlePeriods(ctx IInstanceIntervalSigStrategyContext) (insts []string, iss *types.IntervalState[int16]) { insts = []string{"BTC_USDT", "SOL_USDT"} iss = types.NewIntervalState[int16]() iss.Set(types.Interval5m, 1) diff --git a/pkg/strategy/sig_strategy.go b/pkg/strategy/sig_strategy.go index b420ae4..89f5492 100644 --- a/pkg/strategy/sig_strategy.go +++ b/pkg/strategy/sig_strategy.go @@ -62,7 +62,7 @@ type IIntervalSigStrategyContext interface { type IInstanceIntervalSigStrategy interface { ISigStrategy // 需要的各周期最小数据k线数, 回测时用 - CandlePeriods(ctx IIntervalSigStrategyContext) (tradeInsts []string, iPeriods *types.IntervalState[int16]) + CandlePeriods(ctx IInstanceIntervalSigStrategyContext) (tradeInsts []string, iPeriods *types.IntervalState[int16]) Update(ctx IInstanceIntervalSigStrategyContext) (sides []types.SideInst) } diff --git a/pkg/trade/risk_strategy.go b/pkg/trade/risk_strategy.go index 24467f5..fd5b2a7 100644 --- a/pkg/trade/risk_strategy.go +++ b/pkg/trade/risk_strategy.go @@ -31,7 +31,7 @@ func NewRiskStrategy(param RiskStrategyParam) (rs *RiskStrategy, err error) { // 对交易方向进行信心分数评估, 后续开仓仓位 // 1.当前持有反方向单时, 不进行开仓 // 2.当前持有同方向单时, 根据信心分数评估是否加仓 -func (s *RiskStrategy) SigRiskAnalyze(signalSide types.Side) (doTrade bool, causes []Cause, err error) { +func (s *RiskStrategy) SigRiskAnalyze(account ITradeAccount, signalSide types.Side) (doTrade bool, causes []Cause, err error) { return true, nil, nil } diff --git a/pkg/trade/trade_account.go b/pkg/trade/trade_account.go index cc1dcc2..b46e29a 100644 --- a/pkg/trade/trade_account.go +++ b/pkg/trade/trade_account.go @@ -1,10 +1,13 @@ package trade +import "sig-pub/pkg/types" + // sig -> close strategy // sig -> risk strategy -> trade strategy -> tarde account +// position, trades type ITradeAccount interface { // 获取未平仓交易单 - OpenPositions() []*Position + OpenPositions() map[int64]*Position // 获取未平仓交易单数 CountOpenPositions() int @@ -13,7 +16,7 @@ type ITradeAccount interface { TradeOrder(ta TradeArg) (ok bool, cause Cause, err error) // 将仓位进行平仓 - ClosePosition(*Position, Cause) (err error) + ClosePosition(*Position, types.Kline, Cause) (err error) // // 根据当前价格对仓位进行 mark-to-market，返回账户净值 // GetCurrentEquity() decimal.Decimal diff --git a/pkg/trade/types.go b/pkg/trade/types.go index eee7cdc..0131dc2 100644 --- a/pkg/trade/types.go +++ b/pkg/trade/types.go @@ -41,12 +41,35 @@ func (c Cause) String() string { // Position 持仓仓位 type Position struct { TradeId int64 // 交易订单id + InstId string // 交易产品id Status int32 // 1.交易中 2.持仓中 3.已平仓 Side types.Side // 交易方向 Qty float64 // 交易量 EntryPx float64 // 入场价格 EntryTs int64 // 入场时间 PeakPx float64 // highest (for long) or lowest (for short) observed price since entry + AvgPx float64 // 平均持仓价格 Fee float64 // 手续费 FeeRate float64 // 手续费率 } + +// TradeOrder 交易订单 +type TradeOrder struct { + InstId string // 交易产品id + Status int32 // 1.交易中 2.持仓中 3.已平仓 + Ctime int64 // 创建时间 + Side types.Side // 交易方向 + Qty float64 // 交易量 + Price float64 // 开仓价格 + Fee float64 // 开仓手续费 + Leverage int32 // 杠杆倍数 + Time int64 // 开仓时间 + // ClosePrice float64 // 平仓价格 + // CloseFee float64 // 平仓手续费 + // CloseTime int64 // 平仓时间 + // CloseCause Cause // 平仓原因 + Pnl float64 // 盈利/亏损 pnl = (t.ClosePrice-t.Price)*t.Qty - t.Fee - t.CloseFee + Cash float64 // 平仓后账户净值 + HoldTime string // 持仓时间 + PeakPx float64 // highest (for long) or lowest (for short) observed price since entry +} diff --git a/pkg/types/input.go b/pkg/types/input.go index 91c8296..0f95e82 100644 --- a/pkg/types/input.go +++ b/pkg/types/input.go @@ -91,16 +91,17 @@ func (in Input) Int16(k string) (v int16) { } func (in Input) String(k string) (v string) { - if r, ok := in.getCache(k); ok { - if v, ok = r.(string); ok { - return - } - } + // cacheK := "str:" + k + // if r, ok := in.getCache(cacheK); ok { + // if v, ok = r.(string); ok { + // return + // } + // } v, err := cast.ToStringE(in.get(k, "string")) if err != nil { panic(fmt.Errorf("input string parse error: %s", k)) } - in.setCache(k, v) + // in.setCache(cacheK, v) return }