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/trade"
	"sig-pub/pkg/types"
	"sig-pub/pkg/types/decimals"
	"sig-pub/pkg/utils/collect"
	"sig-pub/pkg/utils/times"
	"sig-pub/pkg/zlog"

	"google.golang.org/grpc"
)

type Backtest struct {
	exchangeClient pb.ExchangeServiceClient
	indReg         *indicator.IndicatorRegistry

	riskStrategy *trade.RiskStrategy
}

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

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

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

	// 回测账户
	account := NewAccount(10000, NewSimulator(0.0005, 0.0008))
	// 平仓管理器
	closeManager := NewCloseManager(0.02, 0)
	closeManager.SetDynamicParams(0.1, 0.02, 0)

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

	watch := times.NewWatch()
	recvTimes, recvTotal := 0, 0
	var lastK *types.Kline
	var msg *pb.RspHistoryKlineStream
	for {
		select {
		case <-ctx.Done():
			err = ctx.Err()
			return
		default:
		}
		msg, err = stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			return
		}
		recvTimes++
		recvTotal += len(msg.Klines)
		for _, k := range msg.Klines {
			kline := new(types.Kline)
			kline.ParsePBKline(seriesRange.Exchange, k)
			lastK = kline
			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
			}

			// 平仓策略
			closeManager.OnKline(*kline, account)

			sigSide := tradingPlan.Update(strategy.StrategyTypeSig)
			if sigSide.IsValid() {
				b.onSigSideSignalWithAccount(sigSide, *kline, account, closeManager)
			}
		}
	}

	_ = lastK
	zlog.Debugf("recv=%d, total=%d, use %s", recvTimes, recvTotal, watch.ElapsedFmt("."))
	collect.SortDesc(account.Trades, func(t *Trade) float64 { return t.Pnl })
	exposure := account.Cash + account.PositionCost()
	_ = exposure
	return
}

// onSigSideSignalWithAccount 交易策略发出交易信号（使用指定的账户和平仓管理器）
func (b *Backtest) onSigSideSignalWithAccount(sigSide types.Side, k types.Kline, account *Account, closeManager *CloseManager) {
	// risk check before executing
	side := b.riskStrategy.SideAssess(sigSide)
	if !side.IsValid() {
		zlog.Debugf("risk strategy filter sig side: %s", sigSide.String())
		return
	}
	// zlog.Debugf("apply market order:  ts=%d, side=%s", k.Ts, side.String())
	price := decimals.MustToFloat64(k.Close)
	account.ApplyMarketOrder(side, 0.01, price, k.Ts)

	// 根据信号方向平掉相反方向的仓位：如果信号是买入，平掉所有卖出仓位；如果信号是卖出，平掉所有买入仓位
	closeManager.CloseBySignal(sigSide, account, k)
}