feat(cninfo): 新增巨潮公告检索 — 三层接入（API/CLI/Web），独立数据源

README.md

@@ -132,6 +132,16 @@ easy-tdx server-info --table
 easy-tdx symbol-info SZ 000001 --table
 ```
 
+### 公告检索（巨潮资讯网）
+
+```bash
+easy-tdx announcement 688017                          # 默认 30 条，JSON 输出
+easy-tdx announcement 600519 --count 50 --page 2      # 翻页
+easy-tdx announcement 000001 --table                  # 表格输出
+```
+
+> 独立数据源（巨潮资讯网），无需连接 TDX 行情服务器即可使用。
+
 ### 技术指标
 
 ```bash
@@ -855,6 +865,9 @@ curl "http://localhost:8000/api/v1/mac/symbol-info?market=SZ&code=000001"
 # 服务器交易时段信息
 curl "http://localhost:8000/api/v1/mac/server-info"
 
+# ── 公告检索（巨潮资讯网，独立数据源）──
+curl "http://localhost:8000/api/v1/announcements?code=688017&count=30&page=1"
+
 # ── 排行 / 竞价 / 异动 ──
 # 全 A 涨幅排行前 20
 curl "http://localhost:8000/api/v1/mac/quote-list?category=A&count=20&sort_type=CHANGE_PCT"
@@ -1515,6 +1528,16 @@ ruff format --check src/ tests/                              # format check
 
 ## Changelog
 
+### 1.13.0 (2026-06-14)
+
+**新增巨潮公告检索** — 三层接入（编程 API / CLI / Web API），独立数据源，无需连接 TDX 行情服务器。
+
+- 新模块 `easy_tdx.cninfo`：`CninfoClient().get_announcements(code, count=, page=)` 返回 `DataFrame[title, type, date, url]`
+- CLI：`easy-tdx announcement 688017 [--count N --page N --table]`
+- Web：`GET /api/v1/announcements?code=&count=&page=`
+- 标准库 urllib 实现，零新依赖
+- 沿用 #19 修复的 orgId 动态映射 + 三段硬编码 fallback（保证 601xxx 段可查）
+
 ### 1.12.0 (2026-06-14)
 
 **新增 4 个技术指标（30 → 34）** — 按"语义空白"补齐三类现有指标库缺失的维度：止损位、机构成本价、趋势启动时机。均为纯 numpy 实现，零新依赖。

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "easy-tdx"
-version = "1.12.0"
+version = "1.13.0"
 description = "通达信 TCP 协议行情数据客户端，支持在线行情、离线数据读取与写入同步"
 readme = "README.md"
 requires-python = ">=3.10"

src/easy_tdx/cli/__init__.py

@@ -7,6 +7,7 @@ import click
 from ..backtest.cli import backtest, portfolio
 from ..screen.cli import screen
 from .cmd_admin import ping, version
