sig-pub/internal/trading/backtest/backtest.go

package backtest

import (
	"context"
	"fmt"
	"io"
	"sig-pub/api/pb"
	"sig-pub/internal/trading/sig"
	"sig-pub/pkg/indicator"
	"sig-pub/pkg/strategy"
	"sig-pub/pkg/types"
	"sig-pub/pkg/types/decimals"
	"sig-pub/pkg/utils/times"
	"sig-pub/pkg/zlog"

	"google.golang.org/grpc"
)

type Backtest struct {
	exchangeClient pb.ExchangeServiceClient
	indReg         *indicator.IndicatorRegistry
	sigStrategyReg *strategy.SigStrategyRegistry
}

func NewBacktest(exchangeClient pb.ExchangeServiceClient, indReg *indicator.IndicatorRegistry, sigStrategyReg *strategy.SigStrategyRegistry) *Backtest {
	return &Backtest{exchangeClient: exchangeClient, indReg: indReg}
}

func (b *Backtest) RunTradingPlan(ctx context.Context, tradingPlan *sig.TradingPlan, stime, etime int64, sigKlineSeries *sig.KlineSeries) (err error) {
	var sim *Simulator
	_ = sim
	plan := tradingPlan.Plan
	exchange := pb.ExchangeType(plan.Exchange)
	interval := types.Interval(plan.Interval)
	instId := plan.InstId

	sigStrategy := tradingPlan.GetSigStrategy()
	maxWindow := sigStrategy.MaxWindow()
	if maxWindow < 0 || maxWindow > indicator.MaxWindow {
		err = fmt.Errorf("invalid window %d 0-%d, planId=%d", maxWindow, indicator.MaxWindow, plan.Id)
		return
	}

	seriesRange := &pb.SeriesRange{
		Exchange: exchange,
		InstId:   instId,
		Interval: string(interval),
		Before:   stime,
		After:    etime,
		Open:     false,
		Live:     false,
		Desc:     false,
		Window:   uint32(maxWindow),
	}
	// fetch history klines via stream
	req := &pb.ReqHistoryKlineStream{Series: seriesRange}
	stream, err := b.exchangeClient.HistoryKlineStream(context.Background(), req, grpc.UseCompressor("snappy"))
	if err != nil {
		return
	}

	recvTimes, total := 0, 0
	watch := times.NewWatch()
	for {
		msg, err0 := stream.Recv()
		if err0 == io.EOF {
			break
		}
		if err0 != nil {
			err = err0
			return
		}
		recvTimes++
		total += len(msg.Klines)
		for _, k := range msg.Klines {
			kline := new(types.Kline)
			kline.ParsePBKline(seriesRange.Exchange, k)

			if lastTs, serial := sigKlineSeries.Update(kline); !serial {
				err = fmt.Errorf("kline not series: %s(%s), interval=%s, lastTs=%d", instId, exchange, interval, lastTs)
				return
			}
			length := sigKlineSeries.Length()
			if length <= maxWindow {
				continue
			}

			signalSide := tradingPlan.Update(strategy.StrategyTypeSig)
			switch signalSide {
			case types.SideBuy, types.SideSell:
				k, _ := sigKlineSeries.Get(0)
				_ = k
				// zlog.Infof("sigSide: ts=%d, %s", k.Ts, sigSide)
			}
		}
	}

	zlog.Debugf("recv=%d, total=%d, use %s", recvTimes, total, watch.ElapsedFmt("."))
	return
}

// Run 执行回测
// seriesRange: 回测的交易产品/周期/时间区间
// sigStrategy: 已创建的策略实例（将调用 New() 并 Init）
// params: 策略参数
func (b *Backtest) Run0(ctx context.Context, seriesRange *pb.SeriesRange, sigStrategy strategy.ISigStrategy, params strategy.StrategyParam, initialCash float64) (res BacktestResult, err error) {
	// prepare strategy
	strat := sigStrategy.New()
	if err = strat.Init(params); err != nil {
		return
	}

	// fetch history klines via stream
	req := &pb.ReqHistoryKlineStream{Series: seriesRange}
	stream, err := b.exchangeClient.HistoryKlineStream(context.Background(), req, grpc.UseCompressor("snappy"))
	if err != nil {
		return
	}

	var klines []types.Kline
	var firstTs, lastTs int64
	for {
		msg, err0 := stream.Recv()
		if err0 == io.EOF {
			break
		}
		if err0 != nil {
			err = err0
			return
		}
		for _, k := range msg.Klines {
			kk := new(types.Kline)
			kk.ParsePBKline(seriesRange.Exchange, k)
			klines = append(klines, *kk)
			if firstTs == 0 {
				firstTs = kk.Ts
			}
			lastTs = kk.Ts
		}
	}

	// prepare account and set risk limits from params if provided
	acct := NewAccount(initialCash, nil)
	if v, ok := params.GetFloat64("max_pos_pct"); ok && v > 0 {
		acct.MaxPosPct = v
	}
	if v, ok := params.GetFloat64("max_exposure_pct"); ok && v > 0 {
		acct.MaxExposurePct = v
	}

	// prepare close manager from params
	var cm *CloseManager
	if sl, ok := params.GetFloat64("stoploss_pct"); ok || true {
		tp, _ := params.GetFloat64("takeprofit_pct")
		cm = NewCloseManager(sl, tp)
	}

	// iterate klines in chronological order
	for i := 0; i < len(klines); i++ {
		// build context with klines up to i
		window := klines[:i+1]
		ctxSig := NewSigStrategyContext(window, b.indReg)
		// first, evaluate stoploss/takeprofit on this kline
		if cm != nil {
			cm.OnKline(klines[i], acct)
		}
		side := strat.Update(ctxSig)
		// simple position sizing: param 'size' as fraction of cash; else use fixed qty 1
		sizePct, ok := params.GetFloat64("size")
		var qty float64
		if ok && sizePct > 0 {
			price := decimals.MustToFloat64(klines[i].Close)
			qty = (acct.Cash * sizePct) / price
		} else {
			qty = 1
		}

		if side == types.SideBuy || side == types.SideSell {
			// signal-based close: close opposite positions first
			if cm != nil {
				cm.CloseBySignal(pb.Side_BUY, acct, klines[i])
			}
			if _, ok := acct.ApplyMarketOrder(pb.Side_BUY, qty, klines[i].Ts, klines[i]); ok {
				// trade recorded
			} else {
				zlog.Debugf("order rejected or insufficient cash at ts=%d", klines[i].Ts)
			}
		}
	}

	// build result
	res.StartTs = firstTs
	res.EndTs = lastTs
	res.Trades = acct.Trades
	for _, p := range acct.Positions {
		res.Positions = append(res.Positions, *p)
	}
	res.Cash = acct.Cash
	// estimate equity using last close price
	if len(klines) > 0 {
		last := decimals.MustToFloat64(klines[len(klines)-1].Close)
		equity := acct.Cash
		// naive mark-to-market of positions
		for _, p := range acct.Positions {
			if p.Side == pb.Side_BUY {
				equity += (last - p.EntryPx) * p.Qty
			} else {
				equity += (p.EntryPx - last) * p.Qty
			}
		}
		res.Equity = equity
	}
	return
}