feat(web): FastAPI REST + WebSocket API layer (v1.10.0)

README.md

@@ -59,6 +59,9 @@ easy-tdx --help
 
 ```bash
 pip install -e ".[dev]"
+
+# 开发 Web API 模式（含 FastAPI + Uvicorn）
+pip install -e ".[web]"
 ```
 
 ## CLI 参考
@@ -735,6 +738,95 @@ easy-tdx offline sync-all
 
 > 建议在通达信关闭时执行 sync 命令，避免文件被锁定。空文件自动全量下载，已有数据只做增量追加。
 
+## Web API
+
+将 easy-tdx 暴露为 REST + WebSocket 服务，供前端、其他语言或远程调用。无需额外注册，零配置启动。
+
+### 安装
+
+```bash
+# 标准安装
+pip install easy-tdx[web]
+
+# 开发模式（从源码安装，支持热重载）
+pip install -e ".[web]"
+```
+
+### 快速启动
+
+```bash
+# 启动 Web API 服务器（自动连接最优 TDX 服务器）
+easy-tdx serve
+
+# 启动后浏览器打开 http://127.0.0.1:8000/docs 查看完整 API 文档（Swagger UI）
+# 也可以访问 http://127.0.0.1:8000/redoc 查看 ReDoc 格式文档
+
+# 指定端口和 TDX 服务器
+easy-tdx serve --port 8080 --tdx-host 119.147.212.81
+
+# 开发模式（代码修改后自动重载）
+easy-tdx serve --reload
+```
+
+> 💡 启动后访问 **http://127.0.0.1:8000/docs** 可以看到完整的交互式 API 文档，支持在线调试每个接口。
+
+### REST API 示例
+
+```bash
+# 获取深圳市场证券数量
+curl "http://localhost:8000/api/v1/security/count?market=SZ"
+
+# 获取股票K线
+curl "http://localhost:8000/api/v1/bars?market=SZ&code=000001&category=DAY&count=100"
+
+# 批量获取实时行情
+curl -X POST "http://localhost:8000/api/v1/quotes" \
+  -H "Content-Type: application/json" \
+  -d '{"stocks": [{"market": "SZ", "code": "000001"}, {"market": "SH", "code": "600000"}]}'
+
+# 缠论分析
+curl -X POST "http://localhost:8000/api/v1/chanlun/analyze" \
+  -H "Content-Type: application/json" \
+  -d '{"market": "SZ", "code": "000001", "category": "DAY", "count": 200}'
+
+# 市场统计
+curl "http://localhost:8000/api/v1/market/stat"
+
+# 板块信息
+curl "http://localhost:8000/api/v1/block?filename=block_gn.dat"
+```
+
+### WebSocket 实时行情
+
+```javascript
+// JavaScript 示例
+const ws = new WebSocket("ws://localhost:8000/ws/realtime/SZ000001");
+
+ws.onmessage = (event) => {
+    const data = JSON.parse(event.data);
+    console.log(data);  // {type: "tick", market: "SZ", code: "000001", price: 10.5, ...}
+};
+
+// 动态订阅更多标的
+ws.send(JSON.stringify({action: "subscribe", symbol: "SH600000"}));
+```
+
+### API 文档
+
+启动服务后访问：
+- Swagger UI: http://localhost:8000/docs
+- ReDoc: http://localhost:8000/redoc
+
+### 编程 API
+
+```python
+from easy_tdx.web import create_app
+import uvicorn
+
+app = create_app(host="119.147.212.81", port=7709)
+uvicorn.run(app, host="0.0.0.0", port=8000)
+```
+
 ## CLI 命令汇总
 
 | 命令 | 说明 |
@@ -765,6 +857,7 @@ easy-tdx offline sync-all
 | `run-all` | 批量运行所有策略并排名（绩效排名 + 综合评分 + 可选图表） |
 | `screen scan` | 策略选股扫描（纯离线，全市场信号扫描） |
 | `screen rank` | 扫描结果回测排名（按夏普/回撤等指标排序） |
+| `serve` | 启动 Web API 服务器（REST + WebSocket，需 `easy-tdx[web]`） |
 | `f10` | F10 公司信息 |
 | `fund-flow` | 历史资金流向 |
 | `ex kline` | 扩展市场 K 线 |
@@ -1293,6 +1386,7 @@ src/easy_tdx/
 ├── backtest/          # 回测引擎（Strategy基类/向量化引擎/多因子组合/组合回测/绩效分析）
 ├── screen/            # 策略选股扫描（scan信号扫描/rank回测排名/并发扫描/增量缓存）
 ├── realtime/          # 实时数据推送框架（EventBus/事件驱动/asyncio）
+├── web/               # Web API（FastAPI REST + WebSocket）
 ├── models/            # 纯 dataclass，无业务逻辑
 ├── offline/           # 离线数据读写模块（读取 + 写入同步）
 └── cli/               # easy-tdx CLI（click）
@@ -1321,6 +1415,20 @@ ruff format --check src/ tests/                              # format check
 
 ## Changelog
 
