From ae25409e567f9fddd612759fe8f2eecf26a1a638 Mon Sep 17 00:00:00 2001 From: ryan Date: Sun, 16 Aug 2026 00:19:50 +0800 Subject: [PATCH] =?UTF-8?q?feat(=E9=80=9A=E7=9F=A5=E6=A8=A1=E6=9D=BF):=20?= =?UTF-8?q?=E6=94=AF=E6=8C=81=E7=BC=BA=E5=AD=97=E6=AE=B5=E6=8C=89=E7=A9=BA?= =?UTF-8?q?=E5=80=BC=E6=B8=B2=E6=9F=93=E4=B8=8E=E7=A9=BA=E8=A1=8C=E7=9C=81?= =?UTF-8?q?=E7=95=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Motivation: 统一 crypto-strategy 与 trade-signal 两类交易信号的模板字段，使同一套通知模板可复用；字段缺失或为空时不再导致渲染报错或输出空行，通知内容更整洁。 Changes: * 渲染器缺失字段改为按空值处理，新增 line 函数实现空值整行省略 * crypto-strategy 信号补充止盈价、止损价、平均价、收益额等模板字段 * trade-signal 信号保留原始报文中的额外字段以适配统一模板 * 修复结构化日志将 error 作为值直接输出导致的格式问题 --- README.md | 13 +++- internal/engine/renderer.go | 37 +++++++++- internal/engine/renderer_test.go | 39 +++++++++-- internal/subscriber/cryptostrategy/convert.go | 69 ++++++++++++------- .../subscriber/cryptostrategy/convert_test.go | 33 +++++++++ internal/subscriber/handle.go | 17 +++-- internal/subscriber/subscriber.go | 4 +- internal/subscriber/tradesignal/convert.go | 15 ++-- .../subscriber/tradesignal/convert_test.go | 13 ++++ 9 files changed, 197 insertions(+), 43 deletions(-) diff --git a/README.md b/README.md index 522dcf8..e4a16bf 100644 --- a/README.md +++ b/README.md @@ -299,13 +299,22 @@ Body 同创建。成功：`{"ok": true}` 前缀：`/api/v1/templates` **鉴权：** Admin Key -模板语法为 Go `text/template`，变量来自解析后的 data。 +模板语法为 Go `text/template`，变量来自解析后的 data。缺字段不再报错，按空值处理。 + +可用 `line` 把前缀和值包在一起：值为空则整行不输出（含前缀和换行）。 + +``` +### {{.symbol}} {{.action}} +{{line "币种" .symbol}}{{line "周期" .period}}{{line "方向" .side}}{{line "价格" .price}}{{line "平均价" .totalAvgPx}}{{line "止盈价" .takeProfitPrice}}{{line "止损价" .stopLossPrice}} +``` + +等价写法：`{{with .totalAvgPx}}平均价：{{.}}{{end}}` 示例： ``` ### {{.symbol}} 开仓 -价格: {{.price}} +{{line "价格" .price}} ``` Text 模式示例：`{{.Body}}` diff --git a/internal/engine/renderer.go b/internal/engine/renderer.go index 15dde2b..f049dbf 100644 --- a/internal/engine/renderer.go +++ b/internal/engine/renderer.go @@ -3,6 +3,7 @@ package engine import ( "bytes" "fmt" + "strings" "text/template" ) @@ -13,7 +14,10 @@ func NewRenderer() *Renderer { } func (r *Renderer) Render(tmplContent string, data map[string]interface{}) (string, error) { - tmpl, err := template.New("notify").Option("missingkey=error").Parse(tmplContent) + tmpl, err := template.New("notify"). + Option("missingkey=zero"). + Funcs(template.FuncMap{"line": templateLine}). + Parse(tmplContent) if err != nil { return "", fmt.Errorf("parse template: %w", err) } @@ -23,3 +27,34 @@ func (r *Renderer) Render(tmplContent string, data map[string]interface{}) (stri } return buf.String(), nil } + +// templateLine renders "label：value\n" or empty if value is missing/zero. +func templateLine(label string, v any) string { + if isEmptyValue(v) { + return "" + } + return label + "：" + fmt.Sprint(v) + "\n" +} + +func isEmptyValue(v any) bool { + switch t := v.(type) { + case nil: + return true + case string: + return strings.TrimSpace(t) == "" + case bool: + return !t + case int: + return t == 0 + case int32: + return t == 0 + case int64: + return t == 0 + case float32: + return t == 0 + case float64: + return t == 0 + default: + return false + } +} diff --git a/internal/engine/renderer_test.go b/internal/engine/renderer_test.go index ab70d75..4b90a83 100644 --- a/internal/engine/renderer_test.go +++ b/internal/engine/renderer_test.go @@ -18,10 +18,41 @@ func TestRenderer(t *testing.T) { } } -func TestRenderer_Error(t *testing.T) { +func TestRendererMissingKeyEmpty(t *testing.T) { r := NewRenderer() - _, err := r.Render("{{.nonexistent}}", map[string]interface{}{}) - if err == nil { - t.Error("expected error for missing field, got nil") + _, err := r.Render("x{{.nonexistent}}y", map[string]interface{}{}) + if err != nil { + t.Fatal(err) + } +} + +func TestRendererLineOmitsEmpty(t *testing.T) { + r := NewRenderer() + tmpl := `{{line "币种" .symbol}}{{line "平均价" .totalAvgPx}}{{line "价格" .price}}{{line "止损价" .stopLossPrice}}` + out, err := r.Render(tmpl, map[string]interface{}{ + "symbol": "ICP", + "price": 2.273, + "totalAvgPx": "", + "stopLossPrice": 0, + }) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(out, "币种：ICP") || !strings.Contains(out, "价格：2.273") { + t.Fatalf("out=%q", out) + } + if strings.Contains(out, "平均价") || strings.Contains(out, "止损价") { + t.Fatalf("empty lines should be omitted, out=%q", out) + } +} + +func TestRendererLineMissingKey(t *testing.T) { + r := NewRenderer() + out, err := r.Render(`{{line "平均价" .totalAvgPx}}{{line "币种" .symbol}}`, map[string]interface{}{"symbol": "ICP"}) + if err != nil { + t.Fatal(err) + } + if strings.Contains(out, "平均价") || !strings.Contains(out, "币种：ICP") { + t.Fatalf("out=%q", out) } } diff --git a/internal/subscriber/cryptostrategy/convert.go b/internal/subscriber/cryptostrategy/convert.go index 39114ef..c5f3a16 100644 --- a/internal/subscriber/cryptostrategy/convert.go +++ b/internal/subscriber/cryptostrategy/convert.go @@ -35,6 +35,7 @@ type payload struct { type remark struct { OrderID string `json:"orderId"` + Revenue string `json:"revenue"` } type Converter struct{} @@ -60,31 +61,39 @@ func Convert(body []byte) (string, map[string]interface{}, error) { text := format(env, p, action) data := map[string]interface{}{ - "eventType": env.EventType, - "correlationId": env.CorrelationID, - "symbol": firstNonEmpty(env.Symbol, p.Currency), - "direction": env.Direction, - "side": strings.ToUpper(env.Direction), - "action": action, - "eventTime": env.EventTime, - "strategyCode": p.StrategyCode, - "period": p.Period, - "currency": p.Currency, - "isSale": p.IsSale, - "isClose": p.IsClose, - "isGain": p.IsGain, - "gainTarget": p.GainTarget, - "price": p.Price, - "lossPrice": p.LossPrice, - "gainPrices": p.GainPrices, - "leverage": p.Leverage, - "formatted": text, + "eventType": env.EventType, + "correlationId": env.CorrelationID, + "symbol": firstNonEmpty(env.Symbol, p.Currency), + "direction": env.Direction, + "side": strings.ToUpper(env.Direction), + "action": action, + "eventTime": env.EventTime, + "strategyCode": p.StrategyCode, + "period": p.Period, + "currency": p.Currency, + "isSale": p.IsSale, + "isClose": p.IsClose, + "isGain": p.IsGain, + "gainTarget": p.GainTarget, + "price": p.Price, + "lossPrice": p.LossPrice, + "gainPrices": p.GainPrices, + "leverage": p.Leverage, + "formatted": text, + "stopLossPrice": p.LossPrice, + "takeProfitPrice": takeProfitPrice(p), + "totalAvgPx": "", } if p.TotalGainTarget != 0 { data["totalGainTarget"] = p.TotalGainTarget } - if oid := parseOrderID(p.Remark); oid != "" { - data["orderId"] = oid + if r := parseRemark(p.Remark); r.OrderID != "" || r.Revenue != "" { + if r.OrderID != "" { + data["orderId"] = r.OrderID + } + if r.Revenue != "" { + data["revenue"] = r.Revenue + } } return event, data, nil } @@ -128,16 +137,26 @@ func inferAction(p payload) string { } } -func parseOrderID(raw string) string { +func takeProfitPrice(p payload) interface{} { + if gp := strings.TrimSpace(p.GainPrices); gp != "" { + return gp + } + if p.IsGain && p.Price > 0 { + return p.Price + } + return "" +} + +func parseRemark(raw string) remark { raw = strings.TrimSpace(raw) if raw == "" { - return "" + return remark{} } var r remark if err := json.Unmarshal([]byte(raw), &r); err != nil { - return "" + return remark{} } - return r.OrderID + return r } func format(env envelope, p payload, action string) string { diff --git a/internal/subscriber/cryptostrategy/convert_test.go b/internal/subscriber/cryptostrategy/convert_test.go index 6f1c7d4..dc4f560 100644 --- a/internal/subscriber/cryptostrategy/convert_test.go +++ b/internal/subscriber/cryptostrategy/convert_test.go @@ -3,6 +3,8 @@ package cryptostrategy import ( "strings" "testing" + + "aiaa-notification-service/internal/engine" ) const sampleBody = `{ @@ -124,6 +126,37 @@ func TestConvertPayloadObject(t *testing.T) { } } +func TestConvertRendersSharedSignalTemplate(t *testing.T) { + body := []byte(`{ + "eventType": "SIGNAL_RECEIVED", + "correlationId": "0_0_0", + "symbol": "ICP", + "direction": "LONG", + "payload": "{\"strategyCode\":\"ai-crypto-signals\",\"period\":\"1h\",\"currency\":\"ICP\",\"isSale\":false,\"isClose\":true,\"isGain\":true,\"gainTarget\":1,\"price\":2.273,\"remark\":\"{\\\"orderId\\\":\\\"uJs3zQI8IaRV16n8NERq\\\",\\\"revenue\\\":\\\"14.1088%\\\"}\",\"totalGainTarget\":5}", + "eventTime": 1786808929754 + }`) + _, data, err := Convert(body) + if err != nil { + t.Fatal(err) + } + for _, key := range []string{"totalAvgPx", "takeProfitPrice", "stopLossPrice", "side", "action", "symbol", "period", "price"} { + if _, ok := data[key]; !ok { + t.Fatalf("missing template key %q in %v", key, data) + } + } + if data["orderId"] != "uJs3zQI8IaRV16n8NERq" || data["revenue"] != "14.1088%" { + t.Fatalf("remark=%v", data) + } + tmpl := "### {{.symbol}} {{.action}}\n币种：{{.symbol}}\n周期：{{.period}}\n方向：{{.side}}\n价格：{{.price}}\n平均价：{{.totalAvgPx}}\n止盈价：{{.takeProfitPrice}}\n止损价：{{.stopLossPrice}}" + out, err := engine.NewRenderer().Render(tmpl, data) + if err != nil { + t.Fatalf("render: %v", err) + } + if !strings.Contains(out, "ICP") || !strings.Contains(out, "CLOSE") { + t.Fatalf("out=%s", out) + } +} + func TestConvertInvalidJSON(t *testing.T) { _, _, err := Convert([]byte(`{not json`)) if err == nil { diff --git a/internal/subscriber/handle.go b/internal/subscriber/handle.go index ccc49a7..ac34b3d 100644 --- a/internal/subscriber/handle.go +++ b/internal/subscriber/handle.go @@ -70,6 +70,13 @@ type MessageConverter interface { Convert(body []byte) (event string, data map[string]interface{}, err error) } +func errText(err error) string { + if err == nil { + return "" + } + return err.Error() +} + func HandleMessage(ctx context.Context, in HandleInput, conv MessageConverter, lookup SourceLookup, process ProcessFunc) Disposition { owned := false hash := "" @@ -77,7 +84,7 @@ func HandleMessage(ctx context.Context, in HandleInput, conv MessageConverter, l hash = MessageHash(in.Body) ok, err := in.Deduper.Claim(ctx, hash) if err != nil { - slog.Warn("dedup claim failed, processing anyway", "hash", hash, "error", err) + slog.Warn("dedup claim failed, processing anyway", "hash", hash, "error", errText(err)) } else if !ok { slog.Info("duplicate message, ack", "hash", hash) return DispositionAck @@ -94,13 +101,13 @@ func HandleMessage(ctx context.Context, in HandleInput, conv MessageConverter, l event, data, err := conv.Convert(in.Body) if err != nil { - slog.Warn("invalid signal, ack", "hash", hash, "raw", raw, "error", err) + slog.Warn("invalid signal, ack", "hash", hash, "raw", raw, "error", errText(err)) return DispositionAck } src, err := lookup(ctx, in.SourceName) if err != nil || src == nil || src.Status != 1 { - slog.Warn("source unavailable, ack", "source", in.SourceName, "hash", hash, "raw", raw, "error", err) + slog.Warn("source unavailable, ack", "source", in.SourceName, "hash", hash, "raw", raw, "error", errText(err)) return DispositionAck } @@ -116,14 +123,14 @@ func HandleMessage(ctx context.Context, in HandleInput, conv MessageConverter, l return DispositionAck } if errors.Is(err, notify.ErrUnprocessable) { - slog.Warn("unprocessable notify, ack", "source", src.Name, "event", event, "hash", hash, "raw", raw, "error", err) + slog.Warn("unprocessable notify, ack", "source", src.Name, "event", event, "hash", hash, "raw", raw, "error", errText(err)) return DispositionAck } disp := DecideRetry(RetryCount(in.Headers), in.MaxRetry) if owned && in.Deduper != nil { if relErr := in.Deduper.Release(ctx, hash); relErr != nil { - slog.Warn("dedup release failed", "hash", hash, "error", relErr) + slog.Warn("dedup release failed", "hash", hash, "error", errText(relErr)) } } return disp diff --git a/internal/subscriber/subscriber.go b/internal/subscriber/subscriber.go index e429023..f70dc88 100644 --- a/internal/subscriber/subscriber.go +++ b/internal/subscriber/subscriber.go @@ -122,7 +122,7 @@ func (s *Subscriber) ensureQueue(ch *amqp.Channel) error { if s.cfg.DeadLetterQueue != "" { if _, err := ch.QueueDeclare(s.cfg.DeadLetterQueue, true, false, false, false, nil); err != nil { - slog.Warn("declare dead letter queue failed", "queue", s.cfg.DeadLetterQueue, "error", err) + slog.Warn("declare dead letter queue failed", "queue", s.cfg.DeadLetterQueue, "error", err.Error()) } } return nil @@ -158,7 +158,7 @@ func (s *Subscriber) republish(ch *amqp.Channel, d amqp.Delivery, queue string) headers := copyAMQPHeaders(d.Headers) headers[retryHeader] = RetryCount(map[string]any(d.Headers)) + 1 if err := publishToQueue(ch, queue, d.Body, headers); err != nil { - slog.Error("requeue failed", "queue", queue, "error", err) + slog.Error("requeue failed", "queue", queue, "error", err.Error()) _ = d.Nack(false, true) return } diff --git a/internal/subscriber/tradesignal/convert.go b/internal/subscriber/tradesignal/convert.go index 1ccb159..852600a 100644 --- a/internal/subscriber/tradesignal/convert.go +++ b/internal/subscriber/tradesignal/convert.go @@ -39,7 +39,7 @@ func (c *Converter) Convert(body []byte) (string, map[string]interface{}, error) opts.AvgPrice = &avg } text := Format(out, opts) - data, err := toData(out) + data, err := toData(body, out) if err != nil { return "", nil, err } @@ -61,14 +61,21 @@ func (c *Converter) overrideFor(code string) *config.StrategyOverride { return &override } -func toData(sig *Signal) (map[string]interface{}, error) { +func toData(body []byte, sig *Signal) (map[string]interface{}, error) { + data := make(map[string]interface{}) + if err := json.Unmarshal(body, &data); err != nil { + return nil, err + } raw, err := json.Marshal(sig) if err != nil { return nil, err } - data := make(map[string]interface{}) - if err := json.Unmarshal(raw, &data); err != nil { + overlay := make(map[string]interface{}) + if err := json.Unmarshal(raw, &overlay); err != nil { return nil, err } + for k, v := range overlay { + data[k] = v + } return data, nil } diff --git a/internal/subscriber/tradesignal/convert_test.go b/internal/subscriber/tradesignal/convert_test.go index 41ce72d..6de94a8 100644 --- a/internal/subscriber/tradesignal/convert_test.go +++ b/internal/subscriber/tradesignal/convert_test.go @@ -36,6 +36,19 @@ func TestConvertOpen(t *testing.T) { } } +func TestConvertKeepsExtraJSONFields(t *testing.T) { + _, data, err := NewConverter(nil).Convert([]byte(`{ + "action":"OPEN","symbol":"BTCUSDT","price":63014.61, + "totalAvgPx":63014.61,"investmentAmount":100095.24 + }`)) + if err != nil { + t.Fatal(err) + } + if data["totalAvgPx"] != 63014.61 { + t.Fatalf("totalAvgPx=%v", data["totalAvgPx"]) + } +} + func TestConvertInvalidJSON(t *testing.T) { _, _, err := NewConverter(nil).Convert([]byte(`{`)) if !errors.Is(err, ErrInvalidSignal) {