sig-pub/internal/trading/trading_grpc_server.go

package trading

import (
	"context"
	"fmt"
	"sig-pub/api/pb"
	"sig-pub/pkg/data"
	"sig-pub/pkg/types"
	"sig-pub/pkg/utils/collect"
	"sig-pub/pkg/utils/misc"
	"sig-pub/pkg/utils/times"
	"sig-pub/pkg/zlog"

	"github.com/bytedance/sonic"
	"google.golang.org/protobuf/types/known/structpb"
)

type TradingGrpcServer struct {
	pb.UnimplementedTradingServiceServer
	tradingService *TradingService
}

func NewTradingGrpcServer(tradingService *TradingService) *TradingGrpcServer {
	return &TradingGrpcServer{
		tradingService: tradingService,
	}
}

func (svr *TradingGrpcServer) Init() (err error) {
	return
}

func (svr *TradingGrpcServer) Indicators(ctx context.Context, req *pb.ReqIndicators) (rsp *pb.RspIndicators, err error) {
	rsp = new(pb.RspIndicators)
	rsp.Indicators = svr.tradingService.Indicators()
	return
}

func (svr *TradingGrpcServer) IndicatorMetas(ctx context.Context, req *pb.ReqIndicatorMetas) (rsp *pb.RspIndicatorMetas, err error) {
	indMetas, err := svr.tradingService.IndicatorMeta(req.Indicators...)
	if err != nil {
		return
	}
	indPlots, err := svr.tradingService.IndicatorPlots(req.Indicators...)
	if err != nil {
		return
	}
	rsp = new(pb.RspIndicatorMetas)
	for _, indName := range req.Indicators {
		// 指标元数据
		meta := indMetas[indName]
		p := &pb.IndicatorMeta{
			Name:  indName,
			Desc:  meta.Desc,
			State: meta.State,
		}
		// 指标输入参数
		for _, input := range meta.Input {
			p.Input = append(p.Input, &pb.InputArg{
				Name:    input.Name,
				Desc:    input.Desc,
				Type:    int32(input.Type),
				Default: misc.Ternary(input.Default == nil, "", fmt.Sprintf("%v", input.Default)),
				Options: collect.Mapping(input.Options, func(opt types.InputOption) *pb.InputOption {
					return &pb.InputOption{
						Name: opt.Name,
						Desc: opt.Desc,
					}
				}),
			})
		}
		// 指标绘图属性
		for _, plot := range indPlots[indName] {
			ps := &pb.IndicatorPlotSeries{
				Name:  plot.Name,
				State: plot.State,
				Type:  int32(plot.Type),
			}
			if ps.Props, err = structpb.NewStruct(plot.Props); err != nil {
				return
			}
			for _, exp := range plot.Exps {
				pe := &pb.IndicatorPlotExp{
					Exp: exp.Exp,
				}
				if pe.Props, err = structpb.NewStruct(exp.Props); err != nil {
					return
				}
				ps.Exps = append(ps.Exps, pe)
			}
			p.Plots = append(p.Plots, ps)
		}
		rsp.Metas = append(rsp.Metas, p)
	}
	return
}

func (svr *TradingGrpcServer) IndicatorPlots(ctx context.Context, req *pb.ReqIndicatorPlots) (rsp *pb.RspIndicatorPlots, err error) {
	indPlots, err := svr.tradingService.IndicatorPlots(req.Indicators...)
	if err != nil {
		return
	}
	rsp = new(pb.RspIndicatorPlots)
	for _, indName := range req.Indicators {
		plots := indPlots[indName]
		p := &pb.IndicatorPlot{Indicator: indName}
		for _, plot := range plots {
			ps := &pb.IndicatorPlotSeries{
				Name:  plot.Name,
				State: plot.State,
				Type:  int32(plot.Type),
			}
			if ps.Props, err = structpb.NewStruct(plot.Props); err != nil {
				return
			}
			for _, exp := range plot.Exps {
				pe := &pb.IndicatorPlotExp{
					Exp: exp.Exp,
				}
				if pe.Props, err = structpb.NewStruct(exp.Props); err != nil {
					return
				}
				ps.Exps = append(ps.Exps, pe)
			}
			p.Plots = append(p.Plots, ps)
		}
		rsp.Plots = append(rsp.Plots, p)
	}
	return
}

