From 6f846a0a3cda78ca842dbd24882fa972770fa2e1 Mon Sep 17 00:00:00 2001 From: ryan Date: Sat, 15 Aug 2026 17:34:49 +0800 Subject: [PATCH] feat: subscribe to RabbitMQ trade signals and notify by rules Consume configurable queues, format signals (including period), share NotifyService with HTTP, and drop duplicate bodies within 1h. --- README.md | 6 + cmd/server/main.go | 38 +- config/config.yaml | 23 + .../plans/2026-08-15-rabbitmq-subscriber.md | 1159 +++++++++++++++++ go.mod | 1 + go.sum | 2 + internal/cache/redis.go | 12 + internal/config/config.go | 108 +- internal/config/config_test.go | 61 + internal/handler/notify.go | 120 +- internal/notify/service.go | 130 ++ internal/notify/service_test.go | 118 ++ internal/subscriber/dedup.go | 70 + internal/subscriber/dedup_test.go | 89 ++ internal/subscriber/handle.go | 117 ++ internal/subscriber/handle_test.go | 135 ++ internal/subscriber/subscriber.go | 182 +++ internal/subscriber/tradesignal/convert.go | 74 ++ .../subscriber/tradesignal/convert_test.go | 51 + internal/subscriber/tradesignal/format.go | 198 +++ .../subscriber/tradesignal/format_test.go | 60 + internal/subscriber/tradesignal/override.go | 29 + internal/subscriber/tradesignal/position.go | 204 +++ internal/subscriber/tradesignal/signal.go | 36 + .../subscriber/tradesignal/transform_test.go | 220 ++++ 25 files changed, 3129 insertions(+), 114 deletions(-) create mode 100644 docs/superpowers/plans/2026-08-15-rabbitmq-subscriber.md create mode 100644 internal/config/config_test.go create mode 100644 internal/notify/service.go create mode 100644 internal/notify/service_test.go create mode 100644 internal/subscriber/dedup.go create mode 100644 internal/subscriber/dedup_test.go create mode 100644 internal/subscriber/handle.go create mode 100644 internal/subscriber/handle_test.go create mode 100644 internal/subscriber/subscriber.go create mode 100644 internal/subscriber/tradesignal/convert.go create mode 100644 internal/subscriber/tradesignal/convert_test.go create mode 100644 internal/subscriber/tradesignal/format.go create mode 100644 internal/subscriber/tradesignal/format_test.go create mode 100644 internal/subscriber/tradesignal/override.go create mode 100644 internal/subscriber/tradesignal/position.go create mode 100644 internal/subscriber/tradesignal/signal.go create mode 100644 internal/subscriber/tradesignal/transform_test.go diff --git a/README.md b/README.md index 6327840..8c80682 100644 --- a/README.md +++ b/README.md @@ -95,6 +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` | 多队列重复消息（body SHA-256）去重窗口 | `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`。 健康检查：`GET /health` → `{"status":"ok"}` diff --git a/cmd/server/main.go b/cmd/server/main.go index 503e0f8..05dfdb5 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -2,6 +2,7 @@ package main import ( "context" + "errors" "fmt" "log/slog" "net/http" @@ -15,8 +16,11 @@ import ( "aiaa-notification-service/internal/config" "aiaa-notification-service/internal/engine" "aiaa-notification-service/internal/handler" + "aiaa-notification-service/internal/model" + "aiaa-notification-service/internal/notify" "aiaa-notification-service/internal/safew" "aiaa-notification-service/internal/store" + "aiaa-notification-service/internal/subscriber" "github.com/gin-gonic/gin" "github.com/logbull/logbull-go/logbull" @@ -123,7 +127,8 @@ func main() { router := engine.NewRouter(st, redisCache, senderFactory) // Build handlers - notifyH := handler.NewNotifyHandler(st, redisCache, matcher, renderer, router) + notifySvc := notify.NewService(matcher, st, renderer, router, st) + notifyH := handler.NewNotifyHandler(notifySvc) sourceH := handler.NewSourceHandler(st, redisCache) templateH := handler.NewTemplateHandler(st, redisCache) @@ -217,6 +222,28 @@ func main() { Handler: r, } + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + lookup := func(ctx context.Context, name string) (*model.Source, error) { + return st.GetSourceByName(ctx, name) + } + deduper := subscriber.NewCacheDeduper(redisCache, cfg.SubscriptionDedupTTL) + for _, sub := range cfg.ActiveSubscriptions() { + sub := sub + cons, err := subscriber.New(sub, lookup, notifySvc.Process, deduper) + if err != nil { + slog.Error("subscriber init", "name", sub.Name, "error", err) + os.Exit(1) + } + go func() { + if err := cons.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { + slog.Error("subscriber stopped", "name", sub.Name, "error", err) + } + }() + slog.Info("subscriber started", "name", sub.Name, "queue", sub.Queue, "source", sub.Source) + } + go func() { slog.Info("server starting", "port", cfg.Server.Port) if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { @@ -225,15 +252,12 @@ func main() { } }() - // Graceful shutdown - quit := make(chan os.Signal, 1) - signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) - <-quit + <-ctx.Done() slog.Info("shutting down...") - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - if err := srv.Shutdown(ctx); err != nil { + if err := srv.Shutdown(shutdownCtx); err != nil { slog.Error("forced shutdown", "error", err) } slog.Info("server stopped") diff --git a/config/config.yaml b/config/config.yaml index 35f4946..11cf88f 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -33,3 +33,26 @@ logbull: project_id: "42a3fef0-2fd6-4ce4-80c5-f6bb6ecc2013" api_key: "lb_60701971723797ed0374aa3896078fe5" log_level: "INFO" + +# 多队列重复消息按 body SHA-256 去重；有 Redis 时跨进程共享 +subscription_dedup_ttl: 1h + +subscriptions: + - name: trade-signal + url: "${RABBITMQ_URL}" + queue: trade.signal.notify.queue + dead_letter_queue: trade.signal.notify.dlq + exchange: trade.signal.executor.queue + exchange_type: fanout + routing_key: "" + max_retry: 3 + source: trade-signal + formatter: trade_signal + strategy_overrides: + BLONG: + quantity_multipliers: + open: 100 + add: 100 + reduce: 100 + close: 100 + leverage: 100 diff --git a/docs/superpowers/plans/2026-08-15-rabbitmq-subscriber.md b/docs/superpowers/plans/2026-08-15-rabbitmq-subscriber.md new file mode 100644 index 0000000..7a8b152 --- /dev/null +++ b/docs/superpowers/plans/2026-08-15-rabbitmq-subscriber.md @@ -0,0 +1,1159 @@ +# RabbitMQ Subscriber Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 增加可配置 RabbitMQ 订阅：消费交易信号、格式化后走内部 `NotifyService` 按规则发到各渠道。 + +**Architecture:** 从现有 `/notify` 抽出 `notify.Service.Process`。`subscriber` 声明/绑定/重连/重试/DLQ，把 body 交给 `tradesignal.Converter`（覆盖 → 均价 → 格式化），再 `Process`。HTTP 与 MQ 共用同一入口。 + +**Tech Stack:** Go 1.22+、Viper、`github.com/rabbitmq/amqp091-go`、stdlib `testing`。不连真实 broker。 + +## Global Constraints + +- Spec: `docs/superpowers/specs/2026-08-15-rabbitmq-subscriber-design.md` +- 每条订阅绑一个 Source 名；`event = "trade." + strings.ToLower(action)` +- `data` 用覆盖后的信号字段（camelCase）+ `formatted`；有均价才写 `avgPrice` +- `period` 原样使用；非空则在「交易品种」后输出 `周期: {period}` +- AMQP URL 只用 `${RABBITMQ_URL}`，禁止把账号写进任何文件 +- 未知 `formatter` 在配置规范化时失败；空则默认 `trade_signal` +- `max_retry<=0` → 3；`exchange_type` 空 → `fanout`；`name` 空 → 用 `queue` +- `url` 为空的订阅不启动；全部未启动时 HTTP 不受影响 +- JSON 无效 / 缺 `action` / Source 不存在或禁用 / 无规则 / 条件未过 / `ErrUnprocessable` → Ack，不进 DLQ +- `Process` 其它 error → 重投；超过 `max_retry` → DLQ（未配置则丢弃 Ack） +- 不移植钉钉直发、按目标过滤、`x-dingtalk-sent` +- 不写连真实 CloudAMQP 的测试 +- 不新增管理 API 或表 +- 用户未明确要求时不要 commit + +--- + +## File Structure + +| File | Responsibility | +|------|----------------| +| `internal/config/config.go` | `SubscriptionConfig`、`StrategyOverride`、`NormalizeSubscriptions`、`ActiveSubscriptions` | +| `internal/config/config_test.go` | 默认值、跳过空 URL、未知 formatter | +| `config/config.yaml` | 第一条订阅，URL 为 `${RABBITMQ_URL}` | +| `internal/notify/service.go` | `Process`、`ErrUnprocessable`、`parseChannelID` | +| `internal/notify/service_test.go` | 无规则 / 过滤 / 命中 / 不可处理 / 内部错误 | +| `internal/handler/notify.go` | 解析 body 后调用 `Process` | +| `cmd/server/main.go` | 构造 `notify.Service`；Task 7 再启动订阅 | +| `internal/subscriber/tradesignal/signal.go` | `Signal`、`ParsedEventTime` | +| `internal/subscriber/tradesignal/format.go` | 文案格式化（含周期） | +| `internal/subscriber/tradesignal/format_test.go` | 开/平/加/减、周期、均价 | +| `internal/subscriber/tradesignal/override.go` | 数量倍数、杠杆覆盖 | +| `internal/subscriber/tradesignal/position.go` | 进程内均价 | +| `internal/subscriber/tradesignal/transform_test.go` | 覆盖 + 均价 | +| `internal/subscriber/tradesignal/convert.go` | `Convert(body) (event, data, error)` | +| `internal/subscriber/tradesignal/convert_test.go` | event / formatted / period / 无效消息 | +| `internal/subscriber/handle.go` | `HandleMessage`、`DecideRetry`（无 AMQP） | +| `internal/subscriber/handle_test.go` | Ack / 重试 / DLQ | +| `internal/subscriber/subscriber.go` | `Run`：连 MQ、声明、消费、重连 | +| `README.md` | 订阅配置说明 | +| `go.mod` | 增加 `amqp091-go` | + +--- + +### Task 1: 订阅配置 + +**Files:** +- Modify: `internal/config/config.go` +- Create: `internal/config/config_test.go` +- Modify: `config/config.yaml` + +**Interfaces:** +- Produces: + - `type QuantityMultipliers struct { Open, Add, Reduce, Close float64 }` tags `open,add,reduce,close` + - `type StrategyOverride struct { QuantityMultipliers QuantityMultipliers; Leverage *int }` + - `func (o StrategyOverride) QuantityMultiplierFor(action string) float64` — `<=0` 视为 1 + - `type SubscriptionConfig` 字段见下方 + - `func (c *Config) NormalizeSubscriptions() error` + - `func (c *Config) ActiveSubscriptions() []SubscriptionConfig` — 仅 `url != ""` + - `Load` 在 Unmarshal 之后调用 `NormalizeSubscriptions` + +- [ ] **Step 1: Write the failing test** + +Create `internal/config/config_test.go`: + +```go +package config + +import "testing" + +func TestNormalizeSubscriptionDefaults(t *testing.T) { + cfg := &Config{Subscriptions: []SubscriptionConfig{{ + URL: "amqps://example.invalid/vhost", + Queue: "trade.signal.notify.queue", + Source: "trade-signal", + }}} + if err := cfg.NormalizeSubscriptions(); err != nil { + t.Fatal(err) + } + s := cfg.Subscriptions[0] + if s.Name != "trade.signal.notify.queue" { + t.Fatalf("name=%q", s.Name) + } + if s.MaxRetry != 3 { + t.Fatalf("max_retry=%d", s.MaxRetry) + } + if s.ExchangeType != "fanout" { + t.Fatalf("exchange_type=%q", s.ExchangeType) + } + if s.Formatter != "trade_signal" { + t.Fatalf("formatter=%q", s.Formatter) + } +} + +func TestNormalizeSkipsEmptyURL(t *testing.T) { + cfg := &Config{Subscriptions: []SubscriptionConfig{{ + Queue: "q", Source: "s", + }}} + if err := cfg.NormalizeSubscriptions(); err != nil { + t.Fatal(err) + } + if n := len(cfg.ActiveSubscriptions()); n != 0 { + t.Fatalf("active=%d", n) + } +} + +func TestNormalizeUnknownFormatter(t *testing.T) { + cfg := &Config{Subscriptions: []SubscriptionConfig{{ + URL: "amqps://example.invalid/vhost", Queue: "q", Source: "s", Formatter: "other", + }}} + if err := cfg.NormalizeSubscriptions(); err == nil { + t.Fatal("expected error") + } +} + +func TestQuantityMultiplierFor(t *testing.T) { + o := StrategyOverride{QuantityMultipliers: QuantityMultipliers{Open: 100, Add: 0}} + if o.QuantityMultiplierFor("OPEN") != 100 { + t.Fatalf("open=%v", o.QuantityMultiplierFor("OPEN")) + } + if o.QuantityMultiplierFor("ADD") != 1 { + t.Fatalf("add<=0 should be 1, got %v", o.QuantityMultiplierFor("ADD")) + } + if o.QuantityMultiplierFor("UNKNOWN") != 1 { + t.Fatalf("unknown=%v", o.QuantityMultiplierFor("UNKNOWN")) + } +} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `go test ./internal/config/ -count=1` + +Expected: FAIL — `SubscriptionConfig` / `NormalizeSubscriptions` undefined + +- [ ] **Step 3: Implement config types and normalize** + +In `internal/config/config.go`, add to `Config`: + +```go +Subscriptions []SubscriptionConfig `mapstructure:"subscriptions"` +``` + +Add types and methods: + +```go +type SubscriptionConfig struct { + Name string `mapstructure:"name"` + URL string `mapstructure:"url"` + Queue string `mapstructure:"queue"` + DeadLetterQueue string `mapstructure:"dead_letter_queue"` + Exchange string `mapstructure:"exchange"` + ExchangeType string `mapstructure:"exchange_type"` + RoutingKey string `mapstructure:"routing_key"` + MaxRetry int `mapstructure:"max_retry"` + Source string `mapstructure:"source"` + Formatter string `mapstructure:"formatter"` + StrategyOverrides map[string]StrategyOverride `mapstructure:"strategy_overrides"` +} + +type StrategyOverride struct { + QuantityMultipliers QuantityMultipliers `mapstructure:"quantity_multipliers"` + Leverage *int `mapstructure:"leverage"` +} + +type QuantityMultipliers struct { + Open float64 `mapstructure:"open"` + Add float64 `mapstructure:"add"` + Reduce float64 `mapstructure:"reduce"` + Close float64 `mapstructure:"close"` +} + +func (o StrategyOverride) QuantityMultiplierFor(action string) float64 { + var v float64 + switch strings.ToUpper(action) { + case "OPEN": + v = o.QuantityMultipliers.Open + case "ADD": + v = o.QuantityMultipliers.Add + case "REDUCE": + v = o.QuantityMultipliers.Reduce + case "CLOSE": + v = o.QuantityMultipliers.Close + default: + return 1 + } + if v <= 0 { + return 1 + } + return v +} + +func (c *Config) NormalizeSubscriptions() error { + for i := range c.Subscriptions { + s := &c.Subscriptions[i] + if s.URL == "" { + continue + } + if s.Queue == "" { + return fmt.Errorf("subscriptions[%d]: queue is required", i) + } + if s.Source == "" { + return fmt.Errorf("subscriptions[%d]: source is required", i) + } + if s.Name == "" { + s.Name = s.Queue + } + if s.MaxRetry <= 0 { + s.MaxRetry = 3 + } + if s.ExchangeType == "" { + s.ExchangeType = "fanout" + } + if s.Formatter == "" { + s.Formatter = "trade_signal" + } + if s.Formatter != "trade_signal" { + return fmt.Errorf("subscriptions[%d]: unknown formatter %q", i, s.Formatter) + } + } + return nil +} + +func (c *Config) ActiveSubscriptions() []SubscriptionConfig { + out := make([]SubscriptionConfig, 0, len(c.Subscriptions)) + for _, s := range c.Subscriptions { + if s.URL != "" { + out = append(out, s) + } + } + return out +} +``` + +In `Load`, after `v.Unmarshal(&cfg)`: + +```go +if err := cfg.NormalizeSubscriptions(); err != nil { + return nil, err +} +``` + +Append to `config/config.yaml`（**不要**写入真实账号）： + +```yaml +subscriptions: + - name: trade-signal + url: "${RABBITMQ_URL}" + queue: trade.signal.notify.queue + dead_letter_queue: trade.signal.notify.dlq + exchange: trade.signal.executor.queue + exchange_type: fanout + routing_key: "" + max_retry: 3 + source: trade-signal + formatter: trade_signal + strategy_overrides: + BLONG: + quantity_multipliers: + open: 100 + add: 100 + reduce: 100 + close: 100 + leverage: 100 +``` + +- [ ] **Step 4: Run tests** + +Run: `go test ./internal/config/ -count=1` + +Expected: PASS + +- [ ] **Step 5: Commit**（仅当用户要求） + +```bash +git add internal/config/config.go internal/config/config_test.go config/config.yaml +git commit -m "feat: add configurable RabbitMQ subscription settings" +``` + +--- + +### Task 2: 抽出 NotifyService + +**Files:** +- Create: `internal/notify/service.go` +- Create: `internal/notify/service_test.go` +- Modify: `internal/handler/notify.go` +- Modify: `cmd/server/main.go` + +**Interfaces:** +- Consumes: 现有 `engine.Matcher.Match`、`engine.Renderer.Render`、`engine.Router.Route`、`store.GetTemplate`、`store.CreateMessageLog`、`condition.Evaluate` +- Produces: + - `var ErrUnprocessable error` + - `type Request struct { Source *model.Source; Event string; Data map[string]interface{} }` + - `type Result struct { Matched bool; Filtered bool; Channels []string; Reason string }` + - `type RuleMatcher interface { Match(ctx context.Context, sourceID int, event string) (*model.Rule, error) }` + - `type TemplateStore interface { GetTemplate(ctx context.Context, id int) (*model.Template, error) }` + - `type ChannelRouter interface { Route(ctx context.Context, rule *model.Rule, title, content string) []string }` + - `type MessageLogger interface { CreateMessageLog(ctx context.Context, ml *model.MessageLog) error }` + - `func NewService(m RuleMatcher, t TemplateStore, r *engine.Renderer, rt ChannelRouter, logs MessageLogger) *Service` + - `func (s *Service) Process(ctx context.Context, req Request) (Result, error)` + - 标题：`req.Source.Name + ": " + req.Event` + - 无规则（Match error）→ `Result{Matched:false}, nil` + - 条件 JSON 坏 / 渲染失败 → `fmt.Errorf("%w: ...", ErrUnprocessable)` + - 模板查找失败 → 普通 error（可重试） + - `NewNotifyHandler(svc *notify.Service)` + +- [ ] **Step 1: Write the failing tests** + +Create `internal/notify/service_test.go`: + +```go +package notify + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "aiaa-notification-service/internal/engine" + "aiaa-notification-service/internal/model" +) + +type fakeMatcher struct { + rule *model.Rule + err error +} + +func (f *fakeMatcher) Match(context.Context, int, string) (*model.Rule, error) { + return f.rule, f.err +} + +type fakeTemplates struct { + tmpl *model.Template + err error +} + +func (f *fakeTemplates) GetTemplate(context.Context, int) (*model.Template, error) { + return f.tmpl, f.err +} + +type fakeRouter struct{ channels []string } + +func (f *fakeRouter) Route(context.Context, *model.Rule, string, string) []string { + return f.channels +} + +type fakeLogs struct{} + +func (f *fakeLogs) CreateMessageLog(context.Context, *model.MessageLog) error { return nil } + +func newSvc(m *fakeMatcher, t *fakeTemplates, rt *fakeRouter) *Service { + return NewService(m, t, engine.NewRenderer(), rt, &fakeLogs{}) +} + +func TestProcessNoRule(t *testing.T) { + svc := newSvc(&fakeMatcher{err: errors.New("no rule")}, &fakeTemplates{}, &fakeRouter{}) + res, err := svc.Process(context.Background(), Request{ + Source: &model.Source{ID: 1, Name: "trade-signal"}, + Event: "trade.open", + Data: map[string]interface{}{"symbol": "BTCUSDT"}, + }) + if err != nil { + t.Fatal(err) + } + if res.Matched { + t.Fatal("expected unmatched") + } +} + +func TestProcessFiltered(t *testing.T) { + raw := json.RawMessage(`[{"field":"symbol","op":"eq","value":"ETHUSDT"}]`) + svc := newSvc(&fakeMatcher{rule: &model.Rule{ID: 1, TemplateID: 1, Conditions: &raw}}, &fakeTemplates{}, &fakeRouter{}) + res, err := svc.Process(context.Background(), Request{ + Source: &model.Source{ID: 1, Name: "trade-signal"}, + Event: "trade.open", + Data: map[string]interface{}{"symbol": "BTCUSDT"}, + }) + if err != nil { + t.Fatal(err) + } + if !res.Matched || !res.Filtered || res.Reason != "condition not met" { + t.Fatalf("%+v", res) + } +} + +func TestProcessMatched(t *testing.T) { + svc := newSvc( + &fakeMatcher{rule: &model.Rule{ID: 9, TemplateID: 1}}, + &fakeTemplates{tmpl: &model.Template{ID: 1, Content: "{{.formatted}}"}}, + &fakeRouter{channels: []string{"dingtalk:3"}}, + ) + res, err := svc.Process(context.Background(), Request{ + Source: &model.Source{ID: 1, Name: "trade-signal"}, + Event: "trade.open", + Data: map[string]interface{}{"formatted": "多单开仓"}, + }) + if err != nil { + t.Fatal(err) + } + if !res.Matched || res.Filtered || len(res.Channels) != 1 || res.Channels[0] != "dingtalk:3" { + t.Fatalf("%+v", res) + } +} + +func TestProcessInvalidConditions(t *testing.T) { + raw := json.RawMessage(`not-json`) + svc := newSvc(&fakeMatcher{rule: &model.Rule{ID: 1, TemplateID: 1, Conditions: &raw}}, &fakeTemplates{}, &fakeRouter{}) + _, err := svc.Process(context.Background(), Request{ + Source: &model.Source{ID: 1, Name: "s"}, + Event: "e", + Data: map[string]interface{}{}, + }) + if !errors.Is(err, ErrUnprocessable) { + t.Fatalf("err=%v", err) + } +} + +func TestProcessTemplateMissing(t *testing.T) { + svc := newSvc(&fakeMatcher{rule: &model.Rule{ID: 1, TemplateID: 1}}, &fakeTemplates{err: errors.New("nope")}, &fakeRouter{}) + _, err := svc.Process(context.Background(), Request{ + Source: &model.Source{ID: 1, Name: "s"}, + Event: "e", + Data: map[string]interface{}{}, + }) + if err == nil || errors.Is(err, ErrUnprocessable) { + t.Fatalf("want retryable error, got %v", err) + } +} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `go test ./internal/notify/ -count=1` + +Expected: FAIL — package / types undefined + +- [ ] **Step 3: Implement Service and thin handler** + +Create `internal/notify/service.go`：把 `handler.Handle` 的步骤 3–8 搬过来。条件 JSON 坏、渲染失败用 `fmt.Errorf("%w: %s", ErrUnprocessable, msg)`。模板找不到返回普通 error（文案 `template not found`）。message_log 仍异步 `go`。`parseChannelID` 移到本文件。 + +`NewService` 允许 `logs == nil`（不写 log）。 + +改 `internal/handler/notify.go`： + +```go +type NotifyHandler struct { + svc *notify.Service +} + +func NewNotifyHandler(svc *notify.Service) *NotifyHandler { + return &NotifyHandler{svc: svc} +} + +func (h *NotifyHandler) Handle(c *gin.Context) { + src := c.MustGet("source").(*model.Source) + body, err := io.ReadAll(c.Request.Body) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "read body failed"}) + return + } + p, err := parser.NewParser(src.ParseMode, src.ParsePattern) + if err != nil { + c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "parser setup: " + err.Error()}) + return + } + msg, err := p.Parse(body) + if err != nil { + c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "parse failed: " + err.Error()}) + return + } + res, err := h.svc.Process(c.Request.Context(), notify.Request{ + Source: src, Event: msg.Event, Data: msg.Data, + }) + if err != nil { + if errors.Is(err, notify.ErrUnprocessable) { + c.JSON(http.StatusUnprocessableEntity, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + if !res.Matched { + c.JSON(http.StatusOK, gin.H{"matched": false}) + return + } + if res.Filtered { + c.JSON(http.StatusOK, gin.H{"matched": true, "filtered": true, "reason": res.Reason}) + return + } + c.JSON(http.StatusOK, gin.H{"matched": true, "channels": res.Channels, "accepted": true}) +} +``` + +`cmd/server/main.go` 在构造 router 之后： + +```go +notifySvc := notify.NewService(matcher, st, renderer, router, st) +notifyH := handler.NewNotifyHandler(notifySvc) +``` + +删除未用 import。 + +- [ ] **Step 4: Run tests** + +Run: `go test ./internal/notify/ ./internal/handler/ ./internal/engine/ -count=1` + +Expected: PASS；`go build ./cmd/server` 成功 + +- [ ] **Step 5: Commit**（仅当用户要求） + +```bash +git add internal/notify internal/handler/notify.go cmd/server/main.go +git commit -m "refactor: extract shared NotifyService from HTTP handler" +``` + +--- + +### Task 3: 交易信号格式化（含周期） + +**Files:** +- Create: `internal/subscriber/tradesignal/signal.go` +- Create: `internal/subscriber/tradesignal/format.go` +- Create: `internal/subscriber/tradesignal/format_test.go` + +**Interfaces:** +- Produces: + - `type Signal` — JSON 标签与参考项目一致（`signalId`、`strategyCode`、`period` 等） + - `func (s *Signal) ParsedEventTime() time.Time` + - `type FormatOptions struct { AvgPrice *float64 }` + - `func Format(signal *Signal, opts ...FormatOptions) string` + - 在 `交易品种` 下一行：`period` trim 后非空则 `周期: {period}` + +- [ ] **Step 1: Write the failing tests** + +Create `internal/subscriber/tradesignal/format_test.go`： + +```go +package tradesignal + +import ( + "strings" + "testing" +) + +func ptr(v float64) *float64 { return &v } + +func TestFormatOpenIncludesPeriodAfterSymbol(t *testing.T) { + out := Format(&Signal{ + Symbol: "BTCUSDT", Side: "LONG", Action: "OPEN", + Quantity: ptr(0.01), Price: 64000.5, Leverage: 10, + Period: "1h", EventTime: "2026-06-23T01:30:00Z", + }) + if !strings.Contains(out, "多单开仓") || !strings.Contains(out, "交易品种: BTC") { + t.Fatalf("%s", out) + } + idxSym := strings.Index(out, "交易品种: BTC") + idxPer := strings.Index(out, "周期: 1h") + idxPx := strings.Index(out, "开仓价格:") + if idxPer < 0 || idxPer < idxSym || idxPx < idxPer { + t.Fatalf("period placement:\n%s", out) + } +} + +func TestFormatOmitsEmptyPeriod(t *testing.T) { + out := Format(&Signal{ + Symbol: "BTCUSDT", Side: "LONG", Action: "OPEN", + Price: 1, EventTime: "2026-06-23T01:30:00Z", + }) + if strings.Contains(out, "周期:") { + t.Fatalf("%s", out) + } +} + +func TestFormatCloseLong(t *testing.T) { + pnl, bal := 941.0, 74744.90 + out := Format(&Signal{ + Symbol: "BTCUSDT", Side: "LONG", Action: "CLOSE", + Quantity: ptr(3), Price: 63175.76, + EventTime: "2026-07-07T05:52:14Z", PnL: &pnl, AccountBalance: &bal, + }) + for _, want := range []string{"多单平仓", "平仓价格: 63175.76", "平仓盈亏: 941.00"} { + if !strings.Contains(out, want) { + t.Fatalf("missing %q in\n%s", want, out) + } + } +} + +func TestFormatWithAvgPrice(t *testing.T) { + avg := 150.0 + out := Format(&Signal{ + Symbol: "BTCUSDT", Side: "LONG", Action: "ADD", + Quantity: ptr(1), Price: 200, EventTime: "2026-07-07T05:52:14Z", + }, FormatOptions{AvgPrice: &avg}) + if !strings.Contains(out, "平均单价: 150.00") { + t.Fatalf("%s", out) + } +} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `go test ./internal/subscriber/tradesignal/ -count=1` + +Expected: FAIL — `Format` undefined + +- [ ] **Step 3: Port formatter and add period** + +`signal.go`：从 `/Users/ryan/Documents/code/go/test-mq-to-ali/internal/model/signal.go` 原样移植（包名改为 `tradesignal`）。 + +`format.go`：从 `/Users/ryan/Documents/code/go/test-mq-to-ali/internal/formatter/formatter.go` 移植，`Options` 改名为 `FormatOptions`。在 + +```go +lines = append(lines, fmt.Sprintf("交易品种: %s", symbol)) +``` + +之后立刻插入： + +```go +if p := strings.TrimSpace(signal.Period); p != "" { + lines = append(lines, fmt.Sprintf("周期: %s", p)) +} +``` + +其余文案规则保持与参考项目一致。 + +- [ ] **Step 4: Run tests** + +Run: `go test ./internal/subscriber/tradesignal/ -count=1` + +Expected: PASS + +- [ ] **Step 5: Commit**（仅当用户要求） + +```bash +git add internal/subscriber/tradesignal +git commit -m "feat: port trade-signal formatter and include period" +``` + +--- + +### Task 4: 策略覆盖与均价追踪 + +**Files:** +- Create: `internal/subscriber/tradesignal/override.go` +- Create: `internal/subscriber/tradesignal/position.go` +- Create: `internal/subscriber/tradesignal/transform_test.go` + +**Interfaces:** +- Consumes: `config.StrategyOverride.QuantityMultiplierFor` +- Produces: + - `func Apply(signal *Signal, override *config.StrategyOverride) *Signal` — override 为 nil 时返回原指针；有 quantity>0 则乘倍数；`Leverage != nil && *Leverage > 0` 则覆盖杠杆 + - `type Snapshot struct { AvgPrice float64; Size float64; HasAvg bool }` + - `func NewTracker() *Tracker` + - `func (t *Tracker) Apply(signal *Signal) Snapshot` — key=`strategyCode|symbol|side`（大写）；同 `signalId` 只应用一次；CLOSE/REDUCE 快照为减仓前均价 + +- [ ] **Step 1: Write the failing tests** + +Create `internal/subscriber/tradesignal/transform_test.go`，从参考项目移植并改 import： + +- `/Users/ryan/Documents/code/go/test-mq-to-ali/internal/strategy/override_test.go` → `Apply` + `config.StrategyOverride` +- `/Users/ryan/Documents/code/go/test-mq-to-ali/internal/position/tracker_test.go` → `NewTracker().Apply` + +包名 `tradesignal`。`ptr` 已在 `format_test.go` 同包，本文件不要再定义 `ptr`。 + +至少覆盖：OPEN 倍数+杠杆；无 override 返回原指针；仅 margin ratio 不造 quantity；OPEN+ADD 均价 100/200 → 150；CLOSE 报入场均价；同 signalId 不重复计；LONG/SHORT 隔离。 + +- [ ] **Step 2: Run test to verify it fails** + +Run: `go test ./internal/subscriber/tradesignal/ -count=1` + +Expected: FAIL — `Apply` / `NewTracker` undefined + +- [ ] **Step 3: Port override and tracker** + +`override.go`：从 `/Users/ryan/Documents/code/go/test-mq-to-ali/internal/strategy/override.go` 移植，改用本包 `Signal` 与 `config.StrategyOverride`。 + +`position.go`：从 `/Users/ryan/Documents/code/go/test-mq-to-ali/internal/position/tracker.go` 原样移植（包名 `tradesignal`，`model.Signal` 改为 `Signal`）。 + +- [ ] **Step 4: Run tests** + +Run: `go test ./internal/subscriber/tradesignal/ -count=1` + +Expected: PASS + +- [ ] **Step 5: Commit**（仅当用户要求） + +```bash +git add internal/subscriber/tradesignal +git commit -m "feat: port trade-signal overrides and average-price tracker" +``` + +--- + +### Task 5: Convert 管道 + +**Files:** +- Create: `internal/subscriber/tradesignal/convert.go` +- Create: `internal/subscriber/tradesignal/convert_test.go` + +**Interfaces:** +- Consumes: `Format`、`Apply`、`Tracker.Apply`、`config.StrategyOverride` +- Produces: + - `var ErrInvalidSignal error` + - `type Converter struct` 内含 overrides 与 `*Tracker` + - `func NewConverter(overrides map[string]config.StrategyOverride) *Converter` + - `func (c *Converter) Convert(body []byte) (event string, data map[string]interface{}, err error)` + - 顺序：Unmarshal → action 空则 `ErrInvalidSignal` → override → tracker → Format → data + - `event = "trade." + strings.ToLower(action)` + - data：覆盖后信号 JSON 圆整为 map，再设 `formatted`；`HasAvg` 时设 `avgPrice` + +- [ ] **Step 1: Write the failing tests** + +```go +package tradesignal + +import ( + "errors" + "strings" + "testing" + + "aiaa-notification-service/internal/config" +) + +func TestConvertOpen(t *testing.T) { + lev := 100 + c := NewConverter(map[string]config.StrategyOverride{ + "BLONG": {QuantityMultipliers: config.QuantityMultipliers{Open: 100}, Leverage: &lev}, + }) + event, data, err := c.Convert([]byte(`{ + "signalId":"s1","strategyCode":"BLONG","symbol":"BTCUSDT", + "side":"LONG","action":"OPEN","quantity":0.01,"price":64000, + "leverage":10,"period":"1h","eventTime":"2026-06-23T01:30:00Z" + }`)) + if err != nil { + t.Fatal(err) + } + if event != "trade.open" { + t.Fatalf("event=%q", event) + } + formatted, _ := data["formatted"].(string) + if !strings.Contains(formatted, "周期: 1h") || !strings.Contains(formatted, "开仓数量: 1.00") { + t.Fatalf("formatted=\n%s", formatted) + } + if data["period"] != "1h" || data["strategyCode"] != "BLONG" { + t.Fatalf("data=%v", data) + } + if data["leverage"] != float64(100) && data["leverage"] != 100 { + t.Fatalf("leverage=%v", data["leverage"]) + } +} + +func TestConvertInvalidJSON(t *testing.T) { + _, _, err := NewConverter(nil).Convert([]byte(`{`)) + if !errors.Is(err, ErrInvalidSignal) { + t.Fatalf("err=%v", err) + } +} + +func TestConvertMissingAction(t *testing.T) { + _, _, err := NewConverter(nil).Convert([]byte(`{"symbol":"BTCUSDT"}`)) + if !errors.Is(err, ErrInvalidSignal) { + t.Fatalf("err=%v", err) + } +} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `go test ./internal/subscriber/tradesignal/ -run Convert -count=1` + +Expected: FAIL — `Convert` undefined + +- [ ] **Step 3: Implement Convert** + +```go +func (c *Converter) Convert(body []byte) (string, map[string]interface{}, error) { + var sig Signal + if err := json.Unmarshal(body, &sig); err != nil { + return "", nil, fmt.Errorf("%w: %v", ErrInvalidSignal, err) + } + if strings.TrimSpace(sig.Action) == "" { + return "", nil, fmt.Errorf("%w: missing action", ErrInvalidSignal) + } + out := Apply(&sig, c.overrideFor(sig.StrategyCode)) + snap := c.positions.Apply(out) + var opts FormatOptions + if snap.HasAvg { + avg := snap.AvgPrice + opts.AvgPrice = &avg + } + text := Format(out, opts) + data, err := toData(out) + if err != nil { + return "", nil, err + } + data["formatted"] = text + if snap.HasAvg { + data["avgPrice"] = snap.AvgPrice + } + return "trade." + strings.ToLower(out.Action), data, nil +} +``` + +`toData`：`json.Marshal` 信号再 `Unmarshal` 到 `map[string]interface{}`。`overrideFor` 按 `strategyCode` 查 map，没有则 nil。 + +- [ ] **Step 4: Run tests** + +Run: `go test ./internal/subscriber/tradesignal/ -count=1` + +Expected: PASS + +- [ ] **Step 5: Commit**（仅当用户要求） + +```bash +git add internal/subscriber/tradesignal +git commit -m "feat: convert trade-signal messages into notify event and data" +``` + +--- + +### Task 6: 投递处置（Ack / 重试 / DLQ） + +**Files:** +- Create: `internal/subscriber/handle.go` +- Create: `internal/subscriber/handle_test.go` + +**Interfaces:** +- Consumes: `tradesignal.Converter.Convert`、`tradesignal.ErrInvalidSignal`、`notify.Process`、`notify.ErrUnprocessable` +- Produces: + - `type Disposition int` — `DispositionAck`、`DispositionRetry`、`DispositionDLQ` + - `func DecideRetry(retryCount, maxRetry int) Disposition` — `retryCount+1 > maxRetry` → DLQ，否则 Retry + - `func RetryCount(headers map[string]any) int` — 读 `x-retry-count`（int/int32/int64/float64/string） + - `type SourceLookup func(ctx context.Context, name string) (*model.Source, error)` + - `type ProcessFunc func(ctx context.Context, req notify.Request) (notify.Result, error)` + - `type HandleInput struct { Body []byte; Headers map[string]any; SourceName string; MaxRetry int }` + - `func HandleMessage(ctx context.Context, in HandleInput, conv *tradesignal.Converter, lookup SourceLookup, process ProcessFunc) Disposition` + - 无效信号 / lookup error / source nil / `status != 1` / Process 无 error（含 unmatched、filtered、`ErrUnprocessable`）→ Ack + - Process 其它 error → `DecideRetry(RetryCount(headers), maxRetry)` + +- [ ] **Step 1: Write the failing tests** + +```go +package subscriber + +import ( + "context" + "errors" + "strings" + "testing" + + "aiaa-notification-service/internal/model" + "aiaa-notification-service/internal/notify" + "aiaa-notification-service/internal/subscriber/tradesignal" +) + +func TestDecideRetry(t *testing.T) { + if DecideRetry(0, 3) != DispositionRetry { + t.Fatal("first failure should retry") + } + if DecideRetry(3, 3) != DispositionDLQ { + t.Fatal("retry 4 > 3 should dlq") + } +} + +func TestRetryCount(t *testing.T) { + if RetryCount(nil) != 0 { + t.Fatal() + } + if RetryCount(map[string]any{"x-retry-count": int32(2)}) != 2 { + t.Fatal() + } +} + +func enabledSrc() *model.Source { + return &model.Source{ID: 1, Name: "trade-signal", Status: 1} +} + +func TestHandleInvalidJSONAck(t *testing.T) { + d := HandleMessage(context.Background(), HandleInput{Body: []byte(`{`), SourceName: "trade-signal", MaxRetry: 3}, + tradesignal.NewConverter(nil), + func(context.Context, string) (*model.Source, error) { return enabledSrc(), nil }, + func(context.Context, notify.Request) (notify.Result, error) { + t.Fatal("process should not run") + return notify.Result{}, nil + }) + if d != DispositionAck { + t.Fatalf("%v", d) + } +} + +func TestHandleMissingSourceAck(t *testing.T) { + d := HandleMessage(context.Background(), HandleInput{ + Body: []byte(`{"action":"OPEN","symbol":"BTCUSDT"}`), SourceName: "trade-signal", MaxRetry: 3, + }, tradesignal.NewConverter(nil), + func(context.Context, string) (*model.Source, error) { return nil, errors.New("not found") }, + func(context.Context, notify.Request) (notify.Result, error) { t.Fatal("process"); return notify.Result{}, nil }) + if d != DispositionAck { + t.Fatalf("%v", d) + } +} + +func TestHandleDisabledSourceAck(t *testing.T) { + d := HandleMessage(context.Background(), HandleInput{ + Body: []byte(`{"action":"OPEN"}`), SourceName: "trade-signal", MaxRetry: 3, + }, tradesignal.NewConverter(nil), + func(context.Context, string) (*model.Source, error) { + return &model.Source{ID: 1, Name: "trade-signal", Status: 0}, nil + }, + func(context.Context, notify.Request) (notify.Result, error) { t.Fatal("process"); return notify.Result{}, nil }) + if d != DispositionAck { + t.Fatalf("%v", d) + } +} + +func TestHandleProcessUnprocessableAck(t *testing.T) { + d := HandleMessage(context.Background(), HandleInput{ + Body: []byte(`{"action":"OPEN"}`), SourceName: "trade-signal", MaxRetry: 3, + }, tradesignal.NewConverter(nil), + func(context.Context, string) (*model.Source, error) { return enabledSrc(), nil }, + func(context.Context, notify.Request) (notify.Result, error) { + return notify.Result{}, notify.ErrUnprocessable + }) + if d != DispositionAck { + t.Fatalf("%v", d) + } +} + +func TestHandleProcessErrorRetryThenDLQ(t *testing.T) { + process := func(context.Context, notify.Request) (notify.Result, error) { + return notify.Result{}, errors.New("db down") + } + lookup := func(context.Context, string) (*model.Source, error) { return enabledSrc(), nil } + conv := tradesignal.NewConverter(nil) + body := []byte(`{"action":"OPEN"}`) + d := HandleMessage(context.Background(), HandleInput{Body: body, SourceName: "s", MaxRetry: 3}, conv, lookup, process) + if d != DispositionRetry { + t.Fatalf("%v", d) + } + d = HandleMessage(context.Background(), HandleInput{ + Body: body, Headers: map[string]any{"x-retry-count": 3}, SourceName: "s", MaxRetry: 3, + }, conv, lookup, process) + if d != DispositionDLQ { + t.Fatalf("%v", d) + } +} + +func TestHandleSuccessAckPassesEventAndFormatted(t *testing.T) { + var got notify.Request + d := HandleMessage(context.Background(), HandleInput{ + Body: []byte(`{"action":"CLOSE","symbol":"ETHUSDT","period":"4h","price":1}`), + SourceName: "trade-signal", MaxRetry: 3, + }, tradesignal.NewConverter(nil), + func(context.Context, string) (*model.Source, error) { return enabledSrc(), nil }, + func(_ context.Context, req notify.Request) (notify.Result, error) { + got = req + return notify.Result{Matched: true, Channels: []string{"dingtalk:1"}}, nil + }) + if d != DispositionAck { + t.Fatalf("%v", d) + } + if got.Event != "trade.close" { + t.Fatalf("event=%q", got.Event) + } + if got.Data["period"] != "4h" { + t.Fatalf("period=%v", got.Data["period"]) + } + formatted, _ := got.Data["formatted"].(string) + if !strings.Contains(formatted, "周期: 4h") { + t.Fatalf("formatted=%s", formatted) + } +} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `go test ./internal/subscriber/ -count=1` + +Expected: FAIL — `HandleMessage` undefined + +- [ ] **Step 3: Implement handle.go** + +按 Interfaces 实现。`HandleMessage` 伪代码： + +```go +event, data, err := conv.Convert(in.Body) +if err != nil { // 含 ErrInvalidSignal + return DispositionAck +} +src, err := lookup(ctx, in.SourceName) +if err != nil || src == nil || src.Status != 1 { + return DispositionAck +} +_, err = process(ctx, notify.Request{Source: src, Event: event, Data: data}) +if err == nil || errors.Is(err, notify.ErrUnprocessable) { + return DispositionAck +} +return DecideRetry(RetryCount(in.Headers), in.MaxRetry) +``` + +无匹配规则、条件未过由 `Process` 返回 `error=nil`，因此 Ack。 + +- [ ] **Step 4: Run tests** + +Run: `go test ./internal/subscriber/ ./internal/subscriber/tradesignal/ -count=1` + +Expected: PASS + +- [ ] **Step 5: Commit**（仅当用户要求） + +```bash +git add internal/subscriber +git commit -m "feat: decide RabbitMQ ack, retry, and DLQ without a live broker" +``` + +--- + +### Task 7: 接入 AMQP、启动与文档 + +**Files:** +- Create: `internal/subscriber/subscriber.go` +- Modify: `cmd/server/main.go` +- Modify: `README.md` +- Modify: `go.mod` / `go.sum` + +**Interfaces:** +- Consumes: `config.SubscriptionConfig`、`HandleMessage`、`notify.Service.Process`、`store.GetSourceByName` +- Produces: + - `func New(cfg config.SubscriptionConfig, lookup SourceLookup, process ProcessFunc) (*Subscriber, error)` — `formatter` 必须是 `trade_signal`（Normalize 已保证） + - `func (s *Subscriber) Run(ctx context.Context) error` — 断线 5s 重连；`ctx` 取消则返回 + - 声明 durable exchange（若配置）、durable 队列、绑定、durable DLQ；QoS 1；`autoAck=false`；consumer tag = `cfg.Name` + - Retry：`x-retry-count+1` 后 `Publish` 到本队列；DLQ：`Publish` 到 `DeadLetterQueue`（空则 Ack 丢掉） + - main：`signal.NotifyContext`；对 `cfg.ActiveSubscriptions()` 各 `go Run(ctx)`；再等 ctx 取消后 `Shutdown` HTTP + +- [ ] **Step 1: Add dependency** + +Run: `go get github.com/rabbitmq/amqp091-go` + +- [ ] **Step 2: Implement subscriber.go** + +对照 `/Users/ryan/Documents/code/go/test-mq-to-ali/internal/mq/consumer.go` 的 `Run` / `consumeOnce` / `ensureQueue` / `retryOrDLQ` / `publishToQueue`。差别： + +- 用 slog，不用 log +- 每条消息：`HandleMessage` → 按 Disposition Ack / 重投 / DLQ +- `retryOrDLQ` 只在 `DispositionRetry` / `DispositionDLQ` 时调用；先把 header 里的 `x-retry-count` 写成 `RetryCount+1` +- 不要钉钉 sent/abandoned header +- `New` 里 `tradesignal.NewConverter(cfg.StrategyOverrides)` + +`lookup` 包装： + +```go +func (st *store.Store) /* in main */ { + lookup := func(ctx context.Context, name string) (*model.Source, error) { + return st.GetSourceByName(ctx, name) + } +} +``` + +- [ ] **Step 3: Wire main.go** + +把现有 `signal.Notify` + `<-quit` 换成： + +```go +ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) +defer stop() + +lookup := func(ctx context.Context, name string) (*model.Source, error) { + return st.GetSourceByName(ctx, name) +} +process := notifySvc.Process +for _, sub := range cfg.ActiveSubscriptions() { + sub := sub + cons, err := subscriber.New(sub, lookup, process) + if err != nil { + slog.Error("subscriber init", "name", sub.Name, "error", err) + os.Exit(1) + } + go func() { + if err := cons.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { + slog.Error("subscriber stopped", "name", sub.Name, "error", err) + } + }() + slog.Info("subscriber started", "name", sub.Name, "queue", sub.Queue, "source", sub.Source) +} + +// ListenAndServe in goroutine as today +<-ctx.Done() +slog.Info("shutting down...") +shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) +defer cancel() +_ = srv.Shutdown(shutdownCtx) +``` + +`RABBITMQ_URL` 未设时 `url` 展开为空，`ActiveSubscriptions` 为空，不启动消费。 + +- [ ] **Step 4: README** + +在配置表增加： + +| 配置项 | 说明 | 默认 | +|--------|------|------| +| `subscriptions` | MQ 订阅列表；`url` 为空则跳过 | 空 | +| `subscriptions[].source` | 对应已有 Source.name | 必填（有 url 时） | +| `subscriptions[].formatter` | 目前仅 `trade_signal` | `trade_signal` | + +说明：设置环境变量 `RABBITMQ_URL`；需事先创建 Source `trade-signal`、模板（可用 `{{.formatted}}`）、规则 `trade.open` / `trade.add` / `trade.close` / `trade.reduce`、渠道。条件可用 `strategyCode` / `symbol` / `period`。 + +- [ ] **Step 5: Run tests and build** + +Run: + +``` +go test ./internal/... -count=1 +go build -o /tmp/notification-server ./cmd/server +``` + +Expected: 全绿；编译成功 + +- [ ] **Step 6: Commit**（仅当用户要求） + +```bash +git add internal/subscriber cmd/server/main.go README.md go.mod go.sum +git commit -m "feat: subscribe to RabbitMQ trade signals and notify by rules" +``` + +--- + +## Self-review + +**Spec coverage** + +| Spec | Task | +|------|------| +| 订阅列表 + 默认值 + `${RABBITMQ_URL}` | 1 | +| 抽出 Process，HTTP 共用 | 2 | +| 格式化 + period 原样 | 3 | +| 策略覆盖 + 进程内均价 | 4 | +| event / data.formatted / 原始字段 | 5 | +| Ack / 重试 / DLQ 判定 | 6 | +| 声明绑定、重连、main 启动、文档 | 7 | +| 不移植钉钉过滤 | 6/7 不包含 | +| 不连真实 broker | 全程 | + +**Placeholder scan:** 无 TBD；实现步骤含代码或明确对照参考文件。 + +**Type consistency:** `Process(ctx, Request) (Result, error)`、`ErrUnprocessable`、`Convert`、`HandleMessage` / `Disposition*` / `SourceLookup` / `ProcessFunc` 在后续任务中名称一致。 diff --git a/go.mod b/go.mod index 408bb62..830e7f5 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,7 @@ require ( github.com/go-sql-driver/mysql v1.10.0 github.com/jmoiron/sqlx v1.4.0 github.com/logbull/logbull-go v0.2.0 + github.com/rabbitmq/amqp091-go v1.13.0 github.com/redis/go-redis/v9 v9.21.0 github.com/spf13/viper v1.21.0 ) diff --git a/go.sum b/go.sum index 9baccee..88fffaa 100644 --- a/go.sum +++ b/go.sum @@ -81,6 +81,8 @@ github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw= github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= +github.com/rabbitmq/amqp091-go v1.13.0 h1:L8NA1WtF76C6KA3LAoufjfLgbist/If1UQYcsOjtxXA= +github.com/rabbitmq/amqp091-go v1.13.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= github.com/redis/go-redis/v9 v9.21.0 h1:FPBE4hhbAke+TLmcY3WkpbDffJEomdqPn3HYiqAtL9E= github.com/redis/go-redis/v9 v9.21.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= diff --git a/internal/cache/redis.go b/internal/cache/redis.go index 0ead04e..5ab7233 100644 --- a/internal/cache/redis.go +++ b/internal/cache/redis.go @@ -175,3 +175,15 @@ func (c *Cache) CheckRateLimit(ctx context.Context, sourceID int, limitPerSec in func (c *Cache) Close() error { return c.rdb.Close() } + +func dedupKey(hash string) string { + return "notify:dedup:" + hash +} + +func (c *Cache) ClaimDedup(ctx context.Context, hash string, ttl time.Duration) (bool, error) { + return c.rdb.SetNX(ctx, dedupKey(hash), "1", ttl).Result() +} + +func (c *Cache) ReleaseDedup(ctx context.Context, hash string) error { + return c.rdb.Del(ctx, dedupKey(hash)).Err() +} diff --git a/internal/config/config.go b/internal/config/config.go index 6ec9344..6ddee25 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -4,17 +4,107 @@ import ( "fmt" "os" "strings" + "time" "github.com/spf13/viper" ) type Config struct { - Server ServerConfig `mapstructure:"server"` - Database DatabaseConfig `mapstructure:"database"` - Redis RedisConfig `mapstructure:"redis"` - SMTP SMTPConfig `mapstructure:"smtp"` - RateLimit RateLimitConfig `mapstructure:"rate_limit"` - Logbull LogbullConfig `mapstructure:"logbull"` + Server ServerConfig `mapstructure:"server"` + Database DatabaseConfig `mapstructure:"database"` + Redis RedisConfig `mapstructure:"redis"` + SMTP SMTPConfig `mapstructure:"smtp"` + RateLimit RateLimitConfig `mapstructure:"rate_limit"` + Logbull LogbullConfig `mapstructure:"logbull"` + Subscriptions []SubscriptionConfig `mapstructure:"subscriptions"` + SubscriptionDedupTTL time.Duration `mapstructure:"subscription_dedup_ttl"` +} + +type SubscriptionConfig struct { + Name string `mapstructure:"name"` + URL string `mapstructure:"url"` + Queue string `mapstructure:"queue"` + DeadLetterQueue string `mapstructure:"dead_letter_queue"` + Exchange string `mapstructure:"exchange"` + ExchangeType string `mapstructure:"exchange_type"` + RoutingKey string `mapstructure:"routing_key"` + MaxRetry int `mapstructure:"max_retry"` + Source string `mapstructure:"source"` + Formatter string `mapstructure:"formatter"` + StrategyOverrides map[string]StrategyOverride `mapstructure:"strategy_overrides"` +} + +type StrategyOverride struct { + QuantityMultipliers QuantityMultipliers `mapstructure:"quantity_multipliers"` + Leverage *int `mapstructure:"leverage"` +} + +type QuantityMultipliers struct { + Open float64 `mapstructure:"open"` + Add float64 `mapstructure:"add"` + Reduce float64 `mapstructure:"reduce"` + Close float64 `mapstructure:"close"` +} + +func (o StrategyOverride) QuantityMultiplierFor(action string) float64 { + var v float64 + switch strings.ToUpper(action) { + case "OPEN": + v = o.QuantityMultipliers.Open + case "ADD": + v = o.QuantityMultipliers.Add + case "REDUCE": + v = o.QuantityMultipliers.Reduce + case "CLOSE": + v = o.QuantityMultipliers.Close + default: + return 1 + } + if v <= 0 { + return 1 + } + return v +} + +func (c *Config) NormalizeSubscriptions() error { + for i := range c.Subscriptions { + s := &c.Subscriptions[i] + if s.URL == "" { + continue + } + if s.Queue == "" { + return fmt.Errorf("subscriptions[%d]: queue is required", i) + } + if s.Source == "" { + return fmt.Errorf("subscriptions[%d]: source is required", i) + } + if s.Name == "" { + s.Name = s.Queue + } + if s.MaxRetry <= 0 { + s.MaxRetry = 3 + } + if s.ExchangeType == "" { + s.ExchangeType = "fanout" + } + if s.Formatter == "" { + s.Formatter = "trade_signal" + } + if s.Formatter != "trade_signal" { + return fmt.Errorf("subscriptions[%d]: unknown formatter %q", i, s.Formatter) + } + } + return nil +} + +func (c *Config) ActiveSubscriptions() []SubscriptionConfig { + out := make([]SubscriptionConfig, 0, len(c.Subscriptions)) + for _, s := range c.Subscriptions { + if s.URL != "" { + out = append(out, s) + } + } + return out } type ServerConfig struct { @@ -90,6 +180,12 @@ func Load(path string) (*Config, error) { if err := v.Unmarshal(&cfg); err != nil { return nil, fmt.Errorf("unmarshal config: %w", err) } + if err := cfg.NormalizeSubscriptions(); err != nil { + return nil, err + } + if cfg.SubscriptionDedupTTL <= 0 { + cfg.SubscriptionDedupTTL = time.Hour + } return &cfg, nil } diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..4961056 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,61 @@ +package config + +import "testing" + +func TestNormalizeSubscriptionDefaults(t *testing.T) { + cfg := &Config{Subscriptions: []SubscriptionConfig{{ + URL: "amqps://example.invalid/vhost", + Queue: "trade.signal.notify.queue", + Source: "trade-signal", + }}} + if err := cfg.NormalizeSubscriptions(); err != nil { + t.Fatal(err) + } + s := cfg.Subscriptions[0] + if s.Name != "trade.signal.notify.queue" { + t.Fatalf("name=%q", s.Name) + } + if s.MaxRetry != 3 { + t.Fatalf("max_retry=%d", s.MaxRetry) + } + if s.ExchangeType != "fanout" { + t.Fatalf("exchange_type=%q", s.ExchangeType) + } + if s.Formatter != "trade_signal" { + t.Fatalf("formatter=%q", s.Formatter) + } +} + +func TestNormalizeSkipsEmptyURL(t *testing.T) { + cfg := &Config{Subscriptions: []SubscriptionConfig{{ + Queue: "q", Source: "s", + }}} + if err := cfg.NormalizeSubscriptions(); err != nil { + t.Fatal(err) + } + if n := len(cfg.ActiveSubscriptions()); n != 0 { + t.Fatalf("active=%d", n) + } +} + +func TestNormalizeUnknownFormatter(t *testing.T) { + cfg := &Config{Subscriptions: []SubscriptionConfig{{ + URL: "amqps://example.invalid/vhost", Queue: "q", Source: "s", Formatter: "other", + }}} + if err := cfg.NormalizeSubscriptions(); err == nil { + t.Fatal("expected error") + } +} + +func TestQuantityMultiplierFor(t *testing.T) { + o := StrategyOverride{QuantityMultipliers: QuantityMultipliers{Open: 100, Add: 0}} + if o.QuantityMultiplierFor("OPEN") != 100 { + t.Fatalf("open=%v", o.QuantityMultiplierFor("OPEN")) + } + if o.QuantityMultiplierFor("ADD") != 1 { + t.Fatalf("add<=0 should be 1, got %v", o.QuantityMultiplierFor("ADD")) + } + if o.QuantityMultiplierFor("UNKNOWN") != 1 { + t.Fatalf("unknown=%v", o.QuantityMultiplierFor("UNKNOWN")) + } +} diff --git a/internal/handler/notify.go b/internal/handler/notify.go index 44f1765..8ae24ba 100644 --- a/internal/handler/notify.go +++ b/internal/handler/notify.go @@ -1,47 +1,34 @@ package handler import ( - "context" - "encoding/json" + "errors" "io" - "log/slog" "net/http" - "strconv" - "strings" - "aiaa-notification-service/internal/cache" - "aiaa-notification-service/internal/condition" - "aiaa-notification-service/internal/engine" "aiaa-notification-service/internal/model" + "aiaa-notification-service/internal/notify" "aiaa-notification-service/internal/parser" - "aiaa-notification-service/internal/store" "github.com/gin-gonic/gin" ) type NotifyHandler struct { - store *store.Store - cache *cache.Cache // used by engine - matcher *engine.Matcher - renderer *engine.Renderer - router *engine.Router + svc *notify.Service } -func NewNotifyHandler(s *store.Store, c *cache.Cache, m *engine.Matcher, r *engine.Renderer, rt *engine.Router) *NotifyHandler { - return &NotifyHandler{store: s, cache: c, matcher: m, renderer: r, router: rt} +func NewNotifyHandler(svc *notify.Service) *NotifyHandler { + return &NotifyHandler{svc: svc} } func (h *NotifyHandler) Handle(c *gin.Context) { src := c.MustGet("source").(*model.Source) - // 1. Read raw body body, err := io.ReadAll(c.Request.Body) if err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": "read body failed"}) return } - // 2. Parse message p, err := parser.NewParser(src.ParseMode, src.ParsePattern) if err != nil { c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "parser setup: " + err.Error()}) @@ -53,93 +40,24 @@ func (h *NotifyHandler) Handle(c *gin.Context) { return } - // 3. Match rule - rule, err := h.matcher.Match(c.Request.Context(), src.ID, msg.Event) + res, err := h.svc.Process(c.Request.Context(), notify.Request{ + Source: src, Event: msg.Event, Data: msg.Data, + }) if err != nil { - // No matching rule → 200 with matched: false + if errors.Is(err, notify.ErrUnprocessable) { + c.JSON(http.StatusUnprocessableEntity, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + if !res.Matched { c.JSON(http.StatusOK, gin.H{"matched": false}) return } - - // 4. Evaluate conditions - if rule.Conditions != nil { - var conds []model.Condition - if err := json.Unmarshal(*rule.Conditions, &conds); err != nil { - slog.Error("failed to unmarshal rule conditions", "rule_id", rule.ID, "error", err) - c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "invalid rule conditions"}) - return - } - if !condition.Evaluate(conds, msg.Data) { - c.JSON(http.StatusOK, gin.H{ - "matched": true, - "filtered": true, - "reason": "condition not met", - }) - return - } - } - - // 5. Get template content - tmpl, err := h.store.GetTemplate(c.Request.Context(), rule.TemplateID) - if err != nil { - c.JSON(http.StatusInternalServerError, gin.H{"error": "template not found"}) + if res.Filtered { + c.JSON(http.StatusOK, gin.H{"matched": true, "filtered": true, "reason": res.Reason}) return } - - // 6. Render template - content, err := h.renderer.Render(tmpl.Content, msg.Data) - if err != nil { - c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "template render failed: " + err.Error()}) - return - } - - // 7. Route to channels - title := src.Name + ": " + msg.Event - channels := h.router.Route(c.Request.Context(), rule, title, content) - - // 8. Log message (best effort, detached context) - go func() { - payloadJSON, _ := json.Marshal(msg.Data) - ctx := context.Background() - for _, chName := range channels { - chID := parseChannelID(chName) - ml := &model.MessageLog{ - RuleID: rule.ID, - ChannelID: chID, - Source: src.Name, - Event: msg.Event, - Payload: payloadJSON, - Content: content, - Status: "pending", - } - if err := h.store.CreateMessageLog(ctx, ml); err != nil { - slog.Warn("failed to create message log", "error", err) - } - } - }() - - slog.Info("notification accepted", - "source", src.Name, - "event", msg.Event, - "channels", channels, - ) - - c.JSON(http.StatusOK, gin.H{ - "matched": true, - "channels": channels, - "accepted": true, - }) -} - -// parseChannelID extracts the numeric channel ID from a channel name formatted as "type:id". -func parseChannelID(chName string) int { - idx := strings.LastIndex(chName, ":") - if idx < 0 || idx == len(chName)-1 { - return 0 - } - id, err := strconv.Atoi(chName[idx+1:]) - if err != nil { - return 0 - } - return id + c.JSON(http.StatusOK, gin.H{"matched": true, "channels": res.Channels, "accepted": true}) } diff --git a/internal/notify/service.go b/internal/notify/service.go new file mode 100644 index 0000000..84c555a --- /dev/null +++ b/internal/notify/service.go @@ -0,0 +1,130 @@ +package notify + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "strconv" + "strings" + + "aiaa-notification-service/internal/condition" + "aiaa-notification-service/internal/engine" + "aiaa-notification-service/internal/model" +) + +var ErrUnprocessable = errors.New("unprocessable") + +type Request struct { + Source *model.Source + Event string + Data map[string]interface{} +} + +type Result struct { + Matched bool + Filtered bool + Channels []string + Reason string +} + +type RuleMatcher interface { + Match(ctx context.Context, sourceID int, event string) (*model.Rule, error) +} + +type TemplateStore interface { + GetTemplate(ctx context.Context, id int) (*model.Template, error) +} + +type ChannelRouter interface { + Route(ctx context.Context, rule *model.Rule, title, content string) []string +} + +type MessageLogger interface { + CreateMessageLog(ctx context.Context, ml *model.MessageLog) error +} + +type Service struct { + matcher RuleMatcher + templates TemplateStore + renderer *engine.Renderer + router ChannelRouter + logs MessageLogger +} + +func NewService(m RuleMatcher, t TemplateStore, r *engine.Renderer, rt ChannelRouter, logs MessageLogger) *Service { + return &Service{matcher: m, templates: t, renderer: r, router: rt, logs: logs} +} + +func (s *Service) Process(ctx context.Context, req Request) (Result, error) { + rule, err := s.matcher.Match(ctx, req.Source.ID, req.Event) + if err != nil { + return Result{Matched: false}, nil + } + + if rule.Conditions != nil { + var conds []model.Condition + if err := json.Unmarshal(*rule.Conditions, &conds); err != nil { + slog.Error("failed to unmarshal rule conditions", "rule_id", rule.ID, "error", err) + return Result{}, fmt.Errorf("%w: invalid rule conditions", ErrUnprocessable) + } + if !condition.Evaluate(conds, req.Data) { + return Result{Matched: true, Filtered: true, Reason: "condition not met"}, nil + } + } + + tmpl, err := s.templates.GetTemplate(ctx, rule.TemplateID) + if err != nil { + return Result{}, fmt.Errorf("template not found") + } + + content, err := s.renderer.Render(tmpl.Content, req.Data) + if err != nil { + return Result{}, fmt.Errorf("%w: template render failed: %s", ErrUnprocessable, err.Error()) + } + + title := req.Source.Name + ": " + req.Event + channels := s.router.Route(ctx, rule, title, content) + + if s.logs != nil { + go func() { + payloadJSON, _ := json.Marshal(req.Data) + logCtx := context.Background() + for _, chName := range channels { + ml := &model.MessageLog{ + RuleID: rule.ID, + ChannelID: parseChannelID(chName), + Source: req.Source.Name, + Event: req.Event, + Payload: payloadJSON, + Content: content, + Status: "pending", + } + if err := s.logs.CreateMessageLog(logCtx, ml); err != nil { + slog.Warn("failed to create message log", "error", err) + } + } + }() + } + + slog.Info("notification accepted", + "source", req.Source.Name, + "event", req.Event, + "channels", channels, + ) + + return Result{Matched: true, Channels: channels}, nil +} + +func parseChannelID(chName string) int { + idx := strings.LastIndex(chName, ":") + if idx < 0 || idx == len(chName)-1 { + return 0 + } + id, err := strconv.Atoi(chName[idx+1:]) + if err != nil { + return 0 + } + return id +} diff --git a/internal/notify/service_test.go b/internal/notify/service_test.go new file mode 100644 index 0000000..50201d2 --- /dev/null +++ b/internal/notify/service_test.go @@ -0,0 +1,118 @@ +package notify + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "aiaa-notification-service/internal/engine" + "aiaa-notification-service/internal/model" +) + +type fakeMatcher struct { + rule *model.Rule + err error +} + +func (f *fakeMatcher) Match(context.Context, int, string) (*model.Rule, error) { + return f.rule, f.err +} + +type fakeTemplates struct { + tmpl *model.Template + err error +} + +func (f *fakeTemplates) GetTemplate(context.Context, int) (*model.Template, error) { + return f.tmpl, f.err +} + +type fakeRouter struct{ channels []string } + +func (f *fakeRouter) Route(context.Context, *model.Rule, string, string) []string { + return f.channels +} + +type fakeLogs struct{} + +func (f *fakeLogs) CreateMessageLog(context.Context, *model.MessageLog) error { return nil } + +func newSvc(m *fakeMatcher, t *fakeTemplates, rt *fakeRouter) *Service { + return NewService(m, t, engine.NewRenderer(), rt, &fakeLogs{}) +} + +func TestProcessNoRule(t *testing.T) { + svc := newSvc(&fakeMatcher{err: errors.New("no rule")}, &fakeTemplates{}, &fakeRouter{}) + res, err := svc.Process(context.Background(), Request{ + Source: &model.Source{ID: 1, Name: "trade-signal"}, + Event: "trade.open", + Data: map[string]interface{}{"symbol": "BTCUSDT"}, + }) + if err != nil { + t.Fatal(err) + } + if res.Matched { + t.Fatal("expected unmatched") + } +} + +func TestProcessFiltered(t *testing.T) { + raw := json.RawMessage(`[{"field":"symbol","op":"eq","value":"ETHUSDT"}]`) + svc := newSvc(&fakeMatcher{rule: &model.Rule{ID: 1, TemplateID: 1, Conditions: &raw}}, &fakeTemplates{}, &fakeRouter{}) + res, err := svc.Process(context.Background(), Request{ + Source: &model.Source{ID: 1, Name: "trade-signal"}, + Event: "trade.open", + Data: map[string]interface{}{"symbol": "BTCUSDT"}, + }) + if err != nil { + t.Fatal(err) + } + if !res.Matched || !res.Filtered || res.Reason != "condition not met" { + t.Fatalf("%+v", res) + } +} + +func TestProcessMatched(t *testing.T) { + svc := newSvc( + &fakeMatcher{rule: &model.Rule{ID: 9, TemplateID: 1}}, + &fakeTemplates{tmpl: &model.Template{ID: 1, Content: "{{.formatted}}"}}, + &fakeRouter{channels: []string{"dingtalk:3"}}, + ) + res, err := svc.Process(context.Background(), Request{ + Source: &model.Source{ID: 1, Name: "trade-signal"}, + Event: "trade.open", + Data: map[string]interface{}{"formatted": "多单开仓"}, + }) + if err != nil { + t.Fatal(err) + } + if !res.Matched || res.Filtered || len(res.Channels) != 1 || res.Channels[0] != "dingtalk:3" { + t.Fatalf("%+v", res) + } +} + +func TestProcessInvalidConditions(t *testing.T) { + raw := json.RawMessage(`not-json`) + svc := newSvc(&fakeMatcher{rule: &model.Rule{ID: 1, TemplateID: 1, Conditions: &raw}}, &fakeTemplates{}, &fakeRouter{}) + _, err := svc.Process(context.Background(), Request{ + Source: &model.Source{ID: 1, Name: "s"}, + Event: "e", + Data: map[string]interface{}{}, + }) + if !errors.Is(err, ErrUnprocessable) { + t.Fatalf("err=%v", err) + } +} + +func TestProcessTemplateMissing(t *testing.T) { + svc := newSvc(&fakeMatcher{rule: &model.Rule{ID: 1, TemplateID: 1}}, &fakeTemplates{err: errors.New("nope")}, &fakeRouter{}) + _, err := svc.Process(context.Background(), Request{ + Source: &model.Source{ID: 1, Name: "s"}, + Event: "e", + Data: map[string]interface{}{}, + }) + if err == nil || errors.Is(err, ErrUnprocessable) { + t.Fatalf("want retryable error, got %v", err) + } +} diff --git a/internal/subscriber/dedup.go b/internal/subscriber/dedup.go new file mode 100644 index 0000000..06e9eae --- /dev/null +++ b/internal/subscriber/dedup.go @@ -0,0 +1,70 @@ +package subscriber + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "sync" + "time" + + "aiaa-notification-service/internal/cache" +) + +type Deduper interface { + Claim(ctx context.Context, hash string) (bool, error) + Release(ctx context.Context, hash string) error +} + +func MessageHash(body []byte) string { + sum := sha256.Sum256(body) + return hex.EncodeToString(sum[:]) +} + +type MemoryDeduper struct { + mu sync.Mutex + seen map[string]struct{} +} + +func NewMemoryDeduper() *MemoryDeduper { + return &MemoryDeduper{seen: make(map[string]struct{})} +} + +func (d *MemoryDeduper) Claim(_ context.Context, hash string) (bool, error) { + d.mu.Lock() + defer d.mu.Unlock() + if _, ok := d.seen[hash]; ok { + return false, nil + } + d.seen[hash] = struct{}{} + return true, nil +} + +func (d *MemoryDeduper) Release(_ context.Context, hash string) error { + d.mu.Lock() + defer d.mu.Unlock() + delete(d.seen, hash) + return nil +} + +type cacheDeduper struct { + c *cache.Cache + ttl time.Duration +} + +func NewCacheDeduper(c *cache.Cache, ttl time.Duration) Deduper { + if c == nil { + return NewMemoryDeduper() + } + if ttl <= 0 { + ttl = time.Hour + } + return &cacheDeduper{c: c, ttl: ttl} +} + +func (d *cacheDeduper) Claim(ctx context.Context, hash string) (bool, error) { + return d.c.ClaimDedup(ctx, hash, d.ttl) +} + +func (d *cacheDeduper) Release(ctx context.Context, hash string) error { + return d.c.ReleaseDedup(ctx, hash) +} diff --git a/internal/subscriber/dedup_test.go b/internal/subscriber/dedup_test.go new file mode 100644 index 0000000..f3faa08 --- /dev/null +++ b/internal/subscriber/dedup_test.go @@ -0,0 +1,89 @@ +package subscriber + +import ( + "context" + "errors" + "sync/atomic" + "testing" + + "aiaa-notification-service/internal/model" + "aiaa-notification-service/internal/notify" + "aiaa-notification-service/internal/subscriber/tradesignal" +) + +func TestMessageHashStable(t *testing.T) { + a := MessageHash([]byte(`{"action":"OPEN","signalId":"s1"}`)) + b := MessageHash([]byte(`{"action":"OPEN","signalId":"s1"}`)) + c := MessageHash([]byte(`{"action":"CLOSE","signalId":"s1"}`)) + if a == "" || a != b { + t.Fatalf("hash should be stable, a=%q b=%q", a, b) + } + if a == c { + t.Fatal("different bodies should hash differently") + } +} + +func TestMemoryDeduperClaimOnce(t *testing.T) { + d := NewMemoryDeduper() + ok, err := d.Claim(context.Background(), "abc") + if err != nil || !ok { + t.Fatalf("first claim ok=%v err=%v", ok, err) + } + ok, err = d.Claim(context.Background(), "abc") + if err != nil || ok { + t.Fatalf("second claim should miss, ok=%v err=%v", ok, err) + } + if err := d.Release(context.Background(), "abc"); err != nil { + t.Fatal(err) + } + ok, err = d.Claim(context.Background(), "abc") + if err != nil || !ok { + t.Fatalf("after release should claim, ok=%v err=%v", ok, err) + } +} + +func TestHandleDuplicateAckSkipsProcess(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, string) (*model.Source, error) { return enabledSrc(), nil } + conv := tradesignal.NewConverter(nil) + body := []byte(`{"action":"OPEN","signalId":"dup-1"}`) + in := HandleInput{Body: body, SourceName: "trade-signal", MaxRetry: 3, Deduper: dedup} + + if d := HandleMessage(context.Background(), in, conv, lookup, process); d != DispositionAck { + t.Fatalf("first=%v", d) + } + if d := HandleMessage(context.Background(), in, conv, lookup, process); d != DispositionAck { + t.Fatalf("dup=%v", d) + } + if n.Load() != 1 { + t.Fatalf("process called %d times, want 1", n.Load()) + } +} + +func TestHandleProcessErrorReleasesDedup(t *testing.T) { + dedup := NewMemoryDeduper() + var n atomic.Int32 + process := func(context.Context, notify.Request) (notify.Result, error) { + n.Add(1) + return notify.Result{}, errors.New("db down") + } + lookup := func(context.Context, string) (*model.Source, error) { return enabledSrc(), nil } + conv := tradesignal.NewConverter(nil) + body := []byte(`{"action":"OPEN","signalId":"retry-1"}`) + in := HandleInput{Body: body, SourceName: "s", MaxRetry: 3, Deduper: dedup} + + if d := HandleMessage(context.Background(), in, conv, lookup, process); d != DispositionRetry { + t.Fatalf("first=%v", d) + } + if d := HandleMessage(context.Background(), in, conv, lookup, process); d != DispositionRetry { + t.Fatalf("retry should process again, got %v", d) + } + if n.Load() != 2 { + t.Fatalf("process called %d times, want 2", n.Load()) + } +} diff --git a/internal/subscriber/handle.go b/internal/subscriber/handle.go new file mode 100644 index 0000000..df71133 --- /dev/null +++ b/internal/subscriber/handle.go @@ -0,0 +1,117 @@ +package subscriber + +import ( + "context" + "errors" + "log/slog" + "strconv" + + "aiaa-notification-service/internal/model" + "aiaa-notification-service/internal/notify" + "aiaa-notification-service/internal/subscriber/tradesignal" +) + +const retryHeader = "x-retry-count" + +type Disposition int + +const ( + DispositionAck Disposition = iota + DispositionRetry + DispositionDLQ +) + +type SourceLookup func(ctx context.Context, name string) (*model.Source, error) + +type ProcessFunc func(ctx context.Context, req notify.Request) (notify.Result, error) + +type HandleInput struct { + Body []byte + Headers map[string]any + SourceName string + MaxRetry int + Deduper Deduper +} + +func DecideRetry(retryCount, maxRetry int) Disposition { + if retryCount+1 > maxRetry { + return DispositionDLQ + } + return DispositionRetry +} + +func RetryCount(headers map[string]any) int { + if headers == nil { + return 0 + } + v, ok := headers[retryHeader] + if !ok { + return 0 + } + switch n := v.(type) { + case int: + return n + case int32: + return int(n) + case int64: + return int(n) + case float64: + return int(n) + case string: + i, _ := strconv.Atoi(n) + return i + default: + return 0 + } +} + +func HandleMessage(ctx context.Context, in HandleInput, conv *tradesignal.Converter, lookup SourceLookup, process ProcessFunc) Disposition { + owned := false + hash := "" + if in.Deduper != nil { + 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) + } else if !ok { + slog.Info("duplicate message, ack", "hash", hash) + return DispositionAck + } else { + owned = true + } + } + + event, data, err := conv.Convert(in.Body) + if err != nil { + slog.Warn("invalid signal, ack", "error", 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, "error", err) + return DispositionAck + } + + res, err := process(ctx, notify.Request{Source: src, Event: event, Data: data}) + if err == nil { + if !res.Matched { + slog.Info("no matching rule", "source", src.Name, "event", event) + } else if res.Filtered { + slog.Info("rule filtered", "source", src.Name, "event", event, "reason", res.Reason) + } + return DispositionAck + } + if errors.Is(err, notify.ErrUnprocessable) { + slog.Warn("unprocessable notify, ack", "source", src.Name, "event", event, "error", 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) + } + } + return disp +} diff --git a/internal/subscriber/handle_test.go b/internal/subscriber/handle_test.go new file mode 100644 index 0000000..68b02b2 --- /dev/null +++ b/internal/subscriber/handle_test.go @@ -0,0 +1,135 @@ +package subscriber + +import ( + "context" + "errors" + "strings" + "testing" + + "aiaa-notification-service/internal/model" + "aiaa-notification-service/internal/notify" + "aiaa-notification-service/internal/subscriber/tradesignal" +) + +func TestDecideRetry(t *testing.T) { + if DecideRetry(0, 3) != DispositionRetry { + t.Fatal("first failure should retry") + } + if DecideRetry(3, 3) != DispositionDLQ { + t.Fatal("retry 4 > 3 should dlq") + } +} + +func TestRetryCount(t *testing.T) { + if RetryCount(nil) != 0 { + t.Fatal() + } + if RetryCount(map[string]any{"x-retry-count": int32(2)}) != 2 { + t.Fatal() + } +} + +func enabledSrc() *model.Source { + return &model.Source{ID: 1, Name: "trade-signal", Status: 1} +} + +func TestHandleInvalidJSONAck(t *testing.T) { + d := HandleMessage(context.Background(), HandleInput{Body: []byte(`{`), SourceName: "trade-signal", MaxRetry: 3}, + tradesignal.NewConverter(nil), + func(context.Context, string) (*model.Source, error) { return enabledSrc(), nil }, + func(context.Context, notify.Request) (notify.Result, error) { + t.Fatal("process should not run") + return notify.Result{}, nil + }) + if d != DispositionAck { + t.Fatalf("%v", d) + } +} + +func TestHandleMissingSourceAck(t *testing.T) { + d := HandleMessage(context.Background(), HandleInput{ + Body: []byte(`{"action":"OPEN","symbol":"BTCUSDT"}`), SourceName: "trade-signal", MaxRetry: 3, + }, tradesignal.NewConverter(nil), + func(context.Context, string) (*model.Source, error) { return nil, errors.New("not found") }, + func(context.Context, notify.Request) (notify.Result, error) { + t.Fatal("process") + return notify.Result{}, nil + }) + if d != DispositionAck { + t.Fatalf("%v", d) + } +} + +func TestHandleDisabledSourceAck(t *testing.T) { + d := HandleMessage(context.Background(), HandleInput{ + Body: []byte(`{"action":"OPEN"}`), SourceName: "trade-signal", MaxRetry: 3, + }, tradesignal.NewConverter(nil), + func(context.Context, string) (*model.Source, error) { + return &model.Source{ID: 1, Name: "trade-signal", Status: 0}, nil + }, + func(context.Context, notify.Request) (notify.Result, error) { + t.Fatal("process") + return notify.Result{}, nil + }) + if d != DispositionAck { + t.Fatalf("%v", d) + } +} + +func TestHandleProcessUnprocessableAck(t *testing.T) { + d := HandleMessage(context.Background(), HandleInput{ + Body: []byte(`{"action":"OPEN"}`), SourceName: "trade-signal", MaxRetry: 3, + }, tradesignal.NewConverter(nil), + func(context.Context, string) (*model.Source, error) { return enabledSrc(), nil }, + func(context.Context, notify.Request) (notify.Result, error) { + return notify.Result{}, notify.ErrUnprocessable + }) + if d != DispositionAck { + t.Fatalf("%v", d) + } +} + +func TestHandleProcessErrorRetryThenDLQ(t *testing.T) { + process := func(context.Context, notify.Request) (notify.Result, error) { + return notify.Result{}, errors.New("db down") + } + lookup := func(context.Context, string) (*model.Source, error) { return enabledSrc(), nil } + conv := tradesignal.NewConverter(nil) + body := []byte(`{"action":"OPEN"}`) + d := HandleMessage(context.Background(), HandleInput{Body: body, SourceName: "s", MaxRetry: 3}, conv, lookup, process) + if d != DispositionRetry { + t.Fatalf("%v", d) + } + d = HandleMessage(context.Background(), HandleInput{ + Body: body, Headers: map[string]any{"x-retry-count": 3}, SourceName: "s", MaxRetry: 3, + }, conv, lookup, process) + if d != DispositionDLQ { + t.Fatalf("%v", d) + } +} + +func TestHandleSuccessAckPassesEventAndFormatted(t *testing.T) { + var got notify.Request + d := HandleMessage(context.Background(), HandleInput{ + Body: []byte(`{"action":"CLOSE","symbol":"ETHUSDT","period":"4h","price":1}`), + SourceName: "trade-signal", MaxRetry: 3, + }, tradesignal.NewConverter(nil), + func(context.Context, string) (*model.Source, error) { return enabledSrc(), nil }, + func(_ context.Context, req notify.Request) (notify.Result, error) { + got = req + return notify.Result{Matched: true, Channels: []string{"dingtalk:1"}}, nil + }) + if d != DispositionAck { + t.Fatalf("%v", d) + } + if got.Event != "trade.close" { + t.Fatalf("event=%q", got.Event) + } + if got.Data["period"] != "4h" { + t.Fatalf("period=%v", got.Data["period"]) + } + formatted, _ := got.Data["formatted"].(string) + if !strings.Contains(formatted, "周期: 4h") { + t.Fatalf("formatted=%s", formatted) + } +} diff --git a/internal/subscriber/subscriber.go b/internal/subscriber/subscriber.go new file mode 100644 index 0000000..11cce3b --- /dev/null +++ b/internal/subscriber/subscriber.go @@ -0,0 +1,182 @@ +package subscriber + +import ( + "context" + "fmt" + "log/slog" + "time" + + "aiaa-notification-service/internal/config" + "aiaa-notification-service/internal/subscriber/tradesignal" + + amqp "github.com/rabbitmq/amqp091-go" +) + +type Subscriber struct { + cfg config.SubscriptionConfig + conv *tradesignal.Converter + lookup SourceLookup + process ProcessFunc + deduper Deduper +} + +func New(cfg config.SubscriptionConfig, lookup SourceLookup, process ProcessFunc, deduper Deduper) (*Subscriber, error) { + if cfg.Formatter != "trade_signal" { + return nil, fmt.Errorf("unknown formatter %q", cfg.Formatter) + } + return &Subscriber{ + cfg: cfg, + conv: tradesignal.NewConverter(cfg.StrategyOverrides), + lookup: lookup, + process: process, + deduper: deduper, + }, nil +} + +func (s *Subscriber) Run(ctx context.Context) error { + for { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + + if err := s.consumeOnce(ctx); err != nil { + slog.Error("subscriber error, reconnecting", "name", s.cfg.Name, "error", err) + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(5 * time.Second): + } + } + } +} + +func (s *Subscriber) consumeOnce(ctx context.Context) error { + conn, err := amqp.Dial(s.cfg.URL) + if err != nil { + return fmt.Errorf("dial rabbitmq: %w", err) + } + defer conn.Close() + + ch, err := conn.Channel() + if err != nil { + return fmt.Errorf("open channel: %w", err) + } + defer ch.Close() + + if err := ch.Qos(1, 0, false); err != nil { + return fmt.Errorf("set qos: %w", err) + } + if err := s.ensureQueue(ch); err != nil { + return err + } + + deliveries, err := ch.Consume(s.cfg.Queue, s.cfg.Name, false, false, false, false, nil) + if err != nil { + return fmt.Errorf("consume queue %s: %w", s.cfg.Queue, err) + } + slog.Info("listening on queue", "name", s.cfg.Name, "queue", s.cfg.Queue) + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case d, ok := <-deliveries: + if !ok { + return fmt.Errorf("delivery channel closed") + } + s.handleDelivery(ch, d) + } + } +} + +func (s *Subscriber) ensureQueue(ch *amqp.Channel) error { + if s.cfg.Exchange != "" { + if err := ch.ExchangeDeclare(s.cfg.Exchange, s.cfg.ExchangeType, true, false, false, false, nil); err != nil { + return fmt.Errorf("declare exchange %q: %w", s.cfg.Exchange, err) + } + } + + if _, err := ch.QueueDeclare(s.cfg.Queue, true, false, false, false, nil); err != nil { + return fmt.Errorf("declare queue %q: %w", s.cfg.Queue, err) + } + + if s.cfg.Exchange != "" { + if err := ch.QueueBind(s.cfg.Queue, s.cfg.RoutingKey, s.cfg.Exchange, false, nil); err != nil { + return fmt.Errorf("bind queue %q to exchange %q: %w", s.cfg.Queue, s.cfg.Exchange, err) + } + slog.Info("queue bound", "queue", s.cfg.Queue, "exchange", s.cfg.Exchange, "type", s.cfg.ExchangeType, "routing_key", s.cfg.RoutingKey) + } + + 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) + } + } + return nil +} + +func (s *Subscriber) handleDelivery(ch *amqp.Channel, d amqp.Delivery) { + disp := HandleMessage(context.Background(), HandleInput{ + Body: d.Body, + Headers: map[string]any(d.Headers), + SourceName: s.cfg.Source, + MaxRetry: s.cfg.MaxRetry, + Deduper: s.deduper, + }, s.conv, s.lookup, s.process) + + switch disp { + case DispositionRetry: + s.republish(ch, d, s.cfg.Queue) + case DispositionDLQ: + if s.cfg.DeadLetterQueue == "" { + slog.Warn("max retry reached, discarded", "name", s.cfg.Name) + _ = d.Ack(false) + return + } + s.republish(ch, d, s.cfg.DeadLetterQueue) + default: + _ = d.Ack(false) + } +} + +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) + _ = d.Nack(false, true) + return + } + _ = d.Ack(false) + if queue == s.cfg.DeadLetterQueue { + slog.Warn("message moved to dlq", "name", s.cfg.Name, "retries", s.cfg.MaxRetry) + return + } + slog.Info("message requeued", "name", s.cfg.Name, "retry", headers[retryHeader], "max", s.cfg.MaxRetry) +} + +func copyAMQPHeaders(headers amqp.Table) amqp.Table { + out := amqp.Table{} + for k, v := range headers { + out[k] = v + } + return out +} + +func publishToQueue(ch *amqp.Channel, queue string, body []byte, headers amqp.Table) error { + return ch.Publish( + "", + queue, + false, + false, + amqp.Publishing{ + ContentType: "application/json", + DeliveryMode: amqp.Persistent, + Headers: headers, + Body: body, + Timestamp: time.Now(), + }, + ) +} diff --git a/internal/subscriber/tradesignal/convert.go b/internal/subscriber/tradesignal/convert.go new file mode 100644 index 0000000..1ccb159 --- /dev/null +++ b/internal/subscriber/tradesignal/convert.go @@ -0,0 +1,74 @@ +package tradesignal + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + + "aiaa-notification-service/internal/config" +) + +var ErrInvalidSignal = errors.New("invalid signal") + +type Converter struct { + overrides map[string]config.StrategyOverride + positions *Tracker +} + +func NewConverter(overrides map[string]config.StrategyOverride) *Converter { + return &Converter{ + overrides: overrides, + positions: NewTracker(), + } +} + +func (c *Converter) Convert(body []byte) (string, map[string]interface{}, error) { + var sig Signal + if err := json.Unmarshal(body, &sig); err != nil { + return "", nil, fmt.Errorf("%w: %v", ErrInvalidSignal, err) + } + if strings.TrimSpace(sig.Action) == "" { + return "", nil, fmt.Errorf("%w: missing action", ErrInvalidSignal) + } + out := Apply(&sig, c.overrideFor(sig.StrategyCode)) + snap := c.positions.Apply(out) + var opts FormatOptions + if snap.HasAvg { + avg := snap.AvgPrice + opts.AvgPrice = &avg + } + text := Format(out, opts) + data, err := toData(out) + if err != nil { + return "", nil, err + } + data["formatted"] = text + if snap.HasAvg { + data["avgPrice"] = snap.AvgPrice + } + return "trade." + strings.ToLower(out.Action), data, nil +} + +func (c *Converter) overrideFor(code string) *config.StrategyOverride { + if c == nil || len(c.overrides) == 0 || code == "" { + return nil + } + override, ok := c.overrides[code] + if !ok { + return nil + } + return &override +} + +func toData(sig *Signal) (map[string]interface{}, error) { + raw, err := json.Marshal(sig) + if err != nil { + return nil, err + } + data := make(map[string]interface{}) + if err := json.Unmarshal(raw, &data); err != nil { + return nil, err + } + return data, nil +} diff --git a/internal/subscriber/tradesignal/convert_test.go b/internal/subscriber/tradesignal/convert_test.go new file mode 100644 index 0000000..41ce72d --- /dev/null +++ b/internal/subscriber/tradesignal/convert_test.go @@ -0,0 +1,51 @@ +package tradesignal + +import ( + "errors" + "strings" + "testing" + + "aiaa-notification-service/internal/config" +) + +func TestConvertOpen(t *testing.T) { + lev := 100 + c := NewConverter(map[string]config.StrategyOverride{ + "BLONG": {QuantityMultipliers: config.QuantityMultipliers{Open: 100}, Leverage: &lev}, + }) + event, data, err := c.Convert([]byte(`{ + "signalId":"s1","strategyCode":"BLONG","symbol":"BTCUSDT", + "side":"LONG","action":"OPEN","quantity":0.01,"price":64000, + "leverage":10,"period":"1h","eventTime":"2026-06-23T01:30:00Z" + }`)) + if err != nil { + t.Fatal(err) + } + if event != "trade.open" { + t.Fatalf("event=%q", event) + } + formatted, _ := data["formatted"].(string) + if !strings.Contains(formatted, "周期: 1h") || !strings.Contains(formatted, "开仓数量: 1.00") { + t.Fatalf("formatted=\n%s", formatted) + } + if data["period"] != "1h" || data["strategyCode"] != "BLONG" { + t.Fatalf("data=%v", data) + } + if data["leverage"] != float64(100) && data["leverage"] != 100 { + t.Fatalf("leverage=%v", data["leverage"]) + } +} + +func TestConvertInvalidJSON(t *testing.T) { + _, _, err := NewConverter(nil).Convert([]byte(`{`)) + if !errors.Is(err, ErrInvalidSignal) { + t.Fatalf("err=%v", err) + } +} + +func TestConvertMissingAction(t *testing.T) { + _, _, err := NewConverter(nil).Convert([]byte(`{"symbol":"BTCUSDT"}`)) + if !errors.Is(err, ErrInvalidSignal) { + t.Fatalf("err=%v", err) + } +} diff --git a/internal/subscriber/tradesignal/format.go b/internal/subscriber/tradesignal/format.go new file mode 100644 index 0000000..de571a7 --- /dev/null +++ b/internal/subscriber/tradesignal/format.go @@ -0,0 +1,198 @@ +package tradesignal + +import ( + "fmt" + "strings" + "time" +) + +type FormatOptions struct { + AvgPrice *float64 +} + +func Format(signal *Signal, opts ...FormatOptions) string { + var opt FormatOptions + if len(opts) > 0 { + opt = opts[0] + } + + title := actionTitle(signal.Side, signal.Action) + symbol := trimQuote(signal.Symbol) + + lines := []string{title} + lines = append(lines, fmt.Sprintf("交易品种: %s", symbol)) + if p := strings.TrimSpace(signal.Period); p != "" { + lines = append(lines, fmt.Sprintf("周期: %s", p)) + } + + action := strings.ToUpper(signal.Action) + switch action { + case "OPEN": + lines = append(lines, fmt.Sprintf("开仓价格: %.2f", signal.Price)) + if line := sizeLine("OPEN", signal.Quantity, signal.AmountMarginRatio); line != "" { + lines = append(lines, line) + } + lines = appendAvgPrice(lines, opt.AvgPrice) + if signal.Leverage > 0 { + lines = append(lines, fmt.Sprintf("杠杆: %dx", signal.Leverage)) + } + if signal.TakeProfitPrice != nil { + lines = append(lines, fmt.Sprintf("止盈价格: %.2f", *signal.TakeProfitPrice)) + } + if signal.StopLossPrice != nil { + lines = append(lines, fmt.Sprintf("止损价格: %.2f", *signal.StopLossPrice)) + } + case "CLOSE": + lines = append(lines, fmt.Sprintf("平仓价格: %.2f", signal.Price)) + lines = append(lines, closeSizeLine(signal.Quantity, signal.PosMarginRatio)) + lines = appendAvgPrice(lines, opt.AvgPrice) + if signal.PnL != nil { + lines = append(lines, fmt.Sprintf("平仓盈亏: %.2f", *signal.PnL)) + } + if signal.AccountBalance != nil { + lines = append(lines, fmt.Sprintf("账户余额：%.2f", *signal.AccountBalance)) + } + case "ADD": + lines = append(lines, fmt.Sprintf("加仓价格: %.2f", signal.Price)) + if line := sizeLine("ADD", signal.Quantity, signal.AmountMarginRatio); line != "" { + lines = append(lines, line) + } + lines = appendAvgPrice(lines, opt.AvgPrice) + if signal.Leverage > 0 { + lines = append(lines, fmt.Sprintf("杠杆: %dx", signal.Leverage)) + } + case "REDUCE": + lines = append(lines, fmt.Sprintf("减仓价格: %.2f", signal.Price)) + if line := sizeLine("REDUCE", signal.Quantity, signal.PosMarginRatio); line != "" { + lines = append(lines, line) + } + lines = appendAvgPrice(lines, opt.AvgPrice) + if signal.PnL != nil { + lines = append(lines, fmt.Sprintf("减仓盈亏: %.2f", *signal.PnL)) + } + if signal.AccountBalance != nil { + lines = append(lines, fmt.Sprintf("账户余额：%.2f", *signal.AccountBalance)) + } + default: + lines = append(lines, fmt.Sprintf("价格: %.2f", signal.Price)) + if line := sizeLine("", signal.Quantity, signal.AmountMarginRatio); line != "" { + lines = append(lines, line) + } + lines = appendAvgPrice(lines, opt.AvgPrice) + } + + if signal.StrategyCode != "" { + lines = append(lines, fmt.Sprintf("策略: %s", signal.StrategyCode)) + } + + eventTime := signal.ParsedEventTime().In(time.Local) + lines = append(lines, fmt.Sprintf("Time: %s", eventTime.Format("2006.01.02 15:04:05"))) + + return strings.Join(lines, "\n") +} + +func actionTitle(side, action string) string { + side = strings.ToUpper(side) + action = strings.ToUpper(action) + + var pos string + switch side { + case "LONG": + pos = "多单" + case "SHORT": + pos = "空单" + default: + pos = side + } + + var act string + switch action { + case "OPEN": + act = "开仓" + case "ADD": + act = "加仓" + case "CLOSE": + act = "平仓" + case "REDUCE": + act = "减仓" + default: + act = action + } + + return pos + act +} + +func appendAvgPrice(lines []string, avgPrice *float64) []string { + if avgPrice == nil || *avgPrice <= 0 { + return lines + } + return append(lines, fmt.Sprintf("平均单价: %.2f", *avgPrice)) +} + +func closeSizeLine(quantity, posMarginRatio *float64) string { + if quantity != nil && *quantity > 0 { + return fmt.Sprintf("平仓数量: %.2f", *quantity) + } + ratio := 1.0 + if posMarginRatio != nil { + ratio = *posMarginRatio + } + return fmt.Sprintf("平仓比例: %s", formatPercent(ratio)) +} + +func sizeLine(action string, quantity, marginRatio *float64) string { + if quantity != nil && *quantity > 0 { + return fmt.Sprintf("%s: %.2f", quantityLabel(action), *quantity) + } + if marginRatio != nil { + return fmt.Sprintf("%s: %s", ratioLabel(action), formatPercent(*marginRatio)) + } + if quantity != nil { + return fmt.Sprintf("%s: %.2f", quantityLabel(action), *quantity) + } + return "" +} + +func quantityLabel(action string) string { + switch strings.ToUpper(action) { + case "OPEN": + return "开仓数量" + case "ADD": + return "加仓数量" + case "CLOSE": + return "平仓数量" + case "REDUCE": + return "减仓数量" + default: + return "数量" + } +} + +func ratioLabel(action string) string { + switch strings.ToUpper(action) { + case "OPEN": + return "开仓比例" + case "ADD": + return "加仓比例" + case "CLOSE": + return "平仓比例" + case "REDUCE": + return "减仓比例" + default: + return "仓位比例" + } +} + +func formatPercent(ratio float64) string { + return fmt.Sprintf("%.2f%%", ratio*100) +} + +func trimQuote(symbol string) string { + symbol = strings.ToUpper(symbol) + for _, suffix := range []string{"USDT", "USDC", "BUSD", "USD"} { + if strings.HasSuffix(symbol, suffix) && len(symbol) > len(suffix) { + return symbol[:len(symbol)-len(suffix)] + } + } + return symbol +} diff --git a/internal/subscriber/tradesignal/format_test.go b/internal/subscriber/tradesignal/format_test.go new file mode 100644 index 0000000..3714d3a --- /dev/null +++ b/internal/subscriber/tradesignal/format_test.go @@ -0,0 +1,60 @@ +package tradesignal + +import ( + "strings" + "testing" +) + +func ptr(v float64) *float64 { return &v } + +func TestFormatOpenIncludesPeriodAfterSymbol(t *testing.T) { + out := Format(&Signal{ + Symbol: "BTCUSDT", Side: "LONG", Action: "OPEN", + Quantity: ptr(0.01), Price: 64000.5, Leverage: 10, + Period: "1h", EventTime: "2026-06-23T01:30:00Z", + }) + if !strings.Contains(out, "多单开仓") || !strings.Contains(out, "交易品种: BTC") { + t.Fatalf("%s", out) + } + idxSym := strings.Index(out, "交易品种: BTC") + idxPer := strings.Index(out, "周期: 1h") + idxPx := strings.Index(out, "开仓价格:") + if idxPer < 0 || idxPer < idxSym || idxPx < idxPer { + t.Fatalf("period placement:\n%s", out) + } +} + +func TestFormatOmitsEmptyPeriod(t *testing.T) { + out := Format(&Signal{ + Symbol: "BTCUSDT", Side: "LONG", Action: "OPEN", + Price: 1, EventTime: "2026-06-23T01:30:00Z", + }) + if strings.Contains(out, "周期:") { + t.Fatalf("%s", out) + } +} + +func TestFormatCloseLong(t *testing.T) { + pnl, bal := 941.0, 74744.90 + out := Format(&Signal{ + Symbol: "BTCUSDT", Side: "LONG", Action: "CLOSE", + Quantity: ptr(3), Price: 63175.76, + EventTime: "2026-07-07T05:52:14Z", PnL: &pnl, AccountBalance: &bal, + }) + for _, want := range []string{"多单平仓", "平仓价格: 63175.76", "平仓盈亏: 941.00"} { + if !strings.Contains(out, want) { + t.Fatalf("missing %q in\n%s", want, out) + } + } +} + +func TestFormatWithAvgPrice(t *testing.T) { + avg := 150.0 + out := Format(&Signal{ + Symbol: "BTCUSDT", Side: "LONG", Action: "ADD", + Quantity: ptr(1), Price: 200, EventTime: "2026-07-07T05:52:14Z", + }, FormatOptions{AvgPrice: &avg}) + if !strings.Contains(out, "平均单价: 150.00") { + t.Fatalf("%s", out) + } +} diff --git a/internal/subscriber/tradesignal/override.go b/internal/subscriber/tradesignal/override.go new file mode 100644 index 0000000..2bcb915 --- /dev/null +++ b/internal/subscriber/tradesignal/override.go @@ -0,0 +1,29 @@ +package tradesignal + +import "aiaa-notification-service/internal/config" + +func Apply(signal *Signal, override *config.StrategyOverride) *Signal { + if override == nil { + return signal + } + + out := *signal + if signal.Quantity != nil { + q := *signal.Quantity + out.Quantity = &q + } + + if out.Quantity != nil && *out.Quantity > 0 { + multiplier := override.QuantityMultiplierFor(out.Action) + if multiplier != 1 { + q := *out.Quantity * multiplier + out.Quantity = &q + } + } + + if override.Leverage != nil && *override.Leverage > 0 { + out.Leverage = *override.Leverage + } + + return &out +} diff --git a/internal/subscriber/tradesignal/position.go b/internal/subscriber/tradesignal/position.go new file mode 100644 index 0000000..5f5563f --- /dev/null +++ b/internal/subscriber/tradesignal/position.go @@ -0,0 +1,204 @@ +package tradesignal + +import ( + "strings" + "sync" +) + +type mode int + +const ( + modeNone mode = iota + modeQty + modeWeight +) + +type Snapshot struct { + AvgPrice float64 + Size float64 + HasAvg bool +} + +type state struct { + avg float64 + size float64 + mode mode +} + +type Tracker struct { + mu sync.Mutex + positions map[string]*state + applied map[string]Snapshot +} + +func NewTracker() *Tracker { + return &Tracker{ + positions: make(map[string]*state), + applied: make(map[string]Snapshot), + } +} + +func (t *Tracker) Apply(signal *Signal) Snapshot { + if signal == nil { + return Snapshot{} + } + + t.mu.Lock() + defer t.mu.Unlock() + + if signal.SignalID != "" { + if snap, ok := t.applied[signal.SignalID]; ok { + return snap + } + } + + key := positionKey(signal.StrategyCode, signal.Symbol, signal.Side) + action := strings.ToUpper(signal.Action) + st := t.positions[key] + + var snap Snapshot + switch action { + case "OPEN": + st = openPosition(signal) + snap = snapshotFrom(st) + if st != nil { + t.positions[key] = st + } else { + delete(t.positions, key) + } + case "ADD": + st = addPosition(st, signal) + snap = snapshotFrom(st) + if st != nil { + t.positions[key] = st + } + case "REDUCE": + snap = snapshotFrom(st) + st = reducePosition(st, signal) + if st == nil || st.size <= 0 { + delete(t.positions, key) + } else { + t.positions[key] = st + } + case "CLOSE": + snap = snapshotFrom(st) + delete(t.positions, key) + default: + snap = snapshotFrom(st) + } + + if signal.SignalID != "" { + t.applied[signal.SignalID] = snap + } + return snap +} + +func openPosition(signal *Signal) *state { + if qty, ok := positiveQty(signal.Quantity); ok { + return &state{avg: signal.Price, size: qty, mode: modeQty} + } + if w, ok := positiveRatio(signal.AmountMarginRatio); ok { + return &state{avg: signal.Price, size: w, mode: modeWeight} + } + if signal.Price > 0 { + return &state{avg: signal.Price, size: 0, mode: modeNone} + } + return nil +} + +func addPosition(st *state, signal *Signal) *state { + if st == nil || st.size <= 0 { + return openPosition(signal) + } + + if qty, ok := positiveQty(signal.Quantity); ok { + if st.mode == modeWeight { + return st + } + if st.mode == modeNone || st.size == 0 { + st.mode = modeQty + st.size = qty + st.avg = signal.Price + return st + } + st.avg = (st.size*st.avg + qty*signal.Price) / (st.size + qty) + st.size += qty + st.mode = modeQty + return st + } + + if w, ok := positiveRatio(signal.AmountMarginRatio); ok { + if st.mode == modeQty { + return st + } + if st.mode == modeNone || st.size == 0 { + st.mode = modeWeight + st.size = w + st.avg = signal.Price + return st + } + st.avg = (st.size*st.avg + w*signal.Price) / (st.size + w) + st.size += w + st.mode = modeWeight + return st + } + + return st +} + +func reducePosition(st *state, signal *Signal) *state { + if st == nil { + return nil + } + + if qty, ok := positiveQty(signal.Quantity); ok && st.mode == modeQty { + st.size -= qty + if st.size < 0 { + st.size = 0 + } + return st + } + + ratio := 0.0 + if r, ok := positiveRatio(signal.PosMarginRatio); ok { + ratio = r + } else if signal.Quantity == nil && signal.PosMarginRatio == nil { + return st + } + if ratio > 1 { + ratio = 1 + } + if ratio > 0 { + st.size *= (1 - ratio) + } + return st +} + +func snapshotFrom(st *state) Snapshot { + if st == nil || st.avg <= 0 { + return Snapshot{} + } + return Snapshot{ + AvgPrice: st.avg, + Size: st.size, + HasAvg: true, + } +} + +func positiveQty(q *float64) (float64, bool) { + if q == nil || *q <= 0 { + return 0, false + } + return *q, true +} + +func positiveRatio(r *float64) (float64, bool) { + if r == nil || *r <= 0 { + return 0, false + } + return *r, true +} + +func positionKey(strategyCode, symbol, side string) string { + return strings.ToUpper(strategyCode) + "|" + strings.ToUpper(symbol) + "|" + strings.ToUpper(side) +} diff --git a/internal/subscriber/tradesignal/signal.go b/internal/subscriber/tradesignal/signal.go new file mode 100644 index 0000000..387ae9e --- /dev/null +++ b/internal/subscriber/tradesignal/signal.go @@ -0,0 +1,36 @@ +package tradesignal + +import "time" + +type Signal struct { + SignalID string `json:"signalId"` + SourcePosID string `json:"sourcePosId"` + StrategyCode string `json:"strategyCode"` + Symbol string `json:"symbol"` + Side string `json:"side"` + Action string `json:"action"` + Quantity *float64 `json:"quantity"` + AmountMarginRatio *float64 `json:"amountMarginRatio"` + PosMarginRatio *float64 `json:"posMarginRatio"` + Price float64 `json:"price"` + Leverage int `json:"leverage"` + Period string `json:"period"` + EventTime string `json:"eventTime"` + TakeProfitPrice *float64 `json:"takeProfitPrice"` + StopLossPrice *float64 `json:"stopLossPrice"` + TakeProfitRatio *float64 `json:"takeProfitRatio"` + StopLossRatio *float64 `json:"stopLossRatio"` + PnL *float64 `json:"pnl"` + AccountBalance *float64 `json:"accountBalance"` +} + +func (s *Signal) ParsedEventTime() time.Time { + if s.EventTime == "" { + return time.Now().UTC() + } + t, err := time.Parse(time.RFC3339, s.EventTime) + if err != nil { + return time.Now().UTC() + } + return t +} diff --git a/internal/subscriber/tradesignal/transform_test.go b/internal/subscriber/tradesignal/transform_test.go new file mode 100644 index 0000000..debac58 --- /dev/null +++ b/internal/subscriber/tradesignal/transform_test.go @@ -0,0 +1,220 @@ +package tradesignal + +import ( + "math" + "testing" + + "aiaa-notification-service/internal/config" +) + +func TestApplyQuantityMultiplierAndLeverage(t *testing.T) { + leverage := 20 + override := &config.StrategyOverride{ + QuantityMultipliers: config.QuantityMultipliers{ + Open: 2, + Add: 1.5, + Reduce: 0.5, + Close: 3, + }, + Leverage: &leverage, + } + + tests := []struct { + action string + quantity float64 + wantQty float64 + }{ + {"OPEN", 1, 2}, + {"ADD", 2, 3}, + {"REDUCE", 4, 2}, + {"CLOSE", 1, 3}, + } + + for _, tt := range tests { + signal := &Signal{ + Action: tt.action, + Quantity: ptr(tt.quantity), + Leverage: 10, + } + out := Apply(signal, override) + if out.Quantity == nil || *out.Quantity != tt.wantQty { + t.Fatalf("action=%s quantity=%v want %v", tt.action, out.Quantity, tt.wantQty) + } + if out.Leverage != 20 { + t.Fatalf("action=%s leverage=%d want 20", tt.action, out.Leverage) + } + } +} + +func TestApplyKeepsOriginalWhenNoOverride(t *testing.T) { + signal := &Signal{ + Action: "OPEN", + Quantity: ptr(1.5), + Leverage: 8, + } + out := Apply(signal, nil) + if out != signal { + t.Fatalf("expected same signal pointer when override is nil") + } +} + +func TestApplyDoesNotChangeMarginRatioOnlySignals(t *testing.T) { + marginRatio := 0.2 + signal := &Signal{ + Action: "OPEN", + AmountMarginRatio: &marginRatio, + Leverage: 5, + } + override := &config.StrategyOverride{ + QuantityMultipliers: config.QuantityMultipliers{Open: 2}, + } + out := Apply(signal, override) + if out.Quantity != nil { + t.Fatalf("expected quantity unchanged when only margin ratio is set") + } + if out.Leverage != 5 { + t.Fatalf("expected leverage unchanged, got %d", out.Leverage) + } +} + +func TestAvgPriceOpenAndAdd(t *testing.T) { + tr := NewTracker() + + open := &Signal{ + SignalID: "s1", + StrategyCode: "BLONG", + Symbol: "BTCUSDT", + Side: "LONG", + Action: "OPEN", + Quantity: ptr(2), + Price: 100, + } + snap := tr.Apply(open) + if !snap.HasAvg || snap.AvgPrice != 100 { + t.Fatalf("open avg=%v has=%v", snap.AvgPrice, snap.HasAvg) + } + + add := &Signal{ + SignalID: "s2", + StrategyCode: "BLONG", + Symbol: "BTCUSDT", + Side: "LONG", + Action: "ADD", + Quantity: ptr(2), + Price: 200, + } + snap = tr.Apply(add) + if !snap.HasAvg || math.Abs(snap.AvgPrice-150) > 1e-9 { + t.Fatalf("expected avg 150, got %v", snap.AvgPrice) + } +} + +func TestAvgPriceWithMarginRatio(t *testing.T) { + tr := NewTracker() + + open := &Signal{ + SignalID: "r1", + StrategyCode: "BLONG", + Symbol: "ETHUSDT", + Side: "LONG", + Action: "OPEN", + AmountMarginRatio: ptr(0.1), + Price: 100, + } + tr.Apply(open) + + add := &Signal{ + SignalID: "r2", + StrategyCode: "BLONG", + Symbol: "ETHUSDT", + Side: "LONG", + Action: "ADD", + AmountMarginRatio: ptr(0.1), + Price: 200, + } + snap := tr.Apply(add) + if !snap.HasAvg || math.Abs(snap.AvgPrice-150) > 1e-9 { + t.Fatalf("expected weighted avg 150, got %v", snap.AvgPrice) + } +} + +func TestCloseKeepsEntryAvgInSnapshot(t *testing.T) { + tr := NewTracker() + tr.Apply(&Signal{ + SignalID: "c1", + StrategyCode: "BLONG", + Symbol: "BTCUSDT", + Side: "LONG", + Action: "OPEN", + Quantity: ptr(1), + Price: 64000, + }) + + snap := tr.Apply(&Signal{ + SignalID: "c2", + StrategyCode: "BLONG", + Symbol: "BTCUSDT", + Side: "LONG", + Action: "CLOSE", + Price: 65000, + }) + if !snap.HasAvg || snap.AvgPrice != 64000 { + t.Fatalf("close should report entry avg 64000, got %v", snap.AvgPrice) + } + + snap = tr.Apply(&Signal{ + SignalID: "c3", + StrategyCode: "BLONG", + Symbol: "BTCUSDT", + Side: "LONG", + Action: "ADD", + Quantity: ptr(1), + Price: 70000, + }) + if !snap.HasAvg || snap.AvgPrice != 70000 { + t.Fatalf("after close, add should reopen at 70000, got %v", snap.AvgPrice) + } +} + +func TestSignalIDIdempotent(t *testing.T) { + tr := NewTracker() + sig := &Signal{ + SignalID: "dup", + StrategyCode: "BLONG", + Symbol: "BTCUSDT", + Side: "LONG", + Action: "OPEN", + Quantity: ptr(1), + Price: 100, + } + tr.Apply(sig) + tr.Apply(sig) + + snap := tr.Apply(&Signal{ + SignalID: "dup2", + StrategyCode: "BLONG", + Symbol: "BTCUSDT", + Side: "LONG", + Action: "ADD", + Quantity: ptr(1), + Price: 200, + }) + if math.Abs(snap.AvgPrice-150) > 1e-9 { + t.Fatalf("duplicate open should not double size, avg=%v", snap.AvgPrice) + } +} + +func TestDifferentSideIsolated(t *testing.T) { + tr := NewTracker() + tr.Apply(&Signal{ + SignalID: "l1", StrategyCode: "BLONG", Symbol: "BTCUSDT", Side: "LONG", + Action: "OPEN", Quantity: ptr(1), Price: 100, + }) + snap := tr.Apply(&Signal{ + SignalID: "s1", StrategyCode: "BLONG", Symbol: "BTCUSDT", Side: "SHORT", + Action: "OPEN", Quantity: ptr(1), Price: 200, + }) + if snap.AvgPrice != 200 { + t.Fatalf("short should be isolated, got %v", snap.AvgPrice) + } +}