fix(消息去重): 去重键加入来源维度，避免跨来源信号被误去重

Motivation: 不同来源（如 crypto-strategy 与 trade-signal）可能产生策略、币种、周期、方向、价格完全一致的信号，旧去重键仅按这五维计算，导致后到的不同来源信号被误判为重复而漏推，无法支撑跟单策略等新增来源并行推送。 Changes: * 信号去重键新增来源（source）维度，不同来源的相同信号不再互相去重 * 信号哈希计算时透传来源名称，同一来源内的重复消息仍正常去重 * 补充跟单策略文案模板的渲染测试与接口创建文档
2026-08-17 15:39:20 +08:00
parent a2c0590353
commit 33b2a96bcf
7 changed files with 177 additions and 11 deletions
@@ -95,12 +95,12 @@ make build && ./bin/server
 | `smtp.*` | 邮件发送（email 渠道） | — |
 | `rate_limit.default` | 每 source 每秒请求上限 | `100` |
 | `rate_limit.dingtalk_per_min` | 同一钉钉机器人（access_token）每分钟发送上限；超限排队到下一分钟 | `18`（官方 20，留余量） |
-| `subscription_dedup_ttl` | 多队列重复消息（策略/币种/周期/方向/价格）去重窗口 | `1h` |
+| `subscription_dedup_ttl` | 多队列重复消息（来源/策略/币种/周期/方向/价格）去重窗口 | `1h` |
 | `subscriptions` | RabbitMQ 订阅列表；某条 `url` 为空则跳过 | 空 |
 | `subscriptions[].source` | 对应已有 Source.name | 有 url 时必填 |
 | `subscriptions[].formatter` | 目前仅 `trade_signal` | `trade_signal` |

-环境变量 `RABBITMQ_URL` 未设置时不启动消费，HTTP 通知不受影响。交易信号订阅需事先创建 Source（如 `trade-signal`）、模板（可用 `{{.formatted}}`）、规则 `trade.open` / `trade.add` / `trade.close` / `trade.reduce`、以及渠道。规则条件可用 `strategyCode` / `symbol` / `period`。
+环境变量 `RABBITMQ_URL` 未设置时不启动消费，HTTP 通知不受影响。交易信号订阅需事先创建 Source（如 `trade-signal`）、模板（可用 `{{.formatted}}`）、规则 `trade.open` / `trade.add` / `trade.close` / `trade.reduce`、以及渠道。规则条件可用 `strategyCode` / `symbol` / `period`。跟单策略（B龙 `BLONG` 及后续）复用模板 `跟单策略`，规则 `trade.*` + `strategyCode`。

 `crypto-strategy` 开仓（多/空，含原来的 `isSale` 空单）都映射为 `trade.open`，不再发 `trade.sell`。止盈（`isGain`）为 `trade.gain`，止损（`isClose` 且非 `isGain`）为 `trade.close`。高低分 `HLSS`、异动 `AMA`、波段 `BTS`、AG 趋势 `AGTS` 用各自前缀：`HLSS.open` / `AMA.open` / `BTS.close` / `AGTS.open` 等。同一 Source 允许多条相同 event 的规则（用条件区分）；精确 event 优先于通配，条件通过的规则都会发送。现成模板与规则见 `docs/httpie/curls.md`。

@@ -34,7 +34,7 @@ logbull:
  api_key: "lb_60701971723797ed0374aa3896078fe5"
  log_level: "INFO"

-# 多队列重复消息按策略/币种/周期/方向/价格去重；有 Redis 时跨进程共享
+# 多队列重复消息按来源/策略/币种/周期/方向/价格去重；有 Redis 时跨进程共享
 subscription_dedup_ttl: 1h

 subscriptions:
