From 23a6af8ae7633d517bbe9d93b718822e0c9a97ea Mon Sep 17 00:00:00 2001 From: strange Date: Fri, 26 Dec 2025 18:21:00 +0800 Subject: [PATCH] gateway, sig server --- README.md | 6 + api/trading.proto | 29 +++- cmd/gateway/main.go | 57 +++++++- cmd/market/main.go | 2 +- cmd/sig-admin/main.go | 41 ++---- cmd/trading/main.go | 2 +- config/gate.toml | 3 - config/gateway.toml | 5 + go.mod | 12 +- go.sum | 15 ++ internal/gateway/gateway.go | 138 ++++++++++++++++-- internal/market/trade_instance_service.go | 4 +- internal/sig/repoitory/backtest_repository.go | 17 +++ internal/sig/service/backtest_service.go | 26 ++++ internal/sig/service/service.go | 15 ++ internal/sig/sig_server.go | 93 +----------- internal/trading/backtest/trade_simulator.go | 25 +--- internal/trading/backtest/types.go | 28 +--- internal/trading/trading_data_persist.go | 9 +- pkg/indicator/indicator_plot.go | 1 + pkg/indicator/indicator_registry.go | 1 + pkg/indicator/kdj.go | 73 +++++++++ pkg/storage/persist/rdb.go | 20 +-- 23 files changed, 410 insertions(+), 212 deletions(-) delete mode 100644 config/gate.toml create mode 100644 config/gateway.toml create mode 100644 internal/sig/repoitory/backtest_repository.go create mode 100644 internal/sig/service/backtest_service.go create mode 100644 internal/sig/service/service.go create mode 100644 pkg/indicator/kdj.go diff --git a/README.md b/README.md index b372950..2836ffb 100644 --- a/README.md +++ b/README.md @@ -138,3 +138,9 @@ RSI[1,2,3,4] -> RSI[0] exchange_service.go: 100 task/一批, 批量成功后mark, 再发布下一波 viceAccount 负账户对冲交易 {Long: mainAccount, Short: viceAccount} + +page design: + - 指标页: 币种选择 -> 周期选择 -> 指标选择 -> 参数选择 -> 绘图 + - 交易计划: 交易策略 -> 交易所,币种,驱动周期 -> risk,sig,close,trade_strategy_param -> + - 回测页: 交易策略, 跑策略, 结果绘图 + - 参数遍历回测 diff --git a/api/trading.proto b/api/trading.proto index 98015f3..2749409 100644 --- a/api/trading.proto +++ b/api/trading.proto @@ -88,18 +88,35 @@ message RspBacktestLog { repeated BacktestLog logs = 1; } - -message BacktestTrade { - int64 id = 1; - int64 ctime = 2; // 交易时间 - Side side = 3; // 交易方向 +message BacktestTradeOrder { + int64 backtest_id = 1; + int64 trade_id = 2; + int32 trade_type = 3; + string inst_id = 4; + Side side = 5; + double qty = 6; + double price = 7; + double fee = 8; + int32 leverage = 9; + int64 ctime = 10; + int32 status = 11; + double peak_px = 12; + string close_cause = 13; + double equity = 14; + string hold_time = 15; + double profit = 16; + double last_px = 17; + double entry_px = 18; + double entry_fee = 19; + int64 entry_time = 20; + repeated int64 trades = 21; } message ReqBacktestLogTrades { Paging paging = 1; int64 backtest_id = 2; } message RspBacktestLogTrades { - repeated BacktestTrade trades = 1; + repeated BacktestTradeOrder trades = 1; } message ReqBacktestRace { diff --git a/cmd/gateway/main.go b/cmd/gateway/main.go index 7959503..581383a 100644 --- a/cmd/gateway/main.go +++ b/cmd/gateway/main.go @@ -1,21 +1,68 @@ package main import ( + "fmt" + "net/url" "sig-pub/internal/gateway" "sig-pub/pkg/config" + "sig-pub/pkg/grpc/discovery" + "sig-pub/pkg/grpc/generic" + "sig-pub/pkg/utils/exit" + + "github.com/hashicorp/consul/api" + _ "github.com/mostynb/go-grpc-compression/snappy" // 注册grpc snappy compress + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" ) type GateConf struct { - HttpAddr string - WsAddr string + HttpAddr string + WsAddr string + SigServer string } // http 网关 func main() { - gateConf := config.MustLoadConfig(new(GateConf), "config/gate.toml") - server := gateway.Route() - err := server.Run(gateConf.HttpAddr) + // load config + conf := config.MustLoadConfig(new(config.Configuration), "config/config.toml") + gateConf := config.MustLoadConfig(new(GateConf), "config/gateway.toml") + + // consul 配置 + cc := api.DefaultConfig() + cc.Address = conf.Consul.Address + client, err := api.NewClient(cc) if err != nil { + panic(fmt.Errorf("consul client error: %v", err)) + } + + // consul service discovery + dis := discovery.NewConsulDiscovery(client) + resolver := dis.Resolver() + gpcGenericClientFactory := generic.NewGpcGenericClientFactory( + discovery.ConsulSchema, + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithResolvers(resolver), + ) + if err := gpcGenericClientFactory.Init(); err != nil { panic(err) } + if err := dis.WatchServices(gpcGenericClientFactory.RefreshService); err != nil { + panic(err) + } + + sigServerUrl, err := url.Parse(gateConf.SigServer) + if err != nil { + panic(err) + } + gateServer := gateway.NewGateServer(sigServerUrl, gpcGenericClientFactory) + if err := gateServer.Init(); err != nil { + panic(err) + } + go func() { + if err := gateServer.Run(gateConf.HttpAddr); err != nil { + panic(err) + } + }() + + exit.Await() } diff --git a/cmd/market/main.go b/cmd/market/main.go index 6a65932..b477d20 100644 --- a/cmd/market/main.go +++ b/cmd/market/main.go @@ -41,7 +41,7 @@ func main() { if err != nil { panic(err) } - rdb := persist.NewRDB(db) + rdb := persist.NewDB(db) if err := rdb.Init(); err != nil { panic(err) } diff --git a/cmd/sig-admin/main.go b/cmd/sig-admin/main.go index 620dc33..fce2faa 100644 --- a/cmd/sig-admin/main.go +++ b/cmd/sig-admin/main.go @@ -1,54 +1,37 @@ package main import ( - "fmt" "sig-pub/internal/sig" "sig-pub/pkg/config" - "sig-pub/pkg/grpc/discovery" - "sig-pub/pkg/grpc/generic" + "sig-pub/pkg/storage/persist" "sig-pub/pkg/utils/exit" - - "github.com/hashicorp/consul/api" - _ "github.com/mostynb/go-grpc-compression/snappy" // 注册grpc snappy compress - "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" ) +// curl server: /api/sig/backtest/log + func main() { - defer exit.Await() // load config conf := config.MustLoadConfig(new(config.Configuration), "config/config.toml") - // consul 配置 - cc := api.DefaultConfig() - cc.Address = conf.Consul.Address - client, err := api.NewClient(cc) + // database + db, err := conf.Database.Postgres.NewGormDB() if err != nil { - panic(fmt.Errorf("consul client error: %v", err)) - } - - // consul service discovery - dis := discovery.NewConsulDiscovery(client) - resolver := dis.Resolver() - gpcGenericClientFactory := generic.NewGpcGenericClientFactory( - discovery.ConsulSchema, - grpc.WithTransportCredentials(insecure.NewCredentials()), - grpc.WithResolvers(resolver), - ) - if err := gpcGenericClientFactory.Init(); err != nil { panic(err) } - if err := dis.WatchServices(gpcGenericClientFactory.RefreshService); err != nil { + rdb := persist.NewDB(db) + if err := rdb.Init(); err != nil { panic(err) } - sigServer := sig.NewSigServer(gpcGenericClientFactory) - if err := sigServer.Init(); err != nil { + sigServer := sig.NewSigServer(rdb) + if err = sigServer.Init(); err != nil { panic(err) } go func() { - if err := sigServer.Run(":7001"); err != nil { + if err := sigServer.Run(":7009"); err != nil { panic(err) } }() + + exit.Await() } diff --git a/cmd/trading/main.go b/cmd/trading/main.go index c061f76..50252ad 100644 --- a/cmd/trading/main.go +++ b/cmd/trading/main.go @@ -52,7 +52,7 @@ func main() { if err != nil { panic(err) } - rdb := persist.NewRDB(db) + rdb := persist.NewDB(db) if err := rdb.Init(); err != nil { panic(err) } diff --git a/config/gate.toml b/config/gate.toml deleted file mode 100644 index bc2701c..0000000 --- a/config/gate.toml +++ /dev/null @@ -1,3 +0,0 @@ - -httpAddr = ":7001" -wsAddr = ":7101" diff --git a/config/gateway.toml b/config/gateway.toml new file mode 100644 index 0000000..7e78e29 --- /dev/null +++ b/config/gateway.toml @@ -0,0 +1,5 @@ + +httpAddr = ":7001" +wsAddr = ":7101" + +sigServer = "http://127.0.0.1:7009" # 后台服务 diff --git a/go.mod b/go.mod index 54d3103..f67b67b 100644 --- a/go.mod +++ b/go.mod @@ -19,7 +19,7 @@ require ( github.com/influxdata/influxdb-client-go/v2 v2.14.0 github.com/jackc/pgx/v5 v5.7.6 github.com/jhump/protoreflect/v2 v2.0.0-beta.2 - github.com/klauspost/compress v1.18.0 + github.com/klauspost/compress v1.18.2 github.com/lib/pq v1.10.9 github.com/mostynb/go-grpc-compression v1.2.3 github.com/nats-io/nats.go v1.47.0 @@ -29,8 +29,8 @@ require ( 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 - golang.org/x/crypto v0.42.0 - golang.org/x/net v0.44.0 + golang.org/x/crypto v0.43.0 + golang.org/x/net v0.46.0 golang.org/x/sync v0.17.0 golang.org/x/time v0.8.0 gonum.org/v1/gonum v0.16.0 @@ -109,6 +109,8 @@ 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 + github.com/valyala/fasthttp v1.68.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 @@ -116,8 +118,8 @@ require ( go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/arch v0.15.0 // indirect golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 // indirect - golang.org/x/sys v0.36.0 // indirect - golang.org/x/text v0.29.0 // indirect + golang.org/x/sys v0.37.0 // indirect + golang.org/x/text v0.30.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20250303144028-a0af3efb3deb // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20250303144028-a0af3efb3deb // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index f41aeb6..fab1cd1 100644 --- a/go.sum +++ b/go.sum @@ -207,6 +207,8 @@ github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+o github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk= github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= +github.com/klauspost/compress v1.18.2 h1:iiPHWW0YrcFgpBYhsA6D1+fqHssJscY/Tm/y2Uqnapk= +github.com/klauspost/compress v1.18.2/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4= github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= @@ -351,6 +353,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= @@ -396,6 +402,8 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/crypto v0.42.0 h1:chiH31gIWm57EkTXpwnqf8qeuMUi0yekh6mT2AvFlqI= golang.org/x/crypto v0.42.0/go.mod h1:4+rDnOTJhQCx2q7/j6rAN5XDw8kPjeaXEUR2eL94ix8= +golang.org/x/crypto v0.43.0 h1:dduJYIi3A3KOfdGOHX8AVZ/jGiyPa3IbBozJ5kNuE04= +golang.org/x/crypto v0.43.0/go.mod h1:BFbav4mRNlXJL4wNeejLpWxB7wMbc79PdRGhWKncxR0= golang.org/x/exp v0.0.0-20250305212735-054e65f0b394 h1:nDVHiLt8aIbd/VzvPWN6kSOPE7+F/fNFDSXLVYkE/Iw= golang.org/x/exp v0.0.0-20250305212735-054e65f0b394/go.mod h1:sIifuuw/Yco/y6yb6+bDNfyeQ/MdPUy/hKEMYQV17cM= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= @@ -412,6 +420,8 @@ golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1/go.mod h1:9tjilg8BloeKEkVJvy golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.44.0 h1:evd8IRDyfNBMBTTY5XRF1vaZlD+EmWx6x8PkhR04H/I= golang.org/x/net v0.44.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY= +golang.org/x/net v0.46.0 h1:giFlY12I07fugqwPuWJi68oOnpfqFnJIJzaIIm2JVV4= +golang.org/x/net v0.46.0/go.mod h1:Q9BGdFy1y4nkUwiLvT5qtyhAnEHgnQ/zd8PfU6nc210= golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -447,9 +457,12 @@ golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.36.0 h1:KVRy2GtZBrk1cBYA7MKu5bEZFxQk4NIDV6RLVcC8o0k= golang.org/x/sys v0.36.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ= +golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.35.0 h1:bZBVKBudEyhRcajGcNc3jIfWPqV4y/Kt2XcoigOWtDQ= golang.org/x/term v0.35.0/go.mod h1:TPGtkTLesOwf2DE8CgVYiZinHAOuy5AYUYT1lENIZnA= +golang.org/x/term v0.36.0 h1:zMPR+aF8gfksFprF/Nc/rd1wRS1EI6nDBGyWAvDzx2Q= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= @@ -457,6 +470,8 @@ golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk= golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4= +golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k= +golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM= golang.org/x/time v0.8.0 h1:9i3RxcPv3PZnitoVGMPDKZSq1xW1gK1Xy3ArNOGZfEg= golang.org/x/time v0.8.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go index 1f28840..6900fe6 100644 --- a/internal/gateway/gateway.go +++ b/internal/gateway/gateway.go @@ -1,19 +1,50 @@ package gateway import ( + "context" + "encoding/json" + "io" "net/http" + "net/http/httputil" + "net/url" "runtime/debug" - "sig-pub/pkg/utils/resp" + "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/gin-gonic/gin" + "google.golang.org/grpc" + "google.golang.org/protobuf/encoding/protojson" ) -// grpc generic call +type GateServer struct { + engine *gin.Engine + sigServerURL *url.URL + grpcGenericClientFactory *generic.GrpcGenericClientFactory +} + +func NewGateServer( + sigServerURL *url.URL, + grpcGenericClientFactory *generic.GrpcGenericClientFactory, +) *GateServer { + return &GateServer{ + sigServerURL: sigServerURL, + grpcGenericClientFactory: grpcGenericClientFactory, + } +} + +func (s *GateServer) Init() (err error) { + s.initGinServer() + return +} -func Route() *gin.Engine { - router := gin.Default() - router.Use(func(c *gin.Context) { +func (s *GateServer) initGinServer() { + s.engine = gin.Default() + s.engine.Use(func(c *gin.Context) { // global recover defer func() { if r := recover(); r != nil { @@ -23,13 +54,100 @@ func Route() *gin.Engine { c.Abort() } }() - c.Next() }) - router.GET("/api/hello", func(c *gin.Context) { - c.JSON(http.StatusOK, resp.Success("hello")) - }) + 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) +} + +// handleSigServerProxy 代理sig http server请求 +func (s *GateServer) handleSigServerProxy(sigServerURL *url.URL) gin.HandlerFunc { + // 创建反向代理 + proxy := httputil.NewSingleHostReverseProxy(sigServerURL) + // 修改请求头、Host 等 + proxy.Director = func(req *http.Request) { + req.URL.Scheme = sigServerURL.Scheme + req.URL.Host = sigServerURL.Host + req.Host = sigServerURL.Host + } + return func(c *gin.Context) { + proxy.ServeHTTP(c.Writer, c.Request) + } +} + +// handleGrpcGenericCall 代理各个grpc服务请求 +func (s *GateServer) handleGrpcGenericCall(c *gin.Context) { + svr := strs.UpperInitialLetter(c.Param("svr")) + method := strs.UpperInitialLetter(c.Param("method")) + + if svr == "" || method == "" { + c.JSON(http.StatusBadRequest, resp.Error("service not found")) + return + } + if !strings.HasSuffix(svr, "Service") { + svr += "Service" + } + // todo service white list + + // get request body + jsonBody, err := io.ReadAll(c.Request.Body) + if err != nil { + c.JSON(http.StatusBadRequest, resp.Error("parse request body error: "+err.Error())) + return + } + + ctx1, cancel := context.WithTimeout(context.Background(), time.Second*5) + defer cancel() + grpcGenericClient, err := s.grpcGenericClientFactory.GetClient(ctx1, svr) + if err != nil { + zlog.Error(err) + c.JSON(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 { + c.JSON(http.StatusNotFound, resp.Error(err.Error())) + return + } + c.JSON(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 { + c.JSON(http.StatusInternalServerError, resp.Error(err.Error())) + return + } - return router + r := json.RawMessage(bytes) + c.JSON(http.StatusOK, resp.Success(r)) } diff --git a/internal/market/trade_instance_service.go b/internal/market/trade_instance_service.go index ae10429..3cff8a1 100644 --- a/internal/market/trade_instance_service.go +++ b/internal/market/trade_instance_service.go @@ -13,10 +13,10 @@ import ( // TradeInstanceService 交易产品管理 // TODO cache type TradeInstanceService struct { - db *persist.RDB + db *persist.DB } -func NewTradeInstanceService(db *persist.RDB) *TradeInstanceService { +func NewTradeInstanceService(db *persist.DB) *TradeInstanceService { return &TradeInstanceService{ db: db, } diff --git a/internal/sig/repoitory/backtest_repository.go b/internal/sig/repoitory/backtest_repository.go new file mode 100644 index 0000000..63061d2 --- /dev/null +++ b/internal/sig/repoitory/backtest_repository.go @@ -0,0 +1,17 @@ +package repository + +import "sig-pub/pkg/storage/persist" + +type BacktestRepository struct { + rdb *persist.DB +} + +func NewBacktestRepository(rdb *persist.DB) *BacktestRepository { + return &BacktestRepository{ + rdb: rdb, + } +} + +func (r *BacktestRepository) ListLog() { + +} diff --git a/internal/sig/service/backtest_service.go b/internal/sig/service/backtest_service.go new file mode 100644 index 0000000..366298e --- /dev/null +++ b/internal/sig/service/backtest_service.go @@ -0,0 +1,26 @@ +package service + +import ( + "net/http" + repository "sig-pub/internal/sig/repoitory" + + "github.com/gin-gonic/gin" +) + +type BacktestService struct { + repo *repository.BacktestRepository +} + +func NewBacktestService(repo *repository.BacktestRepository) *BacktestService { + return &BacktestService{ + repo: repo, + } +} + +func (svc *BacktestService) Route(group *gin.RouterGroup) { + group.GET("hello", svc.listBacktestLog) +} + +func (svc *BacktestService) listBacktestLog(ctx *gin.Context) { + ctx.JSON(http.StatusOK, "ojbk") +} diff --git a/internal/sig/service/service.go b/internal/sig/service/service.go new file mode 100644 index 0000000..cd3b0c5 --- /dev/null +++ b/internal/sig/service/service.go @@ -0,0 +1,15 @@ +package service + +import ( + repository "sig-pub/internal/sig/repoitory" + "sig-pub/pkg/storage/persist" + + "github.com/gin-gonic/gin" +) + +// Init services list +func Init(group *gin.RouterGroup, rdb *persist.DB) { + backtestRepository := repository.NewBacktestRepository(rdb) + + NewBacktestService(backtestRepository).Route(group.Group("/backtest")) +} diff --git a/internal/sig/sig_server.go b/internal/sig/sig_server.go index 0f00224..8e7fcde 100644 --- a/internal/sig/sig_server.go +++ b/internal/sig/sig_server.go @@ -1,32 +1,24 @@ package sig import ( - "context" - "encoding/json" - "io" "net/http" "runtime/debug" - "sig-pub/pkg/grpc/generic" - "sig-pub/pkg/grpc/session" + "sig-pub/internal/sig/service" "sig-pub/pkg/resp" - "sig-pub/pkg/utils/strs" + "sig-pub/pkg/storage/persist" "sig-pub/pkg/zlog" - "strings" - "time" "github.com/gin-gonic/gin" - "google.golang.org/grpc" - "google.golang.org/protobuf/encoding/protojson" ) type SigServer struct { - engine *gin.Engine - grpcGenericClientFactory *generic.GrpcGenericClientFactory + rdb *persist.DB + engine *gin.Engine } -func NewSigServer(grpcGenericClientFactory *generic.GrpcGenericClientFactory) *SigServer { +func NewSigServer(rdb *persist.DB) *SigServer { return &SigServer{ - grpcGenericClientFactory: grpcGenericClientFactory, + rdb: rdb, } } @@ -50,79 +42,10 @@ func (s *SigServer) initGinServer() { c.Next() }) - routerGroup := s.engine.Group("/api") - routerGroup.POST("/v1/:svr/:method", s.handleGrpcGenericCall) - // inst api - // tradeInstanceApi := inst.NewTradeInstanceApi() - // tradeInstanceApi.InitRoute(routerGroup) + routerGroup := s.engine.Group("/api/sig") + service.Init(routerGroup, s.rdb) } func (s *SigServer) Run(addr string) (err error) { - // todo grpc server run return s.engine.Run(addr) } - -func (s *SigServer) handleGrpcGenericCall(c *gin.Context) { - svr := strs.UpperInitialLetter(c.Param("svr")) - method := strs.UpperInitialLetter(c.Param("method")) - - if svr == "" || method == "" { - c.JSON(http.StatusBadRequest, resp.Error("service not found")) - return - } - if !strings.HasSuffix(svr, "Service") { - svr += "Service" - } - // todo service white list - - // get request body - jsonBody, err := io.ReadAll(c.Request.Body) - if err != nil { - c.JSON(http.StatusBadRequest, resp.Error("parse request body error: "+err.Error())) - return - } - - ctx1, cancel := context.WithTimeout(context.Background(), time.Second*5) - defer cancel() - grpcGenericClient, err := s.grpcGenericClientFactory.GetClient(ctx1, svr) - if err != nil { - zlog.Error(err) - c.JSON(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 { - c.JSON(http.StatusNotFound, resp.Error(err.Error())) - return - } - c.JSON(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 { - c.JSON(http.StatusInternalServerError, resp.Error(err.Error())) - return - } - - r := json.RawMessage(bytes) - c.JSON(http.StatusOK, resp.Success(r)) -} diff --git a/internal/trading/backtest/trade_simulator.go b/internal/trading/backtest/trade_simulator.go index 43226f4..b77f904 100644 --- a/internal/trading/backtest/trade_simulator.go +++ b/internal/trading/backtest/trade_simulator.go @@ -51,27 +51,4 @@ func (s *TradeSimulator) ExecuteMarket(tradeId int64, symbol string, ticket trad return } -// ExecuteLimit 简单实现: 如果limit价格被kline的high/low包含则成交 -func (s *TradeSimulator) ExecuteLimit(side types.Side, qty float64, limitPx float64, k types.Kline, ts int64) (trd *Trade, filled bool) { - h := decimals.MustToFloat64(k.High) - l := decimals.MustToFloat64(k.Low) - switch side { - case types.SideLong: - // buy limit: filled if low <= price - if l <= limitPx { - // assume filled at min(limitPx, open) - px := math.Min(limitPx, decimals.MustToFloat64(k.Open)) - fee := math.Abs(px*qty) * s.FeePct - trd = &Trade{Side: side, Qty: qty, Price: px * (1 + s.SlippagePct), Fee: fee, Time: ts} - return trd, true - } - case types.SideShort: - if h >= limitPx { - px := math.Max(limitPx, decimals.MustToFloat64(k.Open)) - fee := math.Abs(px*qty) * s.FeePct - trd = &Trade{Side: side, Qty: qty, Price: px * (1 - s.SlippagePct), Fee: fee, Time: ts} - return trd, true - } - } - return nil, false -} +// ExecuteLimit todo 简单实现: 如果limit价格被kline的high/low包含则成交 diff --git a/internal/trading/backtest/types.go b/internal/trading/backtest/types.go index 6739d5a..e041ee5 100644 --- a/internal/trading/backtest/types.go +++ b/internal/trading/backtest/types.go @@ -3,13 +3,12 @@ package backtest import ( "sig-pub/api/pb" "sig-pub/pkg/trade" - "sig-pub/pkg/types" ) type BacktestResult struct { StartTs int64 EndTs int64 - Trades []*Trade + Trades []*trade.TradeOrder Positions []*trade.Position Cash float64 Equity float64 @@ -111,29 +110,4 @@ func (BacktestTradingPlan) TableName() string { return "t_backtest_trading_plan" } -// Trade 交易计划回测单 -type Trade struct { - Id int64 `json:"id" gorm:"column:id;primaryKey"` // 交易id - BacktestId int64 `json:"backtestId" gorm:"column:backtest_id;primaryKey"` // 回测id - Ctime int64 `json:"ctime" gorm:"column:ctime"` // 创建时间 - Side types.Side `json:"side" gorm:"column:side"` // 交易方向 - Qty float64 `json:"qty" gorm:"column:qty"` // 交易量 - Price float64 `json:"price" gorm:"column:price"` // 开仓价格 - Fee float64 `json:"fee" gorm:"column:fee"` // 开仓手续费 - Leverage int32 `json:"leverage" gorm:"column:leverage"` // 杠杆倍数 - Time int64 `json:"time" gorm:"column:time"` // 开仓时间 - ClosePrice float64 `json:"closePrice" gorm:"column:close_price"` // 平仓价格 - CloseFee float64 `json:"closeFee" gorm:"column:close_fee"` // 平仓手续费 - CloseTime int64 `json:"closeTime" gorm:"column:close_time"` // 平仓时间 - CloseCause trade.Cause `json:"closeCause" gorm:"column:close_cause"` // 平仓原因 ["stoploss", "takeprofit", "trailing", "retrace", "signal"](“止损”、“止盈”、“动态跟踪”、“回撤”、“信号”) - Pnl float64 `json:"pnl" gorm:"column:pnl"` // 盈利/亏损 pnl = (t.ClosePrice-t.Price)*t.Qty - t.Fee - t.CloseFee - Cash float64 `json:"cash" gorm:"column:cash"` // 平仓后账户净值 - HoldTime string `json:"holdTime" gorm:"column:hold_time"` // 持仓时间 - PeakPx float64 `json:"peakPx" gorm:"column:peak_px"` // highest (for long) or lowest (for short) observed price since entry -} - -func (Trade) TableName() string { - return "t_backtest_trading_trade" -} - // 参数迭代 diff --git a/internal/trading/trading_data_persist.go b/internal/trading/trading_data_persist.go index a9994f1..5d07fb2 100644 --- a/internal/trading/trading_data_persist.go +++ b/internal/trading/trading_data_persist.go @@ -7,16 +7,17 @@ import ( "sig-pub/pkg/data/entity" "sig-pub/pkg/storage/ck" "sig-pub/pkg/storage/persist" + "sig-pub/pkg/trade" ) type TradingDataPersist struct { - db *persist.RDB + db *persist.DB pgBatchWriter *persist.PGBatchWriter ckDB *ck.ClickhouseDB ckBatchWriter *ck.ClickhouseBatchWriter } -func NewTradingDataPersist(db *persist.RDB, pgBatchWriter *persist.PGBatchWriter, +func NewTradingDataPersist(db *persist.DB, pgBatchWriter *persist.PGBatchWriter, ckDB *ck.ClickhouseDB, ckBatchWriter *ck.ClickhouseBatchWriter) *TradingDataPersist { return &TradingDataPersist{ db: db, @@ -67,9 +68,9 @@ func (p *TradingDataPersist) ListBacktestLogs(userId int64) (backtestLogs []*bac return } -func (p *TradingDataPersist) ListBacktestTrades(backtestId int64) (trades []*backtest.Trade, err error) { +func (p *TradingDataPersist) ListBacktestTrades(backtestId int64) (trades []*trade.TradeOrder, err error) { err = p.db.Select(&trades, ` - select * from t_backtest_trading_trade where backtest_id = ? order by id asc + select * from t_backtest_trading_order where backtest_id = ? order by id asc `, backtestId) return } diff --git a/pkg/indicator/indicator_plot.go b/pkg/indicator/indicator_plot.go index fa868e9..1c9abf3 100644 --- a/pkg/indicator/indicator_plot.go +++ b/pkg/indicator/indicator_plot.go @@ -31,4 +31,5 @@ const ( ColorGreen string = "green" ColorYellow string = "yellow" ColorBlue string = "blue" + ColorPurple string = "purple" ) diff --git a/pkg/indicator/indicator_registry.go b/pkg/indicator/indicator_registry.go index 937f800..a99ea91 100644 --- a/pkg/indicator/indicator_registry.go +++ b/pkg/indicator/indicator_registry.go @@ -33,6 +33,7 @@ func (r *IndicatorRegistry) Init() (err error) { r.MustRegistIndicator(&BollLB{}) r.MustRegistIndicator(&SuperTrend{}) r.MustRegistIndicator(&ADX{}) + r.MustRegistIndicator(&KDJ{}) return } diff --git a/pkg/indicator/kdj.go b/pkg/indicator/kdj.go new file mode 100644 index 0000000..0dd9a77 --- /dev/null +++ b/pkg/indicator/kdj.go @@ -0,0 +1,73 @@ +package indicator + +import ( + "sig-pub/pkg/types" +) + +type KDJ struct { +} + +func (c *KDJ) Meta() IndicatorMeta { + return IndicatorMeta{ + Name: "KDJ", + Input: []types.InputArg{ + {Name: "rsvWindow", Type: types.InputTypeUInt, Desc: "周期"}, // 9 + {Name: "kWindow", Type: types.InputTypeUInt, Desc: "K平滑"}, // 3 + {Name: "dWindow", Type: types.InputTypeUInt, Desc: "D平滑"}, // 3 + }, + State: []string{"k", "d", "j"}, + Plots: []Plot{ + {State: "k", Type: PlotLine, Props: PlotProps{"color": ColorBlue}}, + {State: "d", Type: PlotLine, Props: PlotProps{"color": ColorYellow}}, + {State: "j", Type: PlotLine, Props: PlotProps{"color": ColorPurple}}, + }, + } +} + +func (c *KDJ) CandlePeriods(ctx IIndicatorContext) int16 { + return ctx.Input().Int16("rsvWindow") +} + +func (c *KDJ) Calculate(ctx IIndicatorContext) (vector float64) { + rsvWindow := ctx.Input().Int16("rsvWindow") + kWindow := float64(ctx.Input().Int16("kWindow")) + dWindow := float64(ctx.Input().Int16("dWindow")) + + klines := ctx.Series(0, rsvWindow) + if len(klines) < int(rsvWindow) { + return + } + + closePx := klines[0].CloseF64() + low := klines.Low().Min() + high := klines.High().Max() + + var rsv float64 + if high == low { + rsv = 50 + } else { + rsv = (closePx - low) / (high - low) * 100 + } + + // K + prevK, ok := ctx.State().Get("k", 1) + if !ok { + prevK = 50 + } + k := (kWindow-1)/kWindow*prevK + 1/kWindow*rsv + ctx.State().Set("k", k) + + // D + prevD, ok := ctx.State().Get("d", 1) + if !ok { + prevD = 50 + } + d := (dWindow-1)/dWindow*prevD + 1/dWindow*k + ctx.State().Set("d", d) + + // J + j := 3*k - 2*d + ctx.State().Set("j", j) + + return j +} diff --git a/pkg/storage/persist/rdb.go b/pkg/storage/persist/rdb.go index 6247556..70092c9 100644 --- a/pkg/storage/persist/rdb.go +++ b/pkg/storage/persist/rdb.go @@ -6,46 +6,46 @@ import ( "gorm.io/gorm" ) -// RDB gorm关系型数据 -type RDB struct { +// DB gorm关系型数据 +type DB struct { db *gorm.DB } -func NewRDB(db *gorm.DB) *RDB { - return &RDB{ +func NewDB(db *gorm.DB) *DB { + return &DB{ db: db, } } -func (m *RDB) Init() (err error) { +func (m *DB) Init() (err error) { return } -func (m *RDB) Insert(data any) (err error) { +func (m *DB) Insert(data any) (err error) { return m.db.Create(data).Error } // sql查询数据 -func (m *RDB) Select(ret any, sql string, args ...any) (err error) { +func (m *DB) Select(ret any, sql string, args ...any) (err error) { return m.db.Raw(sql, args...).Scan(ret).Error } // sql更新数据 -func (m *RDB) Update(sql string, args ...any) (rowsAffected int64, err error) { +func (m *DB) Update(sql string, args ...any) (rowsAffected int64, err error) { tx := m.db.Exec(sql, args...) rowsAffected, err = tx.RowsAffected, tx.Error return } // sql更新数据 -func (m *RDB) UpdateBy(data any) (rowsAffected int64, err error) { +func (m *DB) UpdateBy(data any) (rowsAffected int64, err error) { tx := m.db.Model(data).Updates(data) rowsAffected, err = tx.RowsAffected, tx.Error return } // AutoMigrateTables 自动对齐表结构，自动根据字段修改数据库表结构，只会加改不会删字段 -func (m *RDB) AutoMigrateTables(gormStructs ...any) (err error) { +func (m *DB) AutoMigrateTables(gormStructs ...any) (err error) { if len(gormStructs) == 0 { return }