From b516bfbfc757ea27ee84b4c481e76e151ad1cdb7 Mon Sep 17 00:00:00 2001 From: strange Date: Wed, 31 Dec 2025 10:26:43 +0800 Subject: [PATCH] gateway reverse proxy sig --- cmd/gateway/main.go | 9 +- go.mod | 4 + go.sum | 8 + internal/gateway/fast_gateway.go | 174 ++++++++++++++++++ internal/gateway/gateway.go | 5 - internal/sig/service/backtest_service.go | 6 +- .../backtest/sig_strategy_backtester.go | 2 +- internal/trading/trading_grpc_server.go | 10 +- internal/trading/trading_service.go | 3 +- pkg/grpc/generic/generic_client_factory.go | 2 +- pkg/indicator/indicator_registry.go | 3 - pkg/indicator/macd.go | 3 + pkg/indicator/macd0.go | 110 ----------- pkg/strategy/super_trend_macd_rsi.go | 57 ++++-- pkg/strategy/super_trend_rsi.go | 12 +- pkg/utils/codec/mapstructure.go | 2 - pkg/utils/lang/concurrent.go | 17 ++ 17 files changed, 272 insertions(+), 155 deletions(-) create mode 100644 internal/gateway/fast_gateway.go delete mode 100644 pkg/indicator/macd0.go create mode 100644 pkg/utils/lang/concurrent.go diff --git a/cmd/gateway/main.go b/cmd/gateway/main.go index 581383a..21815b0 100644 --- a/cmd/gateway/main.go +++ b/cmd/gateway/main.go @@ -38,15 +38,15 @@ func main() { // consul service discovery dis := discovery.NewConsulDiscovery(client) resolver := dis.Resolver() - gpcGenericClientFactory := generic.NewGpcGenericClientFactory( + grpcGenericClientFactory := generic.NewGrpcGenericClientFactory( discovery.ConsulSchema, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(resolver), ) - if err := gpcGenericClientFactory.Init(); err != nil { + if err := grpcGenericClientFactory.Init(); err != nil { panic(err) } - if err := dis.WatchServices(gpcGenericClientFactory.RefreshService); err != nil { + if err := dis.WatchServices(grpcGenericClientFactory.RefreshService); err != nil { panic(err) } @@ -54,7 +54,8 @@ func main() { if err != nil { panic(err) } - gateServer := gateway.NewGateServer(sigServerUrl, gpcGenericClientFactory) + gateServer := gateway.NewFastGatewayServer(sigServerUrl, grpcGenericClientFactory) + // gateServer := gateway.NewGateServer(sigServerUrl, gpcGenericClientFactory) if err := gateServer.Init(); err != nil { panic(err) } diff --git a/go.mod b/go.mod index 7351586..181c07c 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/bytedance/sonic v1.13.2 github.com/dsnet/golib/unitconv v1.0.2 github.com/fanjindong/go-cache v0.0.6 + github.com/fasthttp/router v1.5.4 github.com/gin-gonic/gin v1.10.0 github.com/go-resty/resty/v2 v2.16.5 github.com/go-viper/mapstructure/v2 v2.2.1 @@ -25,8 +26,10 @@ require ( github.com/mostynb/go-grpc-compression v1.2.3 github.com/nats-io/nats.go v1.47.0 github.com/redis/go-redis/v9 v9.7.3 + github.com/savsgio/gotils v0.0.0-20240704082632-aef3928b8a38 github.com/spf13/cast v1.10.0 github.com/spf13/viper v1.20.0 + github.com/valyala/fasthttp v1.68.0 go.etcd.io/etcd/api/v3 v3.6.1 go.etcd.io/etcd/client/v3 v3.6.1 go.uber.org/zap v1.27.0 @@ -109,6 +112,7 @@ require ( github.com/subosito/gotenv v1.6.0 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.2.12 // indirect + github.com/valyala/bytebufferpool v1.0.0 // indirect go.etcd.io/etcd/client/pkg/v3 v3.6.1 // indirect go.opentelemetry.io/otel v1.38.0 // indirect go.opentelemetry.io/otel/trace v1.38.0 // indirect diff --git a/go.sum b/go.sum index e96fad9..05db25b 100644 --- a/go.sum +++ b/go.sum @@ -58,6 +58,8 @@ github.com/dsnet/golib/unitconv v1.0.2 h1:45gXng3Op1vTrnX1PdM9Bla4mEpBFYA5aC8dlq github.com/dsnet/golib/unitconv v1.0.2/go.mod h1:86KTUtTJFLreKjc4sS9xE0rhj4lR44Ox0rEQSEXSWwM= github.com/fanjindong/go-cache v0.0.6 h1:4xl8MnfW8pFLH9cRjs0uNfVbFNqV342yl/pgX3Ql9gM= github.com/fanjindong/go-cache v0.0.6/go.mod h1:gxehZ3SqUVta6eFBJAcDlXDT2Q9piXkUqv7s4E0Vj6o= +github.com/fasthttp/router v1.5.4 h1:oxdThbBwQgsDIYZ3wR1IavsNl6ZS9WdjKukeMikOnC8= +github.com/fasthttp/router v1.5.4/go.mod h1:3/hysWq6cky7dTfzaaEPZGdptwjwx0qzTgFCKEWRjgc= github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4= github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL+zU= github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk= @@ -305,6 +307,8 @@ github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWN github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts= github.com/sagikazarmark/locafero v0.7.0 h1:5MqpDsTGNDhY8sGp0Aowyf0qKsPrhewaLSsFaodPcyo= github.com/sagikazarmark/locafero v0.7.0/go.mod h1:2za3Cg5rMaTMoG/2Ulr9AwtFaIppKXTRYnozin4aB5k= +github.com/savsgio/gotils v0.0.0-20240704082632-aef3928b8a38 h1:D0vL7YNisV2yqE55+q0lFuGse6U8lxlg7fYTctlT5Gc= +github.com/savsgio/gotils v0.0.0-20240704082632-aef3928b8a38/go.mod h1:sM7Mt7uEoCeFSCBM+qBrqvEo+/9vdmj19wzp3yzUhmg= github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUtVbo7ada43DJhG55ua/hjS5I= github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc= github.com/segmentio/asm v1.2.0 h1:9BQrFxC+YOHJlTlHGkTrFWf59nbL3XnCoFLTwDCI7ys= @@ -351,6 +355,10 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE= github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg= +github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw= +github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc= +github.com/valyala/fasthttp v1.68.0 h1:v12Nx16iepr8r9ySOwqI+5RBJ/DqTxhOy1HrHoDFnok= +github.com/valyala/fasthttp v1.68.0/go.mod h1:5EXiRfYQAoiO/khu4oU9VISC/eVY6JqmSpPJoHCKsz4= github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI= github.com/xdg-go/scram v1.1.1/go.mod h1:RaEWvsqvNKKvBPvcKeFjrG2cJqOkHTiyTpzz23ni57g= github.com/xdg-go/stringprep v1.0.3/go.mod h1:W3f5j4i+9rC0kuIEJL0ky1VpHXQU3ocBgklLGvcBnW8= diff --git a/internal/gateway/fast_gateway.go b/internal/gateway/fast_gateway.go new file mode 100644 index 0000000..eb9a7d2 --- /dev/null +++ b/internal/gateway/fast_gateway.go @@ -0,0 +1,174 @@ +package gateway + +import ( + "context" + "encoding/json" + "net/http" + "net/url" + "sig-pub/pkg/grpc/generic" + "sig-pub/pkg/grpc/session" + "sig-pub/pkg/resp" + "sig-pub/pkg/utils/strs" + "sig-pub/pkg/zlog" + "strings" + "time" + + "github.com/bytedance/sonic" + "github.com/fasthttp/router" + "github.com/savsgio/gotils/strconv" + "github.com/valyala/fasthttp" + "google.golang.org/grpc" + "google.golang.org/protobuf/encoding/protojson" +) + +const ( + ReadTimeout = 20 * time.Second + WriteTimeout = 20 * time.Second + IdleTimeout = 120 * time.Second +) + +type FastGatewayServer struct { + sigServerURL *url.URL + grpcGenericClientFactory *generic.GrpcGenericClientFactory + + svr *fasthttp.Server + sigClient *fasthttp.HostClient +} + +func NewFastGatewayServer( + sigServerURL *url.URL, + grpcGenericClientFactory *generic.GrpcGenericClientFactory, +) *FastGatewayServer { + return &FastGatewayServer{ + sigServerURL: sigServerURL, + grpcGenericClientFactory: grpcGenericClientFactory, + } +} + +func (g *FastGatewayServer) Init() (err error) { + // fasthttp router + r := router.New() + r.ANY("/api/sig/{path:*}", g.reverseProxyHandler) + r.POST("/api/v1/{path:*}", g.reverseProxyGrpcGenericCall) + + g.svr = &fasthttp.Server{ + Handler: r.Handler, + ReadTimeout: ReadTimeout, + WriteTimeout: WriteTimeout, + IdleTimeout: IdleTimeout, + MaxConnsPerIP: 1024 * 4, + MaxRequestsPerConn: 0, // 0 = unlimited + ReduceMemoryUsage: true, + } + + // sig http service client + g.sigClient = &fasthttp.HostClient{ + Addr: g.sigServerURL.Host, + Name: "gateway", + MaxConns: 512, // 根据实际业务调 + ReadTimeout: ReadTimeout, + WriteTimeout: WriteTimeout, + MaxConnDuration: 10 * time.Minute, + DisableHeaderNamesNormalizing: true, + } + return +} + +func (g *FastGatewayServer) Run(addr string) (err error) { + return g.svr.ListenAndServe(addr) +} + +func (g *FastGatewayServer) reverseProxyHandler(ctx *fasthttp.RequestCtx) { + path := ctx.UserValue("path") + zlog.Infof("path: %s", path) + + req := &ctx.Request + resp := &ctx.Response + + req.Header.Set("Connection", "keep-alive") + err := g.sigClient.Do(req, resp) + if err != nil { + zlog.Errorf("reverse proxy error: url=%s, %v", strconv.B2S(ctx.RequestURI()), err) + } +} + +func responseJSON(ctx *fasthttp.RequestCtx, code int, data any) { + body, err := sonic.Marshal(data) + if err != nil { + zlog.Errorf("marshal resp data error: url=%s, %v", strconv.B2S(ctx.RequestURI()), err) + ctx.SetStatusCode(http.StatusInternalServerError) + return + } + + ctx.SetContentType("application/json; charset=utf-8") + ctx.SetStatusCode(code) + ctx.SetBody(body) +} + +// reverseProxyGrpcGenericCall 代理各个grpc服务请求 +func (s *FastGatewayServer) reverseProxyGrpcGenericCall(c *fasthttp.RequestCtx) { + path := c.UserValue("path").(string) + paths := strings.Split(path, "/") + if len(paths) != 2 { + responseJSON(c, http.StatusBadRequest, resp.Error("service not specified")) + return + } + svr := strs.UpperInitialLetter(paths[0]) + method := strs.UpperInitialLetter(paths[1]) + + if svr == "" || method == "" { + responseJSON(c, http.StatusBadRequest, resp.Error("service not found")) + return + } + if !strings.HasSuffix(svr, "Service") { + svr += "Service" + } + // todo service white list + + // get request body + jsonBody := c.Request.Body() + + ctx1, cancel := context.WithTimeout(context.Background(), time.Second*5) + defer cancel() + grpcGenericClient, err := s.grpcGenericClientFactory.GetClient(ctx1, svr) + if err != nil { + zlog.Error(err) + responseJSON(c, http.StatusForbidden, resp.Error(err.Error())) + return + } + + // put session + ctx := context.Background() + ctx = session.PutSubject(ctx, session.NewRpcSubject("123456")) + ctx, cancel = context.WithTimeout(ctx, time.Second*20) + defer cancel() + + // todo config call options + var opts []grpc.CallOption + if svr == "ExchangeService" && method == "HistoryKline" { + opts = append(opts, grpc.UseCompressor("snappy")) + } + // generic call with json + rsp, err := grpcGenericClient.InvokeUnaryJsonBytes(ctx, method, jsonBody, opts...) + if err != nil { + if err == generic.ErrorMethodNotExists { + responseJSON(c, http.StatusNotFound, resp.Error(err.Error())) + return + } + responseJSON(c, http.StatusInternalServerError, resp.Error(err.Error())) + return + } + + // encode response + bytes, err := protojson.MarshalOptions{ + UseProtoNames: false, // false:lowerCamelCase, true:snake_case + EmitUnpopulated: false, // 是否包含默认值 + }.Marshal(rsp) + if err != nil { + responseJSON(c, http.StatusInternalServerError, resp.Error(err.Error())) + return + } + + r := json.RawMessage(bytes) + responseJSON(c, http.StatusOK, resp.Success(r)) +} diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go index 6900fe6..9cdaca6 100644 --- a/internal/gateway/gateway.go +++ b/internal/gateway/gateway.go @@ -58,16 +58,11 @@ func (s *GateServer) initGinServer() { }) routerGroup := s.engine.Group("/api") - // routerGroup.Any("/sig", s.handleSigServerProxy(s.sigServerURL)) routerGroup.Group("/sig").Any("/", s.handleSigServerProxy(s.sigServerURL)) routerGroup.POST("/v1/:svr/:method", s.handleGrpcGenericCall) } func (s *GateServer) Run(addr string) (err error) { - // fasthttp.ListenAndServe(addr, func(ctx *fasthttp.RequestCtx) { - // ctx.Path() - // ctx.Method() - // }) return s.engine.Run(addr) } diff --git a/internal/sig/service/backtest_service.go b/internal/sig/service/backtest_service.go index 366298e..6f09a25 100644 --- a/internal/sig/service/backtest_service.go +++ b/internal/sig/service/backtest_service.go @@ -3,6 +3,7 @@ package service import ( "net/http" repository "sig-pub/internal/sig/repoitory" + "sig-pub/pkg/resp" "github.com/gin-gonic/gin" ) @@ -22,5 +23,8 @@ func (svc *BacktestService) Route(group *gin.RouterGroup) { } func (svc *BacktestService) listBacktestLog(ctx *gin.Context) { - ctx.JSON(http.StatusOK, "ojbk") + name := ctx.Query("name") + ctx.JSON(http.StatusOK, resp.H{ + "text": "ojbk: " + name, + }) } diff --git a/internal/trading/backtest/sig_strategy_backtester.go b/internal/trading/backtest/sig_strategy_backtester.go index 52741ed..d294ca9 100644 --- a/internal/trading/backtest/sig_strategy_backtester.go +++ b/internal/trading/backtest/sig_strategy_backtester.go @@ -360,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, last=%d", sr.InstId, sr.Interval, err0, driverIntervalAdder(driverSeries.MustGet(0).Ts, 1)) + zlog.Debugf("driver sr finish with: %s(%s), %v", sr.InstId, sr.Interval, err0) // , last=%d , driverIntervalAdder(driverSeries.MustGet(0).Ts, 1)) if err0 == nil { close(stopCh) } else if err0 != errStop { diff --git a/internal/trading/trading_grpc_server.go b/internal/trading/trading_grpc_server.go index 04fbf69..41b8580 100644 --- a/internal/trading/trading_grpc_server.go +++ b/internal/trading/trading_grpc_server.go @@ -4,7 +4,9 @@ import ( "context" "sig-pub/api/pb" "sig-pub/pkg/types" + "sig-pub/pkg/utils/lang" "sig-pub/pkg/utils/times" + "sig-pub/pkg/zlog" "google.golang.org/protobuf/types/known/structpb" ) @@ -121,7 +123,13 @@ func (svr *TradingGrpcServer) BacktestLog(ctx context.Context, req *pb.ReqBackte // BacktestRace 交易计划参数调试回测 func (svr *TradingGrpcServer) BacktestRace(ctx context.Context, req *pb.ReqBacktestRace) (rsp *pb.RspBacktestRace, err error) { rsp = new(pb.RspBacktestRace) - err = svr.tradingService.BacktestRace(ctx, req) + lang.SafeGo(func() { + c := context.Background() + err := svr.tradingService.BacktestRace(c, req) + if err != nil { + zlog.Error("BacktestRace error:", err) + } + }) return } diff --git a/internal/trading/trading_service.go b/internal/trading/trading_service.go index 3f34113..fa7ff0d 100644 --- a/internal/trading/trading_service.go +++ b/internal/trading/trading_service.go @@ -459,7 +459,8 @@ func (svc *TradingService) BacktestRace(ctx context.Context, req *pb.ReqBacktest results = append(results, r) } for _, r := range results { - zlog.Info(r.Id, r.Cash, r.EndCash, r.Profit, r.Singals, r.TotalTrades, r.WinningTrades, r.LosingTrades, r.MaxDrawdown) + winRate := fmt.Sprintf(" %.2f%% ", float64(r.WinningTrades)/float64(r.TotalTrades)*100) + zlog.Info(r.Id, r.Cash, r.EndCash, r.Profit, r.Fee, r.Singals, r.TotalTrades, winRate, r.MaxDrawdown) } return } diff --git a/pkg/grpc/generic/generic_client_factory.go b/pkg/grpc/generic/generic_client_factory.go index b5feadd..34fd255 100644 --- a/pkg/grpc/generic/generic_client_factory.go +++ b/pkg/grpc/generic/generic_client_factory.go @@ -17,7 +17,7 @@ type GrpcGenericClientFactory struct { clientCache *collect.ConcurrentMap[string, *GrpcGenericClient] } -func NewGpcGenericClientFactory(scheme string, defaultOpts ...grpc.DialOption) *GrpcGenericClientFactory { +func NewGrpcGenericClientFactory(scheme string, defaultOpts ...grpc.DialOption) *GrpcGenericClientFactory { return &GrpcGenericClientFactory{ scheme: scheme, defaultOpts: defaultOpts, diff --git a/pkg/indicator/indicator_registry.go b/pkg/indicator/indicator_registry.go index a99ea91..e564f5a 100644 --- a/pkg/indicator/indicator_registry.go +++ b/pkg/indicator/indicator_registry.go @@ -23,9 +23,6 @@ func (r *IndicatorRegistry) Init() (err error) { r.MustRegistIndicator(&ATR{}) r.MustRegistIndicator(&EMA{}) r.MustRegistIndicator(&MACD{}) - r.MustRegistIndicator(&MacdDIF{}) - r.MustRegistIndicator(&MacdDEA{}) - r.MustRegistIndicator(&Macd{}) r.MustRegistIndicator(&OBV{}) r.MustRegistIndicator(&WOBV{}) r.MustRegistIndicator(&BollMB{}) diff --git a/pkg/indicator/macd.go b/pkg/indicator/macd.go index 0a25b6d..177af7f 100644 --- a/pkg/indicator/macd.go +++ b/pkg/indicator/macd.go @@ -2,6 +2,9 @@ package indicator import "sig-pub/pkg/types" +// MACD 分成: Hist(柱状图), DIF线, DEA(信号线) +// 计算 DIF线: 反映短期趋势与长期趋势的“收敛/散度” +// MACD: https://www.investopedia.com/terms/m/macd.asp type MACD struct { } diff --git a/pkg/indicator/macd0.go b/pkg/indicator/macd0.go deleted file mode 100644 index fe2b170..0000000 --- a/pkg/indicator/macd0.go +++ /dev/null @@ -1,110 +0,0 @@ -package indicator - -import ( - "sig-pub/pkg/types" -) - -// Macd macd柱状图计算 - -// Macd 拆分成: Macd(柱状图), MacdDIF线, MacdDEA(信号线) -// 计算 MacdDIF 线 (DIF): 反映短期趋势与长期趋势的“收敛/散度” -// Macd: https://www.investopedia.com/terms/m/macd.asp -type Macd struct { -} - -func (c *Macd) Meta() IndicatorMeta { - return IndicatorMeta{ - Name: "Macd", - Input: []types.InputArg{ - {Name: "fast", Type: types.InputTypeUInt, Desc: "快线周期"}, - {Name: "slow", Type: types.InputTypeUInt, Desc: "慢线周期"}, - {Name: "singal", Type: types.InputTypeUInt, Desc: "信号线周期"}, - }, - } -} - -func (c *Macd) CandlePeriods(ctx IIndicatorContext) int16 { - return max( - ctx.Indicator("MacdDIF", ctx.Input()).CandlePeriods(), - ctx.Indicator("MacdDEA", ctx.Input()).CandlePeriods(), - ) -} - -func (c *Macd) Calculate(ctx IIndicatorContext) (vector float64) { - macd_dea := ctx.Indicator("MacdDEA", ctx.Input()).Get(0) - macd_dif := ctx.Indicator("MacdDIF", ctx.Input()).Get(0) - vector = (macd_dif - macd_dea) * 2 - return -} - -type MacdDIF struct { -} - -// indicator interface -func (c *MacdDIF) Meta() IndicatorMeta { - return IndicatorMeta{ - Name: "MacdDIF", - Input: []types.InputArg{ - {Name: "fast", Type: types.InputTypeUInt, Desc: "快线周期"}, - {Name: "slow", Type: types.InputTypeUInt, Desc: "慢线周期"}, - }, - } -} - -func (c *MacdDIF) CandlePeriods(ctx IIndicatorContext) int16 { - return max( - ctx.Indicator("EMA", ctx.Input().Int16("fast")).CandlePeriods(), - ctx.Indicator("EMA", ctx.Input().Int16("slow")).CandlePeriods(), - ) -} - -// Calculate 计算单根k线sma指标 -func (c *MacdDIF) Calculate(ctx IIndicatorContext) (vector float64) { - fast := ctx.Input().Int16("fast") // 12 - slow := ctx.Input().Int16("slow") // 26 - - // macd计算从第max(fast, slow)期开始稳定 - fastEma := ctx.Indicator("EMA", fast).Get(0) - slowEma := ctx.Indicator("EMA", slow).Get(0) - macd := fastEma - slowEma - vector = macd - return -} - -// MacdDEA macd信号线计算 -type MacdDEA struct { -} - -func (c *MacdDEA) Meta() IndicatorMeta { - return IndicatorMeta{ - Name: "MacdDEA", - Input: []types.InputArg{ - {Name: "fast", Type: types.InputTypeUInt, Desc: "快线周期"}, - {Name: "slow", Type: types.InputTypeUInt, Desc: "慢线周期"}, - {Name: "singal", Type: types.InputTypeUInt, Desc: "信号线周期"}, - }, - } -} - -func (c *MacdDEA) CandlePeriods(ctx IIndicatorContext) int16 { - return ctx.Indicator("MacdDIF", ctx.Input()).CandlePeriods() + ctx.Input().Int16("singal") + 1 -} - -func (c *MacdDEA) Calculate(ctx IIndicatorContext) (vector float64) { - singal := ctx.Input().Int16("singal") // 9 - - deaPrev, ok := ctx.State().Get("_vector", 1) - if !ok { - // 初始值前9期的 MACD_DIF SMA - macdDifs := ctx.Indicator("MacdDIF", ctx.Input()).Series(1, singal) - deaPrev = macdDifs.Avg() - } - macd_dif := ctx.Indicator("MacdDIF", ctx.Input()).Get(0) - // 计算DEA - beta := 2 / float64(singal+1) - dea := beta*macd_dif + (1-beta)*deaPrev - ctx.State().Set("_vector", dea) - - vector = dea - return -} diff --git a/pkg/strategy/super_trend_macd_rsi.go b/pkg/strategy/super_trend_macd_rsi.go index ac603ab..29737c1 100644 --- a/pkg/strategy/super_trend_macd_rsi.go +++ b/pkg/strategy/super_trend_macd_rsi.go @@ -1,9 +1,16 @@ package strategy -import "sig-pub/pkg/types" +import ( + "fmt" + "sig-pub/pkg/types" +) // SuperTrendMacdRSI 结合super trend和rsi指标策略 type SuperTrendMacdRSI struct { + trendWindow int16 + trendMul float64 // super trend + rsi int16 // rsi + fast, slow, singal int16 // macd } func (s *SuperTrendMacdRSI) New() ISigStrategy { @@ -12,43 +19,53 @@ func (s *SuperTrendMacdRSI) New() ISigStrategy { func (s *SuperTrendMacdRSI) Meta() StrategyMeta { return StrategyMeta{ - Name: "SuperTrendMacdRSI", - Desc: "SuperTrend + MACD + RSI 量化策略", + Name: "SuperTrendMacdRSI", + Desc: "SuperTrend + MACD + RSI 量化策略", Input: []types.InputArg{ - // {Name: "trendWindow", Type: types.InputTypeUInt, Desc: "SuperTrend ATR周期"}, - // {Name: "trendMultipiler", Type: types.InputTypeUInt, Desc: "SuperTrend multipiler"}, - // {Name: "rsiWindow", Type: types.InputTypeUInt, Desc: "rsi周期"}, + {Name: "trendWindow", Type: types.InputTypeUInt, Desc: "SuperTrend ATR周期"}, // 10 + {Name: "trendMul", Type: types.InputTypeUFloat, Desc: "SuperTrend multipiler"}, // 3 + {Name: "rsi", Type: types.InputTypeUInt, Desc: "rsi周期"}, // 14 + {Name: "fast", Type: types.InputTypeUInt, Desc: "macd fast period"}, // 12 + {Name: "slow", Type: types.InputTypeUInt, Desc: "macd slow period"}, // 26 + {Name: "singal", Type: types.InputTypeUInt, Desc: "macd singal period"}, // 9 }, } } // Init 校验参数, 并根据参数初始化策略 func (s *SuperTrendMacdRSI) Init(input types.Input) (err error) { + s.trendWindow = input.Int16("trendWindow") + s.trendMul = input.Float("trendMul") + s.rsi = input.Int16("rsi") + s.fast = input.Int16("fast") + s.slow = input.Int16("slow") + s.singal = input.Int16("singal") + if s.fast >= s.slow { + return fmt.Errorf("macd fast(%d) >= slow(%d)", s.fast, s.slow) + } return } func (s *SuperTrendMacdRSI) CandlePeriods(ctx ISingleSigStrategyContext) int16 { return max( - ctx.Indicator("SuperTrend", types.Input{"window": 10, "mul": 3}).CandlePeriods(), - ctx.Indicator("RSI", 14).CandlePeriods(), - ctx.Indicator("MACD", types.Input{"fast": 12, "slow": 26, "singal": 9}).CandlePeriods(), - // ctx.Indicator("MacdDEA", types.Input{"fast": 12, "slow": 26, "singal": 9}).CandlePeriods(), - // ctx.Indicator("MacdDEA", types.Input{"fast": 12, "slow": 26, "singal": 9}).CandlePeriods(), - // ctx.Indicator("MacdDIF", types.Input{"fast": 12, "slow": 26, "singal": 9}).CandlePeriods(), + ctx.Indicator("SuperTrend", types.Input{"window": s.trendWindow, "mul": s.trendMul}).CandlePeriods(), + ctx.Indicator("RSI", s.rsi).CandlePeriods(), + ctx.Indicator("MACD", types.Input{ + "fast": s.fast, + "slow": s.slow, + "singal": s.singal, + }).CandlePeriods(), 21, ) } func (s *SuperTrendMacdRSI) Update(ctx ISingleSigStrategyContext) (side types.Side) { - superTrend := ctx.Indicator("SuperTrend", types.Input{"window": 10, "mul": 3}) - rsi := ctx.Indicator("RSI", 14).Get(0) - macd := ctx.Indicator("MACD", types.Input{"fast": 12, "slow": 26, "singal": 9}) + superTrend := ctx.Indicator("SuperTrend", types.Input{"window": s.trendWindow, "mul": s.trendMul}) + rsi := ctx.Indicator("RSI", s.rsi).Get(0) + macd := ctx.Indicator("MACD", types.Input{"fast": s.fast, "slow": s.slow, "singal": s.singal}) macdHist := macd.Get(0) macdDea := macd.StateSeries("dea", 0, 2) // macd_dea信号线 macdDif := macd.StateSeries("dif", 0, 2) // macd_dif线 - // macdHist := ctx.Indicator("Macd", types.Input{"fast": 12, "slow": 26, "singal": 9}).Get(0) // macd柱 - // macdDea := ctx.Indicator("MacdDEA", types.Input{"fast": 12, "slow": 26, "singal": 9}).Series(0, 2) // macd_dea信号线 - // macdDif := ctx.Indicator("MacdDIF", types.Input{"fast": 12, "slow": 26, "singal": 9}).Series(0, 2) // macd_dif线 crossover := macdDif[0] > macdDea[0] && macdDif[1] < macdDea[1] // 金叉 crossunder := macdDif[0] < macdDea[0] && macdDif[1] > macdDea[1] // 死叉 @@ -63,7 +80,7 @@ func (s *SuperTrendMacdRSI) Update(ctx ISingleSigStrategyContext) (side types.Si // 金叉状态且正向扩张 if crossover && macdHist > 0 { // RSI 强度过滤 - if rsi > 50 { + if rsi > 50 && rsi < 70 { // SuperTrend 趋势确认 if closeP > trend && trendDirection == 1 { // 成交量过滤 @@ -75,7 +92,7 @@ func (s *SuperTrendMacdRSI) Update(ctx ISingleSigStrategyContext) (side types.Si } if crossunder && macdHist < 0 { - if rsi < 50 { + if rsi > 30 && rsi < 50 { // SuperTrend 趋势确认 if closeP < trend && trendDirection == -1 { // 成交量过滤 diff --git a/pkg/strategy/super_trend_rsi.go b/pkg/strategy/super_trend_rsi.go index e8b118f..8bcde9c 100644 --- a/pkg/strategy/super_trend_rsi.go +++ b/pkg/strategy/super_trend_rsi.go @@ -16,8 +16,8 @@ func (s *SuperTrendRSI) Meta() StrategyMeta { Desc: "超级趋势结合RSI(动能过滤)策略", Input: []types.InputArg{ {Name: "trendWindow", Type: types.InputTypeUInt, Desc: "SuperTrend ATR周期"}, - {Name: "trendMultipiler", Type: types.InputTypeUInt, Desc: "SuperTrend multipiler"}, - {Name: "rsiWindow", Type: types.InputTypeUInt, Desc: "rsi周期"}, + {Name: "trend", Type: types.InputTypeUInt, Desc: "SuperTrend multipiler"}, + {Name: "rsi", Type: types.InputTypeUInt, Desc: "rsi周期"}, }, } } @@ -31,18 +31,18 @@ func (s *SuperTrendRSI) CandlePeriods(ctx ISingleSigStrategyContext) int16 { return max( ctx.Indicator("SuperTrend", types.Input{ "window": ctx.Input().Int16("trendWindow"), - "mul": ctx.Input().Int16("trendMultipiler"), + "mul": ctx.Input().Int16("trendMul"), }).CandlePeriods(), - ctx.Indicator("RSI", types.Input{"window": ctx.Input().Int16("rsiWindow")}).CandlePeriods(), + ctx.Indicator("RSI", types.Input{"window": ctx.Input().Int16("rsi")}).CandlePeriods(), ) } func (s *SuperTrendRSI) Update(ctx ISingleSigStrategyContext) (side types.Side) { superTrend := ctx.Indicator("SuperTrend", types.Input{ "window": ctx.Input().Int16("trendWindow"), - "mul": ctx.Input().Int16("trendMultipiler"), + "mul": ctx.Input().Int16("trendMul"), }) - rsi := ctx.Indicator("RSI", types.Input{"window": ctx.Input().Int16("rsiWindow")}) + rsi := ctx.Indicator("RSI", types.Input{"window": ctx.Input().Int16("rsi")}) // 買入：SuperTrend 轉綠（綠線在價格下方）且 RSI > 50（確認動能向上） // 賣出：SuperTrend 轉紅，或 RSI < 30（超賣退出） diff --git a/pkg/utils/codec/mapstructure.go b/pkg/utils/codec/mapstructure.go index fd6c0f1..249e7d1 100644 --- a/pkg/utils/codec/mapstructure.go +++ b/pkg/utils/codec/mapstructure.go @@ -1,7 +1,6 @@ package codec import ( - "fmt" "reflect" "strconv" "strings" @@ -26,7 +25,6 @@ func MapDecode(input, output any) (err error) { // 方案1：最推荐 - 字符串 → 数字（int/uint/float）全覆盖 func StringToNumberHook() mapstructure.DecodeHookFunc { return mapstructure.DecodeHookFuncType(func(from reflect.Type, to reflect.Type, data interface{}) (interface{}, error) { - fmt.Println("hook1") if from.Kind() == reflect.String { str := data.(string) str = strings.TrimSpace(str) diff --git a/pkg/utils/lang/concurrent.go b/pkg/utils/lang/concurrent.go new file mode 100644 index 0000000..34e8828 --- /dev/null +++ b/pkg/utils/lang/concurrent.go @@ -0,0 +1,17 @@ +package lang + +import ( + "runtime/debug" + "sig-pub/pkg/zlog" +) + +func SafeGo(fn func()) { + go func() { + defer func() { + if err := recover(); err != nil { + zlog.Errorf("safe run error %v, stack info %v", err, string(debug.Stack())) + } + }() + fn() + }() +}