@@ -662,3 +662,59 @@ curl -X POST 'http://localhost:8080/api/v1/rules' \
    "enabled": 1
  }'
 ```
+
+---
+
+## 跟单策略（trade-signal）
+
+Source：`trade-signal`。渠道：`safew_B龙策略`。条件：`strategyCode = BLONG`。
+
+这是后续跟单策略的共用文案模板（开/加/平/减仓一套）。新跟单策略复用模板 `跟单策略`，再加一条 `trade.*` 规则（换 `strategyCode`、渠道，并在模板 `case .strategyCode` 里补中文名）。兜底规则 `rule-11`（测试AI）需 `strategyCode ne` 已拆出去的跟单 code，避免双发。
+
+事件：`OPEN` → `trade.open`，`ADD` → `trade.add`，`CLOSE` → `trade.close`，`REDUCE` → `trade.reduce`；规则用 `trade.*` 全覆盖。
+
+开仓渲染示例：
+
+```
+空单开仓
+交易品种: ETH
+开仓价格: 1898.76
+开仓数量: 2.00
+平均单价: 1898.76
+杠杆: 100x
+策略: B龙策略
+推送时间: 2026.08.13 13:15:42
+```
+
+平仓 / 减仓不输出杠杆；加仓与开仓一样带杠杆。
+
+### 创建跟单模板
+
+```bash
+curl -X POST 'http://localhost:8080/api/v1/templates' \
+  -H 'Authorization: Bearer admin-sk-change-me' \
+  -H 'Content-Type: application/json' \
+  -d '{
+    "name": "跟单策略",
+    "content": "{{case .side \"LONG\" \"多单\" \"SHORT\" \"空单\"}}{{case .action \"OPEN\" \"开仓\" \"ADD\" \"加仓\" \"CLOSE\" \"平仓\" \"REDUCE\" \"减仓\"}}\n交易品种: {{case .symbol \"ETHUSDT\" \"ETH\" \"BTCUSDT\" \"BTC\" \"SOLUSDT\" \"SOL\" \"BNBUSDT\" \"BNB\" .symbol}}\n{{case .action \"OPEN\" \"开仓价格\" \"ADD\" \"加仓价格\" \"CLOSE\" \"平仓价格\" \"REDUCE\" \"减仓价格\"}}: {{printf \"%.2f\" .price}}\n{{case .action \"OPEN\" \"开仓数量\" \"ADD\" \"加仓数量\" \"CLOSE\" \"平仓数量\" \"REDUCE\" \"减仓数量\"}}: {{printf \"%.2f\" .quantity}}\n平均单价: {{printf \"%.2f\" .avgPrice}}\n{{if or (eq .action \"OPEN\") (eq .action \"ADD\")}}{{if .leverage}}杠杆: {{.leverage}}x\n{{end}}{{end}}策略: {{case .strategyCode \"BLONG\" \"B龙策略\" .strategyCode}}\n推送时间: {{.pushedAt}}"
+  }'
+```
+
+### 创建 B龙规则
+
+```bash
+curl -X POST 'http://localhost:8080/api/v1/rules' \
+  -H 'Authorization: Bearer admin-sk-change-me' \
+  -H 'Content-Type: application/json' \
+  -d '{
+    "name": "B龙策略",
+    "source_name": "trade-signal",
+    "event": "trade.*",
+    "template_name": "跟单策略",
+    "channels": ["safew_B龙策略"],
+    "conditions": [
+      {"field": "strategyCode", "op": "eq", "value": "BLONG"}
+    ],
+    "enabled": 1
+  }'
+```
@@ -108,3 +108,72 @@ func TestRendererCaseDefault(t *testing.T) {
 		t.Fatalf("got %q", out)
 	}
 }
+
+func TestCopyTradeTemplate(t *testing.T) {
+	tmpl := "{{case .side \"LONG\" \"多单\" \"SHORT\" \"空单\"}}{{case .action \"OPEN\" \"开仓\" \"ADD\" \"加仓\" \"CLOSE\" \"平仓\" \"REDUCE\" \"减仓\"}}\n交易品种: {{case .symbol \"ETHUSDT\" \"ETH\" \"BTCUSDT\" \"BTC\" \"SOLUSDT\" \"SOL\" \"BNBUSDT\" \"BNB\" .symbol}}\n{{case .action \"OPEN\" \"开仓价格\" \"ADD\" \"加仓价格\" \"CLOSE\" \"平仓价格\" \"REDUCE\" \"减仓价格\"}}: {{printf \"%.2f\" .price}}\n{{case .action \"OPEN\" \"开仓数量\" \"ADD\" \"加仓数量\" \"CLOSE\" \"平仓数量\" \"REDUCE\" \"减仓数量\"}}: {{printf \"%.2f\" .quantity}}\n平均单价: {{printf \"%.2f\" .avgPrice}}\n{{if or (eq .action \"OPEN\") (eq .action \"ADD\")}}{{if .leverage}}杠杆: {{.leverage}}x\n{{end}}{{end}}策略: {{case .strategyCode \"BLONG\" \"B龙策略\" .strategyCode}}\n推送时间: {{.pushedAt}}"
+	r := NewRenderer()
+	cases := []struct {
+		name string
+		data map[string]interface{}
+		want []string
+		not  []string
+	}{
+		{
+			name: "open",
+			data: map[string]interface{}{
+				"side": "SHORT", "action": "OPEN", "symbol": "ETHUSDT",
+				"price": 1898.76, "quantity": 2.0, "avgPrice": 1898.76,
+				"leverage": 100, "strategyCode": "BLONG", "pushedAt": "2026.08.13 13:15:42",
+			},
+			want: []string{"空单开仓", "交易品种: ETH", "开仓价格: 1898.76", "开仓数量: 2.00", "平均单价: 1898.76", "杠杆: 100x", "策略: B龙策略", "推送时间: 2026.08.13 13:15:42"},
+		},
+		{
+			name: "close",
+			data: map[string]interface{}{
+				"side": "SHORT", "action": "CLOSE", "symbol": "ETHUSDT",
+				"price": 1883.35, "quantity": 2.0, "avgPrice": 1898.76,
+				"leverage": 100, "strategyCode": "BLONG", "pushedAt": "2026.08.13 17:14:31",
+			},
+			want: []string{"空单平仓", "平仓价格: 1883.35", "平仓数量: 2.00", "策略: B龙策略", "推送时间: 2026.08.13 17:14:31"},
+			not:  []string{"杠杆:"},
+		},
+		{
+			name: "reduce",
+			data: map[string]interface{}{
+				"side": "SHORT", "action": "REDUCE", "symbol": "ETHUSDT",
+				"price": 1889.43, "quantity": 3.1, "avgPrice": 1899.03,
+				"strategyCode": "BLONG", "pushedAt": "2026.08.12 22:05:02",
+			},
+			want: []string{"空单减仓", "减仓价格: 1889.43", "减仓数量: 3.10", "平均单价: 1899.03", "策略: B龙策略"},
+			not:  []string{"杠杆:"},
+		},
+		{
+			name: "add",
+			data: map[string]interface{}{
+				"side": "SHORT", "action": "ADD", "symbol": "ETHUSDT",
+				"price": 1933.05, "quantity": 3.8, "avgPrice": 1933.05,
+				"leverage": 100, "strategyCode": "BLONG", "pushedAt": "2026.08.10 06:14:28",
+			},
+			want: []string{"空单加仓", "加仓价格: 1933.05", "加仓数量: 3.80", "杠杆: 100x", "策略: B龙策略", "推送时间: 2026.08.10 06:14:28"},
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			out, err := r.Render(tmpl, tc.data)
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Logf("\n%s", out)
+			for _, w := range tc.want {
+				if !strings.Contains(out, w) {
+					t.Errorf("missing %q in\n%s", w, out)
+				}
+			}
+			for _, n := range tc.not {
+				if strings.Contains(out, n) {
+					t.Errorf("unexpected %q in\n%s", n, out)
+				}
+			}
+		})
+	}
+}
@@ -24,12 +24,12 @@ func MessageHash(body []byte) string {
 	return hex.EncodeToString(sum[:])
 }

-func SignalHash(data map[string]interface{}) string {
-	sum := sha256.Sum256([]byte(signalKey(data)))
+func SignalHash(source string, data map[string]interface{}) string {
+	sum := sha256.Sum256([]byte(signalKey(source, data)))
 	return hex.EncodeToString(sum[:])
 }

-func signalKey(data map[string]interface{}) string {
+func signalKey(source string, data map[string]interface{}) string {
 	if data == nil {
 		data = map[string]interface{}{}
 	}
@@ -38,7 +38,7 @@ func signalKey(data map[string]interface{}) string {
 	period := fieldString(data["period"])
 	direction := strings.ToUpper(firstNonEmptyField(fieldString(data["direction"]), fieldString(data["side"])))
 	price := fieldString(data["price"])
-	return strings.Join([]string{strategy, symbol, period, direction, price}, "\x1f")
+	return strings.Join([]string{strings.TrimSpace(source), strategy, symbol, period, direction, price}, "\x1f")
 }

 func firstNonEmptyField(a, b string) string {
@@ -26,7 +26,7 @@ func TestMessageHashStable(t *testing.T) {
 }

 func TestSignalHashIgnoresUnrelatedFields(t *testing.T) {
-	a := SignalHash(map[string]interface{}{
+	a := SignalHash("crypto-strategy", map[string]interface{}{
 		"strategyCode": "ai-crypto-signals",
 		"symbol":       "CRV",
 		"period":       "1h",
@@ -34,7 +34,7 @@ func TestSignalHashIgnoresUnrelatedFields(t *testing.T) {
 		"price":        0.2528,
 		"eventTime":    int64(1),
 	})
-	b := SignalHash(map[string]interface{}{
+	b := SignalHash("crypto-strategy", map[string]interface{}{
 		"strategyCode": "ai-crypto-signals",
 		"currency":     "CRV",
 		"period":       "1h",
@@ -45,7 +45,7 @@ func TestSignalHashIgnoresUnrelatedFields(t *testing.T) {
 	if a == "" || a != b {
 		t.Fatalf("same signal fields should hash equal, a=%q b=%q", a, b)
 	}
-	c := SignalHash(map[string]interface{}{
+	c := SignalHash("crypto-strategy", map[string]interface{}{
 		"strategyCode": "ai-crypto-signals",
 		"symbol":       "CRV",
 		"period":       "1h",
@@ -55,6 +55,16 @@ func TestSignalHashIgnoresUnrelatedFields(t *testing.T) {
 	if a == c {
 		t.Fatal("different price should hash differently")
 	}
+	otherSrc := SignalHash("trade-signal", map[string]interface{}{
+		"strategyCode": "ai-crypto-signals",
+		"symbol":       "CRV",
+		"period":       "1h",
+		"direction":    "LONG",
+		"price":        0.2528,
+	})
+	if a == otherSrc {
+		t.Fatal("different sources should hash differently")
+	}
 }

 func TestMemoryDeduperClaimOnce(t *testing.T) {
@@ -121,6 +131,37 @@ func TestHandleDedupByStrategySymbolPeriodDirectionPrice(t *testing.T) {
 	}
 }

+func TestHandleDedupKeepsDifferentSources(t *testing.T) {
+	dedup := NewMemoryDeduper()
+	var n atomic.Int32
+	process := func(context.Context, notify.Request) (notify.Result, error) {
+		n.Add(1)
+		return notify.Result{Matched: true}, nil
+	}
+	lookup := func(_ context.Context, name string) (*model.Source, error) {
+		return &model.Source{ID: 1, Name: name, Status: 1}, nil
+	}
+	conv := cryptostrategy.NewConverter()
+	body := []byte(`{
+		"eventType":"SIGNAL_RECEIVED","symbol":"CRV","direction":"LONG",
+		"payload":"{\"strategyCode\":\"ai-crypto-signals\",\"period\":\"1h\",\"currency\":\"CRV\",\"isClose\":true,\"isGain\":true,\"price\":0.2528}",
+		"eventTime":1786899538978
+	}`)
+	if d := HandleMessage(context.Background(), HandleInput{
+		Body: body, SourceName: "crypto-strategy", MaxRetry: 3, Deduper: dedup,
+	}, conv, lookup, process); d != DispositionAck {
+		t.Fatalf("first=%v", d)
+	}
+	if d := HandleMessage(context.Background(), HandleInput{
+		Body: body, SourceName: "trade-signal", MaxRetry: 3, Deduper: dedup,
+	}, conv, lookup, process); d != DispositionAck {
+		t.Fatalf("other source=%v", d)
+	}
+	if n.Load() != 2 {
+		t.Fatalf("different sources should both process, got %d", n.Load())
+	}
+}
+
 func TestHandleDuplicateAckSkipsProcess(t *testing.T) {
 	dedup := NewMemoryDeduper()
 	var n atomic.Int32
@@ -87,7 +87,7 @@ func HandleMessage(ctx context.Context, in HandleInput, conv MessageConverter, l
 	}

 	owned := false
-	hash := SignalHash(data)
+	hash := SignalHash(in.SourceName, data)
 	if in.Deduper != nil {
 		ok, err := in.Deduper.Claim(ctx, hash)
 		if err != nil {