+### 1.10.0 (2026-06-12)
+
+**Web API 层** — 新增 FastAPI REST + WebSocket 服务，一键将 easy-tdx 暴露为 HTTP API。
+
+- 新增 `src/easy_tdx/web/` 模块：app factory、6 个路由（market/bars/finance/block/chanlun/realtime）、Pydantic schemas、异常处理
+- 新增 `easy-tdx serve` CLI 命令，支持 `--host`、`--port`、`--tdx-host`、`--reload` 参数
+- REST 端点覆盖全部 `AsyncTdxClient` 方法（K线/报价/资金流向/板块/财务/缠论分析等）
+- WebSocket 端点 `/ws/realtime/{symbol}` 支持实时行情订阅和多标的动态切换
+- 自动生成 Swagger UI (`/docs`) 和 ReDoc (`/redoc`) 文档
+- 可选依赖 `pip install easy-tdx[web]`，核心安装不受影响
+- 20 个离线单元测试覆盖 schemas、路由注册、OpenAPI schema 生成、输入验证
+- 修复 `deps.py` 中 `AsyncTdxClient` 在 `TYPE_CHECKING` 下导致运行时 `NameError`（500 → 正常启动）
+- 修复 market/category 参数不支持小写（`sz`/`sh`）和非法值（`ZZZ`）导致 500 的问题，统一返回 400 Bad Request
+
 ### 1.9.10 (2026-06-11)
 
 **板块 N 日涨跌幅排行** — 新增 `board-change-ranking` 命令，支持按行业/概念/风格板块计算指定日期前 N 个交易日的涨跌幅并排行。

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "easy-tdx"
-version = "1.9.10"
+version = "1.10.0"
 description = "通达信 TCP 协议行情数据客户端，支持在线行情、离线数据读取与写入同步"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -15,6 +15,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"]
+web = ["fastapi>=0.110", "uvicorn[standard]>=0.29"]
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/easy_tdx"]
@@ -31,6 +32,10 @@ ignore_missing_imports = true
 module = "easy_tdx.MyTT"
 # Type stubs provided via MyTT.pyi — keep strict checks enabled
 
+[[tool.mypy.overrides]]
+module = ["fastapi.*", "uvicorn.*", "pydantic.*"]
+ignore_missing_imports = true
+
 [tool.ruff]
 target-version = "py310"
 line-length = 100

src/easy_tdx/cli/__init__.py

@@ -29,6 +29,7 @@ from .cmd_quote import quote, quote_list
 from .cmd_run_all import run_all
 from .cmd_tick import tick
 from .cmd_transaction import transaction
+from .cmd_web import serve
 
 
 @click.group()
@@ -84,3 +85,4 @@ cli.add_command(backtest)
 cli.add_command(portfolio)
 cli.add_command(run_all)
 cli.add_command(screen)
+cli.add_command(serve)

src/easy_tdx/cli/cmd_web.py

@@ -0,0 +1,37 @@
+"""easy-tdx serve — 启动 Web API 服务器。"""
+
+from __future__ import annotations
+
+import click
+
+
+@click.command("serve")
+@click.option("--host", default="0.0.0.0", help="监听地址")
+@click.option("--port", default=8000, type=int, help="监听端口")
+@click.option("--tdx-host", default=None, help="TDX 服务器地址（默认自动选择最优）")
+@click.option("--tdx-port", default=None, type=int, help="TDX 服务器端口")
+@click.option("--reload", is_flag=True, help="开发模式（自动重载）")
+def serve(host: str, port: int, tdx_host: str | None, tdx_port: int | None, reload: bool) -> None:
+    """启动 Web API 服务器（需要安装 easy-tdx[web]）。"""
+    try:
+        import uvicorn
+    except ImportError:
+        click.echo(
+            "错误：缺少 web 依赖。请运行: pip install easy-tdx[web]",
+            err=True,
+        )
+        raise SystemExit(1) from None
+
+    if reload:
+        uvicorn.run(
+            "easy_tdx.web:app_factory",
+            host=host,
+            port=port,
+            reload=True,
+            factory=True,
+        )
+    else:
+        from easy_tdx.web import create_app
+
+        app = create_app(host=tdx_host, port=tdx_port)
+        uvicorn.run(app, host=host, port=port)

src/easy_tdx/web/__init__.py

