From dc313fda13cb88c97cef443004b71dcd75ee314e Mon Sep 17 00:00:00 2001 From: ryan Date: Sat, 15 Aug 2026 23:04:39 +0800 Subject: [PATCH] =?UTF-8?q?feat(=E8=AE=A2=E9=98=85):=20=E6=96=B0=E5=A2=9E?= =?UTF-8?q?=20crypto-strategy=20=E4=BF=A1=E5=8F=B7=E6=A0=BC=E5=BC=8F?= =?UTF-8?q?=E5=8C=96=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Motivation: crypto-strategy 交易信号采用 envelope 外壳与字符串化/嵌套的 payload，原有 trade_signal 格式化器无法解析，需要独立的转换逻辑以生成通知文本和事件数据。 Changes: * 新增 crypto_strategy 消息转换器，解析 envelope 及嵌套、字符串化、扁平三种 payload 形态 * 根据平仓/止盈/卖出等标志推断交易动作，生成对应事件类型与中文通知文本 * 提取订单号、策略代码、价格、杠杆等字段用于通知数据 * 将消息转换器抽象为接口，订阅器按 formatter 配置选择对应实现 * 更新配置校验以支持 crypto_strategy 格式化器，并补充单元测试 --- config/config.yaml | 2 +- internal/config/config.go | 2 +- internal/subscriber/cryptostrategy/convert.go | 235 ++++++++++++++++++ .../subscriber/cryptostrategy/convert_test.go | 132 ++++++++++ internal/subscriber/handle.go | 7 +- internal/subscriber/subscriber.go | 13 +- 6 files changed, 384 insertions(+), 7 deletions(-) create mode 100644 internal/subscriber/cryptostrategy/convert.go create mode 100644 internal/subscriber/cryptostrategy/convert_test.go diff --git a/config/config.yaml b/config/config.yaml index bcf0aa3..44fb07c 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -66,4 +66,4 @@ subscriptions: routing_key: strategy.signal max_retry: 3 source: crypto-strategy - formatter: trade_signal + formatter: crypto_strategy diff --git a/internal/config/config.go b/internal/config/config.go index 6589146..cd924e0 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -97,7 +97,7 @@ func (c *Config) NormalizeSubscriptions() error { if s.Formatter == "" { s.Formatter = "trade_signal" } - if s.Formatter != "trade_signal" { + if s.Formatter != "trade_signal" && s.Formatter != "crypto_strategy" { return fmt.Errorf("subscriptions[%d]: unknown formatter %q", i, s.Formatter) } } diff --git a/internal/subscriber/cryptostrategy/convert.go b/internal/subscriber/cryptostrategy/convert.go new file mode 100644 index 0000000..39114ef --- /dev/null +++ b/internal/subscriber/cryptostrategy/convert.go @@ -0,0 +1,235 @@ +package cryptostrategy + +import ( + "bytes" + "encoding/json" + "fmt" + "strings" + "time" +) + +type envelope struct { + EventType string `json:"eventType"` + CorrelationID string `json:"correlationId"` + Symbol string `json:"symbol"` + Direction string `json:"direction"` + Payload json.RawMessage `json:"payload"` + EventTime int64 `json:"eventTime"` +} + +type payload struct { + StrategyCode string `json:"strategyCode"` + Period string `json:"period"` + Currency string `json:"currency"` + IsSale bool `json:"isSale"` + IsClose bool `json:"isClose"` + IsGain bool `json:"isGain"` + GainTarget float64 `json:"gainTarget"` + Price float64 `json:"price"` + LossPrice float64 `json:"lossPrice"` + GainPrices string `json:"gainPrices"` + Remark string `json:"remark"` + TotalGainTarget float64 `json:"totalGainTarget"` + Leverage int `json:"leverage"` +} + +type remark struct { + OrderID string `json:"orderId"` +} + +type Converter struct{} + +func NewConverter() *Converter { return &Converter{} } + +func (c *Converter) Convert(body []byte) (string, map[string]interface{}, error) { + return Convert(body) +} + +func Convert(body []byte) (string, map[string]interface{}, error) { + var env envelope + if err := json.Unmarshal(body, &env); err != nil { + return "", nil, fmt.Errorf("invalid envelope: %w", err) + } + p, err := parsePayload(body, env.Payload) + if err != nil { + return "", nil, err + } + + action := inferAction(p) + event := "trade." + strings.ToLower(action) + 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, + } + if p.TotalGainTarget != 0 { + data["totalGainTarget"] = p.TotalGainTarget + } + if oid := parseOrderID(p.Remark); oid != "" { + data["orderId"] = oid + } + return event, data, nil +} + +func parsePayload(body []byte, raw json.RawMessage) (payload, error) { + var p payload + raw = bytes.TrimSpace(raw) + if len(raw) == 0 || string(raw) == "null" { + if err := json.Unmarshal(body, &p); err != nil { + return p, fmt.Errorf("invalid payload: %w", err) + } + return p, nil + } + var asString string + if err := json.Unmarshal(raw, &asString); err == nil { + asString = strings.TrimSpace(asString) + if asString == "" { + if err := json.Unmarshal(body, &p); err != nil { + return p, fmt.Errorf("invalid payload: %w", err) + } + return p, nil + } + raw = []byte(asString) + } + if err := json.Unmarshal(raw, &p); err != nil { + return p, fmt.Errorf("invalid payload: %w", err) + } + return p, nil +} + +func inferAction(p payload) string { + switch { + case p.IsClose: + return "CLOSE" + case p.IsGain: + return "GAIN" + case p.IsSale: + return "SELL" + default: + return "OPEN" + } +} + +func parseOrderID(raw string) string { + raw = strings.TrimSpace(raw) + if raw == "" { + return "" + } + var r remark + if err := json.Unmarshal([]byte(raw), &r); err != nil { + return "" + } + return r.OrderID +} + +func format(env envelope, p payload, action string) string { + symbol := firstNonEmpty(env.Symbol, p.Currency) + lines := []string{actionTitle(env.Direction, action)} + if symbol != "" { + lines = append(lines, fmt.Sprintf("交易品种: %s", symbol)) + } + if p.Period != "" { + lines = append(lines, fmt.Sprintf("周期: %s", p.Period)) + } + switch action { + case "CLOSE": + if p.Price > 0 { + lines = append(lines, fmt.Sprintf("平仓价格: %.2f", p.Price)) + } + case "GAIN": + if p.Price > 0 { + lines = append(lines, fmt.Sprintf("止盈价格: %.2f", p.Price)) + } + case "SELL": + if p.Price > 0 { + lines = append(lines, fmt.Sprintf("卖出价格: %.2f", p.Price)) + } + default: + if p.Price > 0 { + lines = append(lines, fmt.Sprintf("开仓价格: %.2f", p.Price)) + } + } + if p.LossPrice > 0 { + lines = append(lines, fmt.Sprintf("止损价格: %.2f", p.LossPrice)) + } + if gp := strings.TrimSpace(p.GainPrices); gp != "" && action != "GAIN" { + lines = append(lines, fmt.Sprintf("止盈价格: %s", strings.Join(splitPrices(gp), ", "))) + } + if p.GainTarget != 0 { + lines = append(lines, fmt.Sprintf("止盈目标: %g", p.GainTarget)) + } + if p.Leverage > 0 { + lines = append(lines, fmt.Sprintf("杠杆: %dx", p.Leverage)) + } + if p.StrategyCode != "" { + lines = append(lines, fmt.Sprintf("策略: %s", p.StrategyCode)) + } + if env.EventTime > 0 { + t := time.UnixMilli(env.EventTime).In(time.Local) + lines = append(lines, fmt.Sprintf("Time: %s", t.Format("2006.01.02 15:04:05"))) + } + return strings.Join(lines, "\n") +} + +func actionTitle(direction, action string) string { + var pos string + switch strings.ToUpper(direction) { + case "LONG": + pos = "多单" + case "SHORT": + pos = "空单" + default: + pos = direction + } + var act string + switch action { + case "OPEN": + act = "开仓" + case "CLOSE": + act = "平仓" + case "GAIN": + act = "止盈" + case "SELL": + act = "卖出" + default: + act = action + } + return pos + act +} + +func splitPrices(s string) []string { + parts := strings.Split(s, ",") + out := make([]string, 0, len(parts)) + for _, p := range parts { + p = strings.TrimSpace(p) + if p != "" { + out = append(out, p) + } + } + return out +} + +func firstNonEmpty(a, b string) string { + if strings.TrimSpace(a) != "" { + return a + } + return b +} diff --git a/internal/subscriber/cryptostrategy/convert_test.go b/internal/subscriber/cryptostrategy/convert_test.go new file mode 100644 index 0000000..6f1c7d4 --- /dev/null +++ b/internal/subscriber/cryptostrategy/convert_test.go @@ -0,0 +1,132 @@ +package cryptostrategy + +import ( + "strings" + "testing" +) + +const sampleBody = `{ + "eventType": "SIGNAL_RECEIVED", + "correlationId": "0_0_0", + "symbol": "QNT", + "direction": "LONG", + "payload": "{\"strategyCode\":\"ai-crypto-signals\",\"period\":\"1h\",\"currency\":\"QNT\",\"isSale\":false,\"isClose\":false,\"isGain\":false,\"gainTarget\":5,\"price\":58.23,\"lossPrice\":57.82,\"gainPrices\":\"58.435,58.64,58.845,59.05,59.255\",\"remark\":\"{\\\"orderId\\\":\\\"jeJY8l5YnYwfJbmj6zb4\\\"}\",\"totalGainTarget\":5,\"leverage\":43}", + "eventTime": 1786802842899 +}` + +func TestConvertParsesNestedPayload(t *testing.T) { + event, data, err := Convert([]byte(sampleBody)) + if err != nil { + t.Fatal(err) + } + if event != "trade.open" { + t.Fatalf("event=%q", event) + } + if data["symbol"] != "QNT" || data["strategyCode"] != "ai-crypto-signals" { + t.Fatalf("data=%v", data) + } + if data["period"] != "1h" || data["direction"] != "LONG" { + t.Fatalf("data=%v", data) + } + if data["orderId"] != "jeJY8l5YnYwfJbmj6zb4" { + t.Fatalf("orderId=%v", data["orderId"]) + } + formatted, _ := data["formatted"].(string) + for _, want := range []string{ + "多单开仓", + "交易品种: QNT", + "周期: 1h", + "开仓价格: 58.23", + "止损价格: 57.82", + "止盈价格: 58.435, 58.64, 58.845, 59.05, 59.255", + "杠杆: 43x", + "策略: ai-crypto-signals", + } { + if !strings.Contains(formatted, want) { + t.Fatalf("missing %q in\n%s", want, formatted) + } + } +} + +func TestConvertCloseFlag(t *testing.T) { + body := []byte(`{ + "eventType":"SIGNAL_RECEIVED","symbol":"BTC","direction":"SHORT", + "payload":"{\"isClose\":true,\"price\":64000,\"strategyCode\":\"x\",\"period\":\"4h\"}", + "eventTime":1786802842899 + }`) + event, data, err := Convert(body) + if err != nil { + t.Fatal(err) + } + if event != "trade.close" { + t.Fatalf("event=%q", event) + } + formatted, _ := data["formatted"].(string) + if !strings.Contains(formatted, "空单平仓") { + t.Fatalf("%s", formatted) + } +} + +func TestConvertFlatOneLayer(t *testing.T) { + body := []byte(`{ + "eventType":"SIGNAL_RECEIVED", + "correlationId":"0_0_0", + "symbol":"QNT", + "direction":"LONG", + "strategyCode":"ai-crypto-signals", + "period":"1h", + "currency":"QNT", + "isSale":false, + "isClose":false, + "isGain":false, + "gainTarget":5, + "price":58.23, + "lossPrice":57.82, + "gainPrices":"58.435,58.64", + "remark":"{\"orderId\":\"jeJY8l5YnYwfJbmj6zb4\"}", + "leverage":43, + "eventTime":1786802842899 + }`) + event, data, err := Convert(body) + if err != nil { + t.Fatal(err) + } + if event != "trade.open" { + t.Fatalf("event=%q", event) + } + if data["strategyCode"] != "ai-crypto-signals" || data["price"] != 58.23 { + t.Fatalf("data=%v", data) + } + if data["orderId"] != "jeJY8l5YnYwfJbmj6zb4" { + t.Fatalf("orderId=%v", data["orderId"]) + } + formatted, _ := data["formatted"].(string) + if !strings.Contains(formatted, "多单开仓") || !strings.Contains(formatted, "开仓价格: 58.23") { + t.Fatalf("%s", formatted) + } +} + +func TestConvertPayloadObject(t *testing.T) { + body := []byte(`{ + "eventType":"SIGNAL_RECEIVED","symbol":"ETH","direction":"SHORT", + "payload":{"isClose":true,"price":3200,"strategyCode":"x","period":"1h"}, + "eventTime":1786802842899 + }`) + event, data, err := Convert(body) + if err != nil { + t.Fatal(err) + } + if event != "trade.close" { + t.Fatalf("event=%q", event) + } + if data["price"] != float64(3200) { + t.Fatalf("data=%v", data) + } +} + +func TestConvertInvalidJSON(t *testing.T) { + _, _, err := Convert([]byte(`{not json`)) + if err == nil { + t.Fatal("expected error") + } +} diff --git a/internal/subscriber/handle.go b/internal/subscriber/handle.go index ccbe65d..ccc49a7 100644 --- a/internal/subscriber/handle.go +++ b/internal/subscriber/handle.go @@ -8,7 +8,6 @@ import ( "aiaa-notification-service/internal/model" "aiaa-notification-service/internal/notify" - "aiaa-notification-service/internal/subscriber/tradesignal" ) const retryHeader = "x-retry-count" @@ -67,7 +66,11 @@ func RetryCount(headers map[string]any) int { } } -func HandleMessage(ctx context.Context, in HandleInput, conv *tradesignal.Converter, lookup SourceLookup, process ProcessFunc) Disposition { +type MessageConverter interface { + Convert(body []byte) (event string, data map[string]interface{}, err error) +} + +func HandleMessage(ctx context.Context, in HandleInput, conv MessageConverter, lookup SourceLookup, process ProcessFunc) Disposition { owned := false hash := "" if in.Deduper != nil { diff --git a/internal/subscriber/subscriber.go b/internal/subscriber/subscriber.go index fcb2d23..e429023 100644 --- a/internal/subscriber/subscriber.go +++ b/internal/subscriber/subscriber.go @@ -8,6 +8,7 @@ import ( "time" "aiaa-notification-service/internal/config" + "aiaa-notification-service/internal/subscriber/cryptostrategy" "aiaa-notification-service/internal/subscriber/tradesignal" amqp "github.com/rabbitmq/amqp091-go" @@ -15,19 +16,25 @@ import ( type Subscriber struct { cfg config.SubscriptionConfig - conv *tradesignal.Converter + conv MessageConverter lookup SourceLookup process ProcessFunc deduper Deduper } func New(cfg config.SubscriptionConfig, lookup SourceLookup, process ProcessFunc, deduper Deduper) (*Subscriber, error) { - if cfg.Formatter != "trade_signal" { + var conv MessageConverter + switch cfg.Formatter { + case "trade_signal": + conv = tradesignal.NewConverter(cfg.StrategyOverrides) + case "crypto_strategy": + conv = cryptostrategy.NewConverter() + default: return nil, fmt.Errorf("unknown formatter %q", cfg.Formatter) } return &Subscriber{ cfg: cfg, - conv: tradesignal.NewConverter(cfg.StrategyOverrides), + conv: conv, lookup: lookup, process: process, deduper: deduper,