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 }