diff --git a/README.md b/README.md index 6c13e4a..40a2c1a 100644 --- a/README.md +++ b/README.md @@ -95,7 +95,7 @@ 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` | diff --git a/internal/subscriber/dedup.go b/internal/subscriber/dedup.go index af77514..2139a65 100644 --- a/internal/subscriber/dedup.go +++ b/internal/subscriber/dedup.go @@ -37,8 +37,9 @@ func signalKey(source string, data map[string]interface{}) string { symbol := firstNonEmptyField(fieldString(data["symbol"]), fieldString(data["currency"])) period := fieldString(data["period"]) direction := strings.ToUpper(firstNonEmptyField(fieldString(data["direction"]), fieldString(data["side"]))) + action := strings.ToUpper(fieldString(data["action"])) price := fieldString(data["price"]) - return strings.Join([]string{strings.TrimSpace(source), strategy, symbol, period, direction, price}, "\x1f") + return strings.Join([]string{strings.TrimSpace(source), strategy, symbol, period, direction, action, price}, "\x1f") } func firstNonEmptyField(a, b string) string { diff --git a/internal/subscriber/dedup_test.go b/internal/subscriber/dedup_test.go index 7398131..98835aa 100644 --- a/internal/subscriber/dedup_test.go +++ b/internal/subscriber/dedup_test.go @@ -67,6 +67,31 @@ func TestSignalHashIgnoresUnrelatedFields(t *testing.T) { } } +func TestSignalHashDistinguishesAction(t *testing.T) { + base := map[string]interface{}{ + "strategyCode": "BLONG", + "symbol": "SOLUSDT", + "period": "30m", + "side": "SHORT", + "price": 101.32, + } + reduce := cloneFields(base) + reduce["action"] = "REDUCE" + closeMsg := cloneFields(base) + closeMsg["action"] = "CLOSE" + if SignalHash("trade-signal", reduce) == SignalHash("trade-signal", closeMsg) { + t.Fatal("REDUCE and CLOSE at the same price should hash differently") + } +} + +func cloneFields(in map[string]interface{}) map[string]interface{} { + out := make(map[string]interface{}, len(in)+1) + for k, v := range in { + out[k] = v + } + return out +} + func TestMemoryDeduperClaimOnce(t *testing.T) { d := NewMemoryDeduper() ok, err := d.Claim(context.Background(), "abc") @@ -131,6 +156,49 @@ func TestHandleDedupByStrategySymbolPeriodDirectionPrice(t *testing.T) { } } +func TestHandleDedupKeepsReduceThenClose(t *testing.T) { + dedup := NewMemoryDeduper() + var events []string + process := func(_ context.Context, req notify.Request) (notify.Result, error) { + events = append(events, req.Event) + return notify.Result{Matched: true}, nil + } + lookup := func(context.Context, string) (*model.Source, error) { return enabledSrc(), nil } + conv := tradesignal.NewConverter(nil) + reduce := []byte(`{ + "signalId":"local-0f6f4b00e143f8ee:update:1787629790828:0.12", + "sourcePosId":"local-0f6f4b00e143f8ee","sourcePosIds":["3340443882114850823"], + "strategyCode":"BLONG","symbol":"SOLUSDT","side":"SHORT","action":"REDUCE", + "quantity":0.11,"price":101.32,"leverage":20,"period":"30m", + "eventTime":"2026-08-25T03:49:51.155Z","addCount":0,"totalPos":0.12, + "totalAvgPx":102.64,"posMarginRatio":0.478261,"oldQuantity":0.23,"deltaQuantity":-0.11 + }`) + closeBody := []byte(`{ + "signalId":"local-0f6f4b00e143f8ee:close:1787629790936:0", + "sourcePosId":"local-0f6f4b00e143f8ee","sourcePosIds":["3340443882114850823"], + "strategyCode":"BLONG","symbol":"SOLUSDT","side":"SHORT","action":"CLOSE", + "quantity":0.12,"price":101.32,"leverage":1,"period":"30m", + "eventTime":"2026-08-25T03:49:51.389Z","addCount":0,"totalPos":0, + "totalAvgPx":0,"posMarginRatio":1,"oldQuantity":0.12 + }`) + if d := HandleMessage(context.Background(), HandleInput{ + Body: reduce, SourceName: "trade-signal", MaxRetry: 3, Deduper: dedup, + }, conv, lookup, process); d != DispositionAck { + t.Fatalf("reduce=%v", d) + } + if d := HandleMessage(context.Background(), HandleInput{ + Body: closeBody, SourceName: "trade-signal", MaxRetry: 3, Deduper: dedup, + }, conv, lookup, process); d != DispositionAck { + t.Fatalf("close=%v", d) + } + if len(events) != 2 { + t.Fatalf("reduce then close should both notify, got %v", events) + } + if events[0] != "trade.reduce" || events[1] != "trade.close" { + t.Fatalf("events=%v", events) + } +} + func TestHandleDedupKeepsDifferentSources(t *testing.T) { dedup := NewMemoryDeduper() var n atomic.Int32