diff --git a/README.md b/README.md index 7b94906..fe84595 100644 --- a/README.md +++ b/README.md @@ -1051,6 +1051,10 @@ curl "http://localhost:8000/api/v1/board-mac/belong?market=SZ&code=000001" # 个股所属行业及行业今日涨跌幅 curl "http://localhost:8000/api/v1/stock/industry?market=SZ&code=000001" +# 同花顺公开网页的行业/概念板块（非官方 API，避免高频调用） +# 688126 本地验证：最多返回 3 个概念板块，完整数量见 concept_total +curl "http://127.0.0.1:8001/api/v1/ths/stock/associations?code=688126&concept_limit=3" + # 板块摘要（含主力净流入、涨跌家数） curl "http://localhost:8000/api/v1/board-mac/summary?board_symbol=881001" diff --git a/docs/api_reference.md b/docs/api_reference.md index db6a573..abdc478 100644 --- a/docs/api_reference.md +++ b/docs/api_reference.md @@ -375,6 +375,46 @@ curl "http://localhost:8000/api/v1/stock/industry?market=SH&code=600519" --- +## Web API：同花顺网页关联板块 + +### GET `/api/v1/ths/stock/associations` + +按同花顺公开 F10 网页的分类，返回股票的三级行业和概念板块，并从各板块公开详情页补充 +当日涨跌幅和涨幅排行首只成分股。适合实现类似同花顺“盘口”中的“行业板块”“概念板块”区域。 + +> 这是网页解析接口，不是同花顺官方开放 API。请遵守同花顺网站规则，勿用于高频抓取、 +> 商业再分发或作为交易决策的唯一数据源。网页未公开或暂时不可读取的行情字段会返回 `null`； +> 接口不会伪造同花顺 App 的“最相关”“对应人气股”等私有排序。 + +**请求参数**： + +| 参数 | 类型 | 必填 | 说明 | +|------|------|------|------| +| `code` | `string` | 是 | 6 位 A 股代码，如 `688126` | +| `concept_limit` | `integer` | 否 | 返回概念板块数量，默认 `10`，最大 `30`；总数见 `concept_total` | + +**本地验证示例（688126）**： + +```bash +curl "http://127.0.0.1:8001/api/v1/ths/stock/associations?code=688126&concept_limit=3" +``` + +**响应字段**： + +| 字段 | 说明 | +|------|------| +| `source` | 固定为 `ths_web`，明确表示来自同花顺公开网页解析 | +| `industries` | 三级行业数组；`level` 为层级，`board_code`、`change_pct`、`leader` 可能为空 | +| `concepts` | 按 `concept_limit` 返回的概念板块数组 | +| `concept_total` | 该股票在同花顺 F10 中的概念板块总数 | +| `leader` | 公开详情页涨幅排行中的首只成分股，非同花顺 App 的“对应人气股” | +| `leader_codes` | F10 页面若公开了板块关联股票代码则返回，否则为空数组 | + +行业归属每次请求重新读取；行业和概念详情行情只在当日内短时缓存 30 秒，并且每次最多并发读取 +4 个板块详情，以降低对来源网页的压力。 + +--- + ## 资金流向 ### get_fund_flow diff --git a/pyproject.toml b/pyproject.toml index 8b9b4bf..78f4c2c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,7 @@ easy-tdx = "easy_tdx.cli:cli" # cli/__init__.py exposes the click group [project.optional-dependencies] dev = ["pytest>=8.0", "pytest-asyncio>=0.23", "pytest-cov", "mypy>=1.9", "ruff>=0.4", "scipy>=1.10,<1.16", "httpx>=0.27"] science = ["scipy>=1.10,<1.16"] -web = ["fastapi>=0.110,<1", "uvicorn[standard]>=0.29"] +web = ["fastapi>=0.110,<1", "uvicorn[standard]>=0.29", "httpx>=0.27"] # 打包成桌面 EXE 用：系统托盘（pystray）+ 图标生成（Pillow）。 # 仅 PyInstaller 打包态需要，开发态 ``pip install -e .[web]`` 不强制装。 packaging = ["pystray>=0.19", "Pillow>=10.0"] diff --git a/src/easy_tdx/ths_web.py b/src/easy_tdx/ths_web.py new file mode 100644 index 0000000..97c76c3 --- /dev/null +++ b/src/easy_tdx/ths_web.py @@ -0,0 +1,281 @@ +"""同花顺公开网页的轻量解析客户端。 + +这不是同花顺官方 API。页面结构、访问策略及字段含义均可能变化，调用方应遵守 +同花顺网站规则，并避免高频或商用再分发场景。 +""" + +from __future__ import annotations + +import asyncio +import html +import re +import time +from datetime import date +from typing import Any + +import httpx + +_BROWSER_HEADERS = { + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", + "Referer": "https://basic.10jqka.com.cn/", + "User-Agent": ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36" + ), +} +_FIELD_URL = "https://basic.10jqka.com.cn/{code}/field.html" +_CONCEPT_URL = "https://basic.10jqka.com.cn/{code}/concept.html" +_INDUSTRY_URL = "https://q.10jqka.com.cn/thshy/" +_CONCEPT_DIRECTORY_URL = "https://q.10jqka.com.cn/gn/" +_BOARD_DETAIL_URLS = { + "industry": "https://q.10jqka.com.cn/thshy/detail/code/{code}/", + "concept": "https://q.10jqka.com.cn/gn/detail/code/{code}/", +} +_TAG_RE = re.compile(r"<[^>]+>", re.IGNORECASE) +_THREE_CATE_RE = re.compile( + r"三级行业分类：.*?]*>\s*(.*?)\s*(?:（|\(|<)", re.IGNORECASE | re.DOTALL +) +_CONCEPT_CELL_RE = re.compile( + r']*\bclass=["\'][^"\']*\bgnName\b[^"\']*["\'][^>]*\bclid=["\'](\d+)["\'][^>]*>(.*?)', + re.IGNORECASE | re.DOTALL, +) +_LEADER_CODES_RE = re.compile( + r']*\btopStock=["\']([^"\']*)["\'][^>]*\bcid=["\'](\d+)["\']', + re.IGNORECASE | re.DOTALL, +) +_ROW_RE = re.compile(r"]*>(.*?)", re.IGNORECASE | re.DOTALL) +_TD_RE = re.compile(r"]*>(.*?)", re.IGNORECASE | re.DOTALL) +_BOARD_LINK_RE = re.compile( + r']*href=["\'][^"\']*/(?:thshy|gn)/detail/code/(\d+)/[^"\']*["\'][^>]*>(.*?)', + re.IGNORECASE | re.DOTALL, +) +_STOCK_LINK_RE = re.compile( + r']*href=["\'][^"\']*stockpage\.10jqka\.com\.cn/(\d+)/?[^"\']*["\'][^>]*>(.*?)', + re.IGNORECASE | re.DOTALL, +) +_BOARD_HEADING_RE = re.compile( + r']*\bclass=["\'][^"\']*\bboard-hq\b[^"\']*["\'][^>]*>.*?' + r"]*>.*?]*>(\d+).*?", + re.IGNORECASE | re.DOTALL, +) +_BOARD_CHANGE_RE = re.compile( + r']*\bclass=["\'][^"\']*\bboard-zdf\b[^"\']*["\'][^>]*>.*?' + r"([+-]?\d+(?:\.\d+)?)%", + re.IGNORECASE | re.DOTALL, +) + + +class ThsWebError(RuntimeError): + """同花顺公开页面请求或解析不可用。""" + + +def _text(value: str) -> str: + return " ".join(html.unescape(_TAG_RE.sub("", value)).replace("\xa0", " ").split()) + + +def _percent(value: str) -> float | None: + normalized = _text(value).replace("%", "").replace(",", "") + try: + return float(normalized) + except ValueError: + return None + + +def parse_industry_hierarchy(page: str) -> list[str]: + """Extract the THS three-level industry labels from a stock field page.""" + match = _THREE_CATE_RE.search(page) + if not match: + return [] + return [part.strip() for part in _text(match.group(1)).split("--") if part.strip()] + + +def parse_concepts(page: str) -> list[dict[str, Any]]: + """Extract concept board memberships and THS-exposed leading-stock codes.""" + leader_codes_by_board: dict[str, list[str]] = {} + for raw_codes, board_code in _LEADER_CODES_RE.findall(page): + leader_codes_by_board[board_code] = [ + code for code in raw_codes.split(",") if code.isdigit() + ] + + concepts: list[dict[str, Any]] = [] + seen: set[str] = set() + for board_code, raw_name in _CONCEPT_CELL_RE.findall(page): + if board_code in seen: + continue + seen.add(board_code) + concepts.append( + { + "board_code": board_code, + "name": _text(raw_name), + "leader_codes": leader_codes_by_board.get(board_code, []), + } + ) + return concepts + + +def parse_board_directory(page: str, board_kind: str) -> dict[str, str]: + """Parse public THS board-directory links into a name-to-code mapping.""" + if board_kind not in {"industry", "concept"}: + raise ValueError("board_kind must be 'industry' or 'concept'") + + return {_text(raw_name): board_code for board_code, raw_name in _BOARD_LINK_RE.findall(page)} + + +def parse_board_detail(page: str) -> dict[str, Any] | None: + """Parse a public THS board detail page and its first ranked constituent.""" + heading = _BOARD_HEADING_RE.search(page) + change = _BOARD_CHANGE_RE.search(page) + if not heading or not change: + return None + + leader: dict[str, Any] | None = None + for row in _ROW_RE.findall(page): + cells = _TD_RE.findall(row) + texts = [_text(cell) for cell in cells] + stock_links = _STOCK_LINK_RE.findall(row) + named_stock = next( + ((code, _text(name)) for code, name in stock_links if _text(name) != code), + None, + ) + if not named_stock: + continue + stock_code, stock_name = named_stock + try: + name_index = texts.index(stock_name) + except ValueError: + continue + leader = { + "code": stock_code, + "name": stock_name, + "change_pct": _percent(texts[name_index + 2]) if len(texts) > name_index + 2 else None, + } + break + return { + "board_code": heading.group(1), + "change_pct": _percent(change.group(1)), + "leader": leader, + } + + +class ThsWebClient: + """Retrieve stock associations from public THS pages with quote-only caching.""" + + def __init__( + self, + *, + timeout: float = 10.0, + transport: httpx.AsyncBaseTransport | None = None, + quote_ttl_seconds: float = 30.0, + ) -> None: + self._timeout = timeout + self._transport = transport + self._quote_ttl_seconds = quote_ttl_seconds + self._quote_cache: dict[tuple[str, str], tuple[date, float, dict[str, Any] | None]] = {} + + async def _get_page(self, url: str) -> str: + try: + async with httpx.AsyncClient( + headers=_BROWSER_HEADERS, + follow_redirects=True, + timeout=self._timeout, + transport=self._transport, + ) as client: + response = await client.get(url) + response.raise_for_status() + except httpx.HTTPError as exc: + raise ThsWebError("同花顺公开页面暂时无法访问") from exc + return response.content.decode("gbk", errors="replace") + + async def _get_board_detail(self, board_kind: str, board_code: str) -> dict[str, Any] | None: + """Fetch one board quote, retaining it only briefly within the current day.""" + if board_kind not in _BOARD_DETAIL_URLS: + raise ValueError("board_kind must be 'industry' or 'concept'") + today = date.today() + cache_key = (board_kind, board_code) + cached = self._quote_cache.get(cache_key) + if cached and cached[0] == today and time.monotonic() - cached[1] < self._quote_ttl_seconds: + return cached[2] + + page = await self._get_page(_BOARD_DETAIL_URLS[board_kind].format(code=board_code)) + detail = parse_board_detail(page) + self._quote_cache[cache_key] = (today, time.monotonic(), detail) + return detail + + async def _optional_board_detail( + self, board_kind: str, board_code: str | None + ) -> dict[str, Any] | None: + if board_code is None: + return None + try: + return await self._get_board_detail(board_kind, board_code) + except ThsWebError: + return None + + async def get_stock_associations(self, code: str, concept_limit: int = 10) -> dict[str, Any]: + """Get THS industry hierarchy, concept memberships, and public board quotes.""" + pages = await asyncio.gather( + self._get_page(_FIELD_URL.format(code=code)), + self._get_page(_CONCEPT_URL.format(code=code)), + self._get_page(_INDUSTRY_URL), + self._get_page(_CONCEPT_DIRECTORY_URL), + ) + field_page, concept_page, industry_directory_page, concept_directory_page = pages + industry_directory = parse_board_directory(industry_directory_page, "industry") + concept_directory = parse_board_directory(concept_directory_page, "concept") + industry_names = parse_industry_hierarchy(field_page) + all_concepts = parse_concepts(concept_page) + selected_concepts = all_concepts[:concept_limit] + detail_semaphore = asyncio.Semaphore(4) + + async def get_limited_detail( + board_kind: str, board_code: str | None + ) -> dict[str, Any] | None: + async with detail_semaphore: + return await self._optional_board_detail(board_kind, board_code) + + industry_details, concept_details = await asyncio.gather( + asyncio.gather( + *( + get_limited_detail("industry", industry_directory.get(name)) + for name in industry_names + ) + ), + asyncio.gather( + *( + get_limited_detail("concept", concept_directory.get(concept["name"])) + for concept in selected_concepts + ) + ), + ) + + industries: list[dict[str, Any]] = [] + for level, (name, detail) in enumerate(zip(industry_names, industry_details), start=1): + industries.append( + { + "level": level, + "name": name, + "board_code": detail["board_code"] if detail else industry_directory.get(name), + "change_pct": detail["change_pct"] if detail else None, + "leader": detail["leader"] if detail else None, + } + ) + + concepts: list[dict[str, Any]] = [] + for concept, detail in zip(selected_concepts, concept_details): + concepts.append( + { + "board_code": detail["board_code"] if detail else concept["board_code"], + "name": concept["name"], + "change_pct": detail["change_pct"] if detail else None, + "leader": detail["leader"] if detail else None, + "leader_codes": concept["leader_codes"], + } + ) + return { + "source": "ths_web", + "code": code, + "industries": industries, + "concepts": concepts, + "concept_total": len(all_concepts), + } diff --git a/src/easy_tdx/web/app.py b/src/easy_tdx/web/app.py index 99bd2a0..59dad31 100644 --- a/src/easy_tdx/web/app.py +++ b/src/easy_tdx/web/app.py @@ -215,6 +215,7 @@ def _create_app( from easy_tdx.web.routers.sina import router as sina_router from easy_tdx.web.routers.stock_industry import router as stock_industry_router from easy_tdx.web.routers.strategies import router as strategies_router + from easy_tdx.web.routers.ths import router as ths_router app.include_router(market_router, prefix="/api/v1") app.include_router(bars_router, prefix="/api/v1") @@ -227,6 +228,7 @@ def _create_app( app.include_router(mac_data_router, prefix="/api/v1") app.include_router(mac_quotes_router, prefix="/api/v1") app.include_router(stock_industry_router, prefix="/api/v1") + app.include_router(ths_router, prefix="/api/v1") # 扩展市场路由 app.include_router(ex_market_router, prefix="/api/v1") # 技术指标路由 diff --git a/src/easy_tdx/web/deps.py b/src/easy_tdx/web/deps.py index 2558634..eff2605 100644 --- a/src/easy_tdx/web/deps.py +++ b/src/easy_tdx/web/deps.py @@ -2,6 +2,7 @@ from __future__ import annotations +from functools import lru_cache from typing import Any from fastapi import Request @@ -9,6 +10,14 @@ from fastapi import Request from easy_tdx.client import AsyncTdxClient +@lru_cache(maxsize=1) +def get_ths_web_client() -> Any: + """返回共享的同花顺公开网页客户端（仅行情快照短时缓存）。""" + from easy_tdx.ths_web import ThsWebClient + + return ThsWebClient() + + def get_client(request: Request) -> AsyncTdxClient: """从 app.state 获取共享的 AsyncTdxClient 实例。""" client: AsyncTdxClient = request.app.state.tdx_client diff --git a/src/easy_tdx/web/routers/ths.py b/src/easy_tdx/web/routers/ths.py new file mode 100644 index 0000000..e8a1786 --- /dev/null +++ b/src/easy_tdx/web/routers/ths.py @@ -0,0 +1,32 @@ +"""同花顺公开网页关联板块路由。""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException, Query + +from easy_tdx.ths_web import ThsWebError +from easy_tdx.web.deps import get_ths_web_client +from easy_tdx.web.schemas import DictResponse + +router = APIRouter(tags=["ths-web"]) + + +@router.get("/ths/stock/associations", response_model=DictResponse) +async def stock_associations( + code: str = Query( + ..., min_length=6, max_length=6, pattern=r"^\d{6}$", description="6位股票代码" + ), + concept_limit: int = Query(10, ge=1, le=30, description="返回概念板块数量，默认 10"), + client: Any = Depends(get_ths_web_client), +) -> DictResponse: + """获取同花顺公开网页中的行业层级、概念板块及板块涨跌幅。 + + 数据从同花顺公开网页解析，并非官方开放 API。行业归属每次请求重新读取； + 行业/概念行情只在当日内短时缓存，页面没有公开报价的层级会返回 ``null``。 + """ + try: + return DictResponse.from_dict(await client.get_stock_associations(code, concept_limit)) + except ThsWebError as exc: + raise HTTPException(status_code=503, detail=str(exc)) from exc diff --git a/tests/unit/test_ths_web.py b/tests/unit/test_ths_web.py new file mode 100644 index 0000000..2c6eac5 --- /dev/null +++ b/tests/unit/test_ths_web.py @@ -0,0 +1,187 @@ +"""同花顺公开网页解析及接口测试。""" + +from __future__ import annotations + +import httpx +import pytest + +pytest.importorskip("fastapi") + +from fastapi import FastAPI # noqa: E402 +from fastapi.testclient import TestClient # noqa: E402 + +from easy_tdx.ths_web import ( # noqa: E402 + ThsWebClient, + parse_board_detail, + parse_board_directory, + parse_concepts, + parse_industry_hierarchy, +) +from easy_tdx.web.deps import get_ths_web_client # noqa: E402 +from easy_tdx.web.routers.ths import router # noqa: E402 + +_FIELD_HTML = """ +

