diff --git a/README.md b/README.md index e408f3f..f398c8b 100644 --- a/README.md +++ b/README.md @@ -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`。 diff --git a/config/config.yaml b/config/config.yaml index a008651..b911936 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -34,7 +34,7 @@ logbull: api_key: "lb_60701971723797ed0374aa3896078fe5" log_level: "INFO" -# 多队列重复消息按策略/币种/周期/方向/价格去重；有 Redis 时跨进程共享 +# 多队列重复消息按来源/策略/币种/周期/方向/价格去重；有 Redis 时跨进程共享 subscription_dedup_ttl: 1h subscriptions: diff --git a/docs/httpie/curls.md b/docs/httpie/curls.md index 30b9d1b..91424d1 100644 --- a/docs/httpie/curls.md +++ b/docs/httpie/curls.md @@ -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 + }' +``` diff --git a/internal/engine/renderer_test.go b/internal/engine/renderer_test.go index c5f3a25..24a297f 100644 --- a/internal/engine/renderer_test.go +++ b/internal/engine/renderer_test.go @@ -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) + } + } + }) + } +} diff --git a/internal/subscriber/dedup.go b/internal/subscriber/dedup.go index 5189678..af77514 100644 --- a/internal/subscriber/dedup.go +++ b/internal/subscriber/dedup.go @@ -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 { diff --git a/internal/subscriber/dedup_test.go b/internal/subscriber/dedup_test.go index 95b1dda..7398131 100644 --- a/internal/subscriber/dedup_test.go +++ b/internal/subscriber/dedup_test.go @@ -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 diff --git a/internal/subscriber/handle.go b/internal/subscriber/handle.go index bfabab5..3e72a21 100644 --- a/internal/subscriber/handle.go +++ b/internal/subscriber/handle.go @@ -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 {