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

@@ -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"}`
 

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
+	<-ctx.Done()
-	quit := make(chan os.Signal, 1)
-	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
-	<-quit
 	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")

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

docs/superpowers/specs/2026-08-15-rabbitmq-subscriber-design.md

@@ -0,0 +1,208 @@
+# RabbitMQ 订阅模块设计
+
+**Date:** 2026-08-15  
+**Status:** Approved
+
+## Goal
+
+在通知服务内增加可配置的 RabbitMQ 订阅：消费交易信号、格式化文案、再走现有规则/模板/渠道发送。第一路订阅 fanout exchange `trade.signal.executor.queue`，与 executor 队列独立消费、互不争抢。配置做成订阅列表，后续加 exchange 只加配置、不改代码。
+
+## Non-goals
+
+- 移植参考项目的钉钉直发、按目标过滤、`x-dingtalk-sent` / `x-dingtalk-abandoned`
+- 在 MQ 层等待渠道发送结果
+- 均价持久化（进程内即可，重启丢失）
+- 连真实 CloudAMQP 的集成测试
+- 把 AMQP 账号写进仓库
+- 新增管理 API 或数据库表
+- 订阅侧解析 Source 的 `parse_mode`（MQ 路径自带 event + data）
+
+## Decisions
+
+| 项 | 选择 |
+|----|------|
+| 进 notify 的方式 | 抽出内部 `NotifyService`，MQ 与 HTTP 共用 |
+| 订阅 ↔ Source | 每条订阅绑定一个 `source` 名 |
+| event | `trade.<action小写>`，如 `OPEN` → `trade.open` |
+| 格式化 | 移植交易信号格式化器；`data.formatted` + 原始字段 |
+| 均价 / 策略覆盖 | 都移植；均价按 `strategyCode + symbol + side` 进程内追踪 |
+| 周期 | MQ `period` 原样使用（如 `1h`），不换算 |
+| 过滤 | 交给 Rule 条件（`strategyCode` / `symbol` 等） |
+| 凭据 | `url: "${RABBITMQ_URL}"` |
+
+## Architecture
+
+进程内 HTTP 与 MQ 消费并行。两者都进入同一个内部入口，规则、模板、渠道只维护一份。
+
+```
+RabbitMQ fanout exchange
+        │
+        ├─ executor 队列（已有，不改）
+        └─ trade.signal.notify.queue（本服务独占）
+                │
+                ▼
+        Subscriber（声明/绑定/重连/消费/重试/DLQ）
+                │
+                ▼
+        trade_signal：解析 → 策略覆盖 → 均价 → 格式化
+                │
+                ▼
+        NotifyService
+          匹配规则 → 条件过滤 → 渲染模板 → 异步发渠道
+```
+
+订阅列表为空，或某条订阅的 `url` 未展开到非空值：该条不启动。全部未启动时 HTTP 通知不受影响。
+
+## Config
+
+`config/config.yaml` 增加 `subscriptions` 列表。第一路示例（URL 必须走环境变量）：
+
+```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
+```
+
+| 字段 | 说明 |
+|------|------|
+| `name` | 订阅标识，日志用；缺省用 `queue` |
+| `url` | AMQP(S) URL，支持 `${ENV}` |
+| `queue` | 本服务独占队列 |
+| `dead_letter_queue` | 超过 `max_retry` 后投入；空则丢弃并 Ack |
+| `exchange` / `exchange_type` / `routing_key` | 声明并绑定；`exchange` 为空则只声明队列 |
+| `max_retry` | `Process` 内部错误的重投次数；`<=0` 时默认 3 |
+| `source` | 已有 `notification_source.name` |
+| `formatter` | 空则默认 `trade_signal`；未知值启动时报错 |
+| `strategy_overrides` | 按 `strategyCode`；数量倍数 `<=0` 视为 1 |
+
+`exchange_type` 为空时默认 `fanout`。
+
+## Modules
+
+| 模块 | 职责 | 接口 |
+|------|------|------|
+| `internal/notify` | 匹配规则、条件、渲染、路由、记 message_log | `Process(ctx, Request) (Result, error)` |
+| `internal/subscriber` | 连 MQ、声明/绑定、重连、重试/DLQ；按订阅启动 goroutine | `Run(ctx)`，依赖 `Process` |
+| `internal/subscriber/tradesignal` | 解析信号、策略覆盖、均价、格式化 | 输入 JSON body，输出 `event` + `data` |
+
+`POST /api/v1/notify` 按 Source `parse_mode` 解析 body 后调用 `notify.Process`，不再内嵌匹配/渲染/路由。
+
+`subscriber` 不碰渠道，不读 `parse_mode`。
+
+### NotifyService
+
+```go
+type Request struct {
+    Source *model.Source
+    Event  string
+    Data   map[string]interface{}
+}
+
+type Result struct {
+    Matched  bool
+    Filtered bool
+    Channels []string
+    Reason   string
+}
+```
+
+行为与现有 Handler 一致：无规则 → `Matched=false`；条件未过 → `Matched=true, Filtered=true`；命中则渲染、`Route`、写 log。标题仍为 `{source.name}: {event}`。`Process` 只在查库/渲染等内部失败时返回 `error`。
+
+### Source 查找
+
+订阅用 `source` 名调用已有 `GetSourceByName`。`status != 1` 或未找到：Ack，记日志，不进 DLQ。
+
+## Data flow（trade_signal）
+
+1. 消费 delivery，JSON 反序列化为信号。
+2. 按 `strategyCode` 套 `strategy_overrides`（数量倍数、杠杆覆盖）。
+3. 按 `strategyCode + symbol + side` 更新进程内均价；同一 `signalId` 只应用一次。
+4. 生成 `formatted` 文本。
+5. `event = "trade." + strings.ToLower(action)`。`action` 为空视为无效消息。
+6. 用订阅的 `source` 名取启用 Source，调用 `Process`。
+
+### `data` 字段
+
+`data` 来自**覆盖后**的信号（与 `formatted` 数字一致），camelCase，并附加：
+
+| 键 | 来源 |
+|----|------|
+| `signalId` / `sourcePosId` / `strategyCode` / `symbol` / `side` / `action` | 信号 |
+| `quantity` / `amountMarginRatio` / `posMarginRatio` | 信号（quantity 可能被倍数改写） |
+| `price` / `leverage` / `period` / `eventTime` | 信号（leverage 可能被覆盖） |
+| `takeProfitPrice` / `stopLossPrice` / `takeProfitRatio` / `stopLossRatio` | 信号 |
+| `pnl` / `accountBalance` | 信号（可选） |
+| `formatted` | 格式化全文 |
+| `avgPrice` | 有均价才写入 |
+
+规则条件可写 `strategyCode`、`symbol`、`period` 等。模板可用 `{{.formatted}}`，也可自己拼字段。
+
+### 格式化文案
+
+移植参考项目 `test-mq-to-ali` 的格式化器，并在「交易品种」后增加周期（`period` 非空才输出）：
+
+```
+多单开仓
+交易品种: BTC
+周期: 1h
+开仓价格: 65000.00
+...
+策略: BLONG
+Time: 2026.08.15 16:39:00
+```
+
+其余规则与参考项目一致：品种去掉报价后缀（`BTCUSDT` → `BTC`）；开/加/减/平仓标题与数量或比例行；可选均价、杠杆、止盈止损、盈亏、余额；时间为本地时区 `2006.01.02 15:04:05`。
+
+## Error handling
+
+| 情况 | 处理 |
+|------|------|
+| JSON 无效、缺 `action` | Ack 丢掉，不重试 |
+| Source 不存在/禁用、无匹配规则、条件未过 | Ack，记日志，不进 DLQ |
+| `Process` 返回 error（查库失败等） | `x-retry-count` +1 后重新投递本队列；超过 `max_retry` 则投入 `dead_letter_queue`（未配置则丢弃并 Ack） |
+| 连接/channel 断开 | 5s 后重连，重新 `consumeOnce` |
+
+启动时声明 durable exchange（若配置了）、durable 队列、绑定、durable DLQ。消费 `autoAck=false`，QoS prefetch=1。consumer tag 用订阅 `name`。
+
+渠道发送仍由 `engine.Router` 异步重试（1s / 5s / 30s）。MQ 在 `Process` 接受路由后即 Ack，不等待渠道结果。
+
+不移植参考项目按钉钉目标的部分成功重试。
+
+## Testing
+
+- **格式化**：开/平/加/减仓文案；`period` 有则出现、空则省略；均价行；策略倍数后的数量
+- **均价追踪**：OPEN/ADD 更新，REDUCE/CLOSE 展示原均价；同 `signalId` 不重复应用
+- **策略覆盖**：数量倍数、杠杆覆盖；`<=0` 的倍数视为 1
+- **订阅 → Notify**：假 `Process`，断言 `event`、`data.formatted`、原始字段与 `period`
+- **重试/DLQ**：超过 `max_retry` 才进 DLQ；坏 JSON 直接 Ack
+- **HTTP `/notify`**：抽 `Process` 后现有解析与匹配行为不变
+
+不写连真实 broker 的测试。声明/绑定逻辑用可注入的 channel 假对象，或抽纯函数测重试计数与 DLQ 判定。
+
+## Boot
+
+`cmd/server/main.go` 在 HTTP server 启动后、等信号退出前，对每条有效订阅 `go subscriber.Run(ctx)`。收到 SIGINT/SIGTERM 时 cancel 该 ctx，再 `Shutdown` HTTP。
+
+运营侧需事先创建 Source（`name` 与订阅 `source` 一致）、Template、Rule（`event` 为 `trade.open` 等）、Channel。本模块不自动建这些记录。
+
+## Out of scope later
+
+- 第二种 `formatter`（有新 exchange 再加）
+- 均价写入 Redis
+- 管理 API 热更新订阅

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
 )

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=

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()
+}