三级行业分类： +电子 -- 半导体 -- 数字芯片设计 （共57家）

+""" + +_CONCEPT_HTML = """ + + + +
芯片概念
比亚迪概念
+芯片概念 +比亚迪概念 +""" + +_INDUSTRY_DIRECTORY_HTML = """ + + +
半导体
+""" + +_CONCEPT_DIRECTORY_HTML = """ + +
芯片概念比亚迪概念
+""" + +_INDUSTRY_DETAIL_HTML = """ +

半导体881121

+

12.34    2.35%

+ + + + + + +
1688981中芯国际50.003.21
+""" + +_CHIP_DETAIL_HTML = """ +

芯片概念301085

+

-0.01    -0.05%

+ +
1002886沃特股份31.009.50
+""" + +_BYD_DETAIL_HTML = """ +

比亚迪概念308972

+

0.00    0.00%

+""" + + +def test_parse_industry_hierarchy() -> None: + assert parse_industry_hierarchy(_FIELD_HTML) == ["电子", "半导体", "数字芯片设计"] + + +def test_parse_concepts_extracts_codes_names_and_leading_stock_codes() -> None: + assert parse_concepts(_CONCEPT_HTML) == [ + {"board_code": "301085", "name": "芯片概念", "leader_codes": ["002886", "003005"]}, + {"board_code": "308972", "name": "比亚迪概念", "leader_codes": ["300750"]}, + ] + + +def test_parse_board_directory_and_detail_extract_quote_and_leader() -> None: + assert parse_board_directory(_INDUSTRY_DIRECTORY_HTML, "industry") == {"半导体": "881121"} + assert parse_board_detail(_INDUSTRY_DETAIL_HTML) == { + "board_code": "881121", + "change_pct": 2.35, + "leader": {"code": "688981", "name": "中芯国际", "change_pct": 3.21}, + } + + +@pytest.mark.asyncio +async def test_client_merges_stock_membership_and_board_quotes() -> None: + pages = { + "/603893/field.html": _FIELD_HTML, + "/603893/concept.html": _CONCEPT_HTML, + "/thshy/": _INDUSTRY_DIRECTORY_HTML, + "/gn/": _CONCEPT_DIRECTORY_HTML, + "/thshy/detail/code/881121/": _INDUSTRY_DETAIL_HTML, + "/gn/detail/code/301085/": _CHIP_DETAIL_HTML, + "/gn/detail/code/308972/": _BYD_DETAIL_HTML, + } + calls: dict[str, int] = {} + + async def handler(request: httpx.Request) -> httpx.Response: + calls[request.url.path] = calls.get(request.url.path, 0) + 1 + return httpx.Response(200, content=pages[request.url.path].encode("gbk")) + + client = ThsWebClient(transport=httpx.MockTransport(handler)) + result = await client.get_stock_associations("603893") + + assert result["code"] == "603893" + assert result["industries"] == [ + {"level": 1, "name": "电子", "board_code": None, "change_pct": None, "leader": None}, + { + "level": 2, + "name": "半导体", + "board_code": "881121", + "change_pct": 2.35, + "leader": {"code": "688981", "name": "中芯国际", "change_pct": 3.21}, + }, + { + "level": 3, + "name": "数字芯片设计", + "board_code": None, + "change_pct": None, + "leader": None, + }, + ] + assert result["concept_total"] == 2 + assert result["concepts"] == [ + { + "board_code": "301085", + "name": "芯片概念", + "change_pct": -0.05, + "leader": {"code": "002886", "name": "沃特股份", "change_pct": 9.5}, + "leader_codes": ["002886", "003005"], + }, + { + "board_code": "308972", + "name": "比亚迪概念", + "change_pct": 0.0, + "leader": None, + "leader_codes": ["300750"], + }, + ] + await client.get_stock_associations("603893") + assert calls == { + "/603893/field.html": 2, + "/603893/concept.html": 2, + "/thshy/": 2, + "/gn/": 2, + "/thshy/detail/code/881121/": 1, + "/gn/detail/code/301085/": 1, + "/gn/detail/code/308972/": 1, + } + + +class _FakeThsClient: + async def get_stock_associations(self, code: str, concept_limit: int) -> dict[str, object]: + assert concept_limit == 10 + return {"source": "ths_web", "code": code, "industries": [], "concepts": []} + + +def _api_client() -> TestClient: + app = FastAPI() + app.include_router(router, prefix="/api/v1") + app.dependency_overrides[get_ths_web_client] = lambda: _FakeThsClient() + return TestClient(app) + + +def test_ths_associations_endpoint() -> None: + with _api_client() as client: + response = client.get("/api/v1/ths/stock/associations", params={"code": "603893"}) + + assert response.status_code == 200 + assert response.json() == { + "data": {"source": "ths_web", "code": "603893", "industries": [], "concepts": []} + } + + +def test_ths_associations_endpoint_validates_code() -> None: + with _api_client() as client: + response = client.get("/api/v1/ths/stock/associations", params={"code": "60389"}) + + assert response.status_code == 422 diff --git a/tests/unit/test_web_api.py b/tests/unit/test_web_api.py index 9e23302..94395ef 100644 --- a/tests/unit/test_web_api.py +++ b/tests/unit/test_web_api.py @@ -8,6 +8,25 @@ from __future__ import annotations import pytest + +def _mounted_paths(app): + """Return routes for both flattened and _IncludedRouter FastAPI versions.""" + paths: list[str] = [] + for route in app.routes: + if hasattr(route, "path"): + paths.append(route.path) + continue + original_router = getattr(route, "original_router", None) + include_context = getattr(route, "include_context", None) + if original_router is not None and include_context is not None: + paths.extend( + f"{include_context.prefix}{child.path}" + for child in original_router.routes + if hasattr(child, "path") + ) + return paths + + # --------------------------------------------------------------------------- # Task 2: Schemas & Error Handling # --------------------------------------------------------------------------- @@ -79,7 +98,7 @@ def test_create_app_returns_fastapi_instance(): assert app.title == "easy-tdx API" # Check routers are mounted - routes = [r.path for r in app.routes] + routes = _mounted_paths(app) assert any("/api/v1/security" in r for r in routes) assert any("/api/v1/bars" in r for r in routes) assert any("/api/v1/chanlun" in r for r in routes) @@ -503,13 +522,14 @@ def test_full_app_routes_registered(): from easy_tdx.web import create_app app = create_app() - all_paths = [r.path for r in app.routes] + all_paths = _mounted_paths(app) expected_prefixes = [ "/api/v1/security", "/api/v1/bars", "/api/v1/xdxr", "/api/v1/block", "/api/v1/stock/industry", + "/api/v1/ths/stock/associations", "/api/v1/chanlun", "/api/v1/announcements", "/api/v1/sina/financial-report", @@ -535,3 +555,4 @@ def test_openapi_schema_generated(): # they are verified in test_full_app_routes_registered instead. # Just ensure REST paths are present. assert "/api/v1/fund-flow" in schema["paths"] + assert "/api/v1/ths/stock/associations" in schema["paths"]