feat: 支持 ntfy 一等通知渠道（P6-A） (#1271)

* feat: add ntfy notification channel * test: add ntfy smoke evidence images * fix: validate ntfy endpoint in structured config * fix: align ntfy endpoint validation
2026-09-20 10:53:33 +08:00 · 2026-05-13 12:47:46 +08:00
parent 754bf48509
commit 9f56238420
30 changed files with 681 additions and 24 deletions
--- a/.env.example
+++ b/.env.example
@@ -410,13 +410,19 @@ AGENT_SKILLS=
 # CUSTOM_WEBHOOK_URLS=https://oapi.dingtalk.com/robot/send?access_token=xxx,https://hooks.slack.com/services/xxx
 # CUSTOM_WEBHOOK_BEARER_TOKEN=  # 可选，用于需要认证的 Webhook (Header Authorization: Bearer <token>)
 # CUSTOM_WEBHOOK_BODY_TEMPLATE=  # 可选，全局 JSON body 模板，会覆盖 Bark/Slack/Discord 等自动 payload；推荐 $content_json/$title_json
-# WEBHOOK_VERIFY_SSL=true   # 默认校验。设为 false 可支持自签名证书。警告：禁用后存在 MITM 劫持风险，仅限可信内网
+# WEBHOOK_VERIFY_SSL=true   # 默认校验；影响读取该配置的 webhook-style HTTPS 通知请求。设为 false 可支持自签名证书。警告：禁用后存在 MITM 劫持风险，仅限可信内网
 #
 # 【方式六】Pushover 配置
 # 注册Pushover账号，并创建应用Token https://pushover.net/apps/build
 # PUSHOVER_USER_KEY=
 # PUSHOVER_API_TOKEN=
 #
+# 【方式六扩展】ntfy 配置
+# NTFY_URL 必须包含 topic path，例如 https://ntfy.sh/my-topic 或 https://self-hosted:port/my-topic
+# 系统会解析 topic，并使用 ntfy JSON publish API 发送 Markdown 文本。
+# NTFY_URL=
+# NTFY_TOKEN=  # 可选，用于需要 Bearer Token 的 topic 或自建 ntfy server
+#
 # 【方式七】PushPlus 配置（国内推送服务，推荐）
 # 注册PushPlus账号并获取Token https://www.pushplus.plus
 # PUSHPLUS_TOKEN=
@@ -480,7 +486,7 @@ AGENT_SKILLS=
 #
 # 【通知路由策略】(Issue #1200 P3)
 # 默认留空：该类型通知发送到所有已配置渠道。填写后仅发送到列出的已配置渠道。
-# 允许值：wechat,feishu,telegram,email,pushover,pushplus,serverchan3,custom,discord,slack,astrbot
+# 允许值：wechat,feishu,telegram,email,pushover,ntfy,pushplus,serverchan3,custom,discord,slack,astrbot
 # NOTIFICATION_REPORT_CHANNELS=
 # NOTIFICATION_ALERT_CHANNELS=
 # NOTIFICATION_SYSTEM_ERROR_CHANNELS=
--- a/.github/workflows/daily_analysis.yml
+++ b/.github/workflows/daily_analysis.yml
@@ -287,6 +287,10 @@ jobs:
          # 方式五：Pushover
          PUSHOVER_USER_KEY: ${{ secrets.PUSHOVER_USER_KEY }}
          PUSHOVER_API_TOKEN: ${{ secrets.PUSHOVER_API_TOKEN }}
+
+          # 方式五扩展：ntfy
+          NTFY_URL: ${{ secrets.NTFY_URL }}
+          NTFY_TOKEN: ${{ secrets.NTFY_TOKEN }}
          
          # 方式六：PushPlus ⬅️ 新增！
          PUSHPLUS_TOKEN: ${{ secrets.PUSHPLUS_TOKEN }}
@@ -422,6 +426,7 @@ jobs:
          echo ""
          echo "【通知渠道】"
          echo "  PushPlus: $([ -n "$PUSHPLUS_TOKEN" ] && echo '✅ 已配置' || echo '⚪ 未配置')"
+          echo "  ntfy: $([ -n "$NTFY_URL" ] && echo '✅ 已配置' || echo '⚪ 未配置')"
          echo "  企业微信: $([ -n "$WECHAT_WEBHOOK_URL" ] && echo '✅ 已配置' || echo '⚪ 未配置')"
          echo "  飞书: $([ -n "$FEISHU_WEBHOOK_URL" ] && echo '✅ 已配置' || echo '⚪ 未配置')"
          echo "  Telegram: $([ -n "$TELEGRAM_BOT_TOKEN" ] && echo '✅ 已配置' || echo '⚪ 未配置')"
--- a/api/v1/schemas/system_config.py
+++ b/api/v1/schemas/system_config.py
@@ -14,6 +14,7 @@ NotificationTestChannel = Literal[
    "telegram",
    "email",
    "pushover",
+    "ntfy",
    "pushplus",
    "serverchan3",
    "custom",
--- a/apps/dsa-web/src/components/settings/NotificationTestPanel.tsx
+++ b/apps/dsa-web/src/components/settings/NotificationTestPanel.tsx
@@ -17,6 +17,7 @@ const CHANNEL_OPTIONS: Array<{ value: NotificationTestChannel; label: string }>
  { value: 'telegram', label: 'Telegram' },
  { value: 'email', label: '邮件' },
  { value: 'pushover', label: 'Pushover' },
+  { value: 'ntfy', label: 'ntfy' },
  { value: 'pushplus', label: 'PushPlus' },
  { value: 'serverchan3', label: 'Server酱3' },
  { value: 'custom', label: '自定义 Webhook' },
--- a/apps/dsa-web/src/components/settings/tests/NotificationTestPanel.test.tsx
+++ b/apps/dsa-web/src/components/settings/tests/NotificationTestPanel.test.tsx
@@ -44,6 +44,7 @@ describe('NotificationTestPanel', () => {
      />,
    );

+    expect(screen.getByRole('option', { name: 'ntfy' })).toBeInTheDocument();
    fireEvent.change(screen.getByLabelText('渠道'), { target: { value: 'custom' } });
    fireEvent.click(screen.getByRole('button', { name: /发送测试/ }));

--- a/apps/dsa-web/src/types/systemConfig.ts
+++ b/apps/dsa-web/src/types/systemConfig.ts
@@ -197,6 +197,7 @@ export type NotificationTestChannel =
  | 'telegram'
  | 'email'
  | 'pushover'
+  | 'ntfy'
  | 'pushplus'
  | 'serverchan3'
  | 'custom'
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).

 <!-- 新条目格式：- [类型] 描述（类型取值：新功能/改进/修复/文档/测试/chore）-->
 <!-- 每条独立一行追加到本段末尾，无需分类标题，合并时冲突最小 -->
+- [新功能] 通知网关新增 ntfy 一等渠道，支持通过 `NTFY_URL` / `NTFY_TOKEN` 推送并接入 Web 测试、路由、Actions 与诊断。
 - [修复] 聚合报告通知按静态渠道隔离发送失败，并补充自定义 Webhook 部分成功诊断与脱敏测试。
 - [修复] 未配置 Tushare / Longbridge 凭据时不再实例化对应可选 fetcher，避免缺失凭据的数据源进入候选集。
 - [修复] Longbridge 遇到连接关闭类异常后会进入冷却期，并在美股/港股实时与日线请求中临时跳过该数据源，避免请求级频繁重连。
--- a/docs/assets/issue-1200-ntfy-curl-smoke.png
+++ b/docs/assets/issue-1200-ntfy-curl-smoke.png
--- a/docs/assets/issue-1200-ntfy-ios-smoke.png
+++ b/docs/assets/issue-1200-ntfy-ios-smoke.png
--- a/docs/full-guide.md
+++ b/docs/full-guide.md
@@ -95,14 +95,16 @@ daily_stock_analysis/
 | `SERVERCHAN3_SENDKEY` | Server酱³ Sendkey（[获取地址](https://sc3.ft07.com/)，手机APP推送服务） | 可选 |
 | `ASTRBOT_URL` | AstrBot Webhook URL | 可选 |
 | `ASTRBOT_TOKEN` | AstrBot Bearer Token（可选） | 可选 |
+| `NTFY_URL` | ntfy 完整 topic endpoint，必须包含 topic path，例如 `https://ntfy.sh/my-topic` | 可选 |
+| `NTFY_TOKEN` | ntfy Bearer Token（可选） | 可选 |
 | `CUSTOM_WEBHOOK_URLS` | 自定义 Webhook（支持钉钉等，多个用逗号分隔） | 可选 |
 | `CUSTOM_WEBHOOK_BEARER_TOKEN` | 自定义 Webhook 的 Bearer Token（用于需要认证的 Webhook） | 可选 |
 | `CUSTOM_WEBHOOK_BODY_TEMPLATE` | 自定义 Webhook JSON body 模板，适配 AstrBot、NapCat、自建服务等特殊 payload | 可选 |
-| `WEBHOOK_VERIFY_SSL` | Webhook HTTPS 证书校验（默认 true）。设为 false 可支持自签名证书。警告：关闭有严重安全风险（MITM），仅限可信内网 | 可选 |
+| `WEBHOOK_VERIFY_SSL` | 读取该配置的 webhook-style HTTPS 通知请求证书校验（默认 true）。设为 false 可支持自签名证书。警告：关闭有严重安全风险（MITM），仅限可信内网 | 可选 |

 > *注：至少配置一个渠道，配置多个则同时推送
 >
-> 当前默认 `daily_analysis.yml` 只显式映射固定 Secret / Variable 名称，不会自动把 `STOCK_GROUP_1`、`EMAIL_GROUP_1` 这类任意编号变量导入运行环境。所以分组邮箱功能目前不适用于仓库自带默认 GitHub Actions workflow；它适用于本地 `.env`、Docker，或你自行显式扩展过 `env:` 映射的运行环境。Actions 已显式映射 `CUSTOM_WEBHOOK_BODY_TEMPLATE`、`WEBHOOK_VERIFY_SSL`、`FEISHU_WEBHOOK_SECRET`、`FEISHU_WEBHOOK_KEYWORD`、`PUSHPLUS_TOPIC`、P3 通知路由键以及 P4 通知降噪键；`MARKDOWN_TO_IMAGE_CHANNELS` 和 `MERGE_EMAIL_NOTIFICATION` 仍作为行为开关不在默认 workflow 中自动映射。
+> 当前默认 `daily_analysis.yml` 只显式映射固定 Secret / Variable 名称，不会自动把 `STOCK_GROUP_1`、`EMAIL_GROUP_1` 这类任意编号变量导入运行环境。所以分组邮箱功能目前不适用于仓库自带默认 GitHub Actions workflow；它适用于本地 `.env`、Docker，或你自行显式扩展过 `env:` 映射的运行环境。Actions 已显式映射 `CUSTOM_WEBHOOK_BODY_TEMPLATE`、`WEBHOOK_VERIFY_SSL`、`FEISHU_WEBHOOK_SECRET`、`FEISHU_WEBHOOK_KEYWORD`、`PUSHPLUS_TOPIC`、`NTFY_URL`、`NTFY_TOKEN`、P3 通知路由键以及 P4 通知降噪键；`MARKDOWN_TO_IMAGE_CHANNELS` 和 `MERGE_EMAIL_NOTIFICATION` 仍作为行为开关不在默认 workflow 中自动映射。

 #### 推送行为配置

@@ -257,14 +259,16 @@ daily_stock_analysis/
 | `STOCK_GROUP_N` / `EMAIL_GROUP_N` | 邮件分组路由（Issue #268）：`STOCK_GROUP_N` 应为 `STOCK_LIST` 子集，仅影响邮件收件人，不改变分析范围或其他通知渠道 | 可选 |
 | `CUSTOM_WEBHOOK_URLS` | 自定义 Webhook（逗号分隔） | 可选 |
 | `CUSTOM_WEBHOOK_BEARER_TOKEN` | 自定义 Webhook Bearer Token | 可选 |
-| `WEBHOOK_VERIFY_SSL` | Webhook HTTPS 证书校验（默认 true）。设为 false 可支持自签名。警告：关闭有严重安全风险 | 可选 |
+| `WEBHOOK_VERIFY_SSL` | 读取该配置的 webhook-style HTTPS 通知请求证书校验（默认 true）。设为 false 可支持自签名。警告：关闭有严重安全风险 | 可选 |
 | `PUSHOVER_USER_KEY` | Pushover 用户 Key | 可选 |
 | `PUSHOVER_API_TOKEN` | Pushover API Token | 可选 |
+| `NTFY_URL` | ntfy 完整 topic endpoint，必须包含 topic path，例如 `https://ntfy.sh/my-topic` | 可选 |
+| `NTFY_TOKEN` | ntfy Bearer Token（可选） | 可选 |
 | `PUSHPLUS_TOKEN` | PushPlus Token（国内推送服务） | 可选 |
 | `SERVERCHAN3_SENDKEY` | Server酱³ Sendkey | 可选 |
 | `ASTRBOT_URL` | AstrBot Webhook URL | 可选 |
 | `ASTRBOT_TOKEN` | AstrBot Bearer Token（可选） | 可选 |
-| `NOTIFICATION_REPORT_CHANNELS` | report 路由渠道，逗号分隔；允许值：wechat,feishu,telegram,email,pushover,pushplus,serverchan3,custom,discord,slack,astrbot | 可选 |
+| `NOTIFICATION_REPORT_CHANNELS` | report 路由渠道，逗号分隔；允许值：wechat,feishu,telegram,email,pushover,ntfy,pushplus,serverchan3,custom,discord,slack,astrbot | 可选 |
 | `NOTIFICATION_ALERT_CHANNELS` | alert 路由渠道，逗号分隔；留空保持全渠道 | 可选 |
 | `NOTIFICATION_SYSTEM_ERROR_CHANNELS` | system_error 预留路由渠道，逗号分隔；留空保持全渠道 | 可选 |
 | `NOTIFICATION_DEDUP_TTL_SECONDS` | 通知去重 TTL 秒数，`0` 关闭 | 可选 |
--- a/docs/full-guide_EN.md
+++ b/docs/full-guide_EN.md
@@ -96,14 +96,16 @@ Go to your forked repo → `Settings` → `Secrets and variables` → `Actions`
 | `SERVERCHAN3_SENDKEY` | ServerChan v3 Sendkey ([Get here](https://sc3.ft07.com/), mobile app push service) | Optional |
 | `ASTRBOT_URL` | AstrBot Webhook URL | Optional |
 | `ASTRBOT_TOKEN` | Optional AstrBot Bearer Token | Optional |
+| `NTFY_URL` | Full ntfy topic endpoint, must include topic path, e.g. `https://ntfy.sh/my-topic` | Optional |
+| `NTFY_TOKEN` | Optional ntfy Bearer Token | Optional |
 | `CUSTOM_WEBHOOK_URLS` | Custom Webhook (supports DingTalk, etc., comma-separated) | Optional |
 | `CUSTOM_WEBHOOK_BEARER_TOKEN` | Bearer Token for custom webhooks (for authenticated webhooks) | Optional |
 | `CUSTOM_WEBHOOK_BODY_TEMPLATE` | Custom Webhook JSON body template for AstrBot, NapCat, or self-hosted services with special payloads | Optional |
-| `WEBHOOK_VERIFY_SSL` | Verify Webhook HTTPS certificates (default true). Set to false for self-signed certs. WARNING: Disabling has serious security risk (MITM), use only on trusted internal networks | Optional |
+| `WEBHOOK_VERIFY_SSL` | HTTPS certificate verification for webhook-style notification requests that read this setting (default true). Set to false for self-signed certs. WARNING: Disabling has serious security risk (MITM), use only on trusted internal networks | Optional |

 > *Note: Configure at least one channel; multiple channels will all receive notifications
 >
-> The default `daily_analysis.yml` in this repository only exports fixed Secret / Variable names. Arbitrary numbered env vars such as `STOCK_GROUP_1` and `EMAIL_GROUP_1` are not auto-injected into the job, so grouped email routing is not available in the stock workflow unless you explicitly extend the workflow's `env:` mapping in your own fork. Actions now maps `CUSTOM_WEBHOOK_BODY_TEMPLATE`, `WEBHOOK_VERIFY_SSL`, `FEISHU_WEBHOOK_SECRET`, `FEISHU_WEBHOOK_KEYWORD`, `PUSHPLUS_TOPIC`, the P3 notification route keys, and the P4 notification noise-control keys; `MARKDOWN_TO_IMAGE_CHANNELS` and `MERGE_EMAIL_NOTIFICATION` remain behavior toggles outside the default workflow mapping.
+> The default `daily_analysis.yml` in this repository only exports fixed Secret / Variable names. Arbitrary numbered env vars such as `STOCK_GROUP_1` and `EMAIL_GROUP_1` are not auto-injected into the job, so grouped email routing is not available in the stock workflow unless you explicitly extend the workflow's `env:` mapping in your own fork. Actions now maps `CUSTOM_WEBHOOK_BODY_TEMPLATE`, `WEBHOOK_VERIFY_SSL`, `FEISHU_WEBHOOK_SECRET`, `FEISHU_WEBHOOK_KEYWORD`, `PUSHPLUS_TOPIC`, `NTFY_URL`, `NTFY_TOKEN`, the P3 notification route keys, and the P4 notification noise-control keys; `MARKDOWN_TO_IMAGE_CHANNELS` and `MERGE_EMAIL_NOTIFICATION` remain behavior toggles outside the default workflow mapping.

 #### Push Behavior Configuration

@@ -229,14 +231,16 @@ For the P0 notification baseline and diagnostics, see [Notification Baseline](no
 | `STOCK_GROUP_N` / `EMAIL_GROUP_N` | Email routing groups (Issue #268): `STOCK_GROUP_N` should stay within `STOCK_LIST` and only changes email recipients | Optional |
 | `CUSTOM_WEBHOOK_URLS` | Custom Webhook (comma-separated) | Optional |
 | `CUSTOM_WEBHOOK_BEARER_TOKEN` | Custom Webhook Bearer Token | Optional |
-| `WEBHOOK_VERIFY_SSL` | Webhook HTTPS certificate verification (default true). Set to false for self-signed certs. WARNING: Disabling has serious security risk | Optional |
+| `WEBHOOK_VERIFY_SSL` | HTTPS certificate verification for webhook-style notification requests that read this setting (default true). Set to false for self-signed certs. WARNING: Disabling has serious security risk | Optional |
 | `PUSHOVER_USER_KEY` | Pushover User Key | Optional |
 | `PUSHOVER_API_TOKEN` | Pushover API Token | Optional |
+| `NTFY_URL` | Full ntfy topic endpoint, must include topic path, e.g. `https://ntfy.sh/my-topic` | Optional |
+| `NTFY_TOKEN` | Optional ntfy Bearer Token | Optional |
 | `PUSHPLUS_TOKEN` | PushPlus Token (Chinese push service) | Optional |
 | `SERVERCHAN3_SENDKEY` | ServerChan v3 Sendkey | Optional |
 | `ASTRBOT_URL` | AstrBot Webhook URL | Optional |
 | `ASTRBOT_TOKEN` | Optional AstrBot Bearer Token | Optional |
-| `NOTIFICATION_REPORT_CHANNELS` | Report route channels, comma-separated. Allowed values: wechat,feishu,telegram,email,pushover,pushplus,serverchan3,custom,discord,slack,astrbot | Optional |
+| `NOTIFICATION_REPORT_CHANNELS` | Report route channels, comma-separated. Allowed values: wechat,feishu,telegram,email,pushover,ntfy,pushplus,serverchan3,custom,discord,slack,astrbot | Optional |
 | `NOTIFICATION_ALERT_CHANNELS` | Alert route channels, comma-separated. Empty keeps all configured channels | Optional |
 | `NOTIFICATION_SYSTEM_ERROR_CHANNELS` | Reserved system_error route channels, comma-separated. Empty keeps all configured channels | Optional |
 | `NOTIFICATION_DEDUP_TTL_SECONDS` | Dedup TTL in seconds. `0` disables dedup | Optional |
--- a/docs/notifications.md
+++ b/docs/notifications.md
@@ -1,6 +1,6 @@
 # 通知能力基线

-本文档记录通知能力 P0-P5 基线：渠道、配置 key、GitHub Actions 映射、Web 设置元数据、CLI 诊断口径、Web 一键测试、自定义 Webhook Body 模板语义、通知路由策略、降噪机制和聚合报告失败隔离。P0 只做基线与只读诊断；P1 增加 Web 单渠道真实测试；P2 产品化现有 Body 模板；P3 增加 report / alert / system_error 路由；P4 增加进程内降噪；P5 强化测试诊断和聚合报告逐渠道失败隔离，不包含 per-URL 模板、跨进程持久化、真实每日摘要、重试循环或新增一等渠道。
+本文档记录通知能力 P0-P6-A 基线：渠道、配置 key、GitHub Actions 映射、Web 设置元数据、CLI 诊断口径、Web 一键测试、自定义 Webhook Body 模板语义、通知路由策略、降噪机制、聚合报告失败隔离和 ntfy 一等渠道。P0 只做基线与只读诊断；P1 增加 Web 单渠道真实测试；P2 产品化现有 Body 模板；P3 增加 report / alert / system_error 路由；P4 增加进程内降噪；P5 强化测试诊断和聚合报告逐渠道失败隔离；P6-A 新增 ntfy，不包含 Gotify、WebPush、Apprise、Bark、per-URL 模板、跨进程持久化、真实每日摘要或重试循环。

 ## 渠道基线

@@ -11,6 +11,7 @@
 | Telegram | 静态配置 | `TELEGRAM_BOT_TOKEN`, `TELEGRAM_CHAT_ID` | `TELEGRAM_MESSAGE_THREAD_ID` | token 与 chat id 必须同时存在 |
 | 邮件 | 静态配置 | `EMAIL_SENDER`, `EMAIL_PASSWORD` | `EMAIL_RECEIVERS`, `EMAIL_SENDER_NAME` | `EMAIL_RECEIVERS` 留空时发给自己 |
 | Pushover | 静态配置 | `PUSHOVER_USER_KEY`, `PUSHOVER_API_TOKEN` | - | 两个 key 必须同时存在 |
+| ntfy | 静态配置 | `NTFY_URL` | `NTFY_TOKEN`, `WEBHOOK_VERIFY_SSL` | `NTFY_URL` 必须包含 topic path，例如 `https://ntfy.sh/my-topic` |
 | PushPlus | 静态配置 | `PUSHPLUS_TOKEN` | `PUSHPLUS_TOPIC` | `PUSHPLUS_TOPIC` 仅在 token 存在时生效 |
 | Server酱3 | 静态配置 | `SERVERCHAN3_SENDKEY` | - | 手机 App 推送 |
 | 自定义 Webhook | 静态配置 | `CUSTOM_WEBHOOK_URLS` | `CUSTOM_WEBHOOK_BEARER_TOKEN`, `CUSTOM_WEBHOOK_BODY_TEMPLATE`, `WEBHOOK_VERIFY_SSL` | 支持多个 URL，逗号分隔 |
@@ -27,7 +28,8 @@
 - Advanced key：只影响认证、安全、格式、线程、群组、证书校验或展示行为，不能单独启用渠道。
 - P3 的 `NOTIFICATION_*_CHANNELS` 属于 Advanced key：只收窄已启用渠道，不会单独启用渠道。
 - P4 的 `NOTIFICATION_DEDUP_TTL_SECONDS`、`NOTIFICATION_COOLDOWN_SECONDS`、`NOTIFICATION_QUIET_HOURS`、`NOTIFICATION_TIMEZONE`、`NOTIFICATION_MIN_SEVERITY`、`NOTIFICATION_DAILY_DIGEST_ENABLED` 属于 Advanced key：只影响已启用静态渠道的发送策略，不会单独启用渠道。
- 长尾渠道、更细粒度路由、跨进程降噪和真实每日摘要不在 P4 范围内；相关配置如未来引入，应先更新本文档、`.env.example`、Web 元数据与回归测试。
+- `WEBHOOK_VERIFY_SSL` 是读取该配置的 webhook-style HTTPS 通知请求共用的证书校验开关。
+- Gotify、WebPush、Apprise、Bark、更细粒度路由、跨进程降噪和真实每日摘要不在 P6-A 范围内；相关配置如未来引入，应先更新本文档、`.env.example`、Web 元数据与回归测试。

 ## GitHub Actions 映射

@@ -54,6 +56,11 @@ P4 补齐以下通知降噪映射：
 - `NOTIFICATION_MIN_SEVERITY`
 - `NOTIFICATION_DAILY_DIGEST_ENABLED`

+P6-A 补齐以下 ntfy 渠道映射：
+
+- `NTFY_URL`
+- `NTFY_TOKEN`
+
 默认 workflow 仍不映射 `MARKDOWN_TO_IMAGE_CHANNELS` 与 `MERGE_EMAIL_NOTIFICATION`。它们是发送形态或聚合行为开关，不是渠道凭证；在 Actions 中自动开始读取同名 Secret/Variable 会引入额外行为变化。

 ## CLI 诊断
@@ -71,7 +78,7 @@ python main.py --check-notify

 Web 设置页的“通知渠道”分类提供单渠道测试入口。测试会使用当前页面草稿值合成临时配置，发送一条真实测试通知，但不会保存 `.env`，也不会修改运行时全局配置。

- 测试范围：11 个静态通知渠道，不包含 `UNKNOWN` 和运行时上下文渠道。
+- 测试范围：12 个静态通知渠道，不包含 `UNKNOWN` 和运行时上下文渠道。
 - 普通渠道：返回单次发送结果、耗时和通用错误码。
 - 自定义 Webhook：按 URL 顺序返回 attempts，展示每个 URL 的成功/失败、HTTP 状态、耗时和错误码；多个 URL 部分成功时，顶层 message 会标出成功数 / 总数。
 - 返回结果会脱敏 token、secret、password、Bearer、完整 webhook query 和疑似 path token。
@@ -105,6 +112,13 @@ AstrBot 已是一等通知渠道，优先使用 `ASTRBOT_URL` 和可选的 `ASTR
 CUSTOM_WEBHOOK_BODY_TEMPLATE={"content":$content_json}
 ```

+ntfy 已是一等通知渠道，优先使用 `NTFY_URL` 和可选的 `NTFY_TOKEN`。`NTFY_URL` 表示完整 topic endpoint，例如 `https://ntfy.sh/my-topic` 或 `https://self-hosted:port/my-topic`；系统会解析最后一个 path segment 作为 topic，并向 server root 发送 JSON publish：
+
+```env
+NTFY_URL=https://ntfy.sh/my-topic
+NTFY_TOKEN=
+```
+
 NapCat / OneBot HTTP API 需要按实际 endpoint 和目标类型调整。下面只是常见 body 形态示例，`user_id`、`group_id`、URL 路径和鉴权方式都应以你的 NapCat 配置为准：

 ```env
@@ -127,7 +141,7 @@ P3 新增三类通知路由配置：
 | `alert` | `NOTIFICATION_ALERT_CHANNELS` | EventMonitor 触发通知 |
 | `system_error` | `NOTIFICATION_SYSTEM_ERROR_CHANNELS` | 预留能力；当前不新增自动系统错误生产者 |

-配置值为逗号分隔渠道枚举：`wechat,feishu,telegram,email,pushover,pushplus,serverchan3,custom,discord,slack,astrbot`。
+配置值为逗号分隔渠道枚举：`wechat,feishu,telegram,email,pushover,ntfy,pushplus,serverchan3,custom,discord,slack,astrbot`。

 - 留空或未配置：保持旧行为，发送到所有已配置静态渠道。
 - 非空：只发送到路由列表与已配置渠道的交集；交集为空时不会 fallback 到全渠道。
--- a/src/config.py
+++ b/src/config.py
@@ -75,6 +75,19 @@ _FIXED_TEMPERATURE_LITELLM_MODELS: Dict[str, Dict[str, float]] = {
        "non_thinking": 0.6,
    },
 }
+
+
+def _has_ntfy_topic_endpoint(value: Optional[str]) -> bool:
+    """Return whether an ntfy URL points at a concrete topic endpoint."""
+    raw_url = (value or "").strip()
+    if not raw_url:
+        return False
+    parsed = urlparse(raw_url)
+    if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc:
+        return False
+    return any(segment for segment in parsed.path.split("/") if segment)
+
+
 AGENT_MAX_STEPS_DEFAULT = 10
 NEWS_STRATEGY_WINDOWS: Dict[str, int] = {
    "ultra_short": 1,
@@ -731,6 +744,10 @@ class Config:
    # Pushover 配置（手机/桌面推送通知）
    pushover_user_key: Optional[str] = None  # 用户 Key（https://pushover.net 获取）
    pushover_api_token: Optional[str] = None  # 应用 API Token
+
+    # ntfy 配置（完整 topic endpoint，例如 https://ntfy.sh/my-topic）
+    ntfy_url: Optional[str] = None
+    ntfy_token: Optional[str] = None
    
    # 自定义 Webhook（支持多个，逗号分隔）
    # 适用于：钉钉、Discord、Slack、自建服务等任意支持 POST JSON 的 Webhook
@@ -1470,6 +1487,8 @@ class Config:
            stock_email_groups=cls._parse_stock_email_groups(),
            pushover_user_key=os.getenv('PUSHOVER_USER_KEY'),
            pushover_api_token=os.getenv('PUSHOVER_API_TOKEN'),
+            ntfy_url=os.getenv('NTFY_URL'),
+            ntfy_token=os.getenv('NTFY_TOKEN'),
            pushplus_token=os.getenv('PUSHPLUS_TOKEN'),
            pushplus_topic=os.getenv('PUSHPLUS_TOPIC'),
            serverchan3_sendkey=os.getenv('SERVERCHAN3_SENDKEY'),
@@ -2462,6 +2481,7 @@ class Config:
            or (self.telegram_bot_token and self.telegram_chat_id)
            or (self.email_sender and self.email_password)
            or (self.pushover_user_key and self.pushover_api_token)
+            or _has_ntfy_topic_endpoint(self.ntfy_url)
            or self.pushplus_token
            or self.serverchan3_sendkey
            or self.custom_webhook_urls
@@ -2479,6 +2499,13 @@ class Config:
                field="WECHAT_WEBHOOK_URL",
            ))

+        if self.ntfy_url and not _has_ntfy_topic_endpoint(self.ntfy_url):
+            issues.append(ConfigIssue(
+                severity="error",
+                message="NTFY_URL 必须包含 topic path，例如 https://ntfy.sh/my-topic",
+                field="NTFY_URL",
+            ))
+
        if self.notification_quiet_hours:
            try:
                parse_notification_quiet_hours(self.notification_quiet_hours)
--- a/src/core/config_registry.py
+++ b/src/core/config_registry.py
@@ -1268,6 +1268,37 @@ _FIELD_DEFINITIONS: Dict[str, Dict[str, Any]] = {
        "validation": {},
        "display_order": 43,
    },
+    "NTFY_URL": {
+        "title": "ntfy URL",
+        "description": "Full ntfy publish endpoint including topic path, e.g. https://ntfy.sh/my-topic.",
+        "category": "notification",
+        "data_type": "string",
+        "ui_control": "password",
+        "is_sensitive": True,
+        "is_required": False,
+        "is_editable": True,
+        "default_value": None,
+        "options": [],
+        "validation": {
+            "item_type": "url",
+            "allowed_schemes": ["http", "https"],
+        },
+        "display_order": 44,
+    },
+    "NTFY_TOKEN": {
+        "title": "ntfy Token",
+        "description": "Optional ntfy bearer token for protected topics or self-hosted servers.",
+        "category": "notification",
+        "data_type": "string",
+        "ui_control": "password",
+        "is_sensitive": True,
+        "is_required": False,
+        "is_editable": True,
+        "default_value": None,
+        "options": [],
+        "validation": {},
+        "display_order": 45,
+    },
    "PUSHPLUS_TOPIC": {
        "title": "PushPlus Topic",
        "description": "PushPlus group topic code for one-to-many push.",
@@ -1297,7 +1328,7 @@ _FIELD_DEFINITIONS: Dict[str, Dict[str, Any]] = {
        "default_value": None,
        "options": [],
        "validation": {},
-        "display_order": 45,
+        "display_order": 46,
    },
    "ASTRBOT_URL": {
        "title": "AstrBot URL",
@@ -1314,7 +1345,7 @@ _FIELD_DEFINITIONS: Dict[str, Dict[str, Any]] = {
            "item_type": "url",
            "allowed_schemes": ["http", "https"],
        },
-        "display_order": 46,
+        "display_order": 47,
    },
    "ASTRBOT_TOKEN": {
        "title": "AstrBot Token",
@@ -1328,7 +1359,7 @@ _FIELD_DEFINITIONS: Dict[str, Dict[str, Any]] = {
        "default_value": None,
        "options": [],
        "validation": {},
-        "display_order": 47,
+        "display_order": 48,
    },
    "SINGLE_STOCK_NOTIFY": {
        "title": "Single Stock Notify",
@@ -2243,6 +2274,7 @@ def _infer_category(key: str) -> str:
        "TELEGRAM",
        "EMAIL",
        "PUSHOVER",
+        "NTFY",
        "PUSHPLUS",
        "SERVERCHAN",
        "DINGTALK",
--- a/src/core/pipeline.py
+++ b/src/core/pipeline.py
@@ -1959,6 +1959,7 @@ class StockAnalysisPipeline:
                channels_needing_image = {
                    ch for ch in channels
                    if ch.value in self.notifier._markdown_to_image_channels
+                    and ch != NotificationChannel.NTFY
                }
                non_wechat_channels_needing_image = {
                    ch for ch in channels_needing_image if ch != NotificationChannel.WECHAT
@@ -2155,6 +2156,11 @@ class StockAnalysisPipeline:
                            channel.value,
                            lambda: self.notifier.send_to_pushover(report),
                        ) or non_wechat_success
+                    elif channel == NotificationChannel.NTFY:
+                        non_wechat_success = _send_channel_safely(
+                            channel.value,
+                            lambda: self.notifier.send_to_ntfy(report),
+                        ) or non_wechat_success
                    elif channel == NotificationChannel.ASTRBOT:
                        non_wechat_success = _send_channel_safely(
                            channel.value,
--- a/src/notification.py
+++ b/src/notification.py
@@ -50,13 +50,15 @@ from src.notification_sender import (
    DiscordSender,
    EmailSender,
    FeishuSender,
+    NtfySender,
    PushoverSender,
    PushplusSender,
    Serverchan3Sender,
    SlackSender,
    TelegramSender,
    WechatSender,
-    WECHAT_IMAGE_MAX_BYTES
+    WECHAT_IMAGE_MAX_BYTES,
+    resolve_ntfy_endpoint,
 )

 logger = logging.getLogger(__name__)
@@ -72,6 +74,7 @@ class NotificationChannel(Enum):
    TELEGRAM = "telegram"  # Telegram
    EMAIL = "email"        # 邮件
    PUSHOVER = "pushover"  # Pushover（手机/桌面推送）
+    NTFY = "ntfy"          # ntfy
    PUSHPLUS = "pushplus"  # PushPlus（国内推送服务）
    SERVERCHAN3 = "serverchan3"  # Server酱3（手机APP推送服务）
    CUSTOM = "custom"      # 自定义 Webhook
@@ -97,6 +100,7 @@ class ChannelDetector:
            NotificationChannel.TELEGRAM: "Telegram",
            NotificationChannel.EMAIL: "邮件",
            NotificationChannel.PUSHOVER: "Pushover",
+            NotificationChannel.NTFY: "ntfy",
            NotificationChannel.PUSHPLUS: "PushPlus",
            NotificationChannel.SERVERCHAN3: "Server酱3",
            NotificationChannel.CUSTOM: "自定义Webhook",
@@ -114,6 +118,7 @@ class NotificationService(
    DiscordSender,
    EmailSender,
    FeishuSender,
+    NtfySender,
    PushoverSender,
    PushplusSender,
    Serverchan3Sender,
@@ -168,6 +173,7 @@ class NotificationService(
        DiscordSender.__init__(self, config)
        EmailSender.__init__(self, config)
        FeishuSender.__init__(self, config)
+        NtfySender.__init__(self, config)
        PushoverSender.__init__(self, config)
        PushplusSender.__init__(self, config)
        Serverchan3Sender.__init__(self, config)
@@ -303,6 +309,10 @@ class NotificationService(
        ):
            channels.append(NotificationChannel.PUSHOVER)

+        ntfy_server_url, ntfy_topic = resolve_ntfy_endpoint(getattr(config, "ntfy_url", None))
+        if ntfy_server_url and ntfy_topic:
+            channels.append(NotificationChannel.NTFY)
+
        if getattr(config, "pushplus_token", None):
            channels.append(NotificationChannel.PUSHPLUS)

@@ -1723,6 +1733,7 @@ class NotificationService(
        channels_needing_image = {
            ch for ch in target_channels
            if ch.value in self._markdown_to_image_channels
+            and ch != NotificationChannel.NTFY
        }
        if channels_needing_image:
            from src.md2img import markdown_to_image
@@ -1783,6 +1794,8 @@ class NotificationService(
                        result = self.send_to_email(content, receivers=receivers)
                elif channel == NotificationChannel.PUSHOVER:
                    result = self.send_to_pushover(content)
+                elif channel == NotificationChannel.NTFY:
+                    result = self.send_to_ntfy(content)
                elif channel == NotificationChannel.PUSHPLUS:
                    result = self.send_to_pushplus(content)
                elif channel == NotificationChannel.SERVERCHAN3:
--- a/src/notification_routing.py
+++ b/src/notification_routing.py
@@ -16,6 +16,7 @@ ROUTABLE_NOTIFICATION_CHANNELS: Tuple[str, ...] = (
    "telegram",
    "email",
    "pushover",
+    "ntfy",
    "pushplus",
    "serverchan3",
    "custom",
--- a/src/notification_sender/init.py
+++ b/src/notification_sender/init.py
@@ -12,6 +12,7 @@ from .custom_webhook_sender import CustomWebhookSender
 from .discord_sender import DiscordSender
 from .email_sender import EmailSender
 from .feishu_sender import FeishuSender
+from .ntfy_sender import NtfySender, resolve_ntfy_endpoint
 from .pushover_sender import PushoverSender
 from .pushplus_sender import PushplusSender
 from .serverchan3_sender import Serverchan3Sender
--- a/src/notification_sender/ntfy_sender.py
+++ b/src/notification_sender/ntfy_sender.py
@@ -0,0 +1,125 @@
+# -*- coding: utf-8 -*-
+"""ntfy notification sender."""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime
+from typing import Optional, Tuple
+from urllib.parse import unquote, urlparse, urlunparse
+
+import requests
+
+from src.config import Config
+
+
+logger = logging.getLogger(__name__)
+
+
+def resolve_ntfy_endpoint(ntfy_url: Optional[str]) -> Tuple[Optional[str], Optional[str]]:
+    """Split NTFY_URL into server root and topic from the final path segment."""
+    raw_url = (ntfy_url or "").strip().rstrip("/")
+    if not raw_url:
+        return None, None
+
+    parsed = urlparse(raw_url)
+    if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc:
+        return None, None
+
+    path_segments = [segment for segment in parsed.path.split("/") if segment]
+    if not path_segments:
+        return None, None
+
+    topic = unquote(path_segments[-1]).strip()
+    if not topic:
+        return None, None
+
+    root_path = "/".join(path_segments[:-1])
+    server_url = urlunparse(
+        parsed._replace(
+            path=f"/{root_path}" if root_path else "",
+            params="",
+            query="",
+            fragment="",
+        )
+    ).rstrip("/")
+
+    return server_url, topic
+
+
+class NtfySender:
+    """Send Markdown text notifications through the ntfy JSON publish API."""
+
+    def __init__(self, config: Config):
+        self._ntfy_url = getattr(config, "ntfy_url", None)
+        self._ntfy_token = getattr(config, "ntfy_token", None)
+        self._webhook_verify_ssl = getattr(config, "webhook_verify_ssl", True)
+
+    def _is_ntfy_configured(self) -> bool:
+        return bool(self._ntfy_url)
+
+    def _resolve_ntfy_endpoint(self) -> Tuple[Optional[str], Optional[str]]:
+        return resolve_ntfy_endpoint(self._ntfy_url)
+
+    def send_to_ntfy(
+        self,
+        content: str,
+        title: Optional[str] = None,
+        *,
+        timeout_seconds: Optional[float] = None,
+    ) -> bool:
+        """Publish a notification to ntfy using a JSON body with UTF-8 text."""
+        if not self._is_ntfy_configured():
+            logger.warning("ntfy URL 未配置，跳过推送")
+            return False
+
+        server_url, topic = self._resolve_ntfy_endpoint()
+        if not server_url or not topic:
+            logger.error("NTFY_URL 必须是包含 topic path 的完整 endpoint，例如 https://ntfy.sh/my-topic")
+            return False
+
+        if title is None:
+            date_str = datetime.now().strftime("%Y-%m-%d")
+            title = f"📈 股票分析报告 - {date_str}"
+
+        headers = {
+            "Content-Type": "application/json; charset=utf-8",
+            "User-Agent": "daily_stock_analysis",
+        }
+        token = (self._ntfy_token or "").strip()
+        if token:
+            headers["Authorization"] = f"Bearer {token}"
+
+        payload = {
+            "topic": topic,
+            "title": title,
+            "message": content,
+            "markdown": True,
+        }
+
+        try:
+            response = requests.post(
+                server_url,
+                json=payload,
+                headers=headers,
+                timeout=timeout_seconds or 10,
+                verify=self._webhook_verify_ssl,
+            )
+            if 200 <= response.status_code < 300:
+                logger.info("ntfy 消息发送成功")
+                return True
+
+            logger.error("ntfy 请求失败: HTTP %s", response.status_code)
+            logger.debug("ntfy 响应内容: %s", response.text)
+            return False
+        except requests.exceptions.Timeout:
+            logger.error("发送 ntfy 消息失败: 请求超时")
+            return False
+        except requests.exceptions.RequestException as exc:
+            logger.error("发送 ntfy 消息失败: 网络请求异常")
+            logger.debug("ntfy 请求异常类型: %s", type(exc).__name__)
+            return False
+        except Exception as exc:
+            logger.error("发送 ntfy 消息失败: 未知异常")
+            logger.debug("ntfy 未知异常类型: %s", type(exc).__name__)
+            return False
--- a/src/services/notification_diagnostics.py
+++ b/src/services/notification_diagnostics.py
@@ -20,6 +20,7 @@ from src.notification_routing import (
    ROUTABLE_NOTIFICATION_CHANNELS,
    split_notification_route_channels,
 )
+from src.notification_sender.ntfy_sender import resolve_ntfy_endpoint

 KeyTier = Literal["minimal", "advanced"]
 IssueSeverity = Literal["error", "warning", "info"]
@@ -108,6 +109,14 @@ CHANNEL_SPECS: Tuple[NotificationChannelSpec, ...] = (
        kind="configured",
        minimal_keys=("PUSHOVER_USER_KEY", "PUSHOVER_API_TOKEN"),
    ),
+    NotificationChannelSpec(
+        channel=NotificationChannel.NTFY.value,
+        display_name=ChannelDetector.get_channel_name(NotificationChannel.NTFY),
+        kind="configured",
+        minimal_keys=("NTFY_URL",),
+        advanced_keys=("NTFY_TOKEN", "WEBHOOK_VERIFY_SSL"),
+        note="NTFY_URL must include the topic path, e.g. https://ntfy.sh/my-topic.",
+    ),
    NotificationChannelSpec(
        channel=NotificationChannel.PUSHPLUS.value,
        display_name=ChannelDetector.get_channel_name(NotificationChannel.PUSHPLUS),
@@ -218,6 +227,11 @@ P3_ROUTE_ENV_KEYS: Tuple[str, ...] = tuple(

 P4_NOISE_ACTIONS_ENV_KEYS: Tuple[str, ...] = P4_NOISE_ENV_KEYS

+P6_CHANNEL_ACTIONS_ENV_KEYS: Tuple[str, ...] = (
+    "NTFY_URL",
+    "NTFY_TOKEN",
+)
+

 def _value(config: Config, attr: str):
    return getattr(config, attr, None)
@@ -291,7 +305,7 @@ def run_notification_diagnostics(config: Config) -> NotificationDiagnosticResult
        _issue(
            "info",
            "phase_scope",
-            "通知诊断会检查渠道基线、只读诊断、Web 测试、P3 路由配置和 P4 降噪配置；长尾渠道留给后续 Phase。",
+            "通知诊断会检查渠道基线、只读诊断、Web 测试、P3 路由配置、P4 降噪配置和 P6-A ntfy 渠道。",
        ),
    ]

@@ -304,6 +318,18 @@ def run_notification_diagnostics(config: Config) -> NotificationDiagnosticResult
            )
        )

+    if _has(config, "ntfy_url"):
+        ntfy_server_url, ntfy_topic = resolve_ntfy_endpoint(getattr(config, "ntfy_url", None))
+        if not ntfy_server_url or not ntfy_topic:
+            errors.append(
+                _issue(
+                    "error",
+                    "invalid_ntfy_url",
+                    "NTFY_URL 必须包含 topic path，例如 https://ntfy.sh/my-topic。",
+                    key="NTFY_URL",
+                )
+            )
+
    _require_pair(
        config,
        left_attr="telegram_bot_token",
@@ -372,6 +398,15 @@ def run_notification_diagnostics(config: Config) -> NotificationDiagnosticResult
                key="PUSHPLUS_TOKEN",
            )
        )
+    if _has(config, "ntfy_token") and not _has(config, "ntfy_url"):
+        warnings.append(
+            _issue(
+                "warning",
+                "advanced_without_minimal",
+                "已配置 NTFY_TOKEN，但缺少 NTFY_URL，ntfy 渠道不会启用。",
+                key="NTFY_URL",
+            )
+        )
    if (
        _has(config, "custom_webhook_bearer_token")
        or _has(config, "custom_webhook_body_template")
--- a/src/services/system_config_service.py
+++ b/src/services/system_config_service.py
@@ -44,6 +44,7 @@ from src.core.config_registry import (
    get_registered_field_keys,
 )
 from src.notification_noise import validate_notification_timezone
+from src.notification_sender.ntfy_sender import resolve_ntfy_endpoint

 logger = logging.getLogger(__name__)

@@ -110,6 +111,7 @@ class SystemConfigService:
        "telegram",
        "email",
        "pushover",
+        "ntfy",
        "pushplus",
        "serverchan3",
        "custom",
@@ -134,6 +136,8 @@ class SystemConfigService:
        "EMAIL_RECEIVERS": ("email_receivers", "csv"),
        "PUSHOVER_USER_KEY": ("pushover_user_key", "string"),
        "PUSHOVER_API_TOKEN": ("pushover_api_token", "string"),
+        "NTFY_URL": ("ntfy_url", "string"),
+        "NTFY_TOKEN": ("ntfy_token", "string"),
        "PUSHPLUS_TOKEN": ("pushplus_token", "string"),
        "PUSHPLUS_TOPIC": ("pushplus_topic", "string"),
        "SERVERCHAN3_SENDKEY": ("serverchan3_sendkey", "string"),
@@ -158,6 +162,7 @@ class SystemConfigService:
        "telegram": (("TELEGRAM_BOT_TOKEN", "TELEGRAM_CHAT_ID"),),
        "email": (("EMAIL_SENDER", "EMAIL_PASSWORD"),),
        "pushover": (("PUSHOVER_USER_KEY", "PUSHOVER_API_TOKEN"),),
+        "ntfy": (("NTFY_URL",),),
        "pushplus": (("PUSHPLUS_TOKEN",),),
        "serverchan3": (("SERVERCHAN3_SENDKEY",),),
        "custom": (("CUSTOM_WEBHOOK_URLS",),),
@@ -171,6 +176,7 @@ class SystemConfigService:
        "telegram": ("TELEGRAM_BOT_TOKEN",),
        "email": ("EMAIL_RECEIVERS", "EMAIL_SENDER"),
        "pushover": ("PUSHOVER_USER_KEY",),
+        "ntfy": ("NTFY_URL",),
        "pushplus": ("PUSHPLUS_TOPIC",),
        "serverchan3": ("SERVERCHAN3_SENDKEY",),
        "custom": ("CUSTOM_WEBHOOK_URLS",),
@@ -335,6 +341,20 @@ class SystemConfigService:
                latency_ms=None,
                attempts=[],
            )
+        invalid_message = self._get_invalid_notification_test_config_message(
+            normalized_channel,
+            effective_map,
+        )
+        if invalid_message:
+            return self._build_notification_test_result(
+                success=False,
+                message=invalid_message,
+                error_code="config_invalid",
+                stage="config_validation",
+                retryable=False,
+                latency_ms=None,
+                attempts=[],
+            )

        config = self._build_notification_test_config(effective_map)
        try:
@@ -1713,6 +1733,22 @@ class SystemConfigService:
                    }
                )

+        if key == "NTFY_URL" and value.strip():
+            allowed_schemes = tuple(validation.get("allowed_schemes", ["http", "https"]))
+            if SystemConfigService._is_valid_url(value.strip(), allowed_schemes=allowed_schemes):
+                ntfy_server_url, ntfy_topic = resolve_ntfy_endpoint(value)
+                if not ntfy_server_url or not ntfy_topic:
+                    issues.append(
+                        {
+                            "key": key,
+                            "code": "invalid_ntfy_url",
+                            "message": "NTFY_URL must include a topic path, e.g. https://ntfy.sh/my-topic",
+                            "severity": "error",
+                            "expected": "ntfy publish endpoint with topic path",
+                            "actual": value,
+                        }
+                    )
+
        return issues

    @staticmethod
@@ -1820,6 +1856,21 @@ class SystemConfigService:

        return missing_by_group[0] if missing_by_group else []

+    @staticmethod
+    def _get_invalid_notification_test_config_message(
+        channel: str,
+        effective_map: Dict[str, str],
+    ) -> Optional[str]:
+        if channel != "ntfy":
+            return None
+        ntfy_url = (effective_map.get("NTFY_URL") or "").strip()
+        if not ntfy_url:
+            return None
+        ntfy_server_url, ntfy_topic = resolve_ntfy_endpoint(ntfy_url)
+        if ntfy_server_url and ntfy_topic:
+            return None
+        return "NTFY_URL 必须包含 topic path，例如 https://ntfy.sh/my-topic。"
+
    def _build_notification_test_config(self, effective_map: Dict[str, str]) -> Config:
        """Build an isolated Config instance for notification testing."""
        kwargs: Dict[str, Any] = {"stock_list": []}
@@ -1863,6 +1914,7 @@ class SystemConfigService:
            DiscordSender,
            EmailSender,
            FeishuSender,
+            NtfySender,
            PushoverSender,
            PushplusSender,
            Serverchan3Sender,
@@ -1906,6 +1958,7 @@ class SystemConfigService:
            "telegram": lambda: TelegramSender(config).send_to_telegram(titled_content, timeout_seconds=timeout_seconds),
            "email": lambda: EmailSender(config).send_to_email(content, subject=title, timeout_seconds=timeout_seconds),
            "pushover": lambda: PushoverSender(config).send_to_pushover(content, title=title, timeout_seconds=timeout_seconds),
+            "ntfy": lambda: NtfySender(config).send_to_ntfy(content, title=title, timeout_seconds=timeout_seconds),
            "pushplus": lambda: PushplusSender(config).send_to_pushplus(content, title=title, timeout_seconds=timeout_seconds),
            "serverchan3": lambda: Serverchan3Sender(config).send_to_serverchan3(content, title=title, timeout_seconds=timeout_seconds),
            "discord": lambda: DiscordSender(config).send_to_discord(titled_content, timeout_seconds=timeout_seconds),
@@ -2018,12 +2071,19 @@ class SystemConfigService:

        safe_netloc = parsed.netloc.rsplit("@", 1)[-1]
        safe_segments: List[str] = []
-        for segment in parsed.path.split("/"):
+        path_segments = parsed.path.split("/")
+        last_non_empty_index = next(
+            (index for index in range(len(path_segments) - 1, -1, -1) if path_segments[index]),
+            -1,
+        )
+        for index, segment in enumerate(path_segments):
            if not segment:
                safe_segments.append(segment)
                continue
            lower = segment.lower()
            looks_secret = (
+                (source_key_upper == "NTFY_URL" and index == last_non_empty_index)
+                or
                len(segment) >= 16
                or lower.startswith("bot")
                or "token" in lower
@@ -2105,6 +2165,7 @@ class SystemConfigService:
            "DINGTALK_",
            "WECHAT_",
            "PUSHOVER_",
+            "NTFY_",
            "PUSHPLUS_",
            "SERVERCHAN",
            "CUSTOM_WEBHOOK",
@@ -2131,6 +2192,11 @@ class SystemConfigService:
    def _has_any_config_value(effective_map: Dict[str, str], keys: Sequence[str]) -> bool:
        return any((effective_map.get(key) or "").strip() for key in keys)

+    @staticmethod
+    def _has_valid_ntfy_endpoint(effective_map: Dict[str, str]) -> bool:
+        ntfy_server_url, ntfy_topic = resolve_ntfy_endpoint(effective_map.get("NTFY_URL"))
+        return bool(ntfy_server_url and ntfy_topic)
+
    @classmethod
    def _anspire_legacy_llm_enabled(cls, effective_map: Dict[str, str]) -> bool:
        if not parse_env_bool(effective_map.get("ANSPIRE_LLM_ENABLED"), default=True):
@@ -2442,6 +2508,7 @@ class SystemConfigService:
                    "ASTRBOT_URL",
                ),
            )
+            or self._has_valid_ntfy_endpoint(effective_map)
            or (
                parse_env_bool(effective_map.get("FEISHU_STREAM_ENABLED"), default=False)
                and self._has_any_config_value(effective_map, ("FEISHU_APP_ID",))
--- a/tests/test_config_validate_structured.py
+++ b/tests/test_config_validate_structured.py
@@ -345,6 +345,20 @@ class TestValidateStructuredNotification:
        issues = cfg.validate_structured()
        assert not any(i.severity == "warning" and "通知渠道" in i.message for i in issues)

+    def test_ntfy_url_without_topic_reports_error_and_does_not_count_as_channel(self):
+        cfg = _make_config(wechat_webhook_url=None, ntfy_url="https://ntfy.sh")
+        issues = cfg.validate_structured()
+
+        assert any(i.severity == "error" and i.field == "NTFY_URL" for i in issues)
+        assert any(i.severity == "warning" and "通知渠道" in i.message for i in issues)
+
+    def test_ntfy_topic_endpoint_counts_as_notification_channel(self):
+        cfg = _make_config(wechat_webhook_url=None, ntfy_url="https://ntfy.sh/dsa-topic")
+        issues = cfg.validate_structured()
+
+        assert not any(i.field == "NTFY_URL" for i in issues)
+        assert not any(i.severity == "warning" and "通知渠道" in i.message for i in issues)
+
    def test_feishu_app_credentials_without_webhook_warns_mode_mismatch(self):
        cfg = _make_config(
            wechat_webhook_url=None,
--- a/tests/test_daily_analysis_workflow_notification_env.py
+++ b/tests/test_daily_analysis_workflow_notification_env.py
@@ -5,7 +5,12 @@ from pathlib import Path

 import yaml

-from src.services.notification_diagnostics import P0_ACTIONS_ENV_KEYS, P3_ROUTE_ENV_KEYS, P4_NOISE_ENV_KEYS
+from src.services.notification_diagnostics import (
+    P0_ACTIONS_ENV_KEYS,
+    P3_ROUTE_ENV_KEYS,
+    P4_NOISE_ENV_KEYS,
+    P6_CHANNEL_ACTIONS_ENV_KEYS,
+)


 ROOT_DIR = Path(__file__).resolve().parent.parent
@@ -50,6 +55,13 @@ def test_daily_analysis_maps_p4_notification_noise_env_keys() -> None:
        assert key in env


+def test_daily_analysis_maps_p6_channel_env_keys() -> None:
+    env = _load_daily_analysis_env()
+
+    for key in P6_CHANNEL_ACTIONS_ENV_KEYS:
+        assert key in env
+
+
 def test_daily_analysis_keeps_deferred_behavior_switches_unmapped() -> None:
    env = _load_daily_analysis_env()

--- a/tests/test_notification.py
+++ b/tests/test_notification.py
@@ -614,6 +614,65 @@ class TestNotificationServiceReportGeneration(unittest.TestCase):
        self.assertTrue(ok)
        self.assertAlmostEqual(mock_post.call_count, 4, delta=1)

+    @mock.patch("src.notification.get_config")
+    @mock.patch("requests.post")
+    def test_send_to_ntfy_via_notification_service(
+        self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock
+    ):
+        cfg = _make_config(ntfy_url="https://ntfy.sh/dsa-topic")
+        mock_get_config.return_value = cfg
+        mock_post.return_value = _make_response(200)
+
+        service = NotificationService()
+        self.assertIn(NotificationChannel.NTFY, service.get_available_channels())
+
+        ok = service.send("ntfy content")
+
+        self.assertTrue(ok)
+        mock_post.assert_called_once()
+        self.assertEqual(mock_post.call_args.args[0], "https://ntfy.sh")
+        self.assertEqual(mock_post.call_args.kwargs["json"]["topic"], "dsa-topic")
+
+    @mock.patch("src.notification.get_config")
+    def test_ntfy_url_without_topic_is_not_available(self, mock_get_config: mock.MagicMock):
+        mock_get_config.return_value = _make_config(ntfy_url="https://ntfy.sh")
+
+        service = NotificationService()
+
+        self.assertNotIn(NotificationChannel.NTFY, service.get_available_channels())
+        self.assertFalse(service.is_available())
+
+    @mock.patch("src.notification.get_config")
+    def test_ntfy_url_with_unsupported_scheme_is_not_available(
+        self, mock_get_config: mock.MagicMock
+    ):
+        mock_get_config.return_value = _make_config(ntfy_url="ntfy://ntfy.sh/dsa-topic")
+
+        service = NotificationService()
+
+        self.assertNotIn(NotificationChannel.NTFY, service.get_available_channels())
+        self.assertFalse(service.is_available())
+
+    @mock.patch("src.notification.get_config")
+    @mock.patch("requests.post")
+    def test_send_to_ntfy_does_not_trigger_markdown_to_image(
+        self, mock_post: mock.MagicMock, mock_get_config: mock.MagicMock
+    ):
+        cfg = _make_config(
+            ntfy_url="https://ntfy.sh/dsa-topic",
+            markdown_to_image_channels=["ntfy"],
+        )
+        mock_get_config.return_value = cfg
+        mock_post.return_value = _make_response(200)
+
+        service = NotificationService()
+        with mock.patch("src.md2img.markdown_to_image", return_value=b"png") as mock_md2img:
+            ok = service.send("ntfy content")
+
+        self.assertTrue(ok)
+        mock_md2img.assert_not_called()
+        mock_post.assert_called_once()
+
    @mock.patch("src.notification.get_config")
    @mock.patch("requests.post")
    def test_send_to_pushover_via_notification_service(
--- a/tests/test_notification_diagnostics.py
+++ b/tests/test_notification_diagnostics.py
@@ -39,6 +39,8 @@ class NotificationDiagnosticsTestCase(unittest.TestCase):

        self.assertIn(("ASTRBOT_URL", "minimal"), key_tiers)
        self.assertIn(("ASTRBOT_TOKEN", "advanced"), key_tiers)
+        self.assertIn(("NTFY_URL", "minimal"), key_tiers)
+        self.assertIn(("NTFY_TOKEN", "advanced"), key_tiers)
        self.assertIn(("CUSTOM_WEBHOOK_BODY_TEMPLATE", "advanced"), key_tiers)
        self.assertIn(("WEBHOOK_VERIFY_SSL", "advanced"), key_tiers)
        for key in P3_ROUTE_ENV_KEYS:
@@ -84,12 +86,29 @@ class NotificationDiagnosticsTestCase(unittest.TestCase):
        result = run_notification_diagnostics(
            _config(
                wechat_webhook_url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=1",
+                ntfy_url="https://ntfy.sh/dsa-topic",
                astrbot_url="https://astrbot.example/webhook",
            )
        )

        self.assertTrue(result.ok)
-        self.assertEqual(result.configured_channels, ("wechat", "astrbot"))
+        self.assertEqual(result.configured_channels, ("wechat", "ntfy", "astrbot"))
+
+    def test_ntfy_url_without_topic_reports_error(self):
+        result = run_notification_diagnostics(_config(ntfy_url="https://ntfy.sh"))
+
+        self.assertFalse(result.ok)
+        self.assertNotIn("ntfy", result.configured_channels)
+        self.assertIn("invalid_ntfy_url", {item.code for item in result.errors})
+        self.assertIn("NTFY_URL", {item.key for item in result.errors})
+
+    def test_ntfy_url_with_unsupported_scheme_reports_error(self):
+        result = run_notification_diagnostics(_config(ntfy_url="ftp://ntfy.example/dsa-topic"))
+
+        self.assertFalse(result.ok)
+        self.assertNotIn("ntfy", result.configured_channels)
+        self.assertIn("invalid_ntfy_url", {item.code for item in result.errors})
+        self.assertIn("NTFY_URL", {item.key for item in result.errors})

    def test_advanced_key_without_minimal_warns_but_is_structured(self):
        result = run_notification_diagnostics(_config(pushplus_topic="topic-only"))
@@ -103,7 +122,7 @@ class NotificationDiagnosticsTestCase(unittest.TestCase):
        result = run_notification_diagnostics(
            _config(
                wechat_webhook_url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=1",
-                notification_report_channels=["wechat", "ntfy"],
+                notification_report_channels=["wechat", "not-a-channel"],
            )
        )

--- a/tests/test_notification_routing.py
+++ b/tests/test_notification_routing.py
@@ -0,0 +1,12 @@
+# -*- coding: utf-8 -*-
+"""Tests for notification route channel parsing."""
+
+from src.notification_routing import ROUTABLE_NOTIFICATION_CHANNELS, split_notification_route_channels
+
+
+def test_ntfy_is_a_routable_notification_channel() -> None:
+    valid, invalid = split_notification_route_channels(["wechat", "ntfy", "not-a-channel"])
+
+    assert "ntfy" in ROUTABLE_NOTIFICATION_CHANNELS
+    assert valid == ["wechat", "ntfy"]
+    assert invalid == ["not-a-channel"]
--- a/tests/test_notification_sender.py
+++ b/tests/test_notification_sender.py
@@ -17,6 +17,8 @@ from email.utils import parseaddr
 from unittest import mock
 from typing import Optional

+import requests
+
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))

 from src.config import Config
@@ -26,6 +28,7 @@ from src.notification_sender import (
    DiscordSender,
    EmailSender,
    FeishuSender,
+    NtfySender,
    PushoverSender,
    PushplusSender,
    Serverchan3Sender,
@@ -342,6 +345,101 @@ class TestEmailSender(unittest.TestCase):
        server.quit.assert_called_once()


+class TestNtfySender(unittest.TestCase):
+    """Unit tests for NtfySender."""
+
+    def test_send_returns_false_when_not_configured(self):
+        cfg = _config()
+        sender = NtfySender(cfg)
+
+        result = sender.send_to_ntfy("hello")
+
+        self.assertFalse(result)
+
+    @mock.patch("src.notification_sender.ntfy_sender.requests.post")
+    def test_send_success_uses_json_publish_with_topic_endpoint(self, mock_post):
+        mock_post.return_value = _response(200)
+        cfg = _config(
+            ntfy_url="https://ntfy.sh/dsa-topic",
+            ntfy_token="secret-token",
+            webhook_verify_ssl=False,
+        )
+        sender = NtfySender(cfg)
+
+        result = sender.send_to_ntfy("正文 **Markdown**", title="中文标题", timeout_seconds=5)
+
+        self.assertTrue(result)
+        mock_post.assert_called_once()
+        self.assertEqual(mock_post.call_args.args[0], "https://ntfy.sh")
+        call_kw = mock_post.call_args.kwargs
+        self.assertEqual(
+            call_kw["json"],
+            {
+                "topic": "dsa-topic",
+                "title": "中文标题",
+                "message": "正文 **Markdown**",
+                "markdown": True,
+            },
+        )
+        self.assertEqual(call_kw["headers"]["Authorization"], "Bearer secret-token")
+        self.assertEqual(call_kw["timeout"], 5)
+        self.assertFalse(call_kw["verify"])
+
+    @mock.patch("src.notification_sender.ntfy_sender.requests.post")
+    def test_send_supports_self_hosted_path_prefix(self, mock_post):
+        mock_post.return_value = _response(200)
+        cfg = _config(ntfy_url="https://example.com/ntfy/dsa-topic")
+        sender = NtfySender(cfg)
+
+        result = sender.send_to_ntfy("body", title="title")
+
+        self.assertTrue(result)
+        self.assertEqual(mock_post.call_args.args[0], "https://example.com/ntfy")
+        self.assertEqual(mock_post.call_args.kwargs["json"]["topic"], "dsa-topic")
+
+    @mock.patch("src.notification_sender.ntfy_sender.requests.post")
+    def test_send_returns_false_when_url_has_no_topic(self, mock_post):
+        cfg = _config(ntfy_url="https://ntfy.sh")
+        sender = NtfySender(cfg)
+
+        result = sender.send_to_ntfy("body")
+
+        self.assertFalse(result)
+        mock_post.assert_not_called()
+
+    @mock.patch("src.notification_sender.ntfy_sender.requests.post")
+    def test_send_returns_false_when_url_scheme_is_not_http(self, mock_post):
+        cfg = _config(ntfy_url="ftp://ntfy.example/dsa-topic")
+        sender = NtfySender(cfg)
+
+        result = sender.send_to_ntfy("body")
+
+        self.assertFalse(result)
+        mock_post.assert_not_called()
+
+    @mock.patch("src.notification_sender.ntfy_sender.requests.post")
+    def test_send_http_error_returns_false(self, mock_post):
+        mock_post.return_value = _response(500)
+        cfg = _config(ntfy_url="https://ntfy.sh/dsa-topic")
+        sender = NtfySender(cfg)
+
+        result = sender.send_to_ntfy("body")
+
+        self.assertFalse(result)
+
+    @mock.patch("src.notification_sender.ntfy_sender.requests.post")
+    def test_send_timeout_does_not_log_token_value(self, mock_post):
+        mock_post.side_effect = requests.exceptions.Timeout("secret-token")
+        cfg = _config(ntfy_url="https://ntfy.sh/dsa-topic", ntfy_token="secret-token")
+        sender = NtfySender(cfg)
+
+        with self.assertLogs("src.notification_sender.ntfy_sender", level="ERROR") as captured:
+            result = sender.send_to_ntfy("body")
+
+        self.assertFalse(result)
+        self.assertNotIn("secret-token", "\n".join(captured.output))
+
+
 class TestAstrbotSender(unittest.TestCase):
    """Unit tests for AstrbotSender."""

--- a/tests/test_pipeline_notification_image_routing.py
+++ b/tests/test_pipeline_notification_image_routing.py
@@ -173,6 +173,7 @@ class _FakeRoutedNotifier:
                NotificationChannel.WECHAT,
                NotificationChannel.TELEGRAM,
                NotificationChannel.EMAIL,
+                NotificationChannel.NTFY,
            ]
        )
        self.get_channels_for_route = MagicMock(return_value=list(routed_channels))
@@ -197,6 +198,7 @@ class _FakeRoutedNotifier:
        self.send_to_telegram = MagicMock(return_value=True)
        self._send_email_with_inline_image = MagicMock(return_value=True)
        self.send_to_email = MagicMock(return_value=True)
+        self.send_to_ntfy = MagicMock(return_value=True)

    @staticmethod
    def _generate_dashboard_report(results):
@@ -218,6 +220,7 @@ class TestPipelineReportRouteFiltering(unittest.TestCase):
                NotificationChannel.WECHAT,
                NotificationChannel.TELEGRAM,
                NotificationChannel.EMAIL,
+                NotificationChannel.NTFY,
            ],
        )
        pipeline.notifier.send_to_telegram.assert_called_once_with("report:000001")
@@ -245,6 +248,23 @@ class TestPipelineReportRouteFiltering(unittest.TestCase):
        pipeline.notifier.send_to_email.assert_called_once_with("report:000001")
        pipeline.notifier.send_to_telegram.assert_not_called()

+    def test_ntfy_route_uses_text_report_without_image_conversion(self):
+        pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline)
+        pipeline.notifier = _FakeRoutedNotifier(
+            [NotificationChannel.NTFY],
+            image_channels={"ntfy"},
+        )
+        pipeline.config = SimpleNamespace(stock_email_groups=[])
+        results = [SimpleNamespace(code="000001")]
+
+        with patch("src.md2img.markdown_to_image", return_value=b"png") as mock_md2img:
+            pipeline._send_notifications(results, ReportType.SIMPLE)
+
+        mock_md2img.assert_not_called()
+        pipeline.notifier.send_to_ntfy.assert_called_once_with("report:000001")
+        pipeline.notifier._send_email_with_inline_image.assert_not_called()
+        pipeline.notifier._send_telegram_photo.assert_not_called()
+
    def test_noise_suppression_happens_before_markdown_to_image(self):
        pipeline = StockAnalysisPipeline.__new__(StockAnalysisPipeline)
        pipeline.notifier = _FakeRoutedNotifier(
--- a/tests/test_system_config_api.py
+++ b/tests/test_system_config_api.py
@@ -614,6 +614,17 @@ class SystemConfigApiTestCase(unittest.TestCase):
        self.assertEqual(mock_test.call_args.kwargs["channel"], "wechat")
        self.assertEqual(mock_test.call_args.kwargs["timeout_seconds"], 5)

+    def test_test_notification_channel_schema_accepts_ntfy(self) -> None:
+        request = TestNotificationChannelRequest(
+            channel="ntfy",
+            items=[{"key": "NTFY_URL", "value": "https://ntfy.sh/dsa-topic"}],
+            title="DSA 通知测试",
+            content="hello",
+            timeout_seconds=5,
+        )
+
+        self.assertEqual(request.channel, "ntfy")
+
    def test_validate_returns_user_facing_model_message_without_internal_env_key_name(self) -> None:
        validation = self.service.validate(
            items=[
--- a/tests/test_system_config_service.py
+++ b/tests/test_system_config_service.py
@@ -189,6 +189,18 @@ class SystemConfigServiceTestCase(unittest.TestCase):
        astrbot_complete = next(check for check in status["checks"] if check["key"] == "notification")
        self.assertEqual(astrbot_complete["status"], "configured")

+        self._rewrite_env(*base_lines, "NTFY_URL=https://ntfy.sh/dsa-topic")
+        with patch.dict(os.environ, {}, clear=True):
+            status = self.service.get_setup_status()
+        ntfy_complete = next(check for check in status["checks"] if check["key"] == "notification")
+        self.assertEqual(ntfy_complete["status"], "configured")
+
+        self._rewrite_env(*base_lines, "NTFY_URL=https://ntfy.sh")
+        with patch.dict(os.environ, {}, clear=True):
+            status = self.service.get_setup_status()
+        ntfy_without_topic = next(check for check in status["checks"] if check["key"] == "notification")
+        self.assertEqual(ntfy_without_topic["status"], "optional")
+
    def test_get_setup_status_uses_runtime_env_without_reloading_singletons(self) -> None:
        self._rewrite_env("")

@@ -351,9 +363,22 @@ class SystemConfigServiceTestCase(unittest.TestCase):
        self.assertFalse(validation["valid"])
        self.assertTrue(any(issue["code"] == "invalid_url" for issue in validation["issues"]))

+    def test_validate_reports_ntfy_url_without_topic(self) -> None:
+        validation = self.service.validate(
+            items=[{"key": "NTFY_URL", "value": "https://ntfy.sh"}]
+        )
+
+        self.assertFalse(validation["valid"])
+        self.assertTrue(
+            any(
+                issue["key"] == "NTFY_URL" and issue["code"] == "invalid_ntfy_url"
+                for issue in validation["issues"]
+            )
+        )
+
    def test_validate_reports_invalid_notification_route_channel(self) -> None:
        validation = self.service.validate(
-            items=[{"key": "NOTIFICATION_REPORT_CHANNELS", "value": "wechat,ntfy,email"}]
+            items=[{"key": "NOTIFICATION_REPORT_CHANNELS", "value": "wechat,not-a-channel,email"}]
        )
        self.assertFalse(validation["valid"])
        self.assertTrue(
@@ -1053,6 +1078,48 @@ class SystemConfigServiceTestCase(unittest.TestCase):
        self.assertNotIn("access_token=first", str(payload))
        self.assertNotIn("token=second", str(payload))

+    @patch("src.notification_sender.ntfy_sender.requests.post")
+    def test_test_notification_channel_supports_ntfy_and_masks_topic_target(self, mock_post) -> None:
+        mock_post.return_value = self._mock_http_response(200)
+
+        with self._notification_test_env():
+            payload = self.service.test_notification_channel(
+                channel="ntfy",
+                items=[
+                    {"key": "NTFY_URL", "value": "https://ntfy.sh/private-topic"},
+                    {"key": "NTFY_TOKEN", "value": "secret-token"},
+                ],
+                title="Test title",
+                content="hello",
+                timeout_seconds=4,
+            )
+
+        self.assertTrue(payload["success"])
+        self.assertEqual(mock_post.call_args.args[0], "https://ntfy.sh")
+        self.assertEqual(mock_post.call_args.kwargs["json"]["topic"], "private-topic")
+        self.assertEqual(mock_post.call_args.kwargs["headers"]["Authorization"], "Bearer secret-token")
+        self.assertEqual(mock_post.call_args.kwargs["timeout"], 4)
+        self.assertIn("https://ntfy.sh/***", payload["attempts"][0]["target"])
+        self.assertNotIn("private-topic", str(payload))
+        self.assertNotIn("NTFY_URL", self.env_path.read_text(encoding="utf-8"))
+
+    @patch("src.notification_sender.ntfy_sender.requests.post")
+    def test_test_notification_channel_rejects_ntfy_url_without_topic(self, mock_post) -> None:
+        with self._notification_test_env():
+            payload = self.service.test_notification_channel(
+                channel="ntfy",
+                items=[{"key": "NTFY_URL", "value": "https://ntfy.sh"}],
+                title="Test title",
+                content="hello",
+                timeout_seconds=4,
+            )
+
+        self.assertFalse(payload["success"])
+        self.assertEqual(payload["error_code"], "config_invalid")
+        self.assertEqual(payload["stage"], "config_validation")
+        self.assertIn("NTFY_URL", payload["message"])
+        mock_post.assert_not_called()
+
    @patch(
        "src.notification_sender.WechatSender.send_to_wechat",
        side_effect=requests.exceptions.Timeout(