multi trade inst

README.md

@@ -132,3 +132,4 @@ RSI[1,2,3,4] -> RSI[0]
 
 [Alphalens 因子评估库](https://github.com/quantopian/alphalens)
 
+[词法解析](https://github.com/alecthomas/participle)

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

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,10 +127,13 @@ 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
 		}
+		// 策略所需周期窗口是否满足检查
+		if !periodsChecked {
 			update := true
 			intervalCandlePeriods.Range(func(interval types.Interval, require int16) {
 				if update && require > 0 {
@@ -128,37 +144,64 @@ func (b *SigStrategyBacktester) intervalStrategySeries(ctx context.Context, inte
 			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 {

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
 	}

internal/trading/sig/indicator_context.go

@@ -29,7 +29,7 @@ type IOffsetIndicatorContext interface {
 type IndicatorContext struct {
 	IOffsetIndicatorContext
 	indicator       indicator.IIndicator
-	indicatorStateKey string
+	indicatorKey    string
 	indicatorsReg   *indicator.IndicatorRegistry
 	input           types.Input
 	indicatorStates map[string]*IndicatorState // {macd{window:0,fast:9,slow:21,single:10}: state} 初始化时与KlineSeries周期同步
@@ -42,11 +42,11 @@ func NewIndicatorContext(indicator indicator.IIndicator, input types.Input, indi
 	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,
+		indicatorKey:    indicatorKey,
 		input:           input,
 		indicatorStates: indicatorStates,
 		kSeries:         kSeries,
@@ -55,6 +55,10 @@ func NewIndicatorContext(indicator indicator.IIndicator, input types.Input, indi
 	}
 }
 
+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)
 	}

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 }),
 	}

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)

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)

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)
 }
 

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
 }

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

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
+}

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
 }