17 changed files with 272 additions and 155 deletions
@ -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)) |
||||
} |
||||
@ -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 |
||||
} |
||||
@ -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() |
||||
}() |
||||
} |
||||
Loading…
Reference in new issue