@@ -0,0 +1,50 @@
+"""easy-tdx Web API — FastAPI REST + WebSocket layer.
+
+Install with: pip install easy-tdx[web]
+
+Usage::
+
+    from easy_tdx.web import create_app
+
+    app = create_app()
+
+    # Run with uvicorn:
+    # uvicorn easy_tdx.web:app --host 0.0.0.0 --port 8000
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from fastapi import FastAPI
+
+
+def create_app(
+    host: str | None = None,
+    port: int | None = None,
+    timeout: float | None = None,
+) -> FastAPI:
+    """Create and configure the FastAPI application.
+
+    Args:
+        host: TDX server host (None = auto-detect best host).
+        port: TDX server port (None = default 7709).
+        timeout: Connection timeout in seconds.
+
+    Returns:
+        Configured FastAPI application instance.
+    """
+    from easy_tdx.web.app import _create_app
+
+    return _create_app(host=host, port=port, timeout=timeout)
+
+
+def app_factory() -> FastAPI:
+    """Factory function for uvicorn --reload mode."""
+    from easy_tdx.web.app import _create_app
+
+    return _create_app()
+
+
+__all__ = ["create_app", "app_factory"]

src/easy_tdx/web/app.py

@@ -0,0 +1,98 @@
+"""FastAPI application factory and lifespan management."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from easy_tdx.web.errors import register_exception_handlers
+
+logger = logging.getLogger(__name__)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
+    """管理 TDX 连接生命周期：启动时连接，关闭时断开。"""
+    from easy_tdx.client import AsyncTdxClient
+
+    host = app.state.tdx_host
+    port = app.state.tdx_port
+    timeout = app.state.tdx_timeout
+
+    client = AsyncTdxClient(host=host, port=port, timeout=timeout)
+    try:
+        await client.connect()
+        logger.info("TDX client connected to %s:%s", host, port)
+    except Exception:
+        logger.warning("TDX client connection failed — endpoints will return 503")
+
+    app.state.tdx_client = client
+    yield
+
+    try:
+        await client.close()
+        logger.info("TDX client disconnected")
+    except Exception:
+        pass
+
+
+def _create_app(
+    host: str | None = None,
+    port: int | None = None,
+    timeout: float | None = None,
+) -> FastAPI:
+    """创建并配置 FastAPI 应用实例。"""
+    from easy_tdx.config import get_best_host, get_port, get_timeout
+
+    if host is None:
+        host = get_best_host()
+    if port is None:
+        port = get_port()
+    if timeout is None:
+        timeout = get_timeout()
+
+    app = FastAPI(
+        title="easy-tdx API",
+        description="通达信行情数据 REST + WebSocket API",
+        version="1.0.0",
+        lifespan=lifespan,
+    )
+
+    # Store connection config in app.state for lifespan to use
+    app.state.tdx_host = host
+    app.state.tdx_port = port
+    app.state.tdx_timeout = timeout
+    app.state.tdx_client = None  # will be set in lifespan
+
+    # CORS middleware (permissive for development)
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    # Register exception handlers
+    register_exception_handlers(app)
+
+    # Mount routers
+    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.chanlun import router as chanlun_router
+    from easy_tdx.web.routers.finance import router as finance_router
+    from easy_tdx.web.routers.market import router as market_router
+    from easy_tdx.web.routers.realtime import router as realtime_router
+
+    app.include_router(market_router, prefix="/api/v1")
+    app.include_router(bars_router, prefix="/api/v1")
+    app.include_router(finance_router, prefix="/api/v1")
+    app.include_router(block_router, prefix="/api/v1")
+    app.include_router(chanlun_router, prefix="/api/v1")
+    app.include_router(realtime_router, prefix="/api/v1")
+
+    return app

src/easy_tdx/web/convert.py

@@ -0,0 +1,41 @@
+"""共享参数转换工具（market/category 字符串 → 枚举）。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from easy_tdx.web.schemas import KlineCategoryEnum, MarketEnum
+
+
+def market_from_str(s: str) -> Any:
+    """将字符串转为 Market 枚举，支持大小写，非法值抛 ValueError。
+
+    >>> market_from_str("SZ")  # 正常
+    >>> market_from_str("sz")  # 也正常（自动转大写）
+    >>> market_from_str("ZZZ")  # ValueError
+    """
+    from easy_tdx.models.enums import Market
+
+    key = s.upper()
+    try:
+        return Market[MarketEnum[key].name]
+    except KeyError:
+        valid = ", ".join(m.name for m in MarketEnum)
+        raise ValueError(f"无效市场代码 '{s}'，可选值: {valid}") from None
+
+
+def category_from_str(s: str) -> Any:
+    """将字符串转为 KlineCategory 枚举，支持大小写和数字字符串。"""
+    from easy_tdx.models.enums import KlineCategory
+
+    key = s.upper()
+    # 支持纯数字（如 "4" 表示日线）
+    try:
+        return KlineCategory(int(key))
+    except (ValueError, TypeError):
+        pass
+    try:
+        return KlineCategory[KlineCategoryEnum[key].name]
+    except KeyError:
+        valid = ", ".join(c.name for c in KlineCategoryEnum)
+        raise ValueError(f"无效K线周期 '{s}'，可选值: {valid}") from None

src/easy_tdx/web/deps.py

@@ -0,0 +1,13 @@
+"""Dependency injection for Web API routers."""
+
+from __future__ import annotations
+
+from fastapi import Request
+
+from easy_tdx.client import AsyncTdxClient
+
+
+def get_client(request: Request) -> AsyncTdxClient:
+    """从 app.state 获取共享的 AsyncTdxClient 实例。"""
+    client: AsyncTdxClient = request.app.state.tdx_client
+    return client

src/easy_tdx/web/errors.py

@@ -0,0 +1,46 @@
+"""Web API error handling."""
+
+from __future__ import annotations
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from easy_tdx.exceptions import TdxConnectionError
+
+
+class ApiErrorResponse(BaseModel):
+    """标准错误响应格式。"""
+
+    error: str
+    detail: str = ""
+
+
+def register_exception_handlers(app: FastAPI) -> None:
+    """注册全局异常处理器到 FastAPI app。"""
+
+    @app.exception_handler(TdxConnectionError)
+    async def tdx_connection_error_handler(
+        request: Request, exc: TdxConnectionError
+    ) -> JSONResponse:
+        del request  # unused but required by FastAPI signature
+        return JSONResponse(
+            status_code=503,
+            content=ApiErrorResponse(error="TDX connection error", detail=str(exc)).model_dump(),
+        )
+
+    @app.exception_handler(ValueError)
+    async def value_error_handler(request: Request, exc: ValueError) -> JSONResponse:
+        del request  # unused but required by FastAPI signature
+        return JSONResponse(
+            status_code=400,
+            content=ApiErrorResponse(error="Bad request", detail=str(exc)).model_dump(),
+        )
+
+    @app.exception_handler(Exception)
+    async def generic_error_handler(request: Request, exc: Exception) -> JSONResponse:
+        del request  # unused but required by FastAPI signature
+        return JSONResponse(
+            status_code=500,
+            content=ApiErrorResponse(error="Internal server error", detail=str(exc)).model_dump(),
+        )

src/easy_tdx/web/routers/__init__.py

@@ -0,0 +1 @@
+"""API route modules."""

src/easy_tdx/web/routers/bars.py

@@ -0,0 +1,104 @@
+"""K线 / 分时 / 逐笔成交路由。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Depends, Query
+
+from easy_tdx.web.convert import category_from_str, market_from_str
+from easy_tdx.web.deps import get_client
+from easy_tdx.web.schemas import DataFrameResponse
+
+router = APIRouter(tags=["bars"])
+
+
+def _df_resp(df: Any) -> DataFrameResponse:
+    return DataFrameResponse.from_dataframe(df)
+
+
+@router.get("/bars", response_model=DataFrameResponse)
+async def security_bars(
+    market: str = Query(..., description="市场: SZ, SH, BJ"),
+    code: str = Query(..., min_length=6, max_length=6),
+    category: str = Query(
+        "DAY",
+        description="K线周期: MIN_1, MIN_5, MIN_15, MIN_30, MIN_60, DAY, WEEK, MONTH, YEAR",
+    ),
+    start: int = Query(0, ge=0),
+    count: int = Query(800, ge=1, le=800),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取股票K线数据。"""
+    df = await client.get_security_bars(
+        market_from_str(market), code, category_from_str(category), start, count
+    )
+    return _df_resp(df)
+
+
+@router.get("/bars/index", response_model=DataFrameResponse)
+async def index_bars(
+    market: str = Query(..., description="市场: SZ, SH"),
+    code: str = Query(..., min_length=6, max_length=6),
+    category: str = Query("DAY", description="K线周期"),
+    start: int = Query(0, ge=0),
+    count: int = Query(800, ge=1, le=800),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取指数K线数据。"""
+    df = await client.get_index_bars(
+        market_from_str(market), code, category_from_str(category), start, count
+    )
+    return _df_resp(df)
+
+
+@router.get("/minute", response_model=DataFrameResponse)
+async def minute_time(
+    market: str = Query(..., description="市场: SZ, SH"),
+    code: str = Query(..., min_length=6, max_length=6),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取今日分时数据。"""
+    df = await client.get_minute_time_data(market_from_str(market), code)
+    return _df_resp(df)
+
+
+@router.get("/minute/history", response_model=DataFrameResponse)
+async def history_minute_time(
+    market: str = Query(..., description="市场: SZ, SH"),
+    code: str = Query(..., min_length=6, max_length=6),
+    date: int = Query(..., description="日期 YYYYMMDD"),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取历史某日分时数据。"""
+    df = await client.get_history_minute_time_data(market_from_str(market), code, date)
+    return _df_resp(df)
+
+
+@router.get("/transaction", response_model=DataFrameResponse)
+async def transaction_data(
+    market: str = Query(..., description="市场: SZ, SH"),
+    code: str = Query(..., min_length=6, max_length=6),
+    start: int = Query(0, ge=0),
+    count: int = Query(800, ge=1, le=800),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取当日逐笔成交。"""
+    df = await client.get_transaction_data(market_from_str(market), code, start, count)
+    return _df_resp(df)
+
+
+@router.get("/transaction/history", response_model=DataFrameResponse)
+async def history_transaction_data(
+    market: str = Query(..., description="市场: SZ, SH"),
+    code: str = Query(..., min_length=6, max_length=6),
+    date: int = Query(..., description="日期 YYYYMMDD"),
+    start: int = Query(0, ge=0),
+    count: int = Query(800, ge=1, le=800),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取历史逐笔成交。"""
+    df = await client.get_history_transaction_data(
+        market_from_str(market), code, date, start, count
+    )
+    return _df_resp(df)

src/easy_tdx/web/routers/block.py

@@ -0,0 +1,29 @@
+"""板块信息路由。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Depends, Query
+
+from easy_tdx.web.deps import get_client
+from easy_tdx.web.schemas import DataFrameResponse
+
+router = APIRouter(tags=["block"])
+
+
+def _df_resp(df: Any) -> DataFrameResponse:
+    return DataFrameResponse.from_dataframe(df)
+
+
+@router.get("/block", response_model=DataFrameResponse)
+async def block_info(
+    filename: str = Query(
+        ...,
+        description=("板块文件名: block_zs.dat(行业指数), block_gn.dat(概念), block_fg.dat(风格)"),
+    ),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取并解析板块文件。"""
+    df = await client.get_block_info(filename)
+    return _df_resp(df)

src/easy_tdx/web/routers/chanlun.py

@@ -0,0 +1,54 @@
+"""缠论分析路由。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Depends
+
+from easy_tdx.web.convert import category_from_str, market_from_str
+from easy_tdx.web.deps import get_client
+from easy_tdx.web.schemas import ChanlunRequest
+
+router = APIRouter(tags=["chanlun"])
+
+
+@router.post("/chanlun/analyze")
+async def chanlun_analyze(
+    req: ChanlunRequest,
+    client: Any = Depends(get_client),
+) -> dict[str, Any]:
+    """执行缠论分析。
+
+    自动从 TDX 服务器获取 K 线数据，运行完整缠论计算管道，
+    返回笔、中枢、线段、买卖点、背驰等分析结果。
+    """
+    from easy_tdx.chanlun import ChanlunAnalyser
+
+    # 1. Fetch kline data
+    df = await client.get_security_bars(
+        market_from_str(req.market),
+        req.code,
+        category_from_str(req.category),
+        req.start,
+        req.count,
+    )
+
+    # 2. Run chanlun analysis
+    symbol = f"{req.market}{req.code}"
+    frequency_map: dict[str, str] = {
+        "MIN_1": "1min",
+        "MIN_5": "5min",
+        "MIN_15": "15min",
+        "MIN_30": "30min",
+        "MIN_60": "60min",
+        "DAY": "daily",
+        "WEEK": "weekly",
+        "MONTH": "monthly",
+        "YEAR": "yearly",
+    }
+    freq = frequency_map.get(req.category.upper(), req.category)
+    analyser = ChanlunAnalyser(code=symbol, frequency=freq)
+    result = analyser.process_klines(df)
+
+    return result.to_dict()

src/easy_tdx/web/routers/finance.py

@@ -0,0 +1,85 @@
+"""财务 / 公司信息路由。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Depends, Query
+
+from easy_tdx.web.convert import market_from_str
+from easy_tdx.web.deps import get_client
+from easy_tdx.web.schemas import DataFrameResponse
+
+router = APIRouter(tags=["finance"])
+
+
+def _df_resp(df: Any) -> DataFrameResponse:
+    return DataFrameResponse.from_dataframe(df)
+
+
+@router.get("/xdxr", response_model=DataFrameResponse)
+async def xdxr_info(
+    market: str = Query(..., description="市场: SZ, SH"),
+    code: str = Query(..., min_length=6, max_length=6),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取除权除息历史记录。"""
+    df = await client.get_xdxr_info(market_from_str(market), code)
+    return _df_resp(df)
+
+
+@router.get("/finance", response_model=DataFrameResponse)
+async def finance_info(
+    market: str = Query(..., description="市场: SZ, SH"),
+    code: str = Query(..., min_length=6, max_length=6),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取最新财务数据。"""
+    df = await client.get_finance_info(market_from_str(market), code)
+    return _df_resp(df)
+
+
+@router.get("/company/category", response_model=DataFrameResponse)
+async def company_info_category(
+    market: str = Query(..., description="市场: SZ, SH"),
+    code: str = Query(..., min_length=6, max_length=6),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取公司信息文件目录。"""
+    df = await client.get_company_info_category(market_from_str(market), code)
+    return _df_resp(df)
+
+
+@router.get("/company/content")
+async def company_info_content(
+    market: str = Query(..., description="市场: SZ, SH"),
+    code: str = Query(..., min_length=6, max_length=6),
+    filename: str = Query(..., description="文件名"),
+    offset: int = Query(0, ge=0),
+    length: int = Query(1024, ge=1),
+    client: Any = Depends(get_client),
+) -> dict[str, str]:
+    """读取公司信息文本。"""
+    content = await client.get_company_info_content(
+        market_from_str(market), code, filename, offset, length
+    )
+    return {"content": content}
+
+
+@router.get("/financial/file-list", response_model=DataFrameResponse)
+async def financial_file_list(
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取可用的历史专业财报文件列表。"""
+    df = await client.get_financial_file_list()
+    return _df_resp(df)
+
+
+@router.get("/financial/records", response_model=DataFrameResponse)
+async def financial_records(
+    filename: str = Query(..., description="财报文件名，如 tdxfin/gpcw20260331.zip"),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """下载财报 zip 并解析为记录列表。"""
+    df = await client.get_financial_records(filename)
+    return _df_resp(df)

src/easy_tdx/web/routers/market.py

@@ -0,0 +1,100 @@
+"""市场信息路由：证券列表、实时行情、市场统计、资金流向。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Depends, Query
+
+from easy_tdx.web.convert import market_from_str
+from easy_tdx.web.deps import get_client
+from easy_tdx.web.schemas import (
+    CountResponse,
+    DataFrameResponse,
+    QuoteRequest,
+)
+
+router = APIRouter(tags=["market"])
+
+
+def _df_response(df: Any) -> DataFrameResponse:
+    """将 DataFrame 转为 API 响应。"""
+    return DataFrameResponse.from_dataframe(df)
+
+
+@router.get("/security/count", response_model=CountResponse)
+async def security_count(
+    market: str = Query(..., description="市场: SZ, SH, BJ"),
+    client: Any = Depends(get_client),
+) -> CountResponse:
+    """获取市场证券总数。"""
+    count = await client.get_security_count(market_from_str(market))
+    return CountResponse(count=count)
+
+
+@router.get("/security/list", response_model=DataFrameResponse)
+async def security_list(
+    market: str = Query(..., description="市场: SZ, SH, BJ"),
+    start: int = Query(0, ge=0, description="分页起始位置"),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取证券列表（每页约1000条）。"""
+    df = await client.get_security_list(market_from_str(market), start)
+    return _df_response(df)
+
+
+@router.get("/security/list-all", response_model=DataFrameResponse)
+async def security_list_all(
+    pages: int = Query(1, ge=1, description="拉取页数（每个市场每页1000条）"),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取沪深 A 股完整列表。"""
+    df = await client.get_security_list_all(pages=pages)
+    return _df_response(df)
+
+
+@router.post("/quotes", response_model=DataFrameResponse)
+async def security_quotes(
+    req: QuoteRequest,
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """批量获取实时五档行情（最多80只/次）。"""
+    stocks_parsed: list[tuple[Any, str]] = []
+    for s in req.stocks:
+        m = market_from_str(s.market)
+        stocks_parsed.append((m, s.code))
+    df = await client.get_security_quotes(stocks_parsed)
+    return _df_response(df)
+
+
+@router.get("/market/stat", response_model=DataFrameResponse)
+async def market_stat(
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取 A 股全市场涨跌统计。"""
+    df = await client.get_market_stat()
+    return _df_response(df)
+
+
+@router.get("/fund-flow", response_model=DataFrameResponse)
+async def fund_flow(
+    market: str = Query(..., description="市场: SZ, SH"),
+    code: str = Query(..., min_length=6, max_length=6, description="6位股票代码"),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取个股当日资金流向。"""
+    df = await client.get_fund_flow(market_from_str(market), code)
+    return _df_response(df)
+
+
+@router.get("/fund-flow/history", response_model=DataFrameResponse)
+async def history_fund_flow(
+    market: str = Query(..., description="市场: SZ, SH"),
+    code: str = Query(..., min_length=6, max_length=6, description="6位股票代码"),
+    start: int = Query(0, ge=0),
+    count: int = Query(100, ge=1, le=800),
+    client: Any = Depends(get_client),
+) -> DataFrameResponse:
+    """获取个股历史日线资金流向。"""
+    df = await client.get_history_fund_flow(market_from_str(market), code, start, count)
+    return _df_response(df)

src/easy_tdx/web/routers/realtime.py

@@ -0,0 +1,103 @@
+"""实时数据 WebSocket 路由。"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from typing import Any
+
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(tags=["realtime"])
+
+
+@router.websocket("/ws/realtime/{symbol}")
+async def realtime_websocket(websocket: WebSocket, symbol: str) -> None:
+    """WebSocket 实时行情订阅。
+
+    连接后自动订阅指定标的的实时事件。
+    symbol 格式: SZ000001, SH600000 等。
+
+    客户端可发送 JSON 消息来控制订阅：
+    - {"action": "subscribe", "symbol": "SZ000001"}
+    - {"action": "unsubscribe", "symbol": "SZ000001"}
+
+    服务端推送消息格式：
+    - {"type": "tick", "market": "SZ", "code": "000001", "price": 10.5, ...}
+    - {"type": "signal", "direction": "BUY", ...}
+    """
+    await websocket.accept()
+    logger.info("WebSocket client connected for symbol: %s", symbol)
+
+    # Try to get EventBus from app state
+    event_bus = getattr(websocket.app.state, "event_bus", None)
+
+    subscribed_symbols: set[str] = {symbol.upper()}
+
+    async def _on_event(event: Any) -> None:
+        """EventBus 回调 → 推送 WebSocket 消息。"""
+        event_symbol = f"{event.market}{event.code}"
+        if event_symbol in subscribed_symbols:
+            try:
+                msg = {
+                    "type": event.event_type.value,
+                    "market": event.market,
+                    "code": event.code,
+                    "price": event.price,
+                    "volume": event.volume,
+                    "timestamp": event.timestamp,
+                    "data": event.data,
+                }
+                await websocket.send_json(msg)
+            except Exception:
+                logger.warning("Failed to send WebSocket message")
+
+    # Subscribe to event bus if available
+    if event_bus is not None:
+        event_bus.subscribe_all(_on_event)
+
+    try:
+        while True:
+            # Receive client messages (subscribe/unsubscribe control)
+            try:
+                raw = await asyncio.wait_for(websocket.receive_text(), timeout=30.0)
+                data = json.loads(raw)
+                action = data.get("action", "")
+
+                if action == "subscribe":
+                    new_symbol = data.get("symbol", "").upper()
+                    if new_symbol:
+                        subscribed_symbols.add(new_symbol)
+                        await websocket.send_json(
+                            {"type": "status", "msg": f"subscribed {new_symbol}"}
+                        )
+
+                elif action == "unsubscribe":
+                    old_symbol = data.get("symbol", "").upper()
+                    subscribed_symbols.discard(old_symbol)
+                    await websocket.send_json(
+                        {"type": "status", "msg": f"unsubscribed {old_symbol}"}
+                    )
+
+            except asyncio.TimeoutError:
+                # Send heartbeat ping
+                try:
+                    await websocket.send_json({"type": "ping"})
+                except Exception:
+                    break
+            except WebSocketDisconnect:
+                break
+            except json.JSONDecodeError:
+                await websocket.send_json({"type": "error", "msg": "invalid JSON"})
+
+    except WebSocketDisconnect:
+        logger.info("WebSocket client disconnected: %s", symbol)
+    except Exception:
+        logger.exception("WebSocket error for %s", symbol)
+    finally:
+        if event_bus is not None:
+            event_bus.unsubscribe(symbol.upper(), _on_event)
+        logger.info("WebSocket connection closed: %s", symbol)

src/easy_tdx/web/schemas.py

@@ -0,0 +1,105 @@
+"""Pydantic request/response schemas for the Web API."""
+
+from __future__ import annotations
+
+from enum import IntEnum
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+# ---------------------------------------------------------------------------
+# Enums — mirror easy_tdx.models.enums but as string-based for REST clarity
+# ---------------------------------------------------------------------------
+
+
+class MarketEnum(IntEnum):
+    """Market identifier."""
+
+    SZ = 0
+    SH = 1
+    BJ = 2
+
+
+class KlineCategoryEnum(IntEnum):
+    """K-line period."""
+
+    MIN_5 = 0
+    MIN_15 = 1
+    MIN_30 = 2
+    MIN_60 = 3
+    DAY = 4
+    WEEK = 5
+    MONTH = 6
+    MIN_1 = 7
+    YEAR = 9
+    SEASON = 10
+
+
+# ---------------------------------------------------------------------------
+# Request models
+# ---------------------------------------------------------------------------
+
+
+class StockIdentifier(BaseModel):
+    """A single stock identified by market + code."""
+
+    market: str = Field(..., pattern=r"^(SZ|SH|BJ)$", description="市场代码")
+    code: str = Field(..., min_length=6, max_length=6, description="6位股票代码")
+
+
+class QuoteRequest(BaseModel):
+    """Batch quote request."""
+
+    stocks: list[StockIdentifier] = Field(
+        ..., min_length=1, max_length=80, description="股票列表（最多80只）"
+    )
+
+
+class ChanlunRequest(BaseModel):
+    """缠论分析请求。"""
+
+    market: str = Field(..., pattern=r"^(SZ|SH|BJ)$")
+    code: str = Field(..., min_length=6, max_length=6)
+    category: str = Field(default="DAY", description="K线周期")
+    count: int = Field(default=800, ge=1, le=800)
+    start: int = Field(default=0, ge=0)
+
+
+# ---------------------------------------------------------------------------
+# Response models
+# ---------------------------------------------------------------------------
+
+
+class DataFrameResponse(BaseModel):
+    """通用 DataFrame 响应（records 格式）。"""
+
+    data: list[dict[str, Any]]
+    count: int
+
+    @classmethod
+    def from_dataframe(cls, df: Any) -> DataFrameResponse:
+        """从 pandas DataFrame 构建响应。"""
+        import pandas as pd
+
+        if isinstance(df, pd.DataFrame):
+            records = df.to_dict(orient="records")
+            cleaned: list[dict[str, Any]] = []
+            for row in records:
+                clean_row: dict[str, Any] = {}
+                for k, v in row.items():
+                    if hasattr(v, "isoformat"):
+                        clean_row[k] = v.isoformat()
+                    elif hasattr(v, "item"):
+                        # numpy scalar → Python native
+                        clean_row[k] = v.item()
+                    else:
+                        clean_row[k] = v
+                cleaned.append(clean_row)
+            return cls(data=cleaned, count=len(cleaned))
+        return cls(data=[], count=0)
+
+
+class CountResponse(BaseModel):
+    """简单计数响应。"""
+
+    count: int

tests/unit/test_web_api.py

@@ -0,0 +1,292 @@
+"""Web API tests (offline, no network).
+
+Covers: schemas, error handling, app factory, DI, all routers,
+        CLI serve command, OpenAPI schema generation.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+# ---------------------------------------------------------------------------
+# Task 2: Schemas & Error Handling
+# ---------------------------------------------------------------------------
+
+
+def test_market_enum_values():
+    """MarketEnum should map string names to int values matching Market enum."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.schemas import MarketEnum
+
+    assert MarketEnum.SZ == 0
+    assert MarketEnum.SH == 1
+    assert MarketEnum.BJ == 2
+
+
+def test_kline_category_enum():
+    """KlineCategoryEnum should map string names to int values."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.schemas import KlineCategoryEnum
+
+    assert KlineCategoryEnum.MIN_5 == 0
+    assert KlineCategoryEnum.DAY == 4
+    assert KlineCategoryEnum.WEEK == 5
+
+
+def test_quote_request_validation():
+    """QuoteRequest should validate stocks list."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.schemas import QuoteRequest
+
+    req = QuoteRequest(stocks=[{"market": "SZ", "code": "000001"}])
+    assert len(req.stocks) == 1
+    assert req.stocks[0].market == "SZ"
+    assert req.stocks[0].code == "000001"
+
+
+def test_chanlun_request_defaults():
+    """ChanlunRequest should have sensible defaults."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.schemas import ChanlunRequest
+
+    req = ChanlunRequest(market="SZ", code="000001")
+    assert req.category == "DAY"
+    assert req.count == 800
+
+
+def test_api_error_response():
+    """ApiErrorResponse should serialize correctly."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.errors import ApiErrorResponse
+
+    err = ApiErrorResponse(error="test error", detail="some detail")
+    d = err.model_dump()
+    assert d["error"] == "test error"
+    assert d["detail"] == "some detail"
+
+
+# ---------------------------------------------------------------------------
+# Task 3: App Factory & Dependency Injection
+# ---------------------------------------------------------------------------
+
+
+def test_create_app_returns_fastapi_instance():
+    """create_app should return a FastAPI app with routers mounted."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web import create_app
+
+    app = create_app()
+    assert app.title == "easy-tdx API"
+
+    # Check routers are mounted
+    routes = [r.path for r in app.routes]
+    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)
+    assert any("/ws/realtime" in r for r in routes)
+
+
+def test_deps_get_client_type():
+    """get_client should be callable (actual client creation needs network)."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.deps import get_client
+
+    assert callable(get_client)
+
+
+# ---------------------------------------------------------------------------
+# Task 4: Market Router
+# ---------------------------------------------------------------------------
+
+
+def test_market_router_endpoints():
+    """Market router should define all expected endpoints."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.routers.market import router
+
+    paths = [r.path for r in router.routes]
+    assert "/security/count" in paths
+    assert "/security/list" in paths
+    assert "/security/list-all" in paths
+    assert "/quotes" in paths
+    assert "/market/stat" in paths
+    assert "/fund-flow" in paths
+    assert "/fund-flow/history" in paths
+
+
+# ---------------------------------------------------------------------------
+# Task 5: Bars Router
+# ---------------------------------------------------------------------------
+
+
+def test_bars_router_endpoints():
+    """Bars router should define all expected endpoints."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.routers.bars import router
+
+    paths = [r.path for r in router.routes]
+    assert "/bars" in paths
+    assert "/bars/index" in paths
+    assert "/minute" in paths
+    assert "/minute/history" in paths
+    assert "/transaction" in paths
+    assert "/transaction/history" in paths
+
+
+# ---------------------------------------------------------------------------
+# Task 6: Finance Router
+# ---------------------------------------------------------------------------
+
+
+def test_finance_router_endpoints():
+    """Finance router should define all expected endpoints."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.routers.finance import router
+
+    paths = [r.path for r in router.routes]
+    assert "/xdxr" in paths
+    assert "/finance" in paths
+    assert "/company/category" in paths
+    assert "/company/content" in paths
+    assert "/financial/file-list" in paths
+    assert "/financial/records" in paths
+
+
+# ---------------------------------------------------------------------------
+# Task 7: Block Router
+# ---------------------------------------------------------------------------
+
+
+def test_block_router_endpoints():
+    """Block router should define expected endpoints."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.routers.block import router
+
+    paths = [r.path for r in router.routes]
+    assert "/block" in paths
+
+
+# ---------------------------------------------------------------------------
+# Task 8: Chanlun Router
+# ---------------------------------------------------------------------------
+
+
+def test_chanlun_router_endpoints():
+    """Chanlun router should define the analyze endpoint."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.routers.chanlun import router
+
+    paths = [r.path for r in router.routes]
+    assert "/chanlun/analyze" in paths
+
+
+# ---------------------------------------------------------------------------
+# Task 9: Realtime Router
+# ---------------------------------------------------------------------------
+
+
+def test_realtime_router_endpoints():
+    """Realtime router should define the WebSocket endpoint."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.routers.realtime import router
+
+    paths = [r.path for r in router.routes]
+    assert any("realtime" in p for p in paths)
+
+
+# ---------------------------------------------------------------------------
+# Task 10: CLI serve command
+# ---------------------------------------------------------------------------
+
+
+def test_serve_command_exists():
+    """CLI should have a serve command registered."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.cli import cli
+
+    assert "serve" in cli.commands
+
+
+# ---------------------------------------------------------------------------
+# Task 11: Integration — route registration & OpenAPI
+# ---------------------------------------------------------------------------
+
+
+# ---------------------------------------------------------------------------
+# Regression: input validation (case-insensitive + invalid → ValueError → 400)
+# ---------------------------------------------------------------------------
+
+
+def test_convert_market_lowercase():
+    """market_from_str should accept lowercase input."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.models.enums import Market
+    from easy_tdx.web.convert import market_from_str
+
+    assert market_from_str("sz") == Market.SZ
+    assert market_from_str("sh") == Market.SH
+    assert market_from_str("Bj") == Market.BJ
+
+
+def test_convert_market_invalid_raises_valueerror():
+    """market_from_str should raise ValueError for invalid market codes."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.convert import market_from_str
+
+    with pytest.raises(ValueError, match="无效市场代码"):
+        market_from_str("ZZZ")
+
+
+def test_convert_category_from_int_string():
+    """category_from_str should accept numeric string like '4'."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.models.enums import KlineCategory
+    from easy_tdx.web.convert import category_from_str
+
+    assert category_from_str("4") == KlineCategory.DAY
+
+
+def test_convert_category_invalid_raises_valueerror():
+    """category_from_str should raise ValueError for invalid period."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web.convert import category_from_str
+
+    with pytest.raises(ValueError, match="无效K线周期"):
+        category_from_str("INVALID_PERIOD")
+
+
+def test_full_app_routes_registered():
+    """All routers should be mounted and accessible."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web import create_app
+
+    app = create_app()
+    all_paths = [r.path for r in app.routes]
+    expected_prefixes = [
+        "/api/v1/security",
+        "/api/v1/bars",
+        "/api/v1/xdxr",
+        "/api/v1/block",
+        "/api/v1/chanlun",
+        "/ws/realtime",
+    ]
+    for prefix in expected_prefixes:
+        matched = any(prefix in p for p in all_paths)
+        assert matched, f"Expected route with prefix '{prefix}' not found in {all_paths}"
+
+
+def test_openapi_schema_generated():
+    """OpenAPI schema should be auto-generated and contain key paths."""
+    pytest.importorskip("fastapi")
+    from easy_tdx.web import create_app
+
+    app = create_app()
+    schema = app.openapi()
+    assert schema["info"]["title"] == "easy-tdx API"
+    assert "/api/v1/security/count" in schema["paths"]
+    assert "/api/v1/bars" in schema["paths"]
+    assert "/api/v1/chanlun/analyze" in schema["paths"]
+    # WebSocket routes are NOT included in OpenAPI schema by default;
+    # they are verified in test_full_app_routes_registered instead.
+    # Just ensure REST paths are present.
+    assert "/api/v1/fund-flow" in schema["paths"]