func (svr *TradingGrpcServer) IndicatorSeries(ctx context.Context, req *pb.ReqIndicatorSeries) (rsp *pb.RspIndicatorSeries, err error) {
	// s, err := structpb.NewStruct(map[string]any{})
	input := req.Input.AsMap()
	matrix, times, states, err := svr.tradingService.IndicatorSeries(ctx, req.Indicator, req.Digit, types.Input(input), req.Series)
	if err != nil {
		return
	}
	rsp = &pb.RspIndicatorSeries{}
	rsp.Matrix = matrix
	rsp.Times = times
	for state, value := range states {
		rsp.States = append(rsp.States, &pb.IndicatorState{
			State:  state,
			Matrix: value,
		})
	}
	return
}

func (svr *TradingGrpcServer) IndicatorSummary(ctx context.Context, req *pb.ReqIndicatorSummary) (rsp *pb.RspIndicatorSummary, err error) {
	input := req.Input.AsMap()
	summary, err := svr.tradingService.IndicatorSummary(ctx, req.Indicator, req.Digit, types.Input(input), req.Series)
	if err != nil {
		return
	}
	summaryJson, err := sonic.Marshal(summary)
	if err != nil {
		err = fmt.Errorf("marshal indicator summary error: %w", err)
		return
	}
	rsp = &pb.RspIndicatorSummary{
		Indicator: req.Indicator,
		Summary:   summaryJson,
	}
	return
}

func (svr *TradingGrpcServer) StrategySeries(ctx context.Context, req *pb.ReqStrategySeries) (rsp *pb.RspStrategySeries, err error) {
	rsp = &pb.RspStrategySeries{}
	err = svr.tradingService.StrategySeries(ctx, req, rsp)
	return
}

func (svr *TradingGrpcServer) Backtest(ctx context.Context, req *pb.ReqBacktest) (rsp *pb.RspBacktest, err error) {
	stime, err := times.ParseFORMAT(req.Stime)
	if err != nil {
		return
	}
	etime, err := times.ParseFORMAT(req.Etime)
	if err != nil {
		return
	}
	taskId, err := svr.tradingService.Backtest(ctx, req.PlanId, stime.UnixMilli(), etime.UnixMilli())
	if err != nil {
		return
	}
	rsp = &pb.RspBacktest{TaskId: taskId}
	return
}

// BacktestLog 回测记录查询
func (svr *TradingGrpcServer) BacktestLog(ctx context.Context, req *pb.ReqBacktestLog) (rsp *pb.RspBacktestLog, err error) {
	if req.Paging == nil {
		req.Paging = &pb.Paging{Page: 1, Size: 20, Asc: true, SortBy: "id"}
	}
	paging := data.PageArgs0(int(req.Paging.Page), int(req.Paging.Size), req.Paging.SortBy, req.Paging.Asc)
	logs, err := svr.tradingService.BacktestLog(ctx, 10001, paging, req.Running)
	if err != nil {
		return
	}
	rsp = new(pb.RspBacktestLog)
	for _, l := range logs {
		log := &pb.BacktestLog{
			PlanId:     l.PlanId,
			Stime:      l.SeriesBefore,
			Etime:      l.SeriesAfter,
			BacktestId: l.Id,
			Status:     l.RunningStatus,
			Progress:   l.RunningProgress,
			Total:      l.RunningTotal,
			Pct:        l.RunningPct,
			TaskId:     l.RunningTaskId,
		}
		rsp.Logs = append(rsp.Logs, log)
	}
	return
}

// BacktestRace 交易计划参数调试回测
func (svr *TradingGrpcServer) BacktestRace(ctx context.Context, req *pb.ReqBacktestRace) (rsp *pb.RspBacktestRace, err error) {
	rsp = new(pb.RspBacktestRace)
	misc.SafeGo(func() {
		c := context.Background()
		err := svr.tradingService.BacktestRace(c, req)
		if err != nil {
			zlog.Error("BacktestRace error:", err)
		}
	})
	return
}