+from .cmd_announcement import announcement
 from .cmd_auction import auction
 from .cmd_board import (
     belong_board,
@@ -59,6 +60,7 @@ def cli() -> None:
 
 cli.add_command(ping)
 cli.add_command(version)
+cli.add_command(announcement)
 cli.add_command(kline)
 cli.add_command(quote)
 cli.add_command(quote_list)

src/easy_tdx/cli/cmd_announcement.py

@@ -0,0 +1,36 @@
+"""公告检索命令（巨潮资讯网数据源，无需 TDX 服务器）。"""
+
+from __future__ import annotations
+
+import click
+
+
+@click.command("announcement")
+@click.argument("code")
+@click.option("--count", default=30, type=int, help="每页数量")
+@click.option("--page", default=1, type=int, help="页码（1 起始）")
+@click.option("--table", "use_table", is_flag=True, help="表格输出")
+@click.option("--output", "output_fmt", type=click.Choice(["json", "table", "csv"]), default="json")
+def announcement(code: str, count: int, page: int, use_table: bool, output_fmt: str) -> None:
+    """检索公司公告（巨潮资讯网，独立数据源，无需连接 TDX）。
+
+    \b
+    示例：
+
+      easy-tdx announcement 688017
+
+      easy-tdx announcement 600519 --count 50 --page 2
+
+      easy-tdx announcement 000001 --table
+    """
+    from ..cninfo import CninfoClient, CninfoError
+    from .output import print_error, print_output
+
+    fmt = "table" if use_table else output_fmt
+    client = CninfoClient()
+    try:
+        df = client.get_announcements(code, count=count, page=page)
+    except CninfoError as e:
+        print_error(str(e))
+        raise SystemExit(1) from e
+    print_output(df, fmt)

src/easy_tdx/cninfo/__init__.py

@@ -0,0 +1,18 @@
+"""巨潮资讯网（cninfo）公告检索 —— 独立于 TDX 协议的 HTTP 数据源。
+
+零额外依赖（标准库 urllib），无需连接 TDX 服务器即可使用。
+
+用法::
+
+    from easy_tdx.cninfo import CninfoClient
+
+    df = CninfoClient().get_announcements("688017", count=30)
+    # → DataFrame[title, type, date, url]
+"""
+
+from __future__ import annotations
+
+from .client import CninfoClient
+from .models import Announcement, CninfoError
+
+__all__ = ["CninfoClient", "Announcement", "CninfoError"]

src/easy_tdx/cninfo/client.py

@@ -0,0 +1,193 @@
+"""巨潮资讯网（cninfo）公告检索客户端。
+
+独立于 TDX 协议的 HTTP 数据源（标准库 urllib，零额外依赖）。
+公开方法返回 ``pd.DataFrame``，遵循项目 ``get_*`` 约定。
+
+参考实现说明（来自 #19 修复）：
+
+- 巨潮 ``orgId`` 并非统一的 ``gssx0{code}`` 格式（如 601318→9900002221、
+  601398→jjxt0000019、688017→9900041602），硬编码会导致大量股票
+  （尤其 601xxx 段）返回 ``totalAnnouncement=0``、查不到公告。
+- 优先动态查官方映射表，查不到再回退硬编码规则。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import datetime
+from typing import Any
+from urllib import parse
+from urllib import request as urlrequest
+
+import pandas as pd
+
+from .models import Announcement, CninfoError
+
+logger = logging.getLogger(__name__)
+
+# 巨潮公告检索请求固定头（Referer/Origin 必填，否则被反爬拦截）
+_UA = (
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
+)
+_STOCK_MAP_URL = "http://www.cninfo.com.cn/new/data/szse_stock.json"
+_QUERY_URL = "https://www.cninfo.com.cn/new/hisAnnouncement/query"
+_DETAIL_URL = "https://www.cninfo.com.cn/new/disclosure/detail?annoId="
+
+# 模块级 orgId 映射缓存：首次拉取后全程复用（Cpython dict 读写原子，
+# 并发下最坏多发一次请求，可接受）
+_ORGID_MAP: dict[str, str] = {}
+
+
+def _ts_to_date(ts: Any) -> str:
+    """巨潮 ``announcementTime`` 返回 Unix 毫秒整数，转 ``YYYY-MM-DD``。"""
+    if isinstance(ts, (int, float)):
+        return datetime.fromtimestamp(ts / 1000).strftime("%Y-%m-%d")
+    return str(ts)[:10] if ts else ""
+
+
+def _http_get_json(url: str, timeout: float = 15.0) -> Any:
+    """GET JSON（stdlib urllib，monkeypatch 点）。"""
+    req = urlrequest.Request(url, headers={"User-Agent": _UA})
+    with urlrequest.urlopen(req, timeout=timeout) as resp:
+        return json.loads(resp.read().decode("utf-8"))
+
+
+def _http_post_form(url: str, payload: dict[str, str], timeout: float = 15.0) -> Any:
+    """POST x-www-form-urlencoded，返回 JSON（stdlib urllib，monkeypatch 点）。"""
+    data = parse.urlencode(payload).encode("utf-8")
+    req = urlrequest.Request(
+        url,
+        data=data,
+        headers={
+            "User-Agent": _UA,
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Referer": "https://www.cninfo.com.cn/new/disclosure",
+            "Origin": "https://www.cninfo.com.cn",
+        },
+        method="POST",
+    )
+    with urlrequest.urlopen(req, timeout=timeout) as resp:
+        return json.loads(resp.read().decode("utf-8"))
+
+
+class CninfoClient:
+    """巨潮公告检索客户端（无状态 HTTP，无需 connect/close）。
+
+    用法::
+
+        from easy_tdx.cninfo import CninfoClient
+
+        df = CninfoClient().get_announcements("688017", count=30)
+        # → DataFrame[title, type, date, url]
+    """
+
+    def __init__(self, *, timeout: float = 15.0) -> None:
+        self.timeout = timeout
+
+    # ------------------------------------------------------------------
+    # orgId 解析（#19 修复逻辑：动态表 + 三段硬编码 fallback）
+    # ------------------------------------------------------------------
+
+    def _fetch_stock_map(self) -> dict[str, str]:
+        """拉取官方 szse_stock.json，构建 code→orgId 映射。"""
+        try:
+            d = _http_get_json(_STOCK_MAP_URL, timeout=self.timeout)
+            stock_list = d.get("stockList", []) if isinstance(d, dict) else []
+            return {s["code"]: s["orgId"] for s in stock_list}
+        except Exception as e:  # noqa: BLE001 — 拉取失败需 graceful fallback
+            logger.warning("巨潮 orgId 映射表拉取失败，回退硬编码规则: %s", e)
+            return {}
+
+    def _resolve_orgid(self, code: str) -> str:
+        """查股票真实 orgId，动态表优先，查不到回退硬编码规则。"""
+        global _ORGID_MAP
+        if not _ORGID_MAP:
+            fetched = self._fetch_stock_map()
+            # 只在确实取到数据时写入缓存；空结果保留以便下次重试
+            if fetched:
+                _ORGID_MAP.update(fetched)
+        org = _ORGID_MAP.get(code)
+        if org:
+            return org
+        # fallback：老格式（仅部分老股票如 600519/600036 适用）
+        if code.startswith("6"):
+            return f"gssh0{code}"
+        if code.startswith("8") or code.startswith("4"):
+            return f"gsbj0{code}"
+        return f"gssz0{code}"
+
+    # ------------------------------------------------------------------
+    # 公开 API
+    # ------------------------------------------------------------------
+
+    def get_announcements(
+        self,
+        code: str,
+        *,
+        count: int = 30,
+        page: int = 1,
+    ) -> pd.DataFrame:
+        """检索指定股票的公告列表。
+
+        Args:
+            code: 6 位股票代码（不含市场前缀），如 ``688017``。
+            count: 每页数量（即 pageSize）。
+            page: 页码（1 起始）。
+
+        Returns:
+            ``DataFrame[title, type, date, url]``，按服务器返回顺序（最新在前）。
+            无结果时返回空 DataFrame（含列名）。
+        """
+        rows = self._query_announcements(code, count=count, page=page)
+        if not rows:
+            return pd.DataFrame(columns=["title", "type", "date", "url"])
+        return pd.DataFrame([r.__dict__ for r in rows])
+
+    def _query_announcements(self, code: str, *, count: int, page: int) -> list[Announcement]:
+        """POST 公告检索接口，解析为 Announcement 列表。
+
+        整个 HTTP + 解析过程统一捕获异常并转为 ``CninfoError``，避免
+        ``announcementTime`` 等字段畸形时 ``_ts_to_date`` 抛出未捕获异常。
+        """
+        org_id = self._resolve_orgid(code)
+        payload = {
+            "stock": f"{code},{org_id}",
+            "tabName": "fulltext",
+            "pageSize": str(count),
+            "pageNum": str(page),
+            "column": "",
+            "category": "",
+            "plate": "",
+            "seDate": "",
+            "searchkey": "",
+            "secid": "",
+            "sortName": "",
+            "sortType": "",
+            "isHLtitle": "true",
+        }
+        try:
+            d = _http_post_form(_QUERY_URL, payload, timeout=self.timeout)
+            items = d.get("announcements", []) if isinstance(d, dict) else None
+            if not items:
+                return []
+
+            result: list[Announcement] = []
+            for item in items:
+                if not isinstance(item, dict):
+                    continue
+                anno_id = item.get("announcementId", "")
+                result.append(
+                    Announcement(
+                        title=item.get("announcementTitle", ""),
+                        type=item.get("announcementTypeName", ""),
+                        date=_ts_to_date(item.get("announcementTime")),
+                        url=f"{_DETAIL_URL}{anno_id}",
+                    )
+                )
+            return result
+        except CninfoError:
+            raise
+        except Exception as e:  # noqa: BLE001 — HTTP 失败 / JSON 解析 / 日期转换统一转领域异常
+            raise CninfoError(f"巨潮公告检索失败: {e}") from e

src/easy_tdx/cninfo/models.py

@@ -0,0 +1,24 @@
+"""巨潮资讯网（cninfo）数据模型。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from easy_tdx.exceptions import TdxError
+
+
+@dataclass(frozen=True)
+class Announcement:
+    """单条公告记录。
+
+    巨潮公告检索接口返回的标准化结构。
+    """
+
+    title: str
+    type: str
+    date: str  # YYYY-MM-DD
+    url: str
+
+
+class CninfoError(TdxError):
+    """巨潮数据请求或解析失败。"""

src/easy_tdx/web/app.py

@@ -140,6 +140,7 @@ def _create_app(
     register_exception_handlers(app)
 
     # Mount routers
+    from easy_tdx.web.routers.announcement import router as announcement_router
     from easy_tdx.web.routers.bars import router as bars_router
     from easy_tdx.web.routers.block import router as block_router
     from easy_tdx.web.routers.board_mac import router as board_mac_router
@@ -166,5 +167,7 @@ def _create_app(
     app.include_router(ex_market_router, prefix="/api/v1")
     # 技术指标路由
     app.include_router(indicator_router, prefix="/api/v1")
+    # 公告检索路由（巨潮资讯网，独立数据源）
+    app.include_router(announcement_router, prefix="/api/v1")
 
     return app

src/easy_tdx/web/routers/announcement.py

@@ -0,0 +1,37 @@
+"""公告检索路由（巨潮资讯网，独立数据源，不依赖 TDX 服务器）。"""
+
+from __future__ import annotations
+
+import asyncio
+
+from fastapi import APIRouter, Query
+
+from easy_tdx.web.schemas import DataFrameResponse
+
+router = APIRouter(tags=["announcement"])
+
+
+@router.get("/announcements", response_model=DataFrameResponse)
+async def announcements(
+    code: str = Query(..., min_length=6, max_length=6, description="6位股票代码"),
+    count: int = Query(30, ge=1, le=100, description="每页数量"),
+    page: int = Query(1, ge=1, description="页码（1 起始）"),
+) -> DataFrameResponse:
+    """检索公司公告列表（巨潮资讯网，独立于 TDX 行情服务器）。
+
+    TDX 行情不可用时本接口仍可调用。
+    """
+    from easy_tdx.cninfo import CninfoClient, CninfoError
+
+    client = CninfoClient()
+
+    def _fetch() -> DataFrameResponse:
+        df = client.get_announcements(code, count=count, page=page)
+        return DataFrameResponse.from_dataframe(df)
+
+    try:
+        return await asyncio.to_thread(_fetch)
+    except CninfoError as e:
+        from fastapi import HTTPException
+
+        raise HTTPException(status_code=503, detail=str(e)) from e

tests/unit/test_cninfo.py

@@ -0,0 +1,456 @@
+"""巨潮（cninfo）模块离线测试 —— mock HTTP，零网络依赖。
+
+覆盖：日期转换、orgId 解析（动态表/三段 fallback）、公告解析、分页、
+错误转换、模块导出。
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+import pandas as pd
+import pytest
+
+# ---------------------------------------------------------------------------
+# 导出
+# ---------------------------------------------------------------------------
+
+
+def test_public_exports() -> None:
+    """模块应导出 CninfoClient / Announcement / CninfoError。"""
+    from easy_tdx import cninfo
+
+    assert hasattr(cninfo, "CninfoClient")
+    assert hasattr(cninfo, "Announcement")
+    assert hasattr(cninfo, "CninfoError")
+
+
+def test_announcement_is_frozen_dataclass() -> None:
+    """Announcement 应为 frozen dataclass，含 title/type/date/url。"""
+    from easy_tdx.cninfo import Announcement
+
+    a = Announcement(title="t", type="ty", date="2026-06-14", url="http://x")
+    assert a.title == "t"
+    assert a.type == "ty"
+    assert a.date == "2026-06-14"
+    assert a.url == "http://x"
+    # frozen
+    with pytest.raises(Exception):
+        a.title = "mutated"  # type: ignore[misc]
+
+
+def test_cninfo_error_is_exception() -> None:
+    from easy_tdx.cninfo import CninfoError
+    from easy_tdx.exceptions import TdxError
+
+    assert issubclass(CninfoError, Exception)
+    # 回归 #1：CninfoError 必须继承 TdxError，保证全局 except TdxError 覆盖
+    assert issubclass(CninfoError, TdxError)
+
+
+# ---------------------------------------------------------------------------
+# 日期转换
+# ---------------------------------------------------------------------------
+
+
+def test_ts_to_date_from_millis() -> None:
+    """Unix 毫秒整数应转为 YYYY-MM-DD。"""
+    from easy_tdx.cninfo.client import _ts_to_date
+
+    # 1718323200000 ms = 2024-06-14 00:00:00 UTC ≈ 当地日期
+    assert _ts_to_date(1718323200000)  # 非空字符串，长度 10
+    assert len(_ts_to_date(1718323200000)) == 10
+
+
+def test_ts_to_date_from_string() -> None:
+    """字符串输入应取前 10 字符。"""
+    from easy_tdx.cninfo.client import _ts_to_date
+
+    assert _ts_to_date("2026-06-14T08:00:00") == "2026-06-14"
+
+
+def test_ts_to_date_empty() -> None:
+    from easy_tdx.cninfo.client import _ts_to_date
+
+    assert _ts_to_date("") == ""
+    assert _ts_to_date(None) == ""
+
+
+# ---------------------------------------------------------------------------
+# orgId 解析（动态表 + 三段 fallback）
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def reset_orgid_cache() -> Any:
+    """每个测试前后清空 orgId 缓存，保证隔离。"""
+    import easy_tdx.cninfo.client as mod
+
+    mod._ORGID_MAP.clear()
+    yield
+    mod._ORGID_MAP.clear()
+
+
+def _patch_stock_map(monkeypatch: pytest.MonkeyPatch, mapping: dict[str, str]) -> None:
+    """让 _fetch_stock_map 返回给定映射（不触网）。"""
+    monkeypatch.setattr(
+        "easy_tdx.cninfo.client._http_get_json",
+        lambda url, timeout=15.0: {
+            "stockList": [{"code": c, "orgId": o} for c, o in mapping.items()]
+        },
+    )
+
+
+def test_resolve_orgid_from_dynamic_map(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """动态表命中应返回表中 orgId。"""
+    from easy_tdx.cninfo import CninfoClient
+
+    _patch_stock_map(monkeypatch, {"688017": "9900041602", "601318": "9900002221"})
+    client = CninfoClient()
+    assert client._resolve_orgid("688017") == "9900041602"
+    assert client._resolve_orgid("601318") == "9900002221"
+
+
+def test_resolve_orgid_fallback_6_prefix(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """动态表无此 code 且 6 开头 → gssh0{code}。"""
+    from easy_tdx.cninfo import CninfoClient
+
+    _patch_stock_map(monkeypatch, {})
+    client = CninfoClient()
+    assert client._resolve_orgid("600519") == "gssh0600519"
+
+
+def test_resolve_orgid_fallback_8_prefix(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """北交所 8/4 开头 → gsbj0{code}。"""
+    from easy_tdx.cninfo import CninfoClient
+
+    _patch_stock_map(monkeypatch, {})
+    client = CninfoClient()
+    assert client._resolve_orgid("830799") == "gsbj0830799"
+    assert client._resolve_orgid("430047") == "gsbj0430047"
+
+
+def test_resolve_orgid_fallback_sz_default(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """其他前缀（深圳）→ gssz0{code}。"""
+    from easy_tdx.cninfo import CninfoClient
+
+    _patch_stock_map(monkeypatch, {})
+    client = CninfoClient()
+    assert client._resolve_orgid("000001") == "gssz0000001"
+
+
+def test_resolve_orgid_empty_map_does_not_cache(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """映射表为空时不写入缓存，下次仍会重试（避免永久 fallback）。"""
+    import easy_tdx.cninfo.client as mod
+    from easy_tdx.cninfo import CninfoClient
+
+    _patch_stock_map(monkeypatch, {})
+    client = CninfoClient()
+    client._resolve_orgid("600519")
+    assert mod._ORGID_MAP == {}
+
+
+def test_resolve_orgid_fetch_failure_fallback(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """映射表拉取异常应 graceful fallback 到硬编码规则，不抛错。"""
+    from easy_tdx.cninfo import CninfoClient
+
+    def _boom(url: str, timeout: float = 15.0) -> Any:
+        raise OSError("network down")
+
+    monkeypatch.setattr("easy_tdx.cninfo.client._http_get_json", _boom)
+    client = CninfoClient()
+    # 不抛错，回退 SH 规则
+    assert client._resolve_orgid("600519") == "gssh0600519"
+
+
+def test_resolve_orgid_cache_reused(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """第二次调用不应再次拉取映射表（缓存命中）。"""
+    call_count = {"n": 0}
+
+    def _fake(url: str, timeout: float = 15.0) -> Any:
+        call_count["n"] += 1
+        return {"stockList": [{"code": "688017", "orgId": "9900041602"}]}
+
+    monkeypatch.setattr("easy_tdx.cninfo.client._http_get_json", _fake)
+    from easy_tdx.cninfo import CninfoClient
+
+    client = CninfoClient()
+    client._resolve_orgid("688017")
+    client._resolve_orgid("688017")
+    assert call_count["n"] == 1
+
+
+# ---------------------------------------------------------------------------
+# 公告查询与解析
+# ---------------------------------------------------------------------------
+
+
+_QUERY_RESPONSE: dict[str, Any] = {
+    "announcements": [
+        {
+            "announcementTitle": "关于召开2025年年度股东大会的通知",
+            "announcementTypeName": "股东大会",
+            "announcementTime": 1749859200000,
+            "announcementId": "abc123",
+        },
+        {
+            "announcementTitle": "2024年年度报告",
+            "announcementTypeName": "定期报告",
+            "announcementTime": 1740614400000,
+            "announcementId": "def456",
+        },
+    ],
+    "totalAnnouncement": 2,
+}
+
+
+def test_get_announcements_returns_dataframe(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """应返回 DataFrame[title, type, date, url]。"""
+    _patch_stock_map(monkeypatch, {"688017": "9900041602"})
+    monkeypatch.setattr(
+        "easy_tdx.cninfo.client._http_post_form",
+        lambda url, payload, timeout=15.0: _QUERY_RESPONSE,
+    )
+    from easy_tdx.cninfo import CninfoClient
+
+    df = CninfoClient().get_announcements("688017", count=30, page=1)
+    assert isinstance(df, pd.DataFrame)
+    assert list(df.columns) == ["title", "type", "date", "url"]
+    assert len(df) == 2
+    assert df.iloc[0]["title"] == "关于召开2025年年度股东大会的通知"
+    assert df.iloc[0]["type"] == "股东大会"
+    assert len(df.iloc[0]["date"]) == 10  # YYYY-MM-DD
+    assert df.iloc[0]["url"].endswith("abc123")
+    assert "cninfo.com.cn" in df.iloc[0]["url"]
+
+
+def test_get_announcements_empty(monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any) -> None:
+    """无公告应返回带列名的空 DataFrame。"""
+    _patch_stock_map(monkeypatch, {"688017": "9900041602"})
+    monkeypatch.setattr(
+        "easy_tdx.cninfo.client._http_post_form",
+        lambda url, payload, timeout=15.0: {"announcements": []},
+    )
+    from easy_tdx.cninfo import CninfoClient
+
+    df = CninfoClient().get_announcements("688017")
+    assert isinstance(df, pd.DataFrame)
+    assert df.empty
+    assert list(df.columns) == ["title", "type", "date", "url"]
+
+
+def test_get_announcements_missing_key(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """响应缺少 announcements 键应视为空结果。"""
+    _patch_stock_map(monkeypatch, {"688017": "9900041602"})
+    monkeypatch.setattr(
+        "easy_tdx.cninfo.client._http_post_form",
+        lambda url, payload, timeout=15.0: {"totalAnnouncement": 0},
+    )
+    from easy_tdx.cninfo import CninfoClient
+
+    df = CninfoClient().get_announcements("688017")
+    assert df.empty
+
+
+def test_get_announcements_request_failure_raises_cninfo_error(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """HTTP 异常应转为 CninfoError。"""
+    _patch_stock_map(monkeypatch, {"688017": "9900041602"})
+
+    def _boom(url: str, payload: dict[str, str], timeout: float = 15.0) -> Any:
+        raise OSError("connection refused")
+
+    monkeypatch.setattr("easy_tdx.cninfo.client._http_post_form", _boom)
+    from easy_tdx.cninfo import CninfoClient, CninfoError
+
+    with pytest.raises(CninfoError):
+        CninfoClient().get_announcements("688017")
+
+
+def test_get_announcements_pagination(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """count/page 应正确传入 payload。"""
+    _patch_stock_map(monkeypatch, {"688017": "9900041602"})
+    captured: dict[str, Any] = {}
+
+    def _capture(url: str, payload: dict[str, str], timeout: float = 15.0) -> Any:
+        captured.update(payload)
+        return {"announcements": []}
+
+    monkeypatch.setattr("easy_tdx.cninfo.client._http_post_form", _capture)
+    from easy_tdx.cninfo import CninfoClient
+
+    CninfoClient().get_announcements("688017", count=50, page=3)
+    assert captured["pageSize"] == "50"
+    assert captured["pageNum"] == "3"
+    assert captured["stock"] == "688017,9900041602"
+    assert captured["tabName"] == "fulltext"
+
+
+def test_get_announcements_uses_fallback_orgid(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """601xxx 段动态表未命中时用 gssh0 fallback，stock 字段含该 orgId。"""
+    _patch_stock_map(monkeypatch, {})  # 空表 → fallback
+    captured: dict[str, Any] = {}
+
+    def _capture(url: str, payload: dict[str, str], timeout: float = 15.0) -> Any:
+        captured.update(payload)
+        return {"announcements": []}
+
+    monkeypatch.setattr("easy_tdx.cninfo.client._http_post_form", _capture)
+    from easy_tdx.cninfo import CninfoClient
+
+    CninfoClient().get_announcements("601318")
+    assert captured["stock"] == "601318,gssh0601318"
+
+
+def test_get_announcements_skips_non_dict_items(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """announcements 列表中混入非 dict 元素应被跳过。"""
+    _patch_stock_map(monkeypatch, {"688017": "9900041602"})
+    monkeypatch.setattr(
+        "easy_tdx.cninfo.client._http_post_form",
+        lambda url, payload, timeout=15.0: {
+            "announcements": [
+                "not a dict",
+                {
+                    "announcementTitle": "ok",
+                    "announcementTime": 1749859200000,
+                    "announcementId": "x",
+                },
+            ]
+        },
+    )
+    from easy_tdx.cninfo import CninfoClient
+
+    df = CninfoClient().get_announcements("688017")
+    assert len(df) == 1
+    assert df.iloc[0]["title"] == "ok"
+
+
+def test_get_announcements_response_not_dict(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """响应非 dict（如 list）应视为空结果，不抛错。"""
+    _patch_stock_map(monkeypatch, {"688017": "9900041602"})
+    monkeypatch.setattr(
+        "easy_tdx.cninfo.client._http_post_form",
+        lambda url, payload, timeout=15.0: ["unexpected", "list"],
+    )
+    from easy_tdx.cninfo import CninfoClient
+
+    df = CninfoClient().get_announcements("688017")
+    assert df.empty
+
+
+def test_get_announcements_malformed_timestamp_wrapped_as_cninfo_error(
+    monkeypatch: pytest.MonkeyPatch, reset_orgid_cache: Any
+) -> None:
+    """announcementTime 畸形（导致 fromtimestamp 溢出）应转 CninfoError，不裸抛。
+
+    回归 #2：修复前 _ts_to_date 在 try 块外，畸形时间戳会抛 OverflowError/
+    ValueError 裸异常；修复后整个解析路径统一转 CninfoError。
+    """
+    _patch_stock_map(monkeypatch, {"688017": "9900041602"})
+    monkeypatch.setattr(
+        "easy_tdx.cninfo.client._http_post_form",
+        lambda url, payload, timeout=15.0: {
+            "announcements": [
+                {"announcementTitle": "x", "announcementTime": 10**30, "announcementId": "y"}
+            ]
+        },
+    )
+    from easy_tdx.cninfo import CninfoClient, CninfoError
+
+    with pytest.raises(CninfoError):
+        CninfoClient().get_announcements("688017")
+
+
+# ---------------------------------------------------------------------------
+# urllib helper 烟雾测试（不触网，仅验证 JSON 解码路径）
+# ---------------------------------------------------------------------------
+
+
+def test_http_post_form_urlencoded_body(monkeypatch: pytest.MonkeyPatch) -> None:
+    """_http_post_form 应以 application/x-www-form-urlencoded 发送。"""
+    import easy_tdx.cninfo.client as mod
+
+    captured: dict[str, Any] = {}
+
+    class _FakeResp:
+        def __init__(self, body: bytes) -> None:
+            self._body = body
+
+        def read(self) -> bytes:
+            return self._body
+
+        def __enter__(self) -> _FakeResp:
+            return self
+
+        def __exit__(self, *args: Any) -> None:
+            pass
+
+    def _fake_urlopen(req: Any, timeout: float = 15.0) -> _FakeResp:
+        captured["data"] = req.data
+        captured["headers"] = {k: v for k, v in req.header_items()}
+        captured["method"] = req.get_method()
+        return _FakeResp(json.dumps({"ok": True}).encode("utf-8"))
+
+    monkeypatch.setattr(mod.urlrequest, "urlopen", _fake_urlopen)
+    result = mod._http_post_form("https://example.com/api", {"pageNum": "2", "pageSize": "30"})
+    assert result == {"ok": True}
+    assert b"pageNum=2" in captured["data"]
+    assert b"pageSize=30" in captured["data"]
+    assert captured["method"] == "POST"
+    headers = {k.lower(): v for k, v in captured["headers"].items()}
+    assert "cninfo.com.cn" in headers.get("referer", "")
+    assert "x-www-form-urlencoded" in headers.get("content-type", "")
+
+
+def test_http_get_json_headers(monkeypatch: pytest.MonkeyPatch) -> None:
+    """_http_get_json 应携带 User-Agent。"""
+    import easy_tdx.cninfo.client as mod
+
+    captured: dict[str, Any] = {}
+
+    class _FakeResp:
+        def read(self) -> bytes:
+            return json.dumps({"ok": 1}).encode("utf-8")
+
+        def __enter__(self) -> _FakeResp:
+            return self
+
+        def __exit__(self, *args: Any) -> None:
+            pass
+
+    def _fake_urlopen(req: Any, timeout: float = 15.0) -> _FakeResp:
+        captured["headers"] = {k: v for k, v in req.header_items()}
+        return _FakeResp()
+
+    monkeypatch.setattr(mod.urlrequest, "urlopen", _fake_urlopen)
+    assert mod._http_get_json("https://example.com/x.json") == {"ok": 1}
+    headers = {k.lower(): v for k, v in captured["headers"].items()}
+    assert "user-agent" in headers

tests/unit/test_web_api.py

@@ -268,6 +268,7 @@ def test_full_app_routes_registered():
         "/api/v1/xdxr",
         "/api/v1/block",
         "/api/v1/chanlun",
+        "/api/v1/announcements",
         "/ws/realtime",
     ]
     for prefix in expected_prefixes: