diff --git a/README.md b/README.md index 5ebed67..d5bceb6 100644 --- a/README.md +++ b/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 个交易日的涨跌幅并排行。 diff --git a/pyproject.toml b/pyproject.toml index fff8203..f9dffce 100644 --- a/pyproject.toml +++ b/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 diff --git a/src/easy_tdx/cli/__init__.py b/src/easy_tdx/cli/__init__.py index e3e0655..c391c15 100644 --- a/src/easy_tdx/cli/__init__.py +++ b/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) diff --git a/src/easy_tdx/cli/cmd_web.py b/src/easy_tdx/cli/cmd_web.py new file mode 100644 index 0000000..b5676e9 --- /dev/null +++ b/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) diff --git a/src/easy_tdx/web/__init__.py b/src/easy_tdx/web/__init__.py new file mode 100644 index 0000000..a989925 --- /dev/null +++ b/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"] diff --git a/src/easy_tdx/web/app.py b/src/easy_tdx/web/app.py new file mode 100644 index 0000000..f3bec70 --- /dev/null +++ b/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 diff --git a/src/easy_tdx/web/convert.py b/src/easy_tdx/web/convert.py new file mode 100644 index 0000000..da9e4de --- /dev/null +++ b/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 diff --git a/src/easy_tdx/web/deps.py b/src/easy_tdx/web/deps.py new file mode 100644 index 0000000..f5c0881 --- /dev/null +++ b/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 diff --git a/src/easy_tdx/web/errors.py b/src/easy_tdx/web/errors.py new file mode 100644 index 0000000..a08737b --- /dev/null +++ b/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(), + ) diff --git a/src/easy_tdx/web/routers/__init__.py b/src/easy_tdx/web/routers/__init__.py new file mode 100644 index 0000000..fb0a2f8 --- /dev/null +++ b/src/easy_tdx/web/routers/__init__.py @@ -0,0 +1 @@ +"""API route modules.""" diff --git a/src/easy_tdx/web/routers/bars.py b/src/easy_tdx/web/routers/bars.py new file mode 100644 index 0000000..3f18839 --- /dev/null +++ b/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) diff --git a/src/easy_tdx/web/routers/block.py b/src/easy_tdx/web/routers/block.py new file mode 100644 index 0000000..5bde092 --- /dev/null +++ b/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) diff --git a/src/easy_tdx/web/routers/chanlun.py b/src/easy_tdx/web/routers/chanlun.py new file mode 100644 index 0000000..09dd277 --- /dev/null +++ b/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() diff --git a/src/easy_tdx/web/routers/finance.py b/src/easy_tdx/web/routers/finance.py new file mode 100644 index 0000000..6ca9bc7 --- /dev/null +++ b/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) diff --git a/src/easy_tdx/web/routers/market.py b/src/easy_tdx/web/routers/market.py new file mode 100644 index 0000000..6e11d4b --- /dev/null +++ b/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) diff --git a/src/easy_tdx/web/routers/realtime.py b/src/easy_tdx/web/routers/realtime.py new file mode 100644 index 0000000..2e519d1 --- /dev/null +++ b/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) diff --git a/src/easy_tdx/web/schemas.py b/src/easy_tdx/web/schemas.py new file mode 100644 index 0000000..9fc7377 --- /dev/null +++ b/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 diff --git a/tests/unit/test_web_api.py b/tests/unit/test_web_api.py new file mode 100644 index 0000000..02e1046 --- /dev/null +++ b/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"]