internal/config/config.go

@@ -4,17 +4,107 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"time"
 
 	"github.com/spf13/viper"
 )
 
 type Config struct {
-	Server    ServerConfig    `mapstructure:"server"`
+	Server               ServerConfig         `mapstructure:"server"`
-	Database  DatabaseConfig  `mapstructure:"database"`
+	Database             DatabaseConfig       `mapstructure:"database"`
-	Redis     RedisConfig     `mapstructure:"redis"`
+	Redis                RedisConfig          `mapstructure:"redis"`
-	SMTP      SMTPConfig      `mapstructure:"smtp"`
+	SMTP                 SMTPConfig           `mapstructure:"smtp"`
-	RateLimit RateLimitConfig `mapstructure:"rate_limit"`
+	RateLimit            RateLimitConfig      `mapstructure:"rate_limit"`
-	Logbull   LogbullConfig   `mapstructure:"logbull"`
+	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
 }
 

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"))
+	}
+}

internal/handler/notify.go

@@ -1,47 +1,34 @@
 package handler
 
 import (
-	"context"
+	"errors"
-	"encoding/json"
 	"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
+	svc *notify.Service
-	cache    *cache.Cache // used by engine
-	matcher  *engine.Matcher
-	renderer *engine.Renderer
-	router   *engine.Router
 }
 
-func NewNotifyHandler(s *store.Store, c *cache.Cache, m *engine.Matcher, r *engine.Renderer, rt *engine.Router) *NotifyHandler {
+func NewNotifyHandler(svc *notify.Service) *NotifyHandler {
-	return &NotifyHandler{store: s, cache: c, matcher: m, renderer: r, router: rt}
+	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
+	res, err := h.svc.Process(c.Request.Context(), notify.Request{
-	rule, err := h.matcher.Match(c.Request.Context(), src.ID, msg.Event)
+		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
 	}
-
+	if res.Filtered {
-	// 4. Evaluate conditions
+		c.JSON(http.StatusOK, gin.H{"matched": true, "filtered": true, "reason": res.Reason})
-	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"})
 		return
 	}
-
+	c.JSON(http.StatusOK, gin.H{"matched": true, "channels": res.Channels, "accepted": true})
-	// 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
 }

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
+}

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)
+	}
+}

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)
+}

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())
+	}
+}

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
+}

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)
+	}
+}

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(),
+		},
+	)
+}

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
+}

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)
+	}
+}

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
+}

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)
+	}
+}

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
+}

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)
+}

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
+}

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)
